"""Sensor platform for Intelligent Heating Pilot.""" from __future__ import annotations import logging from datetime import datetime from typing import Any from homeassistant.components.sensor import ( SensorDeviceClass, SensorEntity, SensorStateClass, ) from homeassistant.config_entries import ConfigEntry from homeassistant.core import HomeAssistant, callback from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.util import dt as dt_util from .const import ( ATTR_ANTICIPATED_START_TIME, ATTR_LEARNED_HEATING_SLOPE, ATTR_NEXT_SCHEDULE_TIME, ATTR_NEXT_TARGET_TEMP, CONF_NAME, DOMAIN, EVENT_DEAD_TIME_UPDATED, ) _LOGGER = logging.getLogger(__name__) async def async_setup_entry( hass: HomeAssistant, config_entry: ConfigEntry, async_add_entities: AddEntitiesCallback, ) -> None: """Set up the Intelligent Heating Pilot sensors.""" coordinator = hass.data[DOMAIN][config_entry.entry_id] name = config_entry.data.get(CONF_NAME, "Intelligent Heating Pilot") sensors = [ IntelligentHeatingPilotAnticipationTimeSensor(coordinator, config_entry, name), IntelligentHeatingPilotNextScheduleSensor(coordinator, config_entry, name), # HMS-only display companions IntelligentHeatingPilotAnticipationTimeHmsSensor(coordinator, config_entry, name), IntelligentHeatingPilotNextScheduleHmsSensor(coordinator, config_entry, name), # Metrics - LHS sensors (Global + Contextual) IntelligentHeatingPilotGlobalLearnedSlopeSensor(coordinator, config_entry, name), IntelligentHeatingPilotContextualLearnedSlopeSensor(coordinator, config_entry, name), IntelligentHeatingPilotPredictionConfidenceSensor( coordinator, config_entry, name ), # Phase 4: New # Dead time learning sensor IntelligentHeatingPilotDeadTimeSensor(coordinator, config_entry, name), ] async_add_entities(sensors, True) class IntelligentHeatingPilotSensorBase(SensorEntity): """Base class for Intelligent Heating Pilot sensors.""" _attr_has_entity_name = True def __init__(self, coordinator: Any, config_entry: ConfigEntry, name: str) -> None: """Initialize the sensor.""" self.coordinator = coordinator self._config_entry = config_entry self._attr_device_info = { "identifiers": {(DOMAIN, config_entry.entry_id)}, "name": name, "manufacturer": "Intelligent Heating Pilot", "model": "Intelligent Preheating with ML", } async def async_added_to_hass(self) -> None: """Register callbacks when entity is added.""" await super().async_added_to_hass() @callback def handle_anticipation_event(event): """Handle anticipation calculated event.""" data = event.data or {} # Filter events to only those coming from our own config entry if data.get("entry_id") != self._config_entry.entry_id: return old_value = self.native_value self._handle_anticipation_result(data) # Only write HA state when the sensor value actually changed to # avoid unnecessary state-machine writes and log noise. if self.native_value != old_value: self.async_write_ha_state() self.async_on_remove( self.hass.bus.async_listen( f"{DOMAIN}_anticipation_calculated", handle_anticipation_event ) ) def _handle_anticipation_result(self, data: dict) -> None: """Handle new anticipation result. Override in subclasses.""" pass class IntelligentHeatingPilotAnticipationTimeSensor(IntelligentHeatingPilotSensorBase): """Sensor for anticipated start time.""" _attr_name = "Anticipated Start Time" _attr_icon = "mdi:clock-start" _attr_device_class = SensorDeviceClass.TIMESTAMP def __init__(self, coordinator: Any, config_entry: ConfigEntry, name: str) -> None: """Initialize the sensor.""" super().__init__(coordinator, config_entry, name) self._attr_unique_id = f"{config_entry.entry_id}_anticipated_start_time" self._anticipated_start: datetime | None = None self._attributes: dict[str, Any] = {} @property def native_value(self) -> datetime | None: """Return the state of the sensor.""" return self._anticipated_start @property def available(self) -> bool: """Return True if entity is available.""" return True @property def extra_state_attributes(self) -> dict: """Return additional attributes.""" return self._attributes def _handle_anticipation_result(self, data: dict) -> None: """Handle new anticipation result.""" # Clear sensor values when scheduler is disabled or no timeslot is available # This sets the sensor to 'unknown' state and clears all attributes anticipated_start = data.get(ATTR_ANTICIPATED_START_TIME) if anticipated_start is None: self._anticipated_start = None self._attributes = {} _LOGGER.info("Anticipated start time cleared (scheduler disabled or no timeslot)") return if anticipated_start: # Event carries ISO string; accept datetime too if isinstance(anticipated_start, str): # Parse with HA helper to preserve timezone correctly parsed = dt_util.parse_datetime(anticipated_start) if parsed is None: try: parsed = datetime.fromisoformat(anticipated_start) except ValueError: parsed = None self._anticipated_start = parsed else: self._anticipated_start = anticipated_start # Store attributes - keep next_schedule_time as ISO string for proper serialization next_sched = data.get(ATTR_NEXT_SCHEDULE_TIME) if isinstance(next_sched, str): next_sched_attr = next_sched # Already ISO string elif isinstance(next_sched, datetime): next_sched_attr = next_sched.isoformat() else: next_sched_attr = None self._attributes = { ATTR_NEXT_SCHEDULE_TIME: next_sched_attr, ATTR_NEXT_TARGET_TEMP: data.get(ATTR_NEXT_TARGET_TEMP), "anticipation_minutes": data.get("anticipation_minutes"), "current_temp": data.get("current_temp"), "scheduler_entity": data.get("scheduler_entity"), ATTR_LEARNED_HEATING_SLOPE: data.get(ATTR_LEARNED_HEATING_SLOPE), "confidence_level": data.get("confidence_level"), # Phase 4: New from domain } _LOGGER.info( "Anticipated start time updated: %s (confidence: %.2f)", self._anticipated_start, data.get("confidence_level", 0.0), ) class IntelligentHeatingPilotAnticipationTimeHmsSensor(IntelligentHeatingPilotSensorBase): """Companion sensor showing only HH:MM:SS for anticipated start time.""" _attr_name = "Anticipated Start Time (HMS)" _attr_icon = "mdi:clock-outline" def __init__(self, coordinator: Any, config_entry: ConfigEntry, name: str) -> None: super().__init__(coordinator, config_entry, name) self._attr_unique_id = f"{config_entry.entry_id}_anticipated_start_time_hms" self._time_str: str | None = None self._attributes: dict[str, Any] = {} @property def native_value(self) -> str | None: return self._time_str @property def available(self) -> bool: return True @property def extra_state_attributes(self) -> dict: return self._attributes def _handle_anticipation_result(self, data: dict) -> None: # Check if anticipated start time is available value = data.get(ATTR_ANTICIPATED_START_TIME) if value is None: self._time_str = None self._attributes = {} return dt_val: datetime | None = None if isinstance(value, str): dt_val = dt_util.parse_datetime(value) or None if dt_val is None: try: dt_val = datetime.fromisoformat(value) except ValueError: dt_val = None elif isinstance(value, datetime): dt_val = value if dt_val is not None: # Ensure local timezone and format HH:MM:SS local_dt = dt_util.as_local(dt_val) self._time_str = local_dt.strftime("%H:%M:%S") # Provide raw timestamp as attribute for completeness self._attributes = { "timestamp": local_dt.isoformat(), } class IntelligentHeatingPilotGlobalLearnedSlopeSensor(IntelligentHeatingPilotSensorBase): """Sensor for global learned heating slope (LHS).""" _attr_name = "Global Learned Heating Slope" _attr_native_unit_of_measurement = "°C/h" _attr_state_class = SensorStateClass.MEASUREMENT _attr_icon = "mdi:chart-line" def __init__(self, coordinator: Any, config_entry: ConfigEntry, name: str) -> None: """Initialize the sensor.""" super().__init__(coordinator, config_entry, name) self._attr_unique_id = f"{config_entry.entry_id}_global_learned_heating_slope" # Tracks the last value written to HA state for threshold comparison. self._last_lhs_displayed: float | None = None @property def native_value(self) -> float | None: """Return the global LHS from coordinator cache.""" value = self.coordinator.get_learned_heating_slope() # Round to 2 decimal places for cleaner display return round(value, 2) if value is not None else None @property def available(self) -> bool: """Return True if entity is available.""" return True @property def extra_state_attributes(self) -> dict: """Return additional attributes.""" return { "description": "Average heating slope across all extracted cycles", } def _handle_anticipation_result(self, data: dict) -> None: """Refresh state only when global LHS changed by >= 0.1 °C/h. The coordinator is updated before this event fires, so we compare the current native_value against the last value written to HA state. Availability transitions (None ↔ value) always trigger a state write. """ current = self.native_value prev = self._last_lhs_displayed if (prev is None) != (current is None) or ( prev is not None and current is not None and abs(current - prev) >= 0.1 ): self._last_lhs_displayed = current self.async_write_ha_state() class IntelligentHeatingPilotContextualLearnedSlopeSensor(IntelligentHeatingPilotSensorBase): """Sensor for contextual learned heating slope (for current hour).""" _attr_name = "Contextual Learned Heating Slope" _attr_native_unit_of_measurement = "°C/h" _attr_icon = "mdi:chart-line" def __init__(self, coordinator: Any, config_entry: ConfigEntry, name: str) -> None: """Initialize the sensor.""" super().__init__(coordinator, config_entry, name) self._attr_unique_id = f"{config_entry.entry_id}_contextual_learned_heating_slope" self._next_schedule_time: datetime | None = None # Tracks the last value written to HA state for threshold comparison. self._last_contextual_lhs_displayed: float | None = None @property def native_value(self) -> float | None: """Return the contextual LHS for next scheduled event hour.""" # If no next schedule time, cannot provide contextual LHS if not self._next_schedule_time: return None try: schedule_hour = self._next_schedule_time.hour except AttributeError: return None # Try to get contextual LHS from coordinator cache for scheduled hour contextual_lhs = self.coordinator.get_contextual_learned_heating_slope(schedule_hour) if contextual_lhs is None or contextual_lhs == 2.0: # Default value means no data return None # Round to 2 decimal places for cleaner display and return as float return float(round(contextual_lhs, 2)) @property def available(self) -> bool: """Return True if entity is available.""" return True @property def extra_state_attributes(self) -> dict: """Return additional attributes.""" if not self._next_schedule_time: return { "description": "Average heating slope for scheduled event hour (no event scheduled)", "scheduled_hour": "–", } try: scheduled_hour = self._next_schedule_time.hour except AttributeError: return { "description": "Average heating slope for scheduled event hour (parse error)", "scheduled_hour": "–", } return { "description": "Average heating slope for scheduled event hour", "scheduled_hour": f"{scheduled_hour:02d}:00", } def _handle_anticipation_result(self, data: dict) -> None: """Refresh state only when contextual LHS changed by >= 0.1 °C/h. Captures the current schedule hour BEFORE updating _next_schedule_time so we can detect a context change (different scheduled hour) and always write state when the LHS context changes, regardless of the value delta. Availability transitions (None ↔ value) always trigger a state write. """ old_schedule_hour = self._next_schedule_time.hour if self._next_schedule_time else None # Update next_schedule_time from event next_schedule = data.get(ATTR_NEXT_SCHEDULE_TIME) if next_schedule is None: self._next_schedule_time = None else: # Event carries ISO string; accept datetime too if isinstance(next_schedule, str): parsed = dt_util.parse_datetime(next_schedule) if parsed is None: try: parsed = datetime.fromisoformat(next_schedule) except ValueError: parsed = None self._next_schedule_time = parsed else: self._next_schedule_time = next_schedule new_schedule_hour = self._next_schedule_time.hour if self._next_schedule_time else None schedule_context_changed = old_schedule_hour != new_schedule_hour new_value = self.native_value prev = self._last_contextual_lhs_displayed if ( schedule_context_changed or (prev is None) != (new_value is None) or (prev is not None and new_value is not None and abs(new_value - prev) >= 0.1) ): self._last_contextual_lhs_displayed = new_value self.async_write_ha_state() class IntelligentHeatingPilotNextScheduleSensor(IntelligentHeatingPilotSensorBase): """Sensor for next schedule time.""" _attr_name = "Next Schedule Time" _attr_icon = "mdi:calendar-clock" _attr_device_class = SensorDeviceClass.TIMESTAMP def __init__(self, coordinator: Any, config_entry: ConfigEntry, name: str) -> None: """Initialize the sensor.""" super().__init__(coordinator, config_entry, name) self._attr_unique_id = f"{config_entry.entry_id}_next_schedule_time" self._next_schedule: datetime | None = None self._attributes: dict[str, Any] = {} @property def native_value(self) -> datetime | None: """Return the state of the sensor.""" return self._next_schedule @property def available(self) -> bool: """Return True if entity is available.""" return True @property def extra_state_attributes(self) -> dict: """Return additional attributes.""" return self._attributes def _handle_anticipation_result(self, data: dict) -> None: """Handle new anticipation result.""" # Check if scheduler is disabled or no timeslot is available next_schedule = data.get(ATTR_NEXT_SCHEDULE_TIME) if next_schedule is None: self._next_schedule = None self._attributes = {} _LOGGER.info("Next schedule time cleared (scheduler disabled or no timeslot)") return if next_schedule: # Event carries ISO string; accept datetime too if isinstance(next_schedule, str): # Parse with HA helper to preserve timezone correctly parsed = dt_util.parse_datetime(next_schedule) if parsed is None: try: parsed = datetime.fromisoformat(next_schedule) except ValueError: parsed = None self._next_schedule = parsed else: self._next_schedule = next_schedule self._attributes = { ATTR_NEXT_TARGET_TEMP: data.get(ATTR_NEXT_TARGET_TEMP), "scheduler_entity": data.get("scheduler_entity"), } _LOGGER.debug("Next schedule time updated: %s", self._next_schedule) class IntelligentHeatingPilotNextScheduleHmsSensor(IntelligentHeatingPilotSensorBase): """Companion sensor showing only HH:MM:SS for next schedule time.""" _attr_name = "Next Schedule Time (HMS)" _attr_icon = "mdi:clock-time-three-outline" def __init__(self, coordinator: Any, config_entry: ConfigEntry, name: str) -> None: super().__init__(coordinator, config_entry, name) self._attr_unique_id = f"{config_entry.entry_id}_next_schedule_time_hms" self._time_str: str | None = None self._attributes: dict[str, Any] = {} @property def native_value(self) -> str | None: return self._time_str @property def available(self) -> bool: return True @property def extra_state_attributes(self) -> dict: return self._attributes def _handle_anticipation_result(self, data: dict) -> None: # Clear when no next schedule time is available value = data.get(ATTR_NEXT_SCHEDULE_TIME) if value is None: self._time_str = None self._attributes = {} return dt_val: datetime | None = None if isinstance(value, str): dt_val = dt_util.parse_datetime(value) or None if dt_val is None: try: dt_val = datetime.fromisoformat(value) except ValueError: dt_val = None elif isinstance(value, datetime): dt_val = value if dt_val is not None: local_dt = dt_util.as_local(dt_val) self._time_str = local_dt.strftime("%H:%M:%S") self._attributes = { "timestamp": local_dt.isoformat(), } class IntelligentHeatingPilotPredictionConfidenceSensor(IntelligentHeatingPilotSensorBase): """Sensor for prediction confidence level. Phase 4: New sensor to display domain prediction confidence (0.0-1.0). This reflects the quality of the prediction based on learned data and available environmental sensors. """ _attr_name = "Prediction Confidence" _attr_icon = "mdi:percent" _attr_native_unit_of_measurement = "%" _attr_state_class = SensorStateClass.MEASUREMENT def __init__(self, coordinator: Any, config_entry: ConfigEntry, name: str) -> None: """Initialize the sensor.""" super().__init__(coordinator, config_entry, name) self._attr_unique_id = f"{config_entry.entry_id}_prediction_confidence" self._confidence: float | None = None self._attributes: dict[str, Any] = {} @property def native_value(self) -> float | None: """Return the state of the sensor as percentage (0-100).""" if self._confidence is not None: return round(self._confidence * 100, 1) return None @property def available(self) -> bool: """Return True if entity is available.""" return True @property def extra_state_attributes(self) -> dict: """Return additional attributes.""" return self._attributes def _handle_anticipation_result(self, data: dict) -> None: """Handle new anticipation result.""" # Clear when no confidence level is available confidence = data.get("confidence_level") if confidence is None: self._confidence = None self._attributes = {} _LOGGER.info("Prediction confidence cleared (scheduler disabled or no timeslot)") return confidence_value = float(confidence) if confidence_value > 1.0: confidence_value = confidence_value / 100.0 self._confidence = confidence_value self._attributes = { ATTR_LEARNED_HEATING_SLOPE: data.get(ATTR_LEARNED_HEATING_SLOPE), "anticipation_minutes": data.get("anticipation_minutes"), "environmental_data_available": { "humidity": data.get("humidity") is not None, "cloud_coverage": data.get("cloud_coverage") is not None, }, } _LOGGER.info("Prediction confidence updated: %.1f%%", self._confidence * 100) class IntelligentHeatingPilotDeadTimeSensor(SensorEntity): """Sensor for displaying learned dead time value. Shows the dead time calculated from heating cycles. Dead time is the delay between heating start and first measurable temperature rise (in minutes). """ _attr_name = "Dead Time" _attr_icon = "mdi:timer-sand" _attr_device_class = SensorDeviceClass.DURATION _attr_state_class = SensorStateClass.MEASUREMENT _attr_native_unit_of_measurement = "min" _attr_has_entity_name = True def __init__( self, coordinator: Any, config_entry: ConfigEntry, name: str, ) -> None: """Initialize the sensor.""" self._coordinator = coordinator self._config_entry = config_entry self._attr_unique_id = f"{config_entry.entry_id}_dead_time" # Device info self._attr_device_info = { "identifiers": {("intelligent_heating_pilot", config_entry.entry_id)}, "name": f"Intelligent Heating Pilot {name}", "manufacturer": "Intelligent Heating Pilot", "model": "IHP", } self._dead_time: float | None = None @property def native_value(self) -> float | None: """Return the current dead time value in minutes.""" return self._dead_time @property def available(self) -> bool: """Return True if sensor is available.""" return True @property def extra_state_attributes(self) -> dict[str, Any]: """Return additional attributes.""" return { "auto_learning": self._coordinator.is_auto_learning_enabled(), } async def async_added_to_hass(self) -> None: """When entity is added to Home Assistant.""" await super().async_added_to_hass() await self._async_refresh_dead_time() @callback def handle_dead_time_updated(event) -> None: """Handle dead time update events.""" data = event.data or {} if data.get("entry_id") != self._config_entry.entry_id: return if self.hass is None: return self.hass.async_create_task(self._async_refresh_dead_time()) if self.hass is not None: self.async_on_remove( self.hass.bus.async_listen(EVENT_DEAD_TIME_UPDATED, handle_dead_time_updated) ) async def _async_refresh_dead_time(self) -> None: """Refresh effective dead time from coordinator.""" dead_time = await self._coordinator.get_effective_dead_time() self._dead_time = dead_time if dead_time is not None else None if self.hass is not None: self.async_write_ha_state()