add Greengrass IPC sample to publish to IoT Core

Author: Kriechi

awsiot/greengrasscoreipc/__init__.py

@@ -0,0 +1,68 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0.
+
+import os
+from typing import Optional
+
+from awscrt.io import (
+    ClientBootstrap,
+    DefaultHostResolver,
+    EventLoopGroup,
+    SocketDomain,
+    SocketOptions,
+)
+from awsiot.eventstreamrpc import (
+    Connection,
+    LifecycleHandler,
+    MessageAmendment,
+)
+from awsiot.greengrasscoreipc.client import GreengrassCoreIPCClient
+
+
+def connect(*,
+            ipc_socket: str=None,
+            authtoken: str=None,
+            lifecycle_handler: Optional[LifecycleHandler]=None,
+            timeout=10.0) -> GreengrassCoreIPCClient:
+    """
+    Creates an IPC client and connects to the GreengrassCoreIPC service.
+
+    Args:
+        ipc_socket: Path to the Unix domain socket of Greengrass Nucleus, defaults to
+            environment variable AWS_GG_NUCLEUS_DOMAIN_SOCKET_FILEPATH_FOR_COMPONENT
+        authtoken: Authentication token, defaults to environment variable SVCUID
+        lifecycle_handler: Handler for events over the course of this
+            network connection. See :class:`LifecycleHandler` for more info.
+            Handler methods will only be invoked if the connect attempt
+            succeeds.
+        timeout: The number of seconds to wait for establishing the connection.
+
+    Returns:
+        Client for the GreengrassCoreIPC service.
+    """
+
+    if not ipc_socket:
+        ipc_socket = os.environ["AWS_GG_NUCLEUS_DOMAIN_SOCKET_FILEPATH_FOR_COMPONENT"]
+    if not authtoken:
+        authtoken = os.environ["SVCUID"]
+    if not lifecycle_handler:
+        lifecycle_handler = LifecycleHandler()
+
+    elg = EventLoopGroup(num_threads=1)
+    resolver = DefaultHostResolver(elg)
+    bootstrap = ClientBootstrap(elg, resolver)
+    socket_options = SocketOptions()
+    socket_options.domain = SocketDomain.Local
+    amender = MessageAmendment.create_static_authtoken_amender(authtoken)
+
+    connection = Connection(
+        host_name=ipc_socket,
+        port=0, # dummy port number, not needed for Unix domain sockets
+        bootstrap=bootstrap,
+        socket_options=socket_options,
+        connect_message_amender=amender,
+    )
+    connect_future = connection.connect(lifecycle_handler)
+    connect_future.result(timeout)
+
+    return GreengrassCoreIPCClient(connection)

samples/README.md

@@ -4,6 +4,7 @@
 * [shadow](#shadow)
 * [fleet provisioning](#fleet-provisioning)
 * [basic discovery](#basic-discovery)
+* [IPC with AWS IoT Greengrass to publish to AWS IoT Core](#ipc-greengrass)
 
 ## pubsub
 This sample uses the
@@ -394,5 +395,11 @@ python3 fleetprovisioning.py --endpoint [your endpoint]-ats.iot.[region].amazona
 
 ## basic discovery
 
-This sample intended for use directly with the
+This sample is intended for use directly with the
 [Getting Started with AWS IoT Greengrass](https://docs.aws.amazon.com/greengrass/latest/developerguide/gg-gs.html) guide.
+
+## IPC with AWS IoT Greengrass to publish to AWS IoT Core
+
+This sample is intended to be deployed as an AWS IoT Greengrass component and it will publish MQTT messages from the device to AWS IoT Core.
+
+Source: `samples/ipc_greengrass.py`

samples/ipc_greengrass.py

@@ -0,0 +1,94 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0.
+
+"""
+This sample uses AWS IoT Greengrass v2 to publish messages from the device to
+the AWS IoT Core MQTT broker.
+
+This example can be deployed as Greengrass v2 component and it will start
+publishing telemetry data as MQTT messages in periodic intervals. The IPC
+integration with Greegrass v2 allows this code to run without additional IoT
+certificates or secrets, because it directly communicates with Greengrass Core
+on the device.
+
+See this page for more information on Greengrass v2 components:
+https://docs.aws.amazon.com/greengrass/v2/developerguide/create-components.html
+
+See this page for more information on IPC in Greengrass v2:
+https://docs.aws.amazon.com/greengrass/v2/developerguide/interprocess-communication.html
+
+To run the sample, create a new AWS IoT Greengrass component and deploy it to
+your device with the following recipe snippet to allow your device to publish to
+the AWS IoT Core MQTT broker:
+
+```json
+...
+"ComponentConfiguration": {
+  "DefaultConfiguration": {
+    "accessControl": {
+      "aws.greengrass.ipc.mqttproxy": {
+        "Your.Component.Name:mqttproxy:1": {
+          "policyDescription": "Allows access to publish to all AWS IoT Core topics.",
+          "operations": [
+            "aws.greengrass#PublishToIoTCore"
+          ],
+          "resources": [
+            "*"
+          ]
+        }
+      }
+    }
+  }
+},
+...
+```
+(replace `Your.Component.Name` with your component name)
+
+You can use this recipe `Manifest` snippet to install Python and `AWS IoT Device
+SDK v2 for Python` as dependency:
+```json
+...
+"Manifests": [{
+  ...
+  "Lifecycle": {
+    "Install": {
+      "RequiresPrivilege": true,
+      "Script": "apt-get update --quiet && apt-get --yes install python3 python3-pip && pip3 install awsiotsdk"
+    },
+  ...
+```
+"""
+
+import json
+import time
+import os
+
+import awsiot.greengrasscoreipc
+import awsiot.greengrasscoreipc.model as model
+
+if __name__ == '__main__':
+    ipc_client = awsiot.greengrasscoreipc.connect()
+
+    while True:
+        telemetry_data = {
+            "timestamp": int(round(time.time() * 1000)),
+            "battery_state_of_charge": 42.5,
+            "location": {
+                "longitude": 48.15743,
+                "latitude": 11.57549,
+            },
+        }
+
+        op = ipc_client.new_publish_to_iot_core()
+        op.activate(model.PublishToIoTCoreRequest(
+            topic_name="my/iot/{}/telemetry".format(os.getenv("AWS_IOT_THING_NAME")),
+            qos=model.QOS.AT_LEAST_ONCE,
+            payload=json.dumps(telemetry_data).encode(),
+        ))
+        try:
+            result = op.get_response().result(timeout=5.0)
+            print("successfully published message:", result)
+        except Exception as e:
+            print("failed to publish message:", e)
+
+        time.sleep(5)