diff --git a/homeassistant/components/smappee/sensor.py b/homeassistant/components/smappee/sensor.py index ec93501a508..af66b788a41 100644 --- a/homeassistant/components/smappee/sensor.py +++ b/homeassistant/components/smappee/sensor.py @@ -1,8 +1,13 @@ """Support for monitoring a Smappee energy sensor.""" +from __future__ import annotations + +from dataclasses import dataclass, field + from homeassistant.components.sensor import ( STATE_CLASS_MEASUREMENT, STATE_CLASS_TOTAL_INCREASING, SensorEntity, + SensorEntityDescription, ) from homeassistant.const import ( DEVICE_CLASS_ENERGY, @@ -16,141 +21,177 @@ from homeassistant.const import ( from .const import DOMAIN -TREND_SENSORS = { - "total_power": [ - "Total consumption - Active power", - None, - POWER_WATT, - "total_power", - DEVICE_CLASS_POWER, - True, # both cloud and local - ], - "alwayson": [ - "Always on - Active power", - None, - POWER_WATT, - "alwayson", - DEVICE_CLASS_POWER, - False, # cloud only - ], - "power_today": [ - "Total consumption - Today", - None, - ENERGY_WATT_HOUR, - "power_today", - DEVICE_CLASS_ENERGY, - False, # cloud only - ], - "power_current_hour": [ - "Total consumption - Current hour", - None, - ENERGY_WATT_HOUR, - "power_current_hour", - DEVICE_CLASS_ENERGY, - False, # cloud only - ], - "power_last_5_minutes": [ - "Total consumption - Last 5 minutes", - None, - ENERGY_WATT_HOUR, - "power_last_5_minutes", - DEVICE_CLASS_ENERGY, - False, # cloud only - ], - "alwayson_today": [ - "Always on - Today", - None, - ENERGY_WATT_HOUR, - "alwayson_today", - DEVICE_CLASS_ENERGY, - False, # cloud only - ], -} -REACTIVE_SENSORS = { - "total_reactive_power": [ - "Total consumption - Reactive power", - None, - POWER_WATT, - "total_reactive_power", - DEVICE_CLASS_POWER, - ] -} -SOLAR_SENSORS = { - "solar_power": [ - "Total production - Active power", - None, - POWER_WATT, - "solar_power", - DEVICE_CLASS_POWER, - True, # both cloud and local - ], - "solar_today": [ - "Total production - Today", - None, - ENERGY_WATT_HOUR, - "solar_today", - DEVICE_CLASS_ENERGY, - False, # cloud only - ], - "solar_current_hour": [ - "Total production - Current hour", - None, - ENERGY_WATT_HOUR, - "solar_current_hour", - DEVICE_CLASS_ENERGY, - False, # cloud only - ], -} -VOLTAGE_SENSORS = { - "phase_voltages_a": [ - "Phase voltages - A", - None, - ELECTRIC_POTENTIAL_VOLT, - "phase_voltage_a", - DEVICE_CLASS_VOLTAGE, - ["ONE", "TWO", "THREE_STAR", "THREE_DELTA"], - ], - "phase_voltages_b": [ - "Phase voltages - B", - None, - ELECTRIC_POTENTIAL_VOLT, - "phase_voltage_b", - DEVICE_CLASS_VOLTAGE, - ["TWO", "THREE_STAR", "THREE_DELTA"], - ], - "phase_voltages_c": [ - "Phase voltages - C", - None, - ELECTRIC_POTENTIAL_VOLT, - "phase_voltage_c", - DEVICE_CLASS_VOLTAGE, - ["THREE_STAR"], - ], - "line_voltages_a": [ - "Line voltages - A", - None, - ELECTRIC_POTENTIAL_VOLT, - "line_voltage_a", - DEVICE_CLASS_VOLTAGE, - ["ONE", "TWO", "THREE_STAR", "THREE_DELTA"], - ], - "line_voltages_b": [ - "Line voltages - B", - None, - ELECTRIC_POTENTIAL_VOLT, - "line_voltage_b", - DEVICE_CLASS_VOLTAGE, - ["TWO", "THREE_STAR", "THREE_DELTA"], - ], - "line_voltages_c": [ - "Line voltages - C", - None, - ELECTRIC_POTENTIAL_VOLT, - "line_voltage_c", - DEVICE_CLASS_VOLTAGE, - ["THREE_STAR", "THREE_DELTA"], - ], -} + +@dataclass +class SmappeeRequiredKeysMixin: + """Mixin for required keys.""" + + sensor_id: str + + +@dataclass +class SmappeeSensorEntityDescription(SensorEntityDescription, SmappeeRequiredKeysMixin): + """Describes Smappee sensor entity.""" + + +@dataclass +class SmappeePollingSensorEntityDescription(SmappeeSensorEntityDescription): + """Describes Smappee sensor entity.""" + + local_polling: bool = False + + +@dataclass +class SmappeeVoltageSensorEntityDescription(SmappeeSensorEntityDescription): + """Describes Smappee sensor entity.""" + + phase_types: set[str] = field(default_factory=set) + + +TREND_SENSORS: tuple[SmappeePollingSensorEntityDescription, ...] = ( + SmappeePollingSensorEntityDescription( + key="total_power", + name="Total consumption - Active power", + native_unit_of_measurement=POWER_WATT, + sensor_id="total_power", + device_class=DEVICE_CLASS_POWER, + state_class=STATE_CLASS_MEASUREMENT, + local_polling=True, # both cloud and local + ), + SmappeePollingSensorEntityDescription( + key="alwayson", + name="Always on - Active power", + native_unit_of_measurement=POWER_WATT, + sensor_id="alwayson", + device_class=DEVICE_CLASS_POWER, + state_class=STATE_CLASS_MEASUREMENT, + ), + SmappeePollingSensorEntityDescription( + key="power_today", + name="Total consumption - Today", + native_unit_of_measurement=ENERGY_WATT_HOUR, + sensor_id="power_today", + device_class=DEVICE_CLASS_ENERGY, + state_class=STATE_CLASS_TOTAL_INCREASING, + ), + SmappeePollingSensorEntityDescription( + key="power_current_hour", + name="Total consumption - Current hour", + native_unit_of_measurement=ENERGY_WATT_HOUR, + sensor_id="power_current_hour", + device_class=DEVICE_CLASS_ENERGY, + state_class=STATE_CLASS_TOTAL_INCREASING, + ), + SmappeePollingSensorEntityDescription( + key="power_last_5_minutes", + name="Total consumption - Last 5 minutes", + native_unit_of_measurement=ENERGY_WATT_HOUR, + sensor_id="power_last_5_minutes", + device_class=DEVICE_CLASS_ENERGY, + state_class=STATE_CLASS_TOTAL_INCREASING, + ), + SmappeePollingSensorEntityDescription( + key="alwayson_today", + name="Always on - Today", + native_unit_of_measurement=ENERGY_WATT_HOUR, + sensor_id="alwayson_today", + device_class=DEVICE_CLASS_ENERGY, + state_class=STATE_CLASS_TOTAL_INCREASING, + ), +) +REACTIVE_SENSORS: tuple[SmappeeSensorEntityDescription, ...] = ( + SmappeeSensorEntityDescription( + key="total_reactive_power", + name="Total consumption - Reactive power", + native_unit_of_measurement=POWER_WATT, + sensor_id="total_reactive_power", + device_class=DEVICE_CLASS_POWER, + state_class=STATE_CLASS_MEASUREMENT, + ), +) +SOLAR_SENSORS: tuple[SmappeePollingSensorEntityDescription, ...] = ( + SmappeePollingSensorEntityDescription( + key="solar_power", + name="Total production - Active power", + native_unit_of_measurement=POWER_WATT, + sensor_id="solar_power", + device_class=DEVICE_CLASS_POWER, + state_class=STATE_CLASS_MEASUREMENT, + local_polling=True, # both cloud and local + ), + SmappeePollingSensorEntityDescription( + key="solar_today", + name="Total production - Today", + native_unit_of_measurement=ENERGY_WATT_HOUR, + sensor_id="solar_today", + device_class=DEVICE_CLASS_ENERGY, + state_class=STATE_CLASS_TOTAL_INCREASING, + ), + SmappeePollingSensorEntityDescription( + key="solar_current_hour", + name="Total production - Current hour", + native_unit_of_measurement=ENERGY_WATT_HOUR, + sensor_id="solar_current_hour", + device_class=DEVICE_CLASS_ENERGY, + state_class=STATE_CLASS_TOTAL_INCREASING, + ), +) +VOLTAGE_SENSORS: tuple[SmappeeVoltageSensorEntityDescription, ...] = ( + SmappeeVoltageSensorEntityDescription( + key="phase_voltages_a", + name="Phase voltages - A", + native_unit_of_measurement=ELECTRIC_POTENTIAL_VOLT, + sensor_id="phase_voltage_a", + device_class=DEVICE_CLASS_VOLTAGE, + state_class=STATE_CLASS_MEASUREMENT, + phase_types={"ONE", "TWO", "THREE_STAR", "THREE_DELTA"}, + ), + SmappeeVoltageSensorEntityDescription( + key="phase_voltages_b", + name="Phase voltages - B", + native_unit_of_measurement=ELECTRIC_POTENTIAL_VOLT, + sensor_id="phase_voltage_b", + device_class=DEVICE_CLASS_VOLTAGE, + state_class=STATE_CLASS_MEASUREMENT, + phase_types={"TWO", "THREE_STAR", "THREE_DELTA"}, + ), + SmappeeVoltageSensorEntityDescription( + key="phase_voltages_c", + name="Phase voltages - C", + native_unit_of_measurement=ELECTRIC_POTENTIAL_VOLT, + sensor_id="phase_voltage_c", + device_class=DEVICE_CLASS_VOLTAGE, + state_class=STATE_CLASS_MEASUREMENT, + phase_types={"THREE_STAR"}, + ), + SmappeeVoltageSensorEntityDescription( + key="line_voltages_a", + name="Line voltages - A", + native_unit_of_measurement=ELECTRIC_POTENTIAL_VOLT, + sensor_id="line_voltage_a", + device_class=DEVICE_CLASS_VOLTAGE, + state_class=STATE_CLASS_MEASUREMENT, + phase_types={"ONE", "TWO", "THREE_STAR", "THREE_DELTA"}, + ), + SmappeeVoltageSensorEntityDescription( + key="line_voltages_b", + name="Line voltages - B", + native_unit_of_measurement=ELECTRIC_POTENTIAL_VOLT, + sensor_id="line_voltage_b", + device_class=DEVICE_CLASS_VOLTAGE, + state_class=STATE_CLASS_MEASUREMENT, + phase_types={"TWO", "THREE_STAR", "THREE_DELTA"}, + ), + SmappeeVoltageSensorEntityDescription( + key="line_voltages_c", + name="Line voltages - C", + native_unit_of_measurement=ELECTRIC_POTENTIAL_VOLT, + sensor_id="line_voltage_c", + device_class=DEVICE_CLASS_VOLTAGE, + state_class=STATE_CLASS_MEASUREMENT, + phase_types={"THREE_STAR", "THREE_DELTA"}, + ), +) async def async_setup_entry(hass, config_entry, async_add_entities): @@ -161,116 +202,125 @@ async def async_setup_entry(hass, config_entry, async_add_entities): for service_location in smappee_base.smappee.service_locations.values(): # Add all basic sensors (realtime values and aggregators) # Some are available in local only env - for sensor, attributes in TREND_SENSORS.items(): - if not service_location.local_polling or attributes[5]: - entities.append( - SmappeeSensor( - smappee_base=smappee_base, - service_location=service_location, - sensor=sensor, - attributes=attributes, - ) - ) - - if service_location.has_reactive_value: - for reactive_sensor, attributes in REACTIVE_SENSORS.items(): - entities.append( - SmappeeSensor( - smappee_base=smappee_base, - service_location=service_location, - sensor=reactive_sensor, - attributes=attributes, - ) - ) - - # Add solar sensors (some are available in local only env) - if service_location.has_solar_production: - for sensor, attributes in SOLAR_SENSORS.items(): - if not service_location.local_polling or attributes[5]: - entities.append( - SmappeeSensor( - smappee_base=smappee_base, - service_location=service_location, - sensor=sensor, - attributes=attributes, - ) - ) - - # Add all CT measurements - for measurement_id, measurement in service_location.measurements.items(): - entities.append( + entities.extend( + [ SmappeeSensor( smappee_base=smappee_base, service_location=service_location, - sensor="load", - attributes=[ - measurement.name, - None, - POWER_WATT, - measurement_id, - DEVICE_CLASS_POWER, - ], + description=description, ) + for description in TREND_SENSORS + if not service_location.local_polling or description.local_polling + ] + ) + + if service_location.has_reactive_value: + entities.extend( + [ + SmappeeSensor( + smappee_base=smappee_base, + service_location=service_location, + description=description, + ) + for description in REACTIVE_SENSORS + ] ) + # Add solar sensors (some are available in local only env) + if service_location.has_solar_production: + entities.extend( + [ + SmappeeSensor( + smappee_base=smappee_base, + service_location=service_location, + description=description, + ) + for description in SOLAR_SENSORS + if not service_location.local_polling or description.local_polling + ] + ) + + # Add all CT measurements + entities.extend( + [ + SmappeeSensor( + smappee_base=smappee_base, + service_location=service_location, + description=SmappeeSensorEntityDescription( + key="load", + name=measurement.name, + sensor_id=measurement_id, + device_class=DEVICE_CLASS_POWER, + state_class=STATE_CLASS_MEASUREMENT, + ), + ) + for measurement_id, measurement in service_location.measurements.items() + ] + ) + # Add phase- and line voltages if available if service_location.has_voltage_values: - for sensor_name, sensor in VOLTAGE_SENSORS.items(): - if service_location.phase_type in sensor[5]: + entities.extend( + [ + SmappeeSensor( + smappee_base=smappee_base, + service_location=service_location, + description=description, + ) + for description in VOLTAGE_SENSORS if ( - sensor_name.startswith("line_") - and service_location.local_polling - ): - continue - entities.append( - SmappeeSensor( - smappee_base=smappee_base, - service_location=service_location, - sensor=sensor_name, - attributes=sensor, + service_location.phase_type in description.phase_types + and not ( + description.key.startswith("line_") + and service_location.local_polling ) ) + ] + ) # Add Gas and Water sensors - for sensor_id, sensor in service_location.sensors.items(): - for channel in sensor.channels: - gw_icon = "mdi:gas-cylinder" - if channel.get("type") == "water": - gw_icon = "mdi:water" - - entities.append( - SmappeeSensor( - smappee_base=smappee_base, - service_location=service_location, - sensor="sensor", - attributes=[ - channel.get("name"), - gw_icon, - channel.get("uom"), - f"{sensor_id}-{channel.get('channel')}", - None, - ], - ) + entities.extend( + [ + SmappeeSensor( + smappee_base=smappee_base, + service_location=service_location, + description=SmappeeSensorEntityDescription( + key="sensor", + name=channel.get("name"), + icon=( + "mdi:water" + if channel.get("type") == "water" + else "mdi:gas-cylinder" + ), + native_unit_of_measurement=channel.get("uom"), + sensor_id=f"{sensor_id}-{channel.get('channel')}", + state_class=STATE_CLASS_MEASUREMENT, + ), ) + for sensor_id, sensor in service_location.sensors.items() + for channel in sensor.channels + ] + ) # Add today_energy_kwh sensors for switches - for actuator_id, actuator in service_location.actuators.items(): - if actuator.type == "SWITCH": - entities.append( - SmappeeSensor( - smappee_base=smappee_base, - service_location=service_location, - sensor="switch", - attributes=[ - f"{actuator.name} - energy today", - None, - ENERGY_KILO_WATT_HOUR, - actuator_id, - DEVICE_CLASS_ENERGY, - False, # cloud only - ], - ) + entities.extend( + [ + SmappeeSensor( + smappee_base=smappee_base, + service_location=service_location, + description=SmappeeSensorEntityDescription( + key="switch", + name=f"{actuator.name} - energy today", + native_unit_of_measurement=ENERGY_KILO_WATT_HOUR, + sensor_id=actuator_id, + device_class=DEVICE_CLASS_ENERGY, + state_class=STATE_CLASS_TOTAL_INCREASING, + ), ) + for actuator_id, actuator in service_location.actuators.items() + if actuator.type == "SWITCH" + ] + ) async_add_entities(entities, True) @@ -278,84 +328,47 @@ async def async_setup_entry(hass, config_entry, async_add_entities): class SmappeeSensor(SensorEntity): """Implementation of a Smappee sensor.""" - def __init__(self, smappee_base, service_location, sensor, attributes): + entity_description: SmappeeSensorEntityDescription + + def __init__( + self, + smappee_base, + service_location, + description: SmappeeSensorEntityDescription, + ): """Initialize the Smappee sensor.""" + self.entity_description = description self._smappee_base = smappee_base self._service_location = service_location - self._sensor = sensor - self.data = None - self._state = None - self._name = attributes[0] - self._icon = attributes[1] - self._unit_of_measurement = attributes[2] - self._sensor_id = attributes[3] - self._device_class = attributes[4] @property def name(self): """Return the name for this sensor.""" - if self._sensor in ("sensor", "load", "switch"): + sensor_key = self.entity_description.key + sensor_name = self.entity_description.name + if sensor_key in ("sensor", "load", "switch"): return ( f"{self._service_location.service_location_name} - " - f"{self._sensor.title()} - {self._name}" + f"{sensor_key.title()} - {sensor_name}" ) - return f"{self._service_location.service_location_name} - {self._name}" + return f"{self._service_location.service_location_name} - {sensor_name}" @property - def icon(self): - """Icon to use in the frontend.""" - return self._icon - - @property - def native_value(self): - """Return the state of the sensor.""" - return self._state - - @property - def device_class(self): - """Return the class of this device, from component DEVICE_CLASSES.""" - return self._device_class - - @property - def state_class(self): - """Return the state class of this device.""" - scm = STATE_CLASS_MEASUREMENT - - if self._sensor in ( - "power_today", - "power_current_hour", - "power_last_5_minutes", - "solar_today", - "solar_current_hour", - "alwayson_today", - "switch", - ): - scm = STATE_CLASS_TOTAL_INCREASING - - return scm - - @property - def native_unit_of_measurement(self): - """Return the unit of measurement of this entity, if any.""" - return self._unit_of_measurement - - @property - def unique_id( - self, - ): + def unique_id(self): """Return the unique ID for this sensor.""" - if self._sensor in ("load", "sensor", "switch"): + sensor_key = self.entity_description.key + if sensor_key in ("load", "sensor", "switch"): return ( f"{self._service_location.device_serial_number}-" f"{self._service_location.service_location_id}-" - f"{self._sensor}-{self._sensor_id}" + f"{sensor_key}-{self.entity_description.sensor_id}" ) return ( f"{self._service_location.device_serial_number}-" f"{self._service_location.service_location_id}-" - f"{self._sensor}" + f"{sensor_key}" ) @property @@ -373,37 +386,38 @@ class SmappeeSensor(SensorEntity): """Get the latest data from Smappee and update the state.""" await self._smappee_base.async_update() - if self._sensor == "total_power": - self._state = self._service_location.total_power - elif self._sensor == "total_reactive_power": - self._state = self._service_location.total_reactive_power - elif self._sensor == "solar_power": - self._state = self._service_location.solar_power - elif self._sensor == "alwayson": - self._state = self._service_location.alwayson - elif self._sensor in ( + sensor_key = self.entity_description.key + if sensor_key == "total_power": + self._attr_native_value = self._service_location.total_power + elif sensor_key == "total_reactive_power": + self._attr_native_value = self._service_location.total_reactive_power + elif sensor_key == "solar_power": + self._attr_native_value = self._service_location.solar_power + elif sensor_key == "alwayson": + self._attr_native_value = self._service_location.alwayson + elif sensor_key in ( "phase_voltages_a", "phase_voltages_b", "phase_voltages_c", ): phase_voltages = self._service_location.phase_voltages if phase_voltages is not None: - if self._sensor == "phase_voltages_a": - self._state = phase_voltages[0] - elif self._sensor == "phase_voltages_b": - self._state = phase_voltages[1] - elif self._sensor == "phase_voltages_c": - self._state = phase_voltages[2] - elif self._sensor in ("line_voltages_a", "line_voltages_b", "line_voltages_c"): + if sensor_key == "phase_voltages_a": + self._attr_native_value = phase_voltages[0] + elif sensor_key == "phase_voltages_b": + self._attr_native_value = phase_voltages[1] + elif sensor_key == "phase_voltages_c": + self._attr_native_value = phase_voltages[2] + elif sensor_key in ("line_voltages_a", "line_voltages_b", "line_voltages_c"): line_voltages = self._service_location.line_voltages if line_voltages is not None: - if self._sensor == "line_voltages_a": - self._state = line_voltages[0] - elif self._sensor == "line_voltages_b": - self._state = line_voltages[1] - elif self._sensor == "line_voltages_c": - self._state = line_voltages[2] - elif self._sensor in ( + if sensor_key == "line_voltages_a": + self._attr_native_value = line_voltages[0] + elif sensor_key == "line_voltages_b": + self._attr_native_value = line_voltages[1] + elif sensor_key == "line_voltages_c": + self._attr_native_value = line_voltages[2] + elif sensor_key in ( "power_today", "power_current_hour", "power_last_5_minutes", @@ -411,21 +425,23 @@ class SmappeeSensor(SensorEntity): "solar_current_hour", "alwayson_today", ): - trend_value = self._service_location.aggregated_values.get(self._sensor) - self._state = round(trend_value) if trend_value is not None else None - elif self._sensor == "load": - self._state = self._service_location.measurements.get( - self._sensor_id + trend_value = self._service_location.aggregated_values.get(sensor_key) + self._attr_native_value = ( + round(trend_value) if trend_value is not None else None + ) + elif sensor_key == "load": + self._attr_native_value = self._service_location.measurements.get( + self.entity_description.sensor_id ).active_total - elif self._sensor == "sensor": - sensor_id, channel_id = self._sensor_id.split("-") + elif sensor_key == "sensor": + sensor_id, channel_id = self.entity_description.sensor_id.split("-") sensor = self._service_location.sensors.get(int(sensor_id)) for channel in sensor.channels: if channel.get("channel") == int(channel_id): - self._state = channel.get("value_today") - elif self._sensor == "switch": + self._attr_native_value = channel.get("value_today") + elif sensor_key == "switch": cons = self._service_location.actuators.get( - self._sensor_id + self.entity_description.sensor_id ).consumption_today if cons is not None: - self._state = round(cons / 1000.0, 2) + self._attr_native_value = round(cons / 1000.0, 2)