Merge pull request #11 from Enapter/feat/telemetry-aggregation

Historical telemetry aggregation control

Author: rnovatorov

.gitignore

@@ -1,3 +1,4 @@
+.codex
 .gemini
 .DS_Store
 

src/enapter_mcp_server/core/application_server.py

@@ -229,9 +229,16 @@ async def get_historical_telemetry(
         time_from: datetime.datetime,
         time_to: datetime.datetime,
         granularity: int,
+        aggregation: domain.AggregationFunction,
     ) -> domain.HistoricalTelemetry:
         return await self._enapter_api.get_historical_telemetry(
-            auth, device_id, attributes, time_from, time_to, granularity
+            auth,
+            device_id,
+            attributes,
+            time_from,
+            time_to,
+            granularity,
+            aggregation,
         )
 
     async def search_command_executions(

src/enapter_mcp_server/core/enapter_api.py

@@ -53,4 +53,5 @@ async def get_historical_telemetry(
         time_from: datetime.datetime,
         time_to: datetime.datetime,
         granularity: int,
+        aggregation: domain.AggregationFunction,
     ) -> domain.HistoricalTelemetry: ...

src/enapter_mcp_server/domain/__init__.py

@@ -1,3 +1,4 @@
+from .aggregation_function import AggregationFunction
 from .alert_declaration import AlertDeclaration
 from .alert_severity import AlertSeverity
 from .blueprint_section import BlueprintSection
@@ -19,6 +20,7 @@
 from .telemetry_attribute_declaration import TelemetryAttributeDeclaration
 
 __all__ = [
+    "AggregationFunction",
     "AlertDeclaration",
     "AlertSeverity",
     "BlueprintSection",

src/enapter_mcp_server/domain/aggregation_function.py

@@ -0,0 +1,8 @@
+import enum
+
+
+class AggregationFunction(str, enum.Enum):
+    AVG = "avg"
+    MIN = "min"
+    MAX = "max"
+    LAST = "last"

src/enapter_mcp_server/http/enapter_api.py

@@ -100,6 +100,7 @@ async def get_historical_telemetry(
         time_from: datetime.datetime,
         time_to: datetime.datetime,
         granularity: int,
+        aggregation: domain.AggregationFunction,
     ) -> domain.HistoricalTelemetry:
         async with self._new_client(auth) as client:
             telemetry = await client.telemetry.wide_timeseries(
@@ -108,7 +109,11 @@ async def get_historical_telemetry(
                 granularity=granularity,
                 selectors=[
                     enapter.http.api.telemetry.Selector(
-                        device=device_id, attributes=attributes
+                        device=device_id,
+                        attributes=attributes,
+                        aggregation=enapter.http.api.telemetry.Aggregation(
+                            aggregation.value.upper()
+                        ),
                     )
                 ],
             )

src/enapter_mcp_server/mcp/models/__init__.py

@@ -1,3 +1,4 @@
+from .aggregation_function import AggregationFunction
 from .alert_declaration import AlertDeclaration
 from .alert_severity import AlertSeverity
 from .blueprint_section import BlueprintSection
@@ -18,6 +19,7 @@
 from .telemetry_attribute_declaration import TelemetryAttributeDeclaration
 
 __all__ = [
+    "AggregationFunction",
     "AlertDeclaration",
     "AlertSeverity",
     "BlueprintSection",

src/enapter_mcp_server/mcp/models/aggregation_function.py

@@ -0,0 +1,3 @@
+from typing import Literal
+
+AggregationFunction = Literal["avg", "min", "max", "last"]

src/enapter_mcp_server/mcp/server.py

@@ -302,15 +302,15 @@ async def get_historical_telemetry(
         attributes: list[str],
         time_from: datetime.datetime,
         time_to: datetime.datetime,
+        aggregation: models.AggregationFunction,
         granularity: int = 60 * 60,
     ) -> models.HistoricalTelemetry:
-        """Retrieve aggregated telemetry data.
+        """Retrieve aggregated historical telemetry data for a specific device.
 
-        Most devices send telemetry data once per second. To reduce the amount
-        of data transferred, the `granularity` parameter can be used to
-        aggregate data over a specified interval (in seconds). For example, a
-        granularity of 3600 seconds (1 hour) will return hourly averages of the
-        telemetry data.
+        The data is divided into time buckets of `granularity` seconds. Each
+        returned timestamp marks the start of a bucket, and its values are the
+        result of applying the `aggregation` function to all data points within
+        that bucket.
         """
         auth = await self._get_auth_config()
         telemetry = await self._app.get_historical_telemetry(
@@ -320,6 +320,7 @@ async def get_historical_telemetry(
             time_from=time_from,
             time_to=time_to,
             granularity=granularity,
+            aggregation=domain.AggregationFunction(aggregation),
         )
         return models.HistoricalTelemetry.from_domain(telemetry)
 

tests/unit/core/test_application_server.py

@@ -102,6 +102,7 @@ async def get_historical_telemetry(
         time_from: datetime.datetime,
         time_to: datetime.datetime,
         granularity: int,
+        aggregation: domain.AggregationFunction,
     ) -> domain.HistoricalTelemetry:
         if self._historical_telemetry is None:
             raise NotImplementedError()
@@ -745,6 +746,7 @@ async def test_get_historical_telemetry(self) -> None:
             datetime.datetime.now(),
             datetime.datetime.now(),
             60,
+            domain.AggregationFunction.AVG,
         )
 
         assert result == historical