move logging from mqtt.api.device.channel to standalone.mqtt_adapter

Author: rnovatorov

src/enapter/mqtt/api/device/channel.py

@@ -1,4 +1,3 @@
-import logging
 from typing import AsyncContextManager, AsyncGenerator
 
 from enapter import async_, mqtt
@@ -9,14 +8,11 @@
 from .properties import Properties
 from .telemetry import Telemetry
 
-LOGGER = logging.getLogger(__name__)
-
 
 class Channel:
 
     def __init__(self, client: mqtt.Client, hardware_id: str, channel_id: str) -> None:
         self._client = client
-        self._logger = self._new_logger(hardware_id, channel_id)
         self._hardware_id = hardware_id
         self._channel_id = channel_id
 
@@ -28,11 +24,6 @@ def hardware_id(self) -> str:
     def channel_id(self) -> str:
         return self._channel_id
 
-    @staticmethod
-    def _new_logger(hardware_id: str, channel_id: str) -> logging.LoggerAdapter:
-        extra = {"hardware_id": hardware_id, "channel_id": channel_id}
-        return logging.LoggerAdapter(LOGGER, extra=extra)
-
     @async_.generator
     async def subscribe_to_command_requests(
         self,
@@ -62,7 +53,4 @@ def _subscribe(
 
     async def _publish(self, path: str, payload: str, **kwargs) -> None:
         topic = f"v1/from/{self._hardware_id}/{self._channel_id}/{path}"
-        try:
-            await self._client.publish(topic, payload, **kwargs)
-        except Exception as e:
-            self._logger.error("failed to publish %s: %r", path, e)
+        await self._client.publish(topic, payload, **kwargs)

src/enapter/standalone/mqtt_adapter.py

@@ -1,23 +1,31 @@
 import asyncio
 import contextlib
+import logging
 import time
 import traceback
 
 from enapter import async_, mqtt
 
 from .device_protocol import DeviceProtocol
 
+LOGGER = logging.getLogger(__name__)
+
 
 class MQTTAdapter(async_.Routine):
 
     def __init__(
         self,
-        device_channel: mqtt.api.device.Channel,
+        hardware_id: str,
+        channel_id: str,
+        mqtt_api_client: mqtt.api.Client,
         device: DeviceProtocol,
         task_group: asyncio.TaskGroup | None,
     ) -> None:
         super().__init__(task_group=task_group)
-        self._device_channel = device_channel
+        self._logger = logging.LoggerAdapter(
+            LOGGER, extra={"hardware_id": hardware_id, "channel_id": channel_id}
+        )
+        self._device_channel = mqtt_api_client.device_channel(hardware_id, channel_id)
         self._device = device
 
     async def _run(self) -> None:
@@ -33,36 +41,63 @@ async def _stream_properties(self) -> None:
             async for properties in stream:
                 properties = properties.copy()
                 timestamp = properties.pop("timestamp", int(time.time()))
-                await self._device_channel.publish_properties(
-                    properties=mqtt.api.device.Properties(
-                        timestamp=timestamp, values=properties
-                    )
+                await self._publish_properties(
+                    mqtt.api.device.Properties(timestamp=timestamp, values=properties)
                 )
 
+    async def _publish_properties(self, properties: mqtt.api.device.Properties) -> None:
+        try:
+            await self._device_channel.publish_properties(properties=properties)
+        except Exception as e:
+            self._logger.error("failed to publish properties: %s", e)
+
     async def _stream_telemetry(self) -> None:
         async with contextlib.aclosing(self._device.stream_telemetry()) as stream:
             async for telemetry in stream:
                 telemetry = telemetry.copy()
                 timestamp = telemetry.pop("timestamp", int(time.time()))
                 alerts = telemetry.pop("alerts", None)
-                await self._device_channel.publish_telemetry(
-                    telemetry=mqtt.api.device.Telemetry(
+                await self._publish_telemetry(
+                    mqtt.api.device.Telemetry(
                         timestamp=timestamp, alerts=alerts, values=telemetry
                     )
                 )
 
+    async def _publish_telemetry(self, telemetry: mqtt.api.device.Telemetry) -> None:
+        try:
+            await self._device_channel.publish_telemetry(telemetry=telemetry)
+        except Exception as e:
+            self._logger.error("failed to publish telemetry: %s", e)
+
     async def _stream_logs(self) -> None:
         async with contextlib.aclosing(self._device.stream_logs()) as stream:
             async for log in stream:
-                await self._device_channel.publish_log(
-                    log=mqtt.api.device.Log(
+                match log.severity:
+                    case "debug":
+                        self._logger.debug(log.message)
+                    case "info":
+                        self._logger.info(log.message)
+                    case "warning":
+                        self._logger.warning(log.message)
+                    case "error":
+                        self._logger.error(log.message)
+                    case _:
+                        raise NotImplementedError(log.severity)
+                await self._publish_log(
+                    mqtt.api.device.Log(
                         timestamp=int(time.time()),
                         severity=mqtt.api.device.LogSeverity(log.severity),
                         message=log.message,
                         persist=log.persist,
                     )
                 )
 
+    async def _publish_log(self, log: mqtt.api.device.Log) -> None:
+        try:
+            await self._device_channel.publish_log(log=log)
+        except Exception as e:
+            self._logger.error("failed to publish log: %s", e)
+
     async def _execute_commands(self) -> None:
         async with asyncio.TaskGroup() as tg:
             async with self._device_channel.subscribe_to_command_requests() as requests:

src/enapter/standalone/run.py

@@ -19,21 +19,19 @@ async def run(device: DeviceProtocol) -> None:
         )
         _ = await stack.enter_async_context(
             MQTTAdapter(
-                device_channel=mqtt_api_client.device_channel(
-                    hardware_id=config.communication.hardware_id,
-                    channel_id=config.communication.channel_id,
-                ),
+                hardware_id=config.communication.hardware_id,
+                channel_id=config.communication.channel_id,
+                mqtt_api_client=mqtt_api_client,
                 device=device,
                 task_group=task_group,
             )
         )
         if config.communication.ucm_needed:
             _ = await stack.enter_async_context(
                 MQTTAdapter(
-                    device_channel=mqtt_api_client.device_channel(
-                        hardware_id=config.communication.hardware_id,
-                        channel_id="ucm",
-                    ),
+                    hardware_id=config.communication.hardware_id,
+                    channel_id="ucm",
+                    mqtt_api_client=mqtt_api_client,
                     device=UCM(),
                     task_group=task_group,
                 )