"""Sensor for Midea Lan.""" import math import time from datetime import timedelta from typing import Any, cast from homeassistant.components.sensor import ( SensorDeviceClass, SensorEntity, SensorStateClass, ) from homeassistant.config_entries import ConfigEntry from homeassistant.const import CONF_DEVICE_ID, CONF_SENSORS, Platform from homeassistant.core import HomeAssistant, callback from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.event import async_track_time_interval from homeassistant.helpers.restore_state import RestoreEntity from homeassistant.helpers.typing import StateType from midealan.device import MideaDevice from .const import DEVICES, DOMAIN, supports_device from .midea_devices import MIDEA_DEVICES from .midea_entity import MideaEntity async def async_setup_entry( hass: HomeAssistant, config_entry: ConfigEntry, async_add_entities: AddEntitiesCallback, ) -> None: """Set up sensors for device.""" device_id = config_entry.data.get(CONF_DEVICE_ID) device = hass.data[DOMAIN][DEVICES].get(device_id) extra_sensors = config_entry.options.get(CONF_SENSORS, []) sensors = [] for entity_key, config in cast( "dict", MIDEA_DEVICES[device.device_type]["entities"], ).items(): if ( config["type"] != Platform.SENSOR or not supports_device(device.model, device.subtype, config) or (not config.get("default") and entity_key not in extra_sensors) ): continue required_attribute = config.get("required_attribute") if ( required_attribute is not None and required_attribute not in device.attributes ): continue if config.get("estimate"): sensor = MideaEstimatedUsageSensor(device, entity_key) elif config.get("duration_from_minutes"): sensor = MideaMinuteDurationSensor(device, entity_key) else: sensor = MideaSensor(device, entity_key) sensors.append(sensor) async_add_entities(sensors) class MideaSensor(MideaEntity, SensorEntity): """Represent a Midea sensor.""" def __init__(self, device: MideaDevice, entity_key: str) -> None: """Initialize Midea sensor.""" super().__init__(device, entity_key) # Timer configuration: "down" for countdown, "up" for countup self._timer = self._config.get("timer") self._timer_base_value: int | None = None self._timer_last_update: float | None = None self._timer_listener = None @property def native_value(self) -> StateType: """Native value of the sensor.""" value = self._device.get_attribute(self._entity_key) # If an options mapping exists, translate the raw value to its enum key. options = self._config.get("options") if options is not None and isinstance(value, int): # For an enum sensor an unmapped value is not a valid state, so # report it as unknown (None) instead of leaking the raw int. return cast("StateType", options.get(value)) # Timer mode: calculate elapsed time if self._timer and isinstance(value, int): now = time.time() if self._timer_last_update is not None: elapsed = int(now - self._timer_last_update) if self._timer == "down": # Countdown: value decreases by elapsed seconds return cast("StateType", max(0, value - elapsed)) if self._timer == "up": # Countup: value increases by elapsed seconds return cast("StateType", value + elapsed) return cast("StateType", value) return cast("StateType", value) @property def device_class(self) -> SensorDeviceClass: """Device class of the sensor.""" return cast("SensorDeviceClass", self._config.get("device_class")) @property def state_class(self) -> SensorStateClass | None: """State class of the sensor.""" return cast("SensorStateClass | None", self._config.get("state_class")) @property def options(self) -> list[str] | None: """List of possible states for an enum sensor.""" if self.device_class != SensorDeviceClass.ENUM: return None options = self._config.get("options") return list(options.values()) if options else None @property def native_unit_of_measurement(self) -> str | None: """Unit of measurement for the sensor.""" return cast("str | None", self._config.get("unit")) @property def suggested_display_precision(self) -> int | None: """Suggested number of decimal digits for display.""" return cast("int | None", self._config.get("suggested_display_precision")) @property def suggested_unit_of_measurement(self) -> str | None: """Preferred display unit for device-class conversion.""" return cast("str | None", self._config.get("suggested_unit")) @property def capability_attributes(self) -> dict[str, Any] | None: """Capability attributes of the sensor.""" if self.options is not None: return {"options": self.options} return {"state_class": self.state_class} if self.state_class else {} async def async_added_to_hass(self) -> None: """Subscribe to device updates and start timer tracking.""" await super().async_added_to_hass() # Start timer tracking if configured if self._timer: value = self._device.get_attribute(self._entity_key) if isinstance(value, int): self._timer_base_value = value self._timer_last_update = time.time() # Register 1-second interval update self._timer_listener = async_track_time_interval( self.hass, self._async_timer_update, timedelta(seconds=1), ) async def async_will_remove_from_hass(self) -> None: """Unsubscribe from device updates and stop timer tracking.""" await super().async_will_remove_from_hass() if self._timer_listener: self._timer_listener() self._timer_listener = None @callback def update_state(self, status: Any) -> None: # ruff:ignore[any-type] """Update entity state.""" if self._timer and self._entity_key in status: value = self._device.get_attribute(self._entity_key) if isinstance(value, int): self._timer_base_value = value self._timer_last_update = time.time() super().update_state(status) @callback def _async_timer_update(self, _now: Any) -> None: # ruff:ignore[any-type] """Update timer state every second.""" self.schedule_update_ha_state() class MideaMinuteDurationSensor(MideaSensor): """Present a whole-minute appliance duration as hours and minutes.""" _HOUR_PRECISION = 6 @property def native_value(self) -> StateType: """Convert source minutes to hours for HA's duration formatter.""" raw_value = self._device.get_attribute(self._entity_key) if raw_value is None: return None minutes = max(0, int(cast("int", raw_value))) if minutes == 0: return 0 # HA's duration formatter decomposes decimal hours with floor(). Round # upward below the appliance's one-minute resolution so values such as # 61 minutes cannot render as 1 hour 0 minutes through float truncation. scale = 10**self._HOUR_PRECISION return math.ceil((minutes / 60) * scale) / scale @property def extra_state_attributes(self) -> dict[str, int]: """Keep the exact appliance value available for automations.""" raw_value = self._device.get_attribute(self._entity_key) return { "remaining_minutes": max(0, int(cast("int", raw_value or 0))), } class MideaEstimatedUsageSensor(MideaSensor, RestoreEntity): """Represent estimated dishwasher usage accumulated per run. The dishwasher does not report actual energy/water usage, so this sensor accumulates a fixed per-mode estimate (from the product manual) once each time a wash run *completes*. Completion is detected by the device ``progress`` attribute reaching ``"Complete"``; a cancelled or errored run never reaches that state and is therefore not counted. """ _COMPLETE_PROGRESS = "Complete" _RUNNING_STATUS = "Running" def __init__(self, device: MideaDevice, entity_key: str) -> None: """Initialize estimated usage sensor.""" super().__init__(device, entity_key) self._native_value: float = 0.0 self._last_progress: str | None = None self._running_mode: str | None = None async def async_added_to_hass(self) -> None: """Register for device updates and restore the accumulated value.""" await super().async_added_to_hass() if last_state := await self.async_get_last_state(): try: self._native_value = float(last_state.state) except (TypeError, ValueError): self._native_value = 0.0 self._last_progress = cast( "str | None", self._device.get_attribute("progress"), ) if self._device.get_attribute("status") == self._RUNNING_STATUS: self._running_mode = cast("str | None", self._device.get_attribute("mode")) @property def native_value(self) -> StateType: """Accumulated estimated usage.""" return round(self._native_value, 3) @property def extra_state_attributes(self) -> dict[str, Any]: """Estimate metadata for the sensor.""" estimate = cast("dict[str, Any]", self._config["estimate"]) return { "estimate_source": "fixed_per_wash_mode", "known_modes": list(cast("dict[str, float]", estimate["values"]).keys()), "last_progress": self._last_progress, "running_mode": self._running_mode, } def update_state(self, status: Any) -> None: # ruff:ignore[any-type] """Accumulate the estimate once when a dishwasher run completes.""" current_status = self._device.get_attribute("status") current_progress = cast( "str | None", self._device.get_attribute("progress"), ) current_mode = cast("str | None", self._device.get_attribute("mode")) # Remember the mode selected while the machine is actually running so we # can still attribute usage after it stops reporting the mode on finish. if current_status == self._RUNNING_STATUS: self._running_mode = current_mode # Count once, on the edge into the "Complete" progress state. A cancel or # error transition never reaches "Complete", so it is not counted. if ( current_progress == self._COMPLETE_PROGRESS and self._last_progress != self._COMPLETE_PROGRESS ): mode = self._running_mode or current_mode values = cast("dict[str, float]", self._config["estimate"]["values"]) if mode in values: self._native_value += values[mode] self._running_mode = None self._last_progress = current_progress super().update_state(status) if self.hass and ( "progress" in status or "status" in status or "mode" in status ): self.schedule_update_if_running()