Allows getting statistics from Aquarea Client (#23)

* Adding models

* Getting stats from client

Author: cjaliaga

aioaquarea/__init__.py

@@ -22,6 +22,7 @@
     ClientError,
     RequestFailedError,
 )
+from .statistics import Consumption, DateType
 
 __all__: Tuple[str, ...] = (
     "Client",
@@ -39,4 +40,6 @@
     "ExtendedOperationMode",
     "UpdateOperationMode",
     "QuietMode"
+    "DateType",
+    "Consumption",
 )

aioaquarea/const.py

@@ -3,8 +3,11 @@
 AQUAREA_SERVICE_BASE = "https://aquarea-smart.panasonic.com/"
 AQUAREA_SERVICE_LOGIN = "remote/v1/api/auth/login"
 AQUAREA_SERVICE_DEVICES = "remote/v1/api/devices"
+AQUAREA_SERVICE_CONSUMPTION = "remote/v1/api/consumption"
 AQUAREA_SERVICE_CONTRACT = "remote/contract"
 
-AQUAREA_SERVICE_A2W_STATUS_DISPLAY = "https://aquarea-smart.panasonic.com/remote/a2wStatusDisplay"
+AQUAREA_SERVICE_A2W_STATUS_DISPLAY = (
+    "https://aquarea-smart.panasonic.com/remote/a2wStatusDisplay"
+)
 
 PANASONIC = "Panasonic"

aioaquarea/core.py

@@ -2,7 +2,7 @@
 from __future__ import annotations
 
 import asyncio
-from datetime import datetime, timezone
+from datetime import date, datetime, timezone
 import functools
 import logging
 from typing import List, Optional
@@ -13,6 +13,7 @@
 from .const import (
     AQUAREA_SERVICE_A2W_STATUS_DISPLAY,
     AQUAREA_SERVICE_BASE,
+    AQUAREA_SERVICE_CONSUMPTION,
     AQUAREA_SERVICE_CONTRACT,
     AQUAREA_SERVICE_DEVICES,
     AQUAREA_SERVICE_LOGIN,
@@ -35,6 +36,7 @@
     ZoneSensor,
 )
 from .errors import ApiError, AuthenticationError, AuthenticationErrorCodes, InvalidData
+from .statistics import Consumption, DateType
 
 
 def auth_required(fn):
@@ -533,6 +535,19 @@ async def post_set_quiet_mode(
             json=data,
         )
 
+    async def get_device_consumption(
+        self, long_id: str, aggregation: DateType, date_input: str
+    ) -> Consumption:
+        """Get device consumption"""
+        response = await self.request(
+            "GET",
+            f"{AQUAREA_SERVICE_CONSUMPTION}/{long_id}?{aggregation}={date_input}",
+            referer=AQUAREA_SERVICE_A2W_STATUS_DISPLAY,
+        )
+
+        date_data = await response.json()
+        return Consumption(date_data.get("dateData")[0])
+
 
 class TankImpl(Tank):
     """Tank implementation"""

aioaquarea/statistics.py

@@ -0,0 +1,99 @@
+"""Statistics models for Aquarea"""
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+try:
+    from enum import StrEnum
+except ImportError:
+    from strenum import StrEnum
+
+
+class DateType(StrEnum):
+    """Date types"""
+
+    DAY = "date"
+    WEEK = "week"
+    MONTH = "month"
+    YEAR = "year"
+
+
+class AggregationType(StrEnum):
+    """Aggregation type"""
+
+    HOURLY = "hourly"
+    DAILY = "daily"
+    MONTHLY = "monthly"
+
+
+class DataType(StrEnum):
+    """Data type"""
+
+    HEAT = "Heat"
+    COOL = "AC"
+    WATER_TANK = "HW"
+    TOTAL = "Consume"
+
+
+class GenerationDataType(StrEnum):
+    """Generation data type"""
+
+    GHT = "GHT"
+    GHR = "GHR"
+    GHW = "GHW"
+
+
+class DataSetName(StrEnum):
+    """Data set name"""
+
+    ENERGY = "energyShowing"
+    GENERATED = "generateEnergyShowing"
+    COST = "costShowing"
+    TEMPERATURE = "temperatureShowing"
+
+
+class Consumption:
+    """Consumption"""
+
+    def __init__(self, date_data: dict[str, object]):
+        self._data = dict(
+            [
+                (
+                    dataset.get("name"),
+                    dict(
+                        [
+                            (data.get("name"), data.get("values"))
+                            for data in dataset.get("data", [])
+                        ]
+                    ),
+                )
+                for dataset in date_data.get("dataSets", [])
+            ]
+        )
+        self._start_date = date_data.get("startDate")
+        self._aggregation = AggregationType(date_data.get("timeline", {}).get("type"))
+
+    @property
+    def energy(self) -> dict[DataType, list[float | None]]:
+        """Energy consumption in kWh"""
+        return self._data.get(DataSetName.ENERGY)
+
+    @property
+    def generation(self) -> dict[GenerationDataType, list[float]]:
+        """Energy generation in kWh"""
+        return self._data.get(DataSetName.GENERATED)
+
+    @property
+    def cost(self) -> dict[DataType, list[float]]:
+        """Energy cost in configured currency"""
+        return self._data.get(DataSetName.COST)
+
+    @property
+    def start_date(self) -> str:
+        """Start date of the period"""
+        return self._start_date
+
+    @property
+    def aggregation(self) -> AggregationType:
+        """Aggregation type"""
+        return self._aggregation