http: api: implement expand device connectivity

Author: rnovatorov

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

@@ -26,6 +26,12 @@ def register(parent: cli.Subparsers) -> None:
             action="store_true",
             help="Expand device properties information",
         )
+        parser.add_argument(
+            "-c",
+            "--connectivity",
+            action="store_true",
+            help="Expand device connectivity information",
+        )
 
     @staticmethod
     async def run(args: argparse.Namespace) -> None:
@@ -34,5 +40,6 @@ async def run(args: argparse.Namespace) -> None:
                 args.id,
                 expand_manifest=args.manifest,
                 expand_properties=args.properties,
+                expand_connectivity=args.connectivity,
             )
             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_connectivity import DeviceConnectivity, DeviceConnectivityStatus
 from .device_type import DeviceType
 from .mqtt_credentials import MQTTCredentials
 from .mqtt_protocol import MQTTProtocol
@@ -14,6 +15,8 @@
     "CommunicationConfig",
     "Device",
     "DeviceType",
+    "DeviceConnectivity",
+    "DeviceConnectivityStatus",
     "MQTTCredentials",
     "MQTTProtocol",
     "MQTTSCredentials",

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

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

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

@@ -3,6 +3,7 @@
 from typing import Any, Self
 
 from .authorized_role import AuthorizedRole
+from .device_connectivity import DeviceConnectivity
 from .device_type import DeviceType
 
 
@@ -19,6 +20,7 @@ class Device:
     authorized_role: AuthorizedRole
     manifest: dict[str, Any] | None = None
     properties: dict[str, Any] | None = None
+    connectivity: DeviceConnectivity | None = None
 
     @classmethod
     def from_dto(cls, dto: dict[str, Any]) -> Self:
@@ -33,6 +35,11 @@ def from_dto(cls, dto: dict[str, Any]) -> Self:
             authorized_role=AuthorizedRole(dto["authorized_role"]),
             manifest=dto.get("manifest"),
             properties=dto.get("properties"),
+            connectivity=(
+                DeviceConnectivity.from_dto(dto["connectivity"])
+                if dto.get("connectivity") is not None
+                else None
+            ),
         )
 
     def to_dto(self) -> dict[str, Any]:
@@ -47,4 +54,7 @@ def to_dto(self) -> dict[str, Any]:
             "authorized_role": self.authorized_role.value,
             "manifest": self.manifest,
             "properties": self.properties,
+            "connectivity": (
+                self.connectivity.to_dto() if self.connectivity is not None else None
+            ),
         }

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

@@ -0,0 +1,22 @@
+import dataclasses
+import enum
+
+
+class DeviceConnectivityStatus(enum.Enum):
+
+    UNKNOWN = "UNKNOWN"
+    ONLINE = "ONLINE"
+    OFFLINE = "OFFLINE"
+
+
+@dataclasses.dataclass
+class DeviceConnectivity:
+
+    status: DeviceConnectivityStatus
+
+    @classmethod
+    def from_dto(cls, dto: dict[str, str]) -> "DeviceConnectivity":
+        return cls(status=DeviceConnectivityStatus(dto.get("status", "UNKNOWN")))
+
+    def to_dto(self) -> dict[str, str]:
+        return {"status": self.status.value}