examples: mqtt: add rl6 implementation using api

Author: rnovatorov

examples/mqtt/api/rl6_simulator.py

@@ -0,0 +1,162 @@
+import asyncio
+import os
+import time
+
+import enapter
+
+
+async def main() -> None:
+    hardware_id = os.environ["HARDWARE_ID"]
+    channel_id = os.environ["CHANNEL_ID"]
+    config = enapter.mqtt.api.Config(
+        host=os.environ["MQTT_HOST"],
+        port=int(os.environ["MQTT_PORT"]),
+        tls_config=enapter.mqtt.api.TLSConfig(
+            secret_key=os.environ["MQTT_TLS_SECRET_KEY"],
+            cert=os.environ["MQTT_TLS_CERT"],
+            ca_cert=os.environ["MQTT_TLS_CA_CERT"],
+        ),
+    )
+    async with enapter.mqtt.api.Client(config=config) as client:
+        device_channel = client.device_channel(
+            hardware_id=hardware_id, channel_id=channel_id
+        )
+        ucm_channel = client.device_channel(
+            hardware_id=hardware_id, channel_id=channel_id
+        )
+        simulator = RL6Simulator(device_channel=device_channel, ucm_channel=ucm_channel)
+        await simulator.run()
+
+
+class RL6Simulator:
+
+    def __init__(
+        self,
+        device_channel: enapter.mqtt.api.device.Channel,
+        ucm_channel: enapter.mqtt.api.device.Channel,
+    ) -> None:
+        self.device_channel = device_channel
+        self.ucm_channel = ucm_channel
+        self.loads = {
+            "r1": False,
+            "r2": False,
+            "r3": False,
+            "r4": False,
+            "r5": False,
+            "r6": False,
+        }
+
+    async def run(self) -> None:
+        async with asyncio.TaskGroup() as tg:
+            tg.create_task(self.command_handler())
+            tg.create_task(self.telemetry_publisher())
+            tg.create_task(self.properties_publisher())
+            # NOTE: The following two tasks are necessary only when connecting
+            # to Cloud v2.
+            tg.create_task(self.ucm_properties_publisher())
+            tg.create_task(self.ucm_telemetry_publisher())
+
+    async def command_handler(self) -> None:
+        async with self.device_channel.subscribe_to_command_requests() as requests:
+            async for request in requests:
+                match request.name:
+                    case "enable_load":
+                        response = self.handle_enable_load_command(request)
+                    case "disable_load":
+                        response = self.handle_disable_load_command(request)
+                    case _:
+                        response = self.handle_unknown_command(request)
+                try:
+                    await self.device_channel.publish_command_response(response)
+                except enapter.mqtt.Error as e:
+                    print("failed to publish command response: " + str(e))
+
+    def handle_enable_load_command(
+        self,
+        request: enapter.mqtt.api.device.CommandRequest,
+    ) -> enapter.mqtt.api.device.CommandResponse:
+        load = request.arguments.get("load")
+        if load not in self.loads:
+            return request.new_response(
+                state=enapter.mqtt.api.device.CommandState.ERROR,
+                payload={"reason": "load invalid or missing"},
+            )
+        self.loads[load] = True
+        return request.new_response(
+            state=enapter.mqtt.api.device.CommandState.COMPLETED, payload={}
+        )
+
+    def handle_disable_load_command(
+        self,
+        request: enapter.mqtt.api.device.CommandRequest,
+    ) -> enapter.mqtt.api.device.CommandResponse:
+        load = request.arguments.get("load")
+        if load not in self.loads:
+            return request.new_response(
+                state=enapter.mqtt.api.device.CommandState.ERROR,
+                payload={"reason": "load invalid or missing"},
+            )
+        self.loads[load] = False
+        return request.new_response(
+            state=enapter.mqtt.api.device.CommandState.COMPLETED, payload={}
+        )
+
+    def handle_unknown_command(
+        self,
+        request: enapter.mqtt.api.device.CommandRequest,
+    ) -> enapter.mqtt.api.device.CommandResponse:
+        return request.new_response(
+            state=enapter.mqtt.api.device.CommandState.ERROR,
+            payload={"reason": "command unknown"},
+        )
+
+    async def telemetry_publisher(self) -> None:
+        while True:
+            telemetry = enapter.mqtt.api.device.Telemetry(
+                timestamp=int(time.time()), alerts=[], values=self.loads.copy()
+            )
+            try:
+                await self.device_channel.publish_telemetry(telemetry)
+            except enapter.mqtt.Error as e:
+                print("failed to publish telemetry: " + str(e))
+            await asyncio.sleep(1)
+
+    async def properties_publisher(self) -> None:
+        while True:
+            properties = enapter.mqtt.api.device.Properties(
+                timestamp=int(time.time()), values={}
+            )
+            try:
+                await self.device_channel.publish_properties(properties)
+            except enapter.mqtt.Error as e:
+                print("failed to publish properties: " + str(e))
+            await asyncio.sleep(10)
+
+    async def ucm_telemetry_publisher(self) -> None:
+        while True:
+            telemetry = enapter.mqtt.api.device.Telemetry(
+                timestamp=int(time.time()), alerts=[], values={}
+            )
+            try:
+                await self.ucm_channel.publish_telemetry(telemetry)
+            except enapter.mqtt.Error as e:
+                print("failed to publish ucm telemetry: " + str(e))
+            await asyncio.sleep(1)
+
+    async def ucm_properties_publisher(self) -> None:
+        while True:
+            properties = enapter.mqtt.api.device.Properties(
+                timestamp=int(time.time()), values={"virtual": True, "lua_api_ver": 1}
+            )
+            try:
+                await self.ucm_channel.publish_properties(properties)
+            except enapter.mqtt.Error as e:
+                print("failed to publish ucm properties: " + str(e))
+            await asyncio.sleep(10)
+
+
+if __name__ == "__main__":
+    try:
+        asyncio.run(main())
+    except KeyboardInterrupt:
+        pass