http: api: telemetry: implement timeseries

Author: rnovatorov

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

@@ -3,6 +3,7 @@
 from enapter import cli
 
 from .telemetry_latest_command import TelemetryLatestCommand
+from .telemetry_timeseries_command import TelemetryTimeseriesCommand
 
 
 class TelemetryCommand(cli.Command):
@@ -15,6 +16,7 @@ def register(parent: cli.Subparsers) -> None:
         subparsers = parser.add_subparsers(dest="telemetry_command", required=True)
         for command in [
             TelemetryLatestCommand,
+            TelemetryTimeseriesCommand,
         ]:
             command.register(subparsers)
 
@@ -23,5 +25,7 @@ async def run(args: argparse.Namespace) -> None:
         match args.telemetry_command:
             case "latest":
                 await TelemetryLatestCommand.run(args)
+            case "timeseries":
+                await TelemetryTimeseriesCommand.run(args)
             case _:
                 raise NotImplementedError(args.telemetry_command)

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

@@ -12,19 +12,19 @@ def register(parent: cli.Subparsers) -> None:
             "latest", formatter_class=argparse.ArgumentDefaultsHelpFormatter
         )
         parser.add_argument(
-            "device_attrs",
+            "selectors",
             metavar="device:attr1,attr2,...,attrN",
             nargs="+",
-            help="Device attributes to get the latest telemetry for",
+            help="Device ID or slug followed by a colon and a comma-separated list of attribute names to fetch the telemetry for",
         )
 
     @staticmethod
     async def run(args: argparse.Namespace) -> None:
+        attributes_by_device = {}
+        for selector in args.selectors:
+            device, attributes = selector.split(":", 1)
+            attributes_by_device[device] = attributes.split(",")
         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: {

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

@@ -0,0 +1,111 @@
+import argparse
+import datetime
+import json
+
+from enapter import cli, http
+
+
+class TelemetryTimeseriesCommand(cli.Command):
+
+    @staticmethod
+    def register(parent: cli.Subparsers) -> None:
+        parser = parent.add_parser(
+            "timeseries", formatter_class=argparse.ArgumentDefaultsHelpFormatter
+        )
+        parser.add_argument(
+            "-f",
+            "--time-from",
+            required=True,
+            help="Start time for the telemetry data (ISO 8601 format)",
+        )
+        parser.add_argument(
+            "-t", "--time-to", help="End time for the telemetry data (ISO 8601 format)"
+        )
+        parser.add_argument(
+            "-s",
+            "--shape",
+            choices=["w", "wide", "l", "long"],
+            default="long",
+            help="Shape of the output data",
+        )
+        parser.add_argument(
+            "-g",
+            "--granularity",
+            type=int,
+            default=60 * 60,
+            help="Granularity of the telemetry data in seconds",
+        )
+        parser.add_argument(
+            "selectors",
+            metavar="device:attr1,attr2,...,attrN",
+            nargs="+",
+            help="Device ID or slug followed by a colon and a comma-separated list of attribute names to fetch the telemetry for",
+        )
+
+    @staticmethod
+    async def run(args: argparse.Namespace) -> None:
+        attributes_by_device = {}
+        for selector in args.selectors:
+            device, attributes = selector.split(":", 1)
+            attributes_by_device[device] = attributes.split(",")
+        async with http.api.Client(http.api.Config.from_env()) as client:
+            time_to = (
+                datetime.datetime.fromisoformat(args.time_to)
+                if args.time_to is not None
+                else datetime.datetime.now(datetime.timezone.utc)
+            )
+            time_from = (
+                datetime.datetime.fromisoformat(args.time_from)
+                if args.time_from is not None
+                else time_to - datetime.timedelta(minutes=10)
+            )
+            match args.shape:
+                case "l" | "long":
+                    await TelemetryTimeseriesCommand._handle_long_timeseries(
+                        args, client, time_from, time_to, attributes_by_device
+                    )
+                case "w" | "wide":
+                    await TelemetryTimeseriesCommand._handle_wide_timeseries(
+                        args, client, time_from, time_to, attributes_by_device
+                    )
+                case _:
+                    raise NotImplementedError(args.shape)
+
+    @staticmethod
+    async def _handle_long_timeseries(
+        args: argparse.Namespace,
+        client: http.api.Client,
+        time_from: datetime.datetime,
+        time_to: datetime.datetime,
+        attributes_by_device: dict[str, list[str]],
+    ) -> None:
+        async with client.telemetry.long_timeseries(
+            from_=time_from,
+            to=time_to,
+            granularity=args.granularity,
+            selectors=[
+                http.api.telemetry.Selector(device=device, attributes=attributes)
+                for device, attributes in attributes_by_device.items()
+            ],
+        ) as stream:
+            async for row in stream:
+                print(json.dumps(row.to_dto()))
+
+    @staticmethod
+    async def _handle_wide_timeseries(
+        args: argparse.Namespace,
+        client: http.api.Client,
+        time_from: datetime.datetime,
+        time_to: datetime.datetime,
+        attributes_by_device: dict[str, list[str]],
+    ) -> None:
+        wide_timeseries = await client.telemetry.wide_timeseries(
+            from_=time_from,
+            to=time_to,
+            granularity=args.granularity,
+            selectors=[
+                http.api.telemetry.Selector(device=device, attributes=attributes)
+                for device, attributes in attributes_by_device.items()
+            ],
+        )
+        print(json.dumps(wide_timeseries.to_dto()))

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

@@ -1,4 +1,33 @@
+from .aggregation import Aggregation
 from .client import Client
+from .data_type import DataType
+from .gap_filling import GapFilling
+from .gap_filling_method import GapFillingMethod
+from .labels import Labels
 from .latest_datapoint import LatestDatapoint
+from .long_timeseries_row import LongTimeseriesRow
+from .long_timeseries_stream import LongTimeseriesStream
+from .raw_timeseries_row import RawTimeseriesRow
+from .raw_timeseries_stream import RawTimeseriesStream
+from .selector import Selector
+from .wide_timeseries import WideTimeseries
+from .wide_timeseries_column import WideTimeseriesColumn
 
-__all__ = ["Client", "LatestDatapoint"]
+__all__ = [
+    "Aggregation",
+    "Client",
+    "DataType",
+    "GapFilling",
+    "GapFillingMethod",
+    "Labels",
+    "LatestDatapoint",
+    "LongTimeseriesRow",
+    "LongTimeseriesStream",
+    "RawTimeseriesRow",
+    "RawTimeseriesStream",
+    "Selector",
+    "TimeseriesStream",
+    "TimeseriesStreamRow",
+    "WideTimeseries",
+    "WideTimeseriesColumn",
+]

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

@@ -0,0 +1,11 @@
+import enum
+
+
+class Aggregation(enum.Enum):
+
+    AUTO = "AUTO"
+    AVG = "AVG"
+    LAST = "LAST"
+    MIN = "MIN"
+    MAX = "MAX"
+    BOOL_OR = "BOOL_OR"

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

@@ -1,10 +1,17 @@
+import contextlib
 import datetime
+from typing import AsyncGenerator
 
 import httpx
 
 from enapter.http import api
 
+from .data_type import DataType
 from .latest_datapoint import LatestDatapoint
+from .long_timeseries_stream import LongTimeseriesStream
+from .raw_timeseries_stream import RawTimeseriesStream
+from .selector import Selector
+from .wide_timeseries import WideTimeseries
 
 
 class Client:
@@ -39,3 +46,60 @@ async def latest(
             }
             for device, datapoints in response.json().get("telemetry", {}).items()
         }
+
+    @contextlib.asynccontextmanager
+    async def long_timeseries(
+        self,
+        from_: datetime.datetime,
+        to: datetime.datetime,
+        granularity: int | datetime.timedelta,
+        selectors: list[Selector],
+    ) -> AsyncGenerator[LongTimeseriesStream, None]:
+        async with self.raw_timeseries(
+            from_=from_, to=to, granularity=granularity, selectors=selectors
+        ) as stream:
+            yield LongTimeseriesStream.from_raw_stream(stream)
+
+    async def wide_timeseries(
+        self,
+        from_: datetime.datetime,
+        to: datetime.datetime,
+        granularity: int | datetime.timedelta,
+        selectors: list[Selector],
+    ) -> WideTimeseries:
+        async with self.raw_timeseries(
+            from_=from_, to=to, granularity=granularity, selectors=selectors
+        ) as stream:
+            return await WideTimeseries.from_raw_stream(stream)
+
+    @contextlib.asynccontextmanager
+    async def raw_timeseries(
+        self,
+        from_: datetime.datetime,
+        to: datetime.datetime,
+        granularity: int | datetime.timedelta,
+        selectors: list[Selector],
+    ) -> AsyncGenerator[RawTimeseriesStream, None]:
+        if to <= from_:
+            raise ValueError("`to` must be greater than `from_`")
+        if isinstance(granularity, datetime.timedelta):
+            granularity = int(granularity.total_seconds())
+        payload = {
+            "from": from_.isoformat(),
+            "to": to.isoformat(),
+            "granularity": str(granularity) + "s",
+            "telemetry": [selector.to_dto() for selector in selectors],
+        }
+        async with self._client.stream(
+            "POST",
+            "v3/telemetry/query_timeseries",
+            headers={"Accept": "text/csv"},
+            json=payload,
+        ) as response:
+            api.check_error(response)
+            data_types = [
+                DataType(dt.strip().upper())
+                for dt in response.headers["X-Enapter-Timeseries-Data-Types"].split(",")
+            ]
+            lines = response.aiter_lines()
+            yield await RawTimeseriesStream.new(data_types, lines)

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

@@ -0,0 +1,30 @@
+import enum
+
+
+class DataType(enum.Enum):
+
+    FLOAT = "FLOAT"
+    INTEGER = "INTEGER"
+    STRING = "STRING"
+    STRING_ARRAY = "STRING_ARRAY"
+    BOOLEAN = "BOOLEAN"
+
+    def parse_value(self, s: str) -> float | int | str | list[str] | bool | None:
+        if not s:
+            return None
+        match self:
+            case DataType.FLOAT:
+                return float(s)
+            case DataType.INTEGER:
+                return int(s)
+            case DataType.STRING:
+                return s
+            case DataType.STRING_ARRAY:
+                raise NotImplementedError(self)
+            case DataType.BOOLEAN:
+                if s.lower() in ("true", "1"):
+                    return True
+                elif s.lower() in ("false", "0"):
+                    return False
+                else:
+                    raise ValueError(f"invalid boolean value: {s}")

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

@@ -0,0 +1,18 @@
+import dataclasses
+import datetime
+from typing import Any
+
+from .gap_filling_method import GapFillingMethod
+
+
+@dataclasses.dataclass
+class GapFilling:
+
+    method: GapFillingMethod
+    look_around: datetime.timedelta
+
+    def to_dto(self) -> dict[str, Any]:
+        return {
+            "method": self.method.value.lower(),
+            "look_around": str(int(self.look_around.total_seconds())) + "s",
+        }

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

@@ -0,0 +1,7 @@
+import enum
+
+
+class GapFillingMethod(enum.Enum):
+
+    NONE = "NONE"
+    LOCF = "LOCF"

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

@@ -0,0 +1,17 @@
+import collections
+from typing import Self
+
+
+class Labels(collections.UserDict):
+
+    @classmethod
+    def parse(cls, s: str) -> Self:
+        return cls(kv.split("=") for kv in s.split(" "))
+
+    @property
+    def device(self) -> str:
+        return self.data["device"]
+
+    @property
+    def telemetry(self) -> str:
+        return self.data["telemetry"]