http: api: implement expand device communication info

Author: rnovatorov

src/enapter/cli/http/api/device_get_command.py

@@ -32,6 +32,12 @@ def register(parent: cli.Subparsers) -> None:
             action="store_true",
             help="Expand device connectivity information",
         )
+        parser.add_argument(
+            "-u",
+            "--communication",
+            action="store_true",
+            help="Expand device communication information",
+        )
 
     @staticmethod
     async def run(args: argparse.Namespace) -> None:
@@ -41,5 +47,6 @@ async def run(args: argparse.Namespace) -> None:
                 expand_manifest=args.manifest,
                 expand_properties=args.properties,
                 expand_connectivity=args.connectivity,
+                expand_communication=args.communication,
             )
             print(json.dumps(device.to_dto()))

src/enapter/http/api/devices/__init__.py

@@ -2,6 +2,7 @@
 from .client import Client
 from .communication_config import CommunicationConfig
 from .device import Device
+from .device_communication import DeviceCommunication, DeviceCommunicationType
 from .device_connectivity import DeviceConnectivity, DeviceConnectivityStatus
 from .device_type import DeviceType
 from .mqtt_credentials import MQTTCredentials
@@ -17,6 +18,8 @@
     "DeviceType",
     "DeviceConnectivity",
     "DeviceConnectivityStatus",
+    "DeviceCommunication",
+    "DeviceCommunicationType",
     "MQTTCredentials",
     "MQTTProtocol",
     "MQTTSCredentials",

src/enapter/http/api/devices/client.py

@@ -27,12 +27,14 @@ async def get(
         expand_manifest: bool = False,
         expand_properties: bool = False,
         expand_connectivity: bool = False,
+        expand_communication: bool = False,
     ) -> Device:
         url = f"v3/devices/{device_id}"
         expand = {
             "manifest": expand_manifest,
             "properties": expand_properties,
             "connectivity": expand_connectivity,
+            "communication": expand_communication,
         }
         params = {"expand": ",".join(k for k, v in expand.items() if v)}
         response = await self._client.get(url, params=params)

src/enapter/http/api/devices/device.py

@@ -3,6 +3,7 @@
 from typing import Any, Self
 
 from .authorized_role import AuthorizedRole
+from .device_communication import DeviceCommunication
 from .device_connectivity import DeviceConnectivity
 from .device_type import DeviceType
 
@@ -21,6 +22,7 @@ class Device:
     manifest: dict[str, Any] | None = None
     properties: dict[str, Any] | None = None
     connectivity: DeviceConnectivity | None = None
+    communication: DeviceCommunication | None = None
 
     @classmethod
     def from_dto(cls, dto: dict[str, Any]) -> Self:
@@ -40,6 +42,11 @@ def from_dto(cls, dto: dict[str, Any]) -> Self:
                 if dto.get("connectivity") is not None
                 else None
             ),
+            communication=(
+                DeviceCommunication.from_dto(dto["communication"])
+                if dto.get("communication") is not None
+                else None
+            ),
         )
 
     def to_dto(self) -> dict[str, Any]:
@@ -57,4 +64,7 @@ def to_dto(self) -> dict[str, Any]:
             "connectivity": (
                 self.connectivity.to_dto() if self.connectivity is not None else None
             ),
+            "communication": (
+                self.communication.to_dto() if self.communication is not None else None
+            ),
         }

src/enapter/http/api/devices/device_communication.py

@@ -0,0 +1,39 @@
+import dataclasses
+import enum
+from typing import Any
+
+
+class DeviceCommunicationType(str, enum.Enum):
+
+    MQTT_V1_PLAINTEXT = "MQTT_V1_PLAINTEXT"
+    MQTT_V1_TLS = "MQTT_V1_TLS"
+    MQTT_V1_LOCALHOST = "MQTT_V1_LOCALHOST"
+    UCM_LUA = "UCM_LUA"
+    UCM_EMBEDDED = "UCM_EMBEDDED"
+    LINK = "LINK"
+
+
+@dataclasses.dataclass
+class DeviceCommunication:
+
+    type: DeviceCommunicationType
+    upstream_id: str | None
+    hardware_id: str | None
+    channel_id: str | None
+
+    @classmethod
+    def from_dto(cls, dto: dict[str, Any]) -> "DeviceCommunication":
+        return cls(
+            type=DeviceCommunicationType(dto["type"]),
+            upstream_id=dto.get("upstream_id"),
+            hardware_id=dto.get("hardware_id"),
+            channel_id=dto.get("channel_id"),
+        )
+
+    def to_dto(self) -> dict[str, Any]:
+        return {
+            "type": self.type.value,
+            "upstream_id": self.upstream_id,
+            "hardware_id": self.hardware_id,
+            "channel_id": self.channel_id,
+        }