From 3d378449e87ece726b5020204cc2e4c1f8bfd8b7 Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Tue, 29 Mar 2022 17:25:14 -1000 Subject: [PATCH] Move envoy last reported attribute to its own sensor (#68360) --- .../components/enphase_envoy/__init__.py | 23 +- .../components/enphase_envoy/const.py | 6 - .../components/enphase_envoy/sensor.py | 264 +++++++++++------- 3 files changed, 166 insertions(+), 127 deletions(-) diff --git a/homeassistant/components/enphase_envoy/__init__.py b/homeassistant/components/enphase_envoy/__init__.py index 7b3765bd25c..696baa31775 100644 --- a/homeassistant/components/enphase_envoy/__init__.py +++ b/homeassistant/components/enphase_envoy/__init__.py @@ -11,7 +11,7 @@ import httpx from homeassistant.config_entries import ConfigEntry from homeassistant.const import CONF_HOST, CONF_NAME, CONF_PASSWORD, CONF_USERNAME from homeassistant.core import HomeAssistant -from homeassistant.exceptions import ConfigEntryAuthFailed +from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady from homeassistant.helpers.httpx_client import get_async_client from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed @@ -38,7 +38,6 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: async def async_update_data(): """Fetch data from API endpoint.""" - data = {} async with async_timeout.timeout(30): try: await envoy_reader.getData() @@ -47,15 +46,11 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: except httpx.HTTPError as err: raise UpdateFailed(f"Error communicating with API: {err}") from err - for description in SENSORS: - if description.key != "inverters": - data[description.key] = await getattr( - envoy_reader, description.key - )() - else: - data[ - "inverters_production" - ] = await envoy_reader.inverters_production() + data = { + description.key: await getattr(envoy_reader, description.key)() + for description in SENSORS + } + data["inverters_production"] = await envoy_reader.inverters_production() _LOGGER.debug("Retrieved data from API: %s", data) @@ -78,8 +73,10 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: if not entry.unique_id: try: serial = await envoy_reader.get_full_serial_number() - except httpx.HTTPError: - pass + except httpx.HTTPError as ex: + raise ConfigEntryNotReady( + f"Could not obtain serial number from envoy: {ex}" + ) from ex else: hass.config_entries.async_update_entry(entry, unique_id=serial) diff --git a/homeassistant/components/enphase_envoy/const.py b/homeassistant/components/enphase_envoy/const.py index 747a4886f15..c79c3af604b 100644 --- a/homeassistant/components/enphase_envoy/const.py +++ b/homeassistant/components/enphase_envoy/const.py @@ -71,10 +71,4 @@ SENSORS = ( state_class=SensorStateClass.TOTAL_INCREASING, device_class=SensorDeviceClass.ENERGY, ), - SensorEntityDescription( - key="inverters", - name="Inverter", - native_unit_of_measurement=POWER_WATT, - state_class=SensorStateClass.MEASUREMENT, - ), ) diff --git a/homeassistant/components/enphase_envoy/sensor.py b/homeassistant/components/enphase_envoy/sensor.py index 03431ddc3ea..c8d791907a6 100644 --- a/homeassistant/components/enphase_envoy/sensor.py +++ b/homeassistant/components/enphase_envoy/sensor.py @@ -1,16 +1,77 @@ """Support for Enphase Envoy solar energy monitor.""" from __future__ import annotations -from homeassistant.components.sensor import SensorEntity +from collections.abc import Callable +from dataclasses import dataclass +import datetime +import logging +from typing import cast + +from homeassistant.components.sensor import ( + SensorDeviceClass, + SensorEntity, + SensorEntityDescription, + SensorStateClass, +) from homeassistant.config_entries import ConfigEntry +from homeassistant.const import POWER_WATT from homeassistant.core import HomeAssistant from homeassistant.helpers.entity import DeviceInfo from homeassistant.helpers.entity_platform import AddEntitiesCallback -from homeassistant.helpers.update_coordinator import CoordinatorEntity +from homeassistant.helpers.update_coordinator import ( + CoordinatorEntity, + DataUpdateCoordinator, +) +from homeassistant.util import dt as dt_util from .const import COORDINATOR, DOMAIN, NAME, SENSORS ICON = "mdi:flash" +_LOGGER = logging.getLogger(__name__) + +INVERTERS_KEY = "inverters" +LAST_REPORTED_KEY = "last_reported" + + +@dataclass +class EnvoyRequiredKeysMixin: + """Mixin for required keys.""" + + value_fn: Callable[[tuple[float, str]], datetime.datetime | float | None] + + +@dataclass +class EnvoySensorEntityDescription(SensorEntityDescription, EnvoyRequiredKeysMixin): + """Describes an Envoy inverter sensor entity.""" + + +def _inverter_last_report_time( + watt_report_time: tuple[float, str] +) -> datetime.datetime | None: + if (report_time := watt_report_time[1]) is None: + return None + if (last_reported_dt := dt_util.parse_datetime(report_time)) is None: + return None + if last_reported_dt.tzinfo is None: + return last_reported_dt.replace(tzinfo=dt_util.UTC) + return last_reported_dt + + +INVERTER_SENSORS = ( + EnvoySensorEntityDescription( + key=INVERTERS_KEY, + native_unit_of_measurement=POWER_WATT, + state_class=SensorStateClass.MEASUREMENT, + value_fn=lambda watt_report_time: watt_report_time[0], + ), + EnvoySensorEntityDescription( + key=LAST_REPORTED_KEY, + name="Last Reported", + device_class=SensorDeviceClass.TIMESTAMP, + entity_registry_enabled_default=False, + value_fn=_inverter_last_report_time, + ), +) async def async_setup_entry( @@ -19,129 +80,116 @@ async def async_setup_entry( async_add_entities: AddEntitiesCallback, ) -> None: """Set up envoy sensor platform.""" - data = hass.data[DOMAIN][config_entry.entry_id] - coordinator = data[COORDINATOR] - name = data[NAME] + data: dict = hass.data[DOMAIN][config_entry.entry_id] + coordinator: DataUpdateCoordinator = data[COORDINATOR] + envoy_data: dict = coordinator.data + envoy_name: str = data[NAME] + envoy_serial_num = config_entry.unique_id + assert envoy_serial_num is not None + _LOGGER.debug("Envoy data: %s", envoy_data) - entities = [] - for sensor_description in SENSORS: - if ( - sensor_description.key == "inverters" - and coordinator.data.get("inverters_production") is not None - ): - for inverter in coordinator.data["inverters_production"]: - entity_name = f"{name} {sensor_description.name} {inverter}" - split_name = entity_name.split(" ") - serial_number = split_name[-1] - entities.append( - Envoy( - sensor_description, - entity_name, - name, - config_entry.unique_id, - serial_number, - coordinator, - ) - ) - elif sensor_description.key != "inverters": - data = coordinator.data.get(sensor_description.key) - if isinstance(data, str) and "not available" in data: - continue - - entity_name = f"{name} {sensor_description.name}" - entities.append( - Envoy( - sensor_description, - entity_name, - name, - config_entry.unique_id, - None, - coordinator, - ) + entities: list[Envoy | EnvoyInverter] = [] + for description in SENSORS: + sensor_data = envoy_data.get(description.key) + if isinstance(sensor_data, str) and "not available" in sensor_data: + continue + entities.append( + Envoy( + coordinator, + description, + envoy_name, + envoy_serial_num, ) + ) + + if production := envoy_data.get("inverters_production"): + entities.extend( + EnvoyInverter( + coordinator, + description, + envoy_name, + envoy_serial_num, + str(inverter), + ) + for description in INVERTER_SENSORS + for inverter in production + ) async_add_entities(entities) class Envoy(CoordinatorEntity, SensorEntity): - """Envoy entity.""" + """Envoy inverter entity.""" + + _attr_icon = ICON def __init__( self, - description, - name, - device_name, - device_serial_number, - serial_number, - coordinator, - ): + coordinator: DataUpdateCoordinator, + description: SensorEntityDescription, + envoy_name: str, + envoy_serial_num: str, + ) -> None: """Initialize Envoy entity.""" self.entity_description = description - self._name = name - self._serial_number = serial_number - self._device_name = device_name - self._device_serial_number = device_serial_number - + self._attr_name = f"{envoy_name} {description.name}" + self._attr_unique_id = f"{envoy_serial_num}_{description.key}" + self._attr_device_info = DeviceInfo( + identifiers={(DOMAIN, envoy_serial_num)}, + manufacturer="Enphase", + model="Envoy", + name=envoy_name, + ) super().__init__(coordinator) @property - def name(self): - """Return the name of the sensor.""" - return self._name - - @property - def unique_id(self): - """Return the unique id of the sensor.""" - if self._serial_number: - return self._serial_number - if self._device_serial_number: - return f"{self._device_serial_number}_{self.entity_description.key}" - - @property - def native_value(self): + def native_value(self) -> float | None: """Return the state of the sensor.""" - if self.entity_description.key != "inverters": - value = self.coordinator.data.get(self.entity_description.key) + if (value := self.coordinator.data.get(self.entity_description.key)) is None: + return None + return cast(float, value) - elif ( - self.entity_description.key == "inverters" - and self.coordinator.data.get("inverters_production") is not None - ): - value = self.coordinator.data.get("inverters_production").get( - self._serial_number - )[0] + +class EnvoyInverter(CoordinatorEntity, SensorEntity): + """Envoy inverter entity.""" + + _attr_icon = ICON + entity_description: EnvoySensorEntityDescription + + def __init__( + self, + coordinator: DataUpdateCoordinator, + description: EnvoySensorEntityDescription, + envoy_name: str, + envoy_serial_num: str, + serial_number: str, + ) -> None: + """Initialize Envoy inverter entity.""" + self.entity_description = description + self._serial_number = serial_number + if description.name: + self._attr_name = ( + f"{envoy_name} Inverter {serial_number} {description.name}" + ) else: - return None - - return value - - @property - def icon(self): - """Icon to use in the frontend, if any.""" - return ICON - - @property - def extra_state_attributes(self): - """Return the state attributes.""" - if ( - self.entity_description.key == "inverters" - and self.coordinator.data.get("inverters_production") is not None - ): - value = self.coordinator.data.get("inverters_production").get( - self._serial_number - )[1] - return {"last_reported": value} - - return None - - @property - def device_info(self) -> DeviceInfo | None: - """Return the device_info of the device.""" - if not self._device_serial_number: - return None - return DeviceInfo( - identifiers={(DOMAIN, str(self._device_serial_number))}, + self._attr_name = f"{envoy_name} Inverter {serial_number}" + if description.key == INVERTERS_KEY: + self._attr_unique_id = serial_number + else: + self._attr_unique_id = f"{serial_number}_{description.key}" + self._attr_device_info = DeviceInfo( + identifiers={(DOMAIN, serial_number)}, + name=f"Inverter {serial_number}", manufacturer="Enphase", - model="Envoy", - name=self._device_name, + model="Inverter", + via_device=(DOMAIN, envoy_serial_num), ) + super().__init__(coordinator) + + @property + def native_value(self) -> datetime.datetime | float | None: + """Return the state of the sensor.""" + watt_report_time: tuple[float, str] = self.coordinator.data[ + "inverters_production" + ][self._serial_number] + return self.entity_description.value_fn(watt_report_time)