Merge pull request #14 from tijsverkoyen/peakpower

Peak Power entities for Premium and PremiumHR users

README.md

@@ -21,6 +21,15 @@ For each meter in EnergyID there will be a sensor for the:
 
 These will be updated every 15 minutes.
 
+### EnergyID Premium and Premium HR
+The following will only be available for Premium and Premium HR users.
+
+For each meter in EnergyID there will be a sensor for the:
+
+* current month Peak Power Power (kW)
+* last month Peak Power Power (kW)
+* current month Peak Power Datetime 
+* last month Peak Power Datetime 
 
 ## Services
 ### set_meter_reading

custom_components/energy_id/energy_id/api.py

@@ -1,5 +1,6 @@
 """API client for EnergyID API."""
 import urllib.parse
+import datetime
 
 from requests import get, post, HTTPError
 
@@ -50,6 +51,18 @@ def get_record_meters(self, record: str, expand: list = None):
 
         return data
 
+    def get_record_analyse_peak_power(self, record: str, start: datetime, end: datetime):
+        params = {
+            "start": start.strftime('%Y-%m-%d'),
+            "end": end.strftime('%Y-%m-%d')
+        }
+
+        return self._do_call(
+            'GET',
+            f'api/v1/records/{record}/analyses/peakPower',
+            params=params
+        )
+
     def get_meter_readings(self, meter: str, take: int = 20, next_row_key: str = None):
         params = {
             "take": take

custom_components/energy_id/peak_power_coordinator.py

@@ -0,0 +1,48 @@
+from datetime import timedelta, datetime
+from dateutil.relativedelta import relativedelta
+
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
+
+from .energy_id.api import EnergyIDApi
+from .energy_id.record import EnergyIDRecord
+
+from .const import RESPONSE_ATTRIBUTE_VALUE
+
+import logging
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class EnergyIDRecordPeakPowerCoordinator(DataUpdateCoordinator):
+    def __init__(self, hass: HomeAssistant, api: EnergyIDApi, record: EnergyIDRecord):
+        super().__init__(
+            hass,
+            _LOGGER,
+            name="EnergyIDRecordPeakPowerCoordinator",
+            update_interval=timedelta(hours=1),
+        )
+        self.api = api
+        self.record = record
+
+    async def _async_update_data(self):
+        def fix_datetime_offset(match):
+            return f'+{match.group(1)}{match.group(2)}'
+
+        today = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)
+        first_day_of_previous_month = today.replace(day=1) - relativedelta(months=1)
+        last_day_of_current_month = today + relativedelta(day=31)
+
+        response = await self.hass.async_add_executor_job(
+            self.api.get_record_analyse_peak_power,
+            self.record.record_number,
+            first_day_of_previous_month,
+            last_day_of_current_month
+        )
+
+        if RESPONSE_ATTRIBUTE_VALUE in response:
+            for value in response[RESPONSE_ATTRIBUTE_VALUE]:
+                if value['id'] == f'/records/{self.record.record_id}/analyses/peakPower':
+                    return value
+
+        return None

custom_components/energy_id/peak_power_sensor.py

@@ -0,0 +1,249 @@
+from datetime import datetime
+from dateutil.relativedelta import relativedelta
+import re
+
+from homeassistant.helpers.update_coordinator import CoordinatorEntity, DataUpdateCoordinator
+from homeassistant.helpers.entity import DeviceInfo
+from homeassistant.components.sensor import SensorEntity, SensorDeviceClass, SensorStateClass
+from homeassistant.core import callback
+from homeassistant.const import UnitOfPower
+
+from .energy_id.meter import EnergyIDMeter
+from .energy_id.record import EnergyIDRecord
+
+import logging
+
+_LOGGER = logging.getLogger(__name__)
+
+
+def fix_datetime_offset(match):
+    return f'+{match.group(1)}{match.group(2)}'
+
+
+class EnergyIDRecordPeakPowerPower(CoordinatorEntity, SensorEntity):
+    _attr_has_entity_name = True
+    _attr_device_class = SensorDeviceClass.POWER
+    _attr_state_class = SensorStateClass.MEASUREMENT
+    _attr_suggested_display_precision = 2
+
+    def __init__(
+            self,
+            coordinator: DataUpdateCoordinator,
+            record: EnergyIDRecord,
+            meter: EnergyIDMeter,
+    ):
+        super().__init__(coordinator)
+        self._record = record
+        self._meter = meter
+        self._value = None
+        self._native_unit_of_measurement = UnitOfPower.KILO_WATT
+
+    @property
+    def device_info(self) -> DeviceInfo | None:
+        return self._meter.device_info
+
+    @property
+    def native_value(self) -> float | None:
+        return self._value
+
+    @property
+    def native_unit_of_measurement(self) -> str:
+        return self._native_unit_of_measurement
+
+
+class EnergyIDRecordCurrentMonthPeakPowerPower(EnergyIDRecordPeakPowerPower):
+    @property
+    def name(self):
+        return f'{self._record.display_name}: {self._meter.display_name} - Current month PeakPower Power'
+
+    @property
+    def unique_id(self) -> str:
+        return f'meter-{self._meter.meter_id}-current-month-peak-power-power'
+
+    @callback
+    def _handle_coordinator_update(self) -> None:
+        """Handle updated data from the coordinator."""
+        data = self.coordinator.data
+
+        if data is not None:
+            self._native_unit_of_measurement = data['unit']
+            value = None
+
+            for serie in data['series']:
+                if serie['name'] != self._meter.meter_id:
+                    continue
+
+                for serie_data in data['series'][0]['data']:
+                    month = datetime.strptime(
+                        re.sub(
+                            '\+([0-2][0-9]):([0-9]{2})$',
+                            fix_datetime_offset,
+                            serie_data['month']
+                        ),
+                        '%Y-%m-%dT%H:%M:%S%z'
+                    )
+
+                    if month.month == datetime.now().month:
+                        value = serie_data['total']
+                        break
+
+            self._value = value
+            self.async_write_ha_state()
+
+
+class EnergyIDRecordLastMonthPeakPowerPower(EnergyIDRecordPeakPowerPower):
+    @property
+    def name(self):
+        return f'{self._record.display_name}: {self._meter.display_name} - Last month PeakPower Power'
+
+    @property
+    def unique_id(self) -> str:
+        return f'meter-{self._meter.meter_id}-last-month-peak-power-power'
+
+    @callback
+    def _handle_coordinator_update(self) -> None:
+        """Handle updated data from the coordinator."""
+        data = self.coordinator.data
+
+        if data is not None:
+            self._native_unit_of_measurement = data['unit']
+            value = None
+            first_day_of_previous_month = datetime.now().replace(day=1) - relativedelta(months=1)
+
+            for serie in data['series']:
+                if serie['name'] != self._meter.meter_id:
+                    continue
+
+                for serie_data in data['series'][0]['data']:
+                    month = datetime.strptime(
+                        re.sub(
+                            '\+([0-2][0-9]):([0-9]{2})$',
+                            fix_datetime_offset,
+                            serie_data['month']
+                        ),
+                        '%Y-%m-%dT%H:%M:%S%z'
+                    )
+
+                    if month.month == first_day_of_previous_month.month:
+                        value = serie_data['total']
+                        break
+
+            self._value = value
+            self.async_write_ha_state()
+
+
+class EnergyIDRecordPeakPowerDatetime(CoordinatorEntity, SensorEntity):
+    _attr_has_entity_name = True
+    _attr_device_class = SensorDeviceClass.TIMESTAMP
+
+    def __init__(
+            self,
+            coordinator: DataUpdateCoordinator,
+            record: EnergyIDRecord,
+            meter: EnergyIDMeter,
+    ):
+        super().__init__(coordinator)
+        self._record = record
+        self._meter = meter
+        self._value = None
+
+    @property
+    def device_info(self) -> DeviceInfo | None:
+        return self._meter.device_info
+
+    @property
+    def native_value(self) -> datetime | None:
+        return self._value
+
+
+class EnergyIDRecordCurrentMonthPeakPowerDatetime(EnergyIDRecordPeakPowerDatetime):
+    @property
+    def name(self):
+        return f'{self._record.display_name}: {self._meter.display_name} - Current month PeakPower Datetime'
+
+    @property
+    def unique_id(self) -> str:
+        return f'meter-{self._meter.meter_id}-current-month-peak-power-datetime'
+
+    @callback
+    def _handle_coordinator_update(self) -> None:
+        """Handle updated data from the coordinator."""
+        data = self.coordinator.data
+
+        if data is not None:
+            value = None
+
+            for serie in data['series']:
+                if serie['name'] != self._meter.meter_id:
+                    continue
+
+                for serie_data in data['series'][0]['data']:
+                    month = datetime.strptime(
+                        re.sub(
+                            '\+([0-2][0-9]):([0-9]{2})$',
+                            fix_datetime_offset,
+                            serie_data['month']
+                        ),
+                        '%Y-%m-%dT%H:%M:%S%z'
+                    )
+
+                    if month.month == datetime.now().month:
+                        value = datetime.strptime(
+                            re.sub(
+                                '\+([0-2][0-9]):([0-9]{2})$',
+                                fix_datetime_offset,
+                                serie_data['timestamp']
+                            ),
+                            '%Y-%m-%dT%H:%M:%S%z'
+                        )
+                        break
+
+            self._value = value
+            self.async_write_ha_state()
+
+
+class EnergyIDRecordLastMonthPeakPowerDatetime(EnergyIDRecordPeakPowerDatetime):
+    @property
+    def name(self):
+        return f'{self._record.display_name}: {self._meter.display_name} - Last month PeakPower Datetime'
+
+    @property
+    def unique_id(self) -> str:
+        return f'meter-{self._meter.meter_id}-last-month-peak-power-datetime'
+
+    @callback
+    def _handle_coordinator_update(self) -> None:
+        """Handle updated data from the coordinator."""
+        data = self.coordinator.data
+
+        if data is not None:
+            value = None
+            first_day_of_previous_month = datetime.now().replace(day=1) - relativedelta(months=1)
+
+            for serie in data['series']:
+                if serie['name'] != self._meter.meter_id:
+                    continue
+
+                for serie_data in data['series'][0]['data']:
+                    month = datetime.strptime(
+                        re.sub(
+                            '\+([0-2][0-9]):([0-9]{2})$',
+                            fix_datetime_offset,
+                            serie_data['month']
+                        ),
+                        '%Y-%m-%dT%H:%M:%S%z'
+                    )
+
+                    if month.month == first_day_of_previous_month.month:
+                        value = datetime.strptime(
+                            re.sub(
+                                '\+([0-2][0-9]):([0-9]{2})$',
+                                fix_datetime_offset,
+                                serie_data['timestamp']
+                            ),
+                            '%Y-%m-%dT%H:%M:%S%z'
+                        )
+                        break
+
+            self._value = value
+            self.async_write_ha_state()