http: api: implement latest telemetry

Author: rnovatorov

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

@@ -6,6 +6,7 @@
 from .command_command import CommandCommand
 from .device_command import DeviceCommand
 from .site_command import SiteCommand
+from .telemetry_command import TelemetryCommand
 
 
 class Command(cli.Command):
@@ -21,6 +22,7 @@ def register(parent: cli.Subparsers) -> None:
             CommandCommand,
             DeviceCommand,
             SiteCommand,
+            TelemetryCommand,
         ]:
             command.register(subparsers)
 
@@ -35,5 +37,7 @@ async def run(args: argparse.Namespace) -> None:
                 await DeviceCommand.run(args)
             case "site":
                 await SiteCommand.run(args)
+            case "telemetry":
+                await TelemetryCommand.run(args)
             case _:
                 raise NotImplementedError(args.device_command)

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

@@ -0,0 +1,27 @@
+import argparse
+
+from enapter import cli
+
+from .telemetry_latest_command import TelemetryLatestCommand
+
+
+class TelemetryCommand(cli.Command):
+
+    @staticmethod
+    def register(parent: cli.Subparsers) -> None:
+        parser = parent.add_parser(
+            "telemetry", formatter_class=argparse.ArgumentDefaultsHelpFormatter
+        )
+        subparsers = parser.add_subparsers(dest="telemetry_command", required=True)
+        for command in [
+            TelemetryLatestCommand,
+        ]:
+            command.register(subparsers)
+
+    @staticmethod
+    async def run(args: argparse.Namespace) -> None:
+        match args.telemetry_command:
+            case "latest":
+                await TelemetryLatestCommand.run(args)
+            case _:
+                raise NotImplementedError(args.telemetry_command)

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

@@ -0,0 +1,36 @@
+import argparse
+import json
+
+from enapter import cli, http
+
+
+class TelemetryLatestCommand(cli.Command):
+
+    @staticmethod
+    def register(parent: cli.Subparsers) -> None:
+        parser = parent.add_parser(
+            "latest", formatter_class=argparse.ArgumentDefaultsHelpFormatter
+        )
+        parser.add_argument(
+            "device_attrs",
+            metavar="device:attr1,attr2,...,attrN",
+            nargs="+",
+            help="Device attributes to get the latest telemetry for",
+        )
+
+    @staticmethod
+    async def run(args: argparse.Namespace) -> None:
+        async with http.api.Client(http.api.Config.from_env()) as client:
+            attributes_by_device = {}
+            for device_attrs in args.device_attrs:
+                device, attrs_str = device_attrs.split(":", 1)
+                attributes_by_device[device] = attrs_str.split(",")
+            telemetry = await client.telemetry.latest(attributes_by_device)
+            dto = {
+                device: {
+                    attribute: datapoint.to_dto() if datapoint is not None else None
+                    for attribute, datapoint in datapoints.items()
+                }
+                for device, datapoints in telemetry.items()
+            }
+            print(json.dumps(dto))

src/enapter/http/api/client.py

@@ -2,7 +2,7 @@
 
 import httpx
 
-from enapter.http.api import blueprints, commands, devices, sites
+from enapter.http.api import blueprints, commands, devices, sites, telemetry
 
 from .config import Config
 
@@ -45,3 +45,7 @@ def commands(self) -> commands.Client:
     @property
     def blueprints(self) -> blueprints.Client:
         return blueprints.Client(client=self._client)
+
+    @property
+    def telemetry(self) -> telemetry.Client:
+        return telemetry.Client(client=self._client)

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

@@ -0,0 +1,4 @@
+from .client import Client
+from .latest_datapoint import LatestDatapoint
+
+__all__ = ["Client", "LatestDatapoint"]

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

@@ -0,0 +1,41 @@
+import datetime
+
+import httpx
+
+from enapter.http import api
+
+from .latest_datapoint import LatestDatapoint
+
+
+class Client:
+
+    def __init__(self, client: httpx.AsyncClient) -> None:
+        self._client = client
+
+    async def latest(
+        self,
+        attributes_by_device: dict[str, list[str]],
+        relevance_interval: int | datetime.timedelta | None = None,
+    ) -> dict[str, dict[str, LatestDatapoint | None]]:
+        url = "v3/telemetry/latest"
+        params = {
+            f"devices[{device}]": ",".join(attributes)
+            for device, attributes in attributes_by_device.items()
+        }
+        if relevance_interval is not None:
+            if isinstance(relevance_interval, datetime.timedelta):
+                relevance_interval = int(relevance_interval.total_seconds())
+            params["relevance_interval"] = str(relevance_interval) + "s"
+        response = await self._client.get(url, params=params)
+        api.check_error(response)
+        return {
+            device: {
+                attribute: (
+                    LatestDatapoint.from_dto(datapoint)
+                    if datapoint.get("timestamp") is not None
+                    else None
+                )
+                for attribute, datapoint in datapoints.items()
+            }
+            for device, datapoints in response.json().get("telemetry", {}).items()
+        }

src/enapter/http/api/telemetry/latest_datapoint.py

@@ -0,0 +1,23 @@
+import dataclasses
+import datetime
+from typing import Any, Self
+
+
+@dataclasses.dataclass
+class LatestDatapoint:
+
+    timestamp: datetime.datetime
+    value: Any
+
+    @classmethod
+    def from_dto(cls, dto: dict[str, Any]) -> Self:
+        return cls(
+            timestamp=datetime.datetime.fromtimestamp(dto["timestamp"]),
+            value=dto.get("value"),
+        )
+
+    def to_dto(self) -> dict[str, Any]:
+        return {
+            "timestamp": int(self.timestamp.timestamp()),
+            "value": self.value,
+        }