Cleanup

Author: Alexwijn

custom_components/sat/area.py

@@ -19,15 +19,14 @@
 from .heating_curve import HeatingCurve
 from .helpers import float_value, is_state_stale, state_age_seconds
 from .pid import PID, PID_UPDATE_INTERVAL
-from .temperature.state import TemperatureStates, TemperatureState
+from .temperature.state import TemperatureState
 
 _LOGGER = logging.getLogger(__name__)
 
 ATTR_TEMPERATURE = "temperature"
 ATTR_CURRENT_TEMPERATURE = "current_temperature"
-ATTR_CURRENT_VALVE_POSITION = "current_valve_position"
-
 ATTR_SENSOR_TEMPERATURE_ID = "sensor_temperature_id"
+ATTR_CURRENT_VALVE_POSITION = "current_valve_position"
 
 COMFORT_BAND = 0.1
 COOLING_SLOPE = 4.0
@@ -81,7 +80,7 @@ def climate_state(self) -> Optional[State]:
 
         return state if state.state not in [STATE_UNKNOWN, STATE_UNAVAILABLE] else None
 
-    def temperature_state(self) -> Optional[State]:
+    def sensor_state(self) -> Optional[State]:
         """Return the source state used to calculate the current temperature."""
         if (self._hass is None) or (climate_state := self.climate_state) is None:
             return None
@@ -105,7 +104,7 @@ def temperature_state(self) -> Optional[State]:
     @property
     def current_temperature(self) -> Optional[float]:
         """Retrieve the current temperature, overridden by a sensor if set."""
-        if (state := self.temperature_state()) is None:
+        if (state := self.sensor_state()) is None:
             return None
 
         if sensor.DOMAIN in state.entity_id:
@@ -125,10 +124,9 @@ def target_temperature(self) -> Optional[float]:
         return float_value(state.attributes.get("temperature"))
 
     @property
-    def error(self) -> Optional[TemperatureState]:
-        """Calculate the temperature error (target - current)."""
-        temperature_state = self.temperature_state()
-        if temperature_state is None:
+    def temperature_state(self) -> Optional[TemperatureState]:
+        """Calculate the temperature state."""
+        if (sensor_state := self.sensor_state()) is None:
             return None
 
         target_temperature = self.target_temperature
@@ -141,9 +139,9 @@ def error(self) -> Optional[TemperatureState]:
             entity_id=self._entity_id,
             setpoint=target_temperature,
             current=current_temperature,
-            last_reported=temperature_state.last_reported,
-            last_updated=temperature_state.last_updated,
-            last_changed=temperature_state.last_changed,
+            last_reported=sensor_state.last_reported,
+            last_updated=sensor_state.last_updated,
+            last_changed=sensor_state.last_changed,
         )
 
     @property
@@ -228,11 +226,11 @@ async def async_outside_entity_changed(self, _event: Event[EventStateChangedData
 
     def control_pid(self, _time: Optional[datetime] = None) -> None:
         """Update the PID controller with the current error and heating curve."""
-        if (error := self.error) is None:
-            _LOGGER.debug("Skipping control loop for %s because error could not be computed", self._entity_id)
+        if (temperature_state := self.temperature_state) is None:
+            _LOGGER.debug("Skipping control loop for %s because the temperature is not available.", self._entity_id)
             return
 
-        self.pid.update(error)
+        self.pid.update(temperature_state)
 
 
 class Areas:
@@ -246,12 +244,6 @@ def __init__(self, areas: list[Area]) -> None:
     def from_config(config: SatConfig) -> "Areas":
         return Areas([Area(config, entity_id) for entity_id in config.rooms])
 
-    @property
-    def errors(self) -> TemperatureStates:
-        """Return a collection of all the error values for all areas."""
-        error_list = [area.error for area in self._areas if area.error is not None]
-        return TemperatureStates(error_list)
-
     @property
     def pids(self) -> "Areas._PIDs":
         """Return an interface to reset PID controllers for all areas."""
@@ -342,15 +334,14 @@ def overshoot_cap(self) -> Optional[float]:
                 if not area.pid.available or not area.requires_heat:
                     continue
 
-                error = area.error
-                if error is None:
+                if (temperature_state := area.temperature_state) is None:
                     continue
 
-                if error.error >= -OVERSHOOT_MARGIN:
+                if temperature_state.error >= -OVERSHOOT_MARGIN:
                     continue
 
                 # Degrees above target (positive number)
-                degrees_over = -error.error
+                degrees_over = -temperature_state.error
 
                 # Start from a “max allowed in cooling” and pull it down with overshoot severity.
                 caps.append(COLD_SETPOINT + COOLING_HEADROOM - COOLING_SLOPE * degrees_over)

custom_components/sat/climate.py

@@ -227,7 +227,7 @@ def temperature_statistics(self) -> TemperatureStatistics:
         return self._temperature_history.statistics
 
     @property
-    def error(self) -> Optional[TemperatureState]:
+    def temperature_state(self) -> Optional[TemperatureState]:
         """Return the error value."""
         if self._config.sensors.inside_sensor_entity_id is None:
             return None
@@ -329,24 +329,25 @@ def requested_setpoint(self) -> float:
     def weighted_error(self) -> Optional[float]:
         errors: list[tuple[float, float]] = []
 
-        if (primary_error := self.error) is not None:
-            errors.append((primary_error.error, 1.0))
+        if (primary_temperature := self.temperature_state) is not None:
+            errors.append((primary_temperature.error, 1.0))
 
         for area in self.areas.items():
-            if (area_error := area.error) is None:
+            if (temperature := area.temperature_state) is None:
                 continue
 
-            errors.append((area_error.error, area.room_weight))
+            errors.append((temperature.error, area.room_weight))
 
         if not errors:
             return None
 
-        weight_sum = sum(weight for _, weight in errors)
-        if weight_sum <= 0.0:
+        total = sum(weight for _, weight in errors)
+
+        if total <= 0.0:
             return None
 
         weighted_error = sum(error * weight for error, weight in errors)
-        return weighted_error / weight_sum
+        return weighted_error / total
 
     @property
     def valves_open(self) -> bool:
@@ -415,15 +416,14 @@ def control_pid(self, _time: Optional[datetime] = None) -> None:
                 self.pid.reset()
                 return
 
-        if (error := self.error) is None:
-            _LOGGER.debug("Skipping control loop for %s because error value is not available.", self.entity_id)
+        if (temperature_state := self.temperature_state) is None:
+            _LOGGER.debug("Skipping control loop for %s because temperature is not available.", self.entity_id)
             return
 
-        if self.hvac_mode == HVACMode.HEAT and self.valves_open:
-            if (weighted_error := self.weighted_error) is not None:
-                self._temperature_history.record(weighted_error, event_timestamp(_time))
+        if self.hvac_mode == HVACMode.HEAT and self.valves_open and (weighted_error := self.weighted_error) is not None:
+            self._temperature_history.record(weighted_error, event_timestamp(_time))
 
-        self.pid.update(error)
+        self.pid.update(temperature_state)
 
     def schedule_heating_control_loop(self, _time: Optional[datetime] = None, force: bool = False) -> None:
         """Schedule a debounced execution of the heating control loop."""

custom_components/sat/pid.py

@@ -19,7 +19,6 @@
 DERIVATIVE_ALPHA1 = 0.2
 DERIVATIVE_ALPHA2 = 0.1
 DERIVATIVE_RAW_CAP = 5.0
-DEADBAND_EPSILON = 1e-6
 
 PID_UPDATE_INTERVAL = 60
 
@@ -184,7 +183,7 @@ def update(self, state: TemperatureState) -> None:
 
     def _update_integral(self, state: TemperatureState) -> None:
         """Update the integral value in the PID controller."""
-        if abs(state.error) > (DEADBAND + DEADBAND_EPSILON):
+        if abs(state.error) > DEADBAND:
             self._integral = 0.0
             return
 

custom_components/sat/sensor.py

@@ -70,7 +70,7 @@ async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry, asyn
 
 class SatRequestedSetpoint(SatClimateEntity, SensorEntity):
     async def async_added_to_hass(self) -> None:
-        def on_pid_updated(entity_id: str) -> None:
+        def on_pid_updated(_) -> None:
             self.schedule_update_ha_state()
 
         await super().async_added_to_hass()
@@ -383,15 +383,15 @@ def native_unit_of_measurement(self):
     @property
     def native_value(self) -> Optional[float]:
         """Return the state of the device in native units."""
-        if (error := self._climate.error) is None:
+        if (temperature_state := self._climate.temperature_state) is None:
             return None
 
-        return error.error
+        return temperature_state.error
 
     @property
     def available(self):
         """Return availability of the sensor."""
-        return self._climate.error is not None
+        return self._climate.temperature_state is not None
 
     @property
     def unique_id(self) -> str:

custom_components/sat/temperature/state.py

@@ -1,37 +1,5 @@
 from dataclasses import dataclass
 from datetime import datetime
-from typing import Optional, Iterable, Iterator
-
-
-class TemperatureStates:
-    """A collection of TemperatureState objects with utility methods."""
-
-    def __init__(self, errors: Optional[Iterable["TemperatureState"]] = None) -> None:
-        self._errors: list["TemperatureState"] = list(errors) if errors else []
-
-    def __iter__(self) -> Iterator["TemperatureState"]:
-        return iter(self._errors)
-
-    def __len__(self) -> int:
-        return len(self._errors)
-
-    def __getitem__(self, index: int) -> "TemperatureState":
-        return self._errors[index]
-
-    def __add__(self, other: "TemperatureStates") -> "TemperatureStates":
-        return self.merge(other)
-
-    def add(self, error: "TemperatureState") -> None:
-        self._errors.append(error)
-
-    def merge(self, other: "TemperatureStates") -> "TemperatureStates":
-        return TemperatureStates(self._errors + other._errors)
-
-    def max(self) -> Optional["TemperatureState"]:
-        if not self._errors:
-            return None
-
-        return max(self._errors, key=lambda e: e.error)
 
 
 @dataclass(frozen=True, slots=True)
@@ -47,4 +15,4 @@ class TemperatureState:
     @property
     def error(self) -> float:
         """Return the temperature error."""
-        return self.setpoint - self.current
+        return round(self.setpoint - self.current, 3)