Refactor python samples

.pre-commit-config.yaml

@@ -40,7 +40,7 @@ repos:
     hooks:
       - id: flake8
         additional_dependencies: [flake8-comprehensions, flake8-docstrings, flake8-import-order, pep8-naming, pydocstyle]
-        args: ["--ignore", "E501,D103,D100,W503,E203", "--import-order-style", "google"]
+        args: ["--ignore", "E501,D103,D100,W503,E203,E402", "--import-order-style", "google"]
 
   - repo: https://github.com/jumanjihouse/pre-commit-hooks
     rev: 3.0.0

samples/python/command-response/README.md

@@ -7,10 +7,6 @@ We are using the [Eclipse Paho](http://eclipse.org/paho/) MQTT Python client lib
 
 Note that the OCI IoT Platform only supports MQTT Secure (MQTTS) on port 8883.
 
-We only illustrate a connection using password-based authentication with MQTT Secure (MQTTS).
-This example can be easily modified to use certificate-based authentication and/or WebSocket
-Secure (WSS) -- see the respective "publish" examples in this repository.
-
 ## Scenario
 
 The `command-response.py` script will establish an MQTT connection with the OCI IoT
@@ -28,10 +24,38 @@ Install the Python dependencies
 pip install -r requirements.txt
 ```
 
-The script assumes a Digital Twin with unstructured telemetry has already been created.
+The script assumes a Digital Twin has already been created.
 
 ## Configure and run the script
 
+### Telemetry payload
+
+- For unstructured telemetry, the content can be arbitrary.
+- For structured telemetry, it must match the Model/Adapter.
+- For structured telemetry in the default format, if a "time" property is specified,
+  it must be an epoch time in microseconds and will override the "time_observed" field
+  in the database.
+- The same applies to structured telemetry in a custom format, but the mapping must be
+  defined in the adapter.
+
+The sample telemetry used by the scripts is compatible with all three Digital Twins
+created in the "Manage Digital Twins" section of this repository:
+
+```json
+telemetry_data = {
+    "time": 1757512025226854,
+    "sht_temperature": 23.8,
+    "qmp_temperature": 24.4,
+    "humidity": 56.1,
+    "pressure": 1012.2,
+    "count": 1,
+}
+```
+
+The `time` field is optional, this can be specified in the configuration file (see below)
+
+### Configuration file
+
 Copy `config.distr.py` to `config.py` and set the following variables:
 
 - `iot_device_host`: The Device Host for your IoT Domain.
@@ -53,9 +77,16 @@ Copy `config.distr.py` to `config.py` and set the following variables:
   here. See the
   [proxy_set](https://eclipse.dev/paho/files/paho.mqtt.python/html/client.html#paho.mqtt.client.Client.proxy_set)
   documentation for more details.
-- `username`: The "externalKey" property of your Digital Twin.
-- `password`: The Digital Twin password; that is, the content of the vault secret
-  corresponding to the authId property of your Digital Twin.
+- `auth_type`: Authentication type: `basic` or `cert`.
+- For basic authentication:
+  - `username`: The "externalKey" property of your Digital Twin.
+  - `password`: The Digital Twin password; that is, the content of the vault secret
+    corresponding to the authId property of your Digital Twin.
+- For certificate authentication:
+  - Set the path to your client certificate and key in the `client_cert` and `client_key`
+    variables.
+  - Keep in mind that the `authId` property of your Digital Twin must match the
+    Common Name (CN) of the certificate.
 
 Run the script:
 
@@ -101,3 +132,6 @@ Waiting 2 seconds to process possible /cmd messages...
 Terminated
 $
 ```
+
+You can check the status of the message delivery and response in the `RAW_COMMAND_DATA`
+database view.

samples/python/command-response/command-response.py

@@ -11,11 +11,17 @@
 """
 
 import json
+import os
 import sys
 import threading
 import time
 
+sys.path.append(
+    os.path.abspath(os.path.join(os.path.dirname(__file__), os.pardir, "shared"))
+)
+
 import config
+import environmental_sensor_simulator
 import paho.mqtt.client as mqtt
 
 MQTT_PORT = 8883
@@ -26,19 +32,6 @@ def current_epoch_microseconds():
     return int(time.time() * 1000 * 1000)
 
 
-# Telemetry data example.
-telemetry_data = {
-    "time": 0,
-    "sht_temperature": 23.8,
-    "qmp_temperature": 24.4,
-    "humidity": 56.1,
-    "pressure": 1012.2,
-    "count": 0,
-}
-
-shutdown_event = threading.Event()
-
-
 # Callback for MQTT connection event.
 def on_connect(client, userdata, flags, reason_code, properties=None):
     print(f"Connected with result code {reason_code}")
@@ -50,6 +43,7 @@ def on_connect(client, userdata, flags, reason_code, properties=None):
 def on_message(client, userdata, message, properties=None):
     topic = message.topic
     payload = message.payload.decode()
+    state = userdata
     if topic.endswith("/cmd"):
         print(f"Received command on {topic}: {payload}")
         # Command handling logic goes here.
@@ -62,31 +56,42 @@ def on_message(client, userdata, message, properties=None):
 
         if cmd and cmd.get("shutdown", False):
             print("Shutdown command received. Preparing to exit...")
-            shutdown_event.set()  # Signal the telemetry loop to stop.
+            state["shutdown_event"].set()  # Signal the telemetry loop to stop.
 
         # Build corresponding /rsp topic
         rsp_topic = topic[:-4] + "/rsp"
         ack_msg = json.dumps(
             {"status": "acknowledged", "time": current_epoch_microseconds()}
         )
-        print(f"Sending ack to {rsp_topic}: {ack_msg}")
-        client.publish(topic=rsp_topic, payload=ack_msg, qos=config.qos)
+        state["ack_msg_info"].append(
+            client.publish(topic=rsp_topic, payload=ack_msg, qos=config.qos)
+        )
+
 
+if config.auth_type not in ("basic", "cert"):
+    raise ValueError("auth_type must be 'basic' or 'cert'")
 
 client = mqtt.Client(
     client_id=config.client_id,  # Ensure client_id is set for persistent sessions.
     clean_session=False,  # Enable persistent session.
     protocol=mqtt.MQTTv311,  # Use v311 unless v5 features are needed.
     callback_api_version=mqtt.CallbackAPIVersion.VERSION2,  # type: ignore
 )
+state = {
+    "ack_msg_info": [],
+    "shutdown_event": threading.Event(),
+}
+client.user_data_set(state)
 client.on_connect = on_connect
 client.on_message = on_message
 
-# TLS/SSL configuration.
-client.tls_set(ca_certs=config.ca_certs)
-
-# Authentication.
-client.username_pw_set(username=config.username, password=config.password)
+if config.auth_type == "basic":
+    client.tls_set(ca_certs=config.ca_certs)
+    client.username_pw_set(username=config.username, password=config.password)
+else:
+    client.tls_set(
+        ca_certs=config.ca_certs, certfile=config.client_cert, keyfile=config.client_key
+    )
 
 # Configure proxy if needed.
 if config.proxy_args:
@@ -101,26 +106,34 @@ def on_message(client, userdata, message, properties=None):
 
 # Send telemetry messages.
 try:
+    telemetry = environmental_sensor_simulator.EnvironmentalSensorSimulator(
+        time_format=config.time_format
+    )
     count = 1
     print("Telemetry loop -- Press Ctrl-C to stop.")
-    while not shutdown_event.is_set():
+    while not state["shutdown_event"].is_set():
         print(f"Sending message #{count}")
-        telemetry_data["time"] = current_epoch_microseconds()
-        telemetry_data["count"] = count
         rc_pub = client.publish(
             topic=config.iot_endpoint,
-            payload=json.dumps(telemetry_data),
+            payload=json.dumps(telemetry.get_telemetry()),
             qos=config.qos,
         )
         rc_pub.wait_for_publish()
         count += 1
         time.sleep(config.message_delay)
+        # Drain ack_msg_info
+        while state["ack_msg_info"]:
+            state["ack_msg_info"].pop(0).wait_for_publish()
+
 except KeyboardInterrupt:
     print("\nInterrupted by user. Exiting...")
 
 # Wait to process any potential /cmd messages before exit.
 print("Waiting 2 seconds to process possible /cmd messages...")
 time.sleep(2)
+# Drain ack_msg_info
+while state["ack_msg_info"]:
+    state["ack_msg_info"].pop(0).wait_for_publish()
 
 # Tear down the client and exit.
 client.loop_stop()

samples/python/command-response/config.distr.py

@@ -16,6 +16,9 @@
 # MQTT client id
 client_id = "your_device_name"
 
+# Format of the "time" field in the payload ("none", "epoch", "iso")
+time_format = "epoch"
+
 # Quality of Service
 qos = 1
 
@@ -42,9 +45,32 @@
 # Authentication
 ###
 
+# Authentication type: basic or cert
+auth_type = "basic"
+
+###
+# For basic authentication
+###
+
 # The username is the "externalKey" property of your Digital Twin.
 username = "your_device_username"
 
 # The Digital Twin password. This should be the content of the vault secret
 # corresponding to the authId property of your Digital Twin.
 password = "your_device_password"
+
+###
+# For certificate authentication (mTLS)
+###
+
+# Path to your client certificate and key.
+# If both the certificate and private key are in the same file, set client_key to None.
+# You can retrieve a certificate bundle from the OCI certificate store with:
+# oci certificates certificate-bundle get \
+#   --certificate-id <Certificate OCID> \
+#   --bundle-type CERTIFICATE_CONTENT_WITH_PRIVATE_KEY |
+#   jq -r '.data."certificate-pem"','.data."private-key-pem"' > client_certificate_bundle.pem
+# Keep in mind that the authId property of your Digital Twin must match the
+# Common Name (CN) of the certificate.
+client_cert = "/path/to/client_certificate.pem"
+client_key = ""

samples/python/command-response/requirements.txt

@@ -1,2 +1,3 @@
+numpy~=2.3.0
 paho-mqtt>=2.0.0
 PySocks

samples/python/publish-https/README.md

@@ -11,27 +11,70 @@ Install the Python dependencies (using a
 pip install -r requirements.txt
 ```
 
-## Using Password-Based Authentication
+The script assumes a Digital Twin has already been created.
+The sample payload sent will be accepted by any of the Digital Twins created by the
+[Manage Digital Twins](../../script/manage-dt/) section of this repository.
+
+## Configure and run the scripts
+
+### Telemetry payload
+
+- For unstructured telemetry, the content can be arbitrary.
+- For structured telemetry, it must match the Model/Adapter.
+- For structured telemetry in the default format, if a "time" property is specified,
+  it must be an epoch time in microseconds and will override the "time_observed" field
+  in the database.
+- The same applies to structured telemetry in a custom format, but the mapping must be
+  defined in the adapter.
+
+The sample telemetry used by the scripts is compatible with all three Digital Twins
+created in the "Manage Digital Twins" section of this repository:
+
+```json
+telemetry_data = {
+    "time": 1757512025226854,
+    "sht_temperature": 23.8,
+    "qmp_temperature": 24.4,
+    "humidity": 56.1,
+    "pressure": 1012.2,
+    "count": 1,
+}
+```
+
+The `time` field is optional, this can be specified in the configuration file (see below)
+
+### Common configuration
 
 Copy `config.distr.py` to `config.py` and set the following variables:
 
-- `iot_device_host`: The Device Host for your IoT Domain
-- `iot_endpoint`: The _path_ for your telemetry (equivalent to an MQTT topic)
-- `username` and `password`: Credentials for your device
+- `iot_device_host`: The Device Host for your IoT Domain.
+- `iot_endpoint`: The  MQTT topic for your telemetry.
+- `time_format`: format of the `time` field in the payload:
+  - `none`: No time information is included in the payload.
+  - `epoch`: Current time as integer microseconds since Unix epoch.
+  - `iso`: Current time as an ISO8601 string in UTC (with 'Z' suffix).
+
+### Using Password-Based Authentication
+
+Set your device credentials in `config.py`:
+
+- `username`: The "externalKey" property of your Digital Twin.
+- `password`: The Digital Twin password, i.e., the content of the vault secret
+  corresponding to the authId property of your Digital Twin.
 
 Run the script:
 
 ```shell
 ./pub-https-basic.py
 ```
 
-## Using Certificate-Based Authentication
+### Using Certificate-Based Authentication
 
-Copy `config.distr.py` to `config.py` and set the following variables:
+Set the path to your client certificate and key in the `client_cert` and `client_key`
+variables of the `config.py` file.
 
-- `iot_device_host`: The Device Host for your IoT Domain
-- `iot_endpoint`: The _path_ for your telemetry (equivalent to an MQTT topic)
-- `client_cert` and `client_key`: Paths to your client certificate and key
+Keep in mind that the `authId` property of your Digital Twin must match the
+Common Name (CN) of the certificate.
 
 Run the script:
 

samples/python/publish-https/config.distr.py

@@ -10,6 +10,8 @@
 # The IoT endpoint can be any value, similar to the MQTT topic.
 iot_endpoint = "iot/v1/http"
 
+# Format of the "time" field in the payload ("none", "epoch", "iso")
+time_format = "epoch"
 
 ###
 # For basic authentication