diff --git a/README.md b/README.md index 92a1a4f..ed22422 100644 --- a/README.md +++ b/README.md @@ -40,6 +40,7 @@ Predictive fan speed control for HVAC systems, designed to work with Versatile T - [`smart_fan_controller.apply_learned_settings`](#smart_fan_controllerapply_learned_settings) - [`smart_fan_controller.reset_learning`](#smart_fan_controllerreset_learning) - [`smart_fan_controller.set_effective_slope`](#smart_fan_controllerset_effective_slope) + - [`smart_fan_controller.force_fan`](#smart_fan_controllerforce_fan) - [Troubleshooting](#troubleshooting) - [License](#license) @@ -339,6 +340,31 @@ data: effective_slope: 0.15 ``` +### `smart_fan_controller.force_fan` + +Force a specific fan mode for a fixed duration, overriding the MPC. The override is applied +immediately and automatically expires after the duration, handing control back to the MPC. Set +`duration_minutes` to `0` to cancel an active override early. + +**Parameters**: + +| Parameter | Required | Example | Description | +| ------------------ | -------- | --------------------- | ----------------------------------------------------------------- | +| `climate_entity` | No* | `climate.living_room` | Required when several Smart Fan Controller entries are configured | +| `fan_mode` | Yes | `high` | The fan mode to force | +| `duration_minutes` | Yes | `30` | How long to hold the forced mode; `0` cancels an active override | + +**Example** (Developer Tools → Services): +```yaml +service: smart_fan_controller.force_fan +data: + climate_entity: climate.living_room + fan_mode: high + duration_minutes: 30 +``` + +While a force is active, `mpc_status` reports `Forced` and `mpc_reason` shows the remaining time. + --- ## Troubleshooting diff --git a/custom_components/smart_fan_controller/__init__.py b/custom_components/smart_fan_controller/__init__.py index cbdc63c..e1f8180 100644 --- a/custom_components/smart_fan_controller/__init__.py +++ b/custom_components/smart_fan_controller/__init__.py @@ -52,6 +52,7 @@ SERVICE_APPLY_LEARNED_SETTINGS = "apply_learned_settings" SERVICE_RESET_LEARNING = "reset_learning" SERVICE_SET_EFFECTIVE_SLOPE = "set_effective_slope" +SERVICE_FORCE_FAN = "force_fan" def _filter_supported_fan_modes(raw_modes: list[str] | None) -> list[str]: @@ -187,6 +188,7 @@ async def _async_migrate_entity_ids(hass: HomeAssistant, entry: ConfigEntry, cli def _should_collect_slope_sample( *, current_fan: str | None, + hvac_mode: str, is_defrost_active: bool, is_hvac_idle: bool, phase: str, @@ -196,6 +198,10 @@ def _should_collect_slope_sample( now: float, ) -> bool: """Return True when slope learning conditions are met.""" + # Only heating/cooling produce meaningful thermal-response samples; skip + # off/dry/fan_only so those cycles don't pollute the learned profiles. + if hvac_mode not in ("heat", "cool"): + return False if current_fan is None or is_defrost_active or is_hvac_idle: return False if phase != PHASE_ESTABLISHED: @@ -312,9 +318,41 @@ async def set_effective_slope(call): "Set effective slope for %s/%s to %.3f", hvac_mode, fan_mode, effective_slope ) + async def force_fan(call): + """Service to force a specific fan mode for a fixed duration (minutes). + + A duration of 0 cancels any active override and hands control back to the MPC. + """ + entry_id, entry_data = _resolve_controller_entry(hass, call.data.get(CONF_CLIMATE_ENTITY)) + mpc_controller = entry_data["mpc_controller"] + climate_id = entry_data["climate_entity"] + fan_mode = call.data["fan_mode"] + duration_minutes = float(call.data["duration_minutes"]) + + if duration_minutes <= 0: + entry_data["force"] = None + _LOGGER.info("Force fan cancelled for %s; resuming MPC control", climate_id) + else: + available = mpc_controller.fan_modes or [] + if available and fan_mode not in available: + raise HomeAssistantError( + f"Fan mode '{fan_mode}' is not available for {climate_id}. " + f"Known fan modes: {', '.join(available) or 'none yet'}" + ) + entry_data["force"] = {"fan_mode": fan_mode, "until": time.time() + duration_minutes * 60.0} + _LOGGER.info( + "Forcing fan '%s' for %s for %.0f min", fan_mode, climate_id, duration_minutes + ) + + # Apply immediately instead of waiting for the next 2-minute cycle. + trigger_cycle = entry_data.get("trigger_cycle") + if callable(trigger_cycle): + trigger_cycle() + hass.services.async_register(DOMAIN, SERVICE_APPLY_LEARNED_SETTINGS, apply_learned_settings) hass.services.async_register(DOMAIN, SERVICE_RESET_LEARNING, reset_learning) hass.services.async_register(DOMAIN, SERVICE_SET_EFFECTIVE_SLOPE, set_effective_slope) + hass.services.async_register(DOMAIN, SERVICE_FORCE_FAN, force_fan) # MPC statuses that mean the controller is paused; the current fan mode is @@ -322,6 +360,38 @@ async def set_effective_slope(call): _MPC_PAUSED_STATUSES = frozenset({"Idle", "Disturbed", "Not ready"}) +def _resolve_active_force( + entry_data: dict, + fan_modes: list[str] | None, + now: float, + climate_id: str, +) -> str | None: + """Return the forced fan mode if a force override is currently active, else None. + + Clears expired or invalid overrides as a side effect so control returns to the MPC. + """ + force = entry_data.get("force") + if not force: + return None + + if now >= force["until"]: + entry_data["force"] = None + _LOGGER.info("Force fan expired for %s; resuming MPC control", climate_id) + return None + + forced_fan = force["fan_mode"] + if fan_modes and forced_fan not in fan_modes: + entry_data["force"] = None + _LOGGER.warning( + "Forced fan '%s' is no longer a valid mode for %s; cancelling override", + forced_fan, + climate_id, + ) + return None + + return forced_fan + + def _read_climate_cycle_data( state, climate_id: str, @@ -350,12 +420,29 @@ def _read_climate_cycle_data( ) return None + # VTherm can briefly expose non-numeric values ("unknown"/"unavailable") + # during restarts; guard the conversion so a transient state skips the + # cycle instead of raising and aborting the control loop. + try: + vtherm_slope_value = float(vtherm_slope) + current_temp_value = float(current_temp) + target_temp_value = float(target_temp) + except (TypeError, ValueError): + _LOGGER.debug( + "Skipping control cycle for %s: non-numeric climate data (slope=%s, current=%s, target=%s)", + climate_id, + vtherm_slope, + current_temp, + target_temp, + ) + return None + window_mgr = attrs.get("window_manager", {}) is_window_open = window_mgr.get("window_state") == "on" or window_mgr.get("window_auto_state") == "on" or attrs.get("specific_states", {}).get("hvac_off_reason") == "Window" return ( - float(vtherm_slope), - float(current_temp), - float(target_temp), + vtherm_slope_value, + current_temp_value, + target_temp_value, str(hvac_mode), current_fan, is_window_open, @@ -488,6 +575,9 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: "sensors": [], "ensure_profile_sensors": None, "store": store, + # Manual override set by the force_fan service: {"fan_mode": str, "until": float} + "force": None, + "trigger_cycle": None, } if not hass.services.has_service(DOMAIN, SERVICE_APPLY_LEARNED_SETTINGS): @@ -510,10 +600,16 @@ def _ensure_profile_sensors() -> None: "previous_slope": None, "last_hvac_mode": None, "defrost": {"active": False, "start_time": 0.0}, + # Fan mode the controller itself just commanded, so the state-change + # listener can tell its own change apart from a genuine manual override. + "controller_commanded_fan": None, + # Re-entrancy guard: True while a cycle is executing so overlapping + # triggers (periodic timer, startup run, force_fan) do not run concurrently. + "cycle_running": False, } - async def run_control_loop(_): - """Main control loop executed every 2 minutes.""" + async def _execute_control_cycle(): + """Run one full control cycle (guarded by run_control_loop).""" current_state = hass.states.get(climate_id) if not current_state: _LOGGER.warning("Climate entity %s not found; skipping control cycle", climate_id) @@ -585,8 +681,23 @@ async def run_control_loop(_): mpc_decision.get("mpc_confidence"), ) - # Determine effective fan: use MPC when status is actionable, else hold current - if mpc_decision.get("mpc_status") not in _MPC_PAUSED_STATUSES and mpc_decision.get("mpc_fan_mode"): + # Manual override: the force_fan service can pin a fan mode for a fixed window. + entry_data = domain_data[entry.entry_id] + forced_fan = _resolve_active_force(entry_data, mpc_controller.fan_modes, now, climate_id) + + # Determine effective fan: forced > MPC (when actionable) > hold current + if forced_fan is not None: + remaining_min = (entry_data["force"]["until"] - now) / 60.0 + effective_fan = forced_fan + effective_reason = f"Forced fan '{forced_fan}' ({remaining_min:.0f} min left)" + mpc_decision = { + **mpc_decision, + "mpc_status": "Forced", + "mpc_fan_mode": forced_fan, + "mpc_reason": effective_reason, + "mpc_would_change_now": "yes" if forced_fan != current_fan else "no", + } + elif mpc_decision.get("mpc_status") not in _MPC_PAUSED_STATUSES and mpc_decision.get("mpc_fan_mode"): effective_fan = mpc_decision["mpc_fan_mode"] effective_reason = f"MPC: {mpc_decision.get('mpc_reason', 'MPC')}" else: @@ -605,6 +716,7 @@ async def run_control_loop(_): # Slope sample collection if _should_collect_slope_sample( current_fan=current_fan, + hvac_mode=hvac_mode, is_defrost_active=is_defrost_active, is_hvac_idle=is_hvac_idle, phase=phase, @@ -646,7 +758,7 @@ async def run_control_loop(_): phase=phase, effective_slope=-float(vtherm_slope) if hvac_mode == "cool" else float(vtherm_slope), effective_timeout=mpc_controller.get_effective_timeout(hvac_mode), - force=False, + force=forced_fan is not None, learning_ready=learning.is_ready(), dead_time=learning.get_dead_time(hvac_mode), mpc_decision=mpc_decision, @@ -672,10 +784,29 @@ def _confirm(): ctrl_state["last_change_time"] = time.time() _LOGGER.debug("Confirmed fan change for %s at %.3f", climate_id, ctrl_state["last_change_time"]) + # Mark this as a controller-initiated change so the resulting + # state-change event is not mislabelled as a manual override. + ctrl_state["controller_commanded_fan"] = effective_fan await _async_apply_fan_change(hass, climate_id, effective_fan, current_fan, effective_reason, _confirm) else: _LOGGER.debug("No fan mode change required for %s (%s)", climate_id, effective_reason) + async def run_control_loop(_=None): + """Guarded entry point: run one cycle unless one is already in progress. + + Skipping (rather than queuing) prevents overlapping triggers from racing + on shared state or issuing duplicate fan commands. A skipped force_fan + trigger is still applied by the next cycle, since the override is stored. + """ + if ctrl_state["cycle_running"]: + _LOGGER.debug("Control cycle already in progress for %s; skipping overlapping trigger", climate_id) + return + ctrl_state["cycle_running"] = True + try: + await _execute_control_cycle() + finally: + ctrl_state["cycle_running"] = False + async def _handle_manual_change(event): """Track manual fan mode changes to reset the controller cooldown.""" new_state = event.data.get("new_state") @@ -689,14 +820,25 @@ async def _handle_manual_change(event): if new_fan is None or new_fan == old_fan: return + # Ignore the state change produced by the controller's own command so it + # is not reported as a manual override. + if new_fan == ctrl_state.get("controller_commanded_fan"): + ctrl_state["controller_commanded_fan"] = None + _LOGGER.debug("Ignoring controller-initiated fan change for %s: %s -> %s", climate_id, old_fan, new_fan) + return + _LOGGER.info("Manual fan mode change detected for %s: %s -> %s", climate_id, old_fan, new_fan) ctrl_state["last_change_time"] = time.time() manual_data = {"fan_mode": new_fan, "minutes_since_last_change": 0.0, "reason": "Manual Override"} _update_sensors(hass, entry.entry_id, manual_data) + # Expose a way for the force_fan service to apply an override immediately. + domain_data[entry.entry_id]["trigger_cycle"] = lambda: hass.async_create_task(run_control_loop(None)) + entry.async_on_unload(async_track_time_interval(hass, run_control_loop, timedelta(minutes=DELTA_TIME_CONTROL_LOOP))) entry.async_on_unload(async_track_state_change_event(hass, [climate_id], _handle_manual_change)) + entry.async_on_unload(entry.add_update_listener(async_reload_entry)) hass.async_create_task(run_control_loop(None)) @@ -743,6 +885,7 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: hass.services.async_remove(DOMAIN, SERVICE_APPLY_LEARNED_SETTINGS) hass.services.async_remove(DOMAIN, SERVICE_RESET_LEARNING) hass.services.async_remove(DOMAIN, SERVICE_SET_EFFECTIVE_SLOPE) + hass.services.async_remove(DOMAIN, SERVICE_FORCE_FAN) if not domain_data: hass.data.pop(DOMAIN, None) _LOGGER.info("Unloaded Smart Fan Controller entry %s", entry.entry_id) @@ -750,7 +893,6 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: async def async_reload_entry(hass: HomeAssistant, entry: ConfigEntry) -> None: - """Reload config entry.""" + """Reload the config entry when its options change.""" _LOGGER.info("Reloading Smart Fan Controller entry %s", entry.entry_id) - await async_unload_entry(hass, entry) - await async_setup_entry(hass, entry) + await hass.config_entries.async_reload(entry.entry_id) diff --git a/custom_components/smart_fan_controller/const.py b/custom_components/smart_fan_controller/const.py index 09413e0..768fe83 100644 --- a/custom_components/smart_fan_controller/const.py +++ b/custom_components/smart_fan_controller/const.py @@ -33,7 +33,6 @@ # Controller thresholds THRESHOLD_SLOPE = 0.1 # °C/h – minimum slope delta to trigger re-evaluation THRESHOLD_TARGET_DROP = -1.0 # °C – setpoint drop that triggers immediate speed cut -MAX_PROJECTION_DELTA = 1.0 # °C – maximum temperature change projected in 10 min DEFAULT_DEAD_TIME = 10.0 # minutes – fallback dead time before learning is ready DEAD_TIME_SAFETY_FACTOR = 1.5 # multiplier applied to learned dead time for effective timeout @@ -46,6 +45,10 @@ MIN_SAMPLES_LEARNING = 240 # Minimum slope samples required for initial readiness MIN_LIMIT_TIMEOUT = 5 # Minimum limit_timeout (minutes) derived from learning MIN_MODE_PROFILE_SAMPLES = 10 # Minimum samples per fan mode to consider profile reliable +REFERENCE_SLOPE_ERROR = 1.0 # °C – reference comfort error at which the representative +# "working" effective slope is reported. The learned slope model is slope(error) = a + b·error; +# evaluating it at this gap yields a value reflecting real cooling/heating power rather than the +# near-equilibrium median, which is structurally diluted by samples taken close to the setpoint. SETPOINT_DROP_LEARNING_COOLDOWN = 30.0 # Minutes to block learning after a setpoint drop MIN_ESTABLISHED_RATIO = 2.0 # Minimum factor × dead_time the fan mode must be active before learning diff --git a/custom_components/smart_fan_controller/manifest.json b/custom_components/smart_fan_controller/manifest.json index e5e67c8..74ad70b 100644 --- a/custom_components/smart_fan_controller/manifest.json +++ b/custom_components/smart_fan_controller/manifest.json @@ -2,10 +2,13 @@ "domain": "smart_fan_controller", "name": "Smart Fan Controller", "documentation": "https://github.com/Gamso/smart_fan_controller", + "issue_tracker": "https://github.com/Gamso/smart_fan_controller/issues", "config_flow": true, "dependencies": [], + "after_dependencies": ["versatile_thermostat"], "integration_type": "service", "iot_class": "calculated", + "loggers": ["custom_components.smart_fan_controller"], "codeowners": [ "@Gamso" ], diff --git a/custom_components/smart_fan_controller/mpc_controller.py b/custom_components/smart_fan_controller/mpc_controller.py index fd582b7..a94552c 100644 --- a/custom_components/smart_fan_controller/mpc_controller.py +++ b/custom_components/smart_fan_controller/mpc_controller.py @@ -11,6 +11,7 @@ PHASE_DEAD_TIME, PHASE_ESTABLISHED, PHASE_TRANSIENT, + REFERENCE_SLOPE_ERROR, THRESHOLD_TARGET_DROP, ) from .thermal_learning import ThermalLearning @@ -24,6 +25,10 @@ FLOOR_VIOLATION_QUADRATIC_WEIGHT = 30.0 MODE_CHANGE_DISTANCE_COST = 0.15 MODE_RANK_COST = 0.05 +# Geometric growth of relative power draw per fan-mode rank. ~6**(1/3) so a +# 4-mode ladder reproduces the legacy [1.0, 1.5, 3.0, 6.0] power scaling while +# extending naturally to any number of modes. +MODE_POWER_RATIO = 1.82 MIN_INTERVAL_CHANGE_PENALTY = 25.0 URGENCY_SENSITIVITY = 2.0 @@ -177,6 +182,7 @@ def evaluate( active_fan = current_fan if current_fan in fan_modes else fan_modes[0] current_effective_slope = -vtherm_slope if hvac_mode == "cool" else vtherm_slope + current_error = self._temperature_error(current_temp, target_temp, hvac_mode) dead_time = self._learning.get_dead_time() change_allowed = minutes_since_change >= self._min_interval phase = self._detect_phase(minutes_since_change, dead_time) @@ -187,9 +193,18 @@ def evaluate( current_effective_slope, fan_modes, ) + # Compare the observed slope against what the gap-dependent model expects + # *at the current error*, not at the reference gap. This keeps the + # disturbance bias clean: it only captures genuine external disturbances + # (solar gain, occupancy) instead of the systematic variation of cooling + # power with the distance to setpoint. + current_mode_gain = ( + self._learning.get_mode_slope_gain(active_fan, hvac_mode) if current_known_profile else 0.0 + ) + expected_slope_now = self._gap_slope(current_mode_slope, current_mode_gain, current_error) self._update_disturbance_bias( observed_effective_slope=current_effective_slope, - expected_effective_slope=current_mode_slope, + expected_effective_slope=expected_slope_now, known_profile=current_known_profile, phase=phase, is_window_open=is_window_open, @@ -230,7 +245,6 @@ def evaluate( # Setpoint drop: when the target moves far away (e.g. night setpoint), # there is no point running the full MPC cost optimisation — the answer # is always the lowest mode. - current_error = self._temperature_error(current_temp, target_temp, hvac_mode) if current_error < THRESHOLD_TARGET_DROP: lowest_fan = fan_modes[0] would_change = "yes" if active_fan != lowest_fan else "no" @@ -268,6 +282,7 @@ def evaluate( fan_modes, monotone_slopes, ) + mode_gain = self._learning.get_mode_slope_gain(fan_mode, hvac_mode) if known_profile else 0.0 sim = self._simulate_mode( current_temp=current_temp, target_temp=target_temp, @@ -276,6 +291,7 @@ def evaluate( candidate_fan=fan_mode, current_effective_slope=current_effective_slope, candidate_mode_slope=mode_slope, + candidate_mode_gain=mode_gain, dead_time=dead_time, candidate_index=fan_modes.index(fan_mode), current_index=current_index, @@ -349,6 +365,10 @@ def evaluate( ) if selection_note: reason += f" | {selection_note}" + # Surface capacity saturation: strongest fan selected yet still well short + # of target means the HVAC system is capacity-bound, not a control issue. + if best.fan_mode == fan_modes[-1] and current_error > self._deadband: + reason += f" | Saturated: max fan, {current_error:.1f}C from target (capacity-bound)" if abs(self._disturbance_bias) >= 0.05: reason += f" | Bias={self._disturbance_bias:+.2f}C/h" if not change_allowed: @@ -420,6 +440,20 @@ def _get_mode_slope( ) return scaled, False + @staticmethod + def _gap_slope(reference_slope: float, gain: float, error: float) -> float: + """Return the modelled effective slope at a given comfort error. + + ``reference_slope`` is the representative slope at REFERENCE_SLOPE_ERROR + (a + b·REF) and ``gain`` is b, so the model at ``error`` is + reference_slope + b·(error − REF). The error is floored at 0 (no driving + force at/below setpoint) and the result is floored at 0 so the model never + projects active cooling/heating away from the setpoint; the additive + disturbance bias is applied separately by the caller. + """ + modelled = reference_slope + gain * (max(error, 0.0) - REFERENCE_SLOPE_ERROR) + return max(0.0, modelled) + @staticmethod def _detect_phase(minutes_since_change: float, dead_time: float) -> str: """Classify the current prediction phase without relying on the live controller.""" @@ -522,6 +556,7 @@ def _simulate_mode( candidate_fan: str, current_effective_slope: float, candidate_mode_slope: float, + candidate_mode_gain: float = 0.0, dead_time: float, candidate_index: int, current_index: int, @@ -529,7 +564,16 @@ def _simulate_mode( known_profile: bool, horizon_minutes: int | None = None, ) -> ModeSimulation: - """Simulate one constant fan mode over the prediction horizon.""" + """Simulate one fan mode over the prediction horizon. + + The candidate's effective slope is gap-dependent: at each step it is + recomputed from the simulated comfort error as + ``candidate_mode_slope + candidate_mode_gain·(error − REFERENCE_SLOPE_ERROR)`` + (floored at 0), plus the disturbance bias. This makes the projection + decelerate realistically as the room approaches the setpoint instead of + cooling/heating at a constant rate, eliminating the phantom overshoot that a + constant-slope model produces past the target. + """ horizon = horizon_minutes if horizon_minutes is not None else self._horizon_minutes steps = max(1, int(horizon / self._cycle_minutes)) step_hours = self._cycle_minutes / 60.0 @@ -540,14 +584,17 @@ def _simulate_mode( cost = 0.0 thermal_power = current_effective_slope change_delay = 0.0 if candidate_fan == current_fan else dead_time - candidate_effective_slope = candidate_mode_slope + self._disturbance_bias for step in range(1, steps + 1): elapsed = step * self._cycle_minutes if elapsed <= change_delay: target_effective_slope = current_effective_slope else: - target_effective_slope = candidate_effective_slope + step_error = self._temperature_error(sim_temp, target_temp, hvac_mode) + target_effective_slope = ( + self._gap_slope(candidate_mode_slope, candidate_mode_gain, step_error) + + self._disturbance_bias + ) thermal_power += blend * (target_effective_slope - thermal_power) raw_slope = -thermal_power if hvac_mode == "cool" else thermal_power @@ -571,13 +618,11 @@ def _simulate_mode( cost += FLOOR_VIOLATION_QUADRATIC_WEIGHT * floor_violation * floor_violation cost += MODE_CHANGE_DISTANCE_COST * abs(candidate_index - current_index) - # Apply non-linear economic mode ranking cost to represent actual physical power scaling - # (Low: 1.0x, Medium: 1.5x, High: 3.0x, Superhigh: 6.0x equivalent) - # This replaces the abstract linear mode rank cost with a real physical relative power draw representation. - power_draw = [1.0, 1.5, 3.0, 6.0] - # Map indices to relative weights - p_index = min(candidate_index, len(power_draw) - 1) - relative_power = power_draw[p_index] + # Apply a non-linear economic mode-ranking cost representing physical power + # scaling. Relative power grows geometrically with the mode rank so every + # mode is differentiated regardless of how many the climate entity exposes + # (a 4-mode system reproduces the previous 1.0 / 1.8 / 3.3 / 6.0 ramp). + relative_power = MODE_POWER_RATIO ** candidate_index cost += MODE_RANK_COST * relative_power if candidate_fan != current_fan and not change_allowed: cost += MIN_INTERVAL_CHANGE_PENALTY @@ -651,16 +696,29 @@ def _step_down_hold_note( def _compute_confidence(self, known_profiles: int, total_profiles: int, phase: str, worst_spread: float = 0.0) -> float: """Return a coarse confidence score for the current recommendation. + Confidence is driven primarily by per-mode profile *coverage* and + *quality* (spread), not by the global readiness flag. Previously the + global ``is_ready()`` threshold (240 samples) halved the score, so a + controller with every per-mode profile fully learned could still be + stuck reporting "Low confidence" until that global count was reached — + which never happened for an HVAC mode used only part of the year. + + - coverage : fraction of fan modes with a learned profile (main driver). + - readiness : small bonus once the global sample threshold is reached. + - phase : transient/dead-time phases attenuate confidence. + - penalties : sustained disturbance bias and high profile spread. + worst_spread is the maximum MAD/median ratio across all known profiles. Profiles with high spread (> 0.15) reduce confidence proportionally, capped at a 0.20 penalty. """ coverage = known_profiles / max(total_profiles, 1) - base = 1.0 if self._learning.is_ready() else 0.45 + coverage_score = 0.3 + 0.7 * coverage + readiness_bonus = 0.1 if self._learning.is_ready() else 0.0 phase_factor = 1.0 if phase == PHASE_ESTABLISHED else 0.85 disturbance_penalty = min(abs(self._disturbance_bias) / MAX_DISTURBANCE_BIAS, 0.35) spread_penalty = min(max(worst_spread - 0.15, 0.0) * 0.4, 0.20) - return max(0.1, min(1.0, (base * (0.5 + 0.5 * coverage) * phase_factor) - disturbance_penalty - spread_penalty)) + return max(0.1, min(1.0, (coverage_score + readiness_bonus) * phase_factor - disturbance_penalty - spread_penalty)) def _payload( self, diff --git a/custom_components/smart_fan_controller/sensor.py b/custom_components/smart_fan_controller/sensor.py index a7aad89..1854b0a 100644 --- a/custom_components/smart_fan_controller/sensor.py +++ b/custom_components/smart_fan_controller/sensor.py @@ -17,6 +17,7 @@ EFFECTIVE_SLOPE_UNIT, MIN_MODE_PROFILE_SAMPLES, PROFILE_HVAC_MODES, + REFERENCE_SLOPE_ERROR, build_scoped_entity_id, build_unique_id, ) @@ -85,7 +86,10 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry, async_add_e climate_entity = data["climate_entity"] sensor_definitions = [ - ("Fan Mode", "fan_mode", "fan_mode", None, SensorDeviceClass.ENUM, "mdi:fan", None), + # No ENUM device_class: it requires a static `options` list, but fan modes + # are discovered at runtime and vary per climate, which would emit HA + # validation warnings. A plain text sensor shows the fan mode just fine. + ("Fan Mode", "fan_mode", "fan_mode", None, None, "mdi:fan", None), ( "Fan Mode Last Change", "fan_mode_last_change", @@ -484,8 +488,9 @@ def native_value(self) -> float | None: @property def extra_state_attributes(self) -> dict: """Expose sampling details for the current profile.""" - samples = self._controller.learning.get_mode_sample_count(self._fan_mode, self._hvac_mode) - spread = self._controller.learning.get_profile_spread(self._fan_mode, self._hvac_mode) + learning = self._controller.learning + samples = learning.get_mode_sample_count(self._fan_mode, self._hvac_mode) + spread = learning.get_profile_spread(self._fan_mode, self._hvac_mode) if spread is None: quality = "unknown" elif spread < 0.15: @@ -494,6 +499,17 @@ def extra_state_attributes(self) -> dict: quality = "fair" else: quality = "poor" + # Gap-dependent slope model: effective_slope(error) = intercept + gain·error. + # The state value is this model evaluated at REFERENCE_SLOPE_ERROR. + model = learning.get_mode_slope_model(self._fan_mode, self._hvac_mode) + if model is None: + slope_intercept = None + slope_gain = None + else: + slope_intercept = round(model[0], 3) + slope_gain = round(model[1], 3) + r_squared = learning.get_mode_slope_r2(self._fan_mode, self._hvac_mode) + time_constant = learning.get_mode_time_constant(self._fan_mode, self._hvac_mode) return { "hvac_mode": self._hvac_mode, "fan_mode": self._fan_mode, @@ -502,6 +518,11 @@ def extra_state_attributes(self) -> dict: "ready": samples >= MIN_MODE_PROFILE_SAMPLES, "spread": spread, "quality": quality, + "slope_intercept": slope_intercept, + "slope_gain": slope_gain, + "reference_error": REFERENCE_SLOPE_ERROR, + "model_r_squared": round(r_squared, 3) if r_squared is not None else None, + "thermal_time_constant_h": round(time_constant, 2) if time_constant is not None else None, } diff --git a/custom_components/smart_fan_controller/services.yaml b/custom_components/smart_fan_controller/services.yaml index 19503af..275a28d 100644 --- a/custom_components/smart_fan_controller/services.yaml +++ b/custom_components/smart_fan_controller/services.yaml @@ -65,3 +65,35 @@ set_effective_slope: step: 0.001 mode: box unit_of_measurement: "°C/h" + +force_fan: + name: Force fan mode + description: Force a specific fan mode for a fixed duration, overriding the MPC. Set the duration to 0 to cancel an active override and hand control back to the MPC. + fields: + climate_entity: + name: Climate entity + description: Optional when only one Smart Fan Controller entry is configured. Required when several climates are managed. + required: false + example: "climate.living_room" + selector: + entity: + domain: climate + fan_mode: + name: Fan mode + description: The fan mode to force (e.g. silent, low, med, high, superhigh). + required: true + example: "high" + selector: + text: + duration_minutes: + name: Duration (minutes) + description: How long to hold the forced fan mode. Use 0 to cancel an active override. + required: true + example: 30 + selector: + number: + min: 0 + max: 1440 + step: 1 + mode: box + unit_of_measurement: "min" diff --git a/custom_components/smart_fan_controller/thermal_learning.py b/custom_components/smart_fan_controller/thermal_learning.py index ecabf55..ea43829 100644 --- a/custom_components/smart_fan_controller/thermal_learning.py +++ b/custom_components/smart_fan_controller/thermal_learning.py @@ -2,7 +2,13 @@ import time import statistics -from .const import MIN_SAMPLES_LEARNING, MIN_LIMIT_TIMEOUT, MIN_MODE_PROFILE_SAMPLES, DEFAULT_DEAD_TIME +from .const import ( + MIN_SAMPLES_LEARNING, + MIN_LIMIT_TIMEOUT, + MIN_MODE_PROFILE_SAMPLES, + DEFAULT_DEAD_TIME, + REFERENCE_SLOPE_ERROR, +) _LOGGER = logging.getLogger(__name__) @@ -12,7 +18,7 @@ class ThermalLearning: def __init__(self): # Data collection with sliding window - self._slope_samples = [] # (timestamp, fan_mode, slope, hvac_mode) + self._slope_samples = [] # (timestamp, fan_mode, slope, hvac_mode, temperature_error) self._response_events = [] # (timestamp, response_time_minutes) - thermal response from fan change to slope change self._learning_window_hours = 168 # 7 days sliding window self._min_samples = MIN_SAMPLES_LEARNING # Minimum samples for initial readiness (48-72h typical activity) @@ -61,7 +67,7 @@ def add_slope_sample(self, fan_mode: str, slope: float, temperature_error: float _LOGGER.debug("Learning: Skipped sample (stagnation, slope=%.2f)", slope) return - self._slope_samples.append((time.time(), fan_mode, slope, hvac_mode)) + self._slope_samples.append((time.time(), fan_mode, slope, hvac_mode, temperature_error)) profile_samples = self.get_mode_sample_count(fan_mode, hvac_mode) _LOGGER.debug( "Learning: Collected slope sample #%d (fan=%s, slope=%.2f, err=%.2f, hvac=%s, profile=%d/%d)", @@ -119,7 +125,7 @@ def trim_with_min_retention( Prevents a rarely-used mode from losing its learned profile solely because it hasn't been active in the past 7 days. - Samples are 4-tuples: (timestamp, fan_mode, slope, hvac_mode). + Samples are 5-tuples: (timestamp, fan_mode, slope, hvac_mode, temperature_error). """ within = [s for s in samples if s[0] > cutoff_time] expired = [s for s in samples if s[0] <= cutoff_time] @@ -128,8 +134,8 @@ def trim_with_min_retention( # Count within-window samples per (fan_mode, hvac_mode) profile profile_counts: dict[tuple, int] = {} - for _, fm, _, hm in within: - key = (fm, hm) + for s in within: + key = (s[1], s[3]) profile_counts[key] = profile_counts.get(key, 0) + 1 # Group expired samples by profile @@ -283,11 +289,129 @@ def get_dead_time(self, hvac_mode: str = "unknown") -> float: return DEFAULT_DEAD_TIME return statistics.median(response_times) + def _fit_mode_slope(self, fan_mode: str, hvac_mode: str) -> tuple[float, float, float | None] | None: + """Fit the gap-dependent slope model and return (intercept_a, gain_b, r_squared). + + The effective cooling/heating rate is not constant: it scales with the + comfort error (distance to the setpoint), per Newton's law of cooling. + We model it as a linear relationship: + + effective_slope(error) = a + b * error + + fitted by ordinary least squares over the profile's (error, effective_slope) + samples. ``error`` is the signed comfort error (positive = needs more + cooling/heating). ``effective_slope`` is positive when moving towards target + (raw VTherm slope is inverted in cooling). + + ``r_squared`` is the coefficient of determination of the fit (0..1); it is + ``None`` for the constant fallback (no real regression was performed). + + Falls back to a constant model ``(median_effective_slope, 0.0, None)`` when + there are too few error-bearing samples or the error has no spread — which + keeps behaviour identical to the previous median estimator for legacy data + and for synthetic profiles seeded via ``set_mode_effective_slope``. + + The gain ``b`` is clamped to be non-negative: a larger gap can only cool/heat + at least as fast, never slower. + + Returns None if fewer than MIN_MODE_PROFILE_SAMPLES are available. + """ + sign = -1.0 if hvac_mode == "cool" else 1.0 + matching = [s for s in self._slope_samples if s[1] == fan_mode and s[3] == hvac_mode] + if len(matching) < MIN_MODE_PROFILE_SAMPLES: + return None + + median_eff = sign * statistics.median([s[2] for s in matching]) + + # Only samples that carry a stored error can drive the regression. + points = [(s[4], sign * s[2]) for s in matching if len(s) > 4 and s[4] is not None] + if len(points) < MIN_MODE_PROFILE_SAMPLES: + return (median_eff, 0.0, None) + + n = len(points) + mean_x = sum(x for x, _ in points) / n + mean_y = sum(y for _, y in points) / n + var_x = sum((x - mean_x) ** 2 for x, _ in points) + if var_x < 1e-6: + # All samples taken at (nearly) the same error: no slope can be fitted. + return (median_eff, 0.0, None) + + cov_xy = sum((x - mean_x) * (y - mean_y) for x, y in points) + gain_b = max(0.0, cov_xy / var_x) + intercept_a = mean_y - gain_b * mean_x + + # Coefficient of determination against the (clamped) fitted line. + ss_tot = sum((y - mean_y) ** 2 for _, y in points) + if ss_tot < 1e-9: + r_squared = None + else: + ss_res = sum((y - (intercept_a + gain_b * x)) ** 2 for x, y in points) + r_squared = max(0.0, 1.0 - ss_res / ss_tot) + return (intercept_a, gain_b, r_squared) + + def get_mode_slope_model(self, fan_mode: str, hvac_mode: str) -> tuple[float, float] | None: + """Return the gap-dependent slope model ``(intercept_a, gain_b)`` for a profile. + + See :meth:`_fit_mode_slope` for the model definition. Returns None if the + profile has fewer than MIN_MODE_PROFILE_SAMPLES samples. + """ + fit = self._fit_mode_slope(fan_mode, hvac_mode) + return None if fit is None else (fit[0], fit[1]) + + def get_mode_slope_gain(self, fan_mode: str, hvac_mode: str) -> float: + """Return the gap gain ``b`` (°C/h per °C of comfort error) for a profile. + + Returns 0.0 when the profile is unknown or the model is constant. + """ + model = self.get_mode_slope_model(fan_mode, hvac_mode) + return 0.0 if model is None else model[1] + + def get_mode_slope_r2(self, fan_mode: str, hvac_mode: str) -> float | None: + """Return the R² (goodness of fit, 0..1) of the gap-dependent slope model. + + Higher values mean the cooling/heating rate is well explained by the + distance to the setpoint, i.e. the learned thermal model is reliable. + Returns None when the model is a constant fallback (no regression). + """ + fit = self._fit_mode_slope(fan_mode, hvac_mode) + return None if fit is None else fit[2] + + def get_mode_time_constant(self, fan_mode: str, hvac_mode: str) -> float | None: + """Return the thermal time constant τ (hours) for a profile. + + In a first-order RC thermal model the rate of approach to the setpoint is + ``error / τ``, so the learned gain ``b`` (°C/h per °C) approximates ``1/τ``. + A larger τ means more thermal inertia / resistance (slower response). + Returns None when the gain is ~0 (constant model: τ is undefined). + """ + gain = self.get_mode_slope_gain(fan_mode, hvac_mode) + if gain < 1e-3: + return None + return 1.0 / gain + + def get_mode_effective_slope_at(self, fan_mode: str, hvac_mode: str, error: float) -> float | None: + """Return the modelled effective slope at a given comfort error. + + The error is floored at 0: at/below the setpoint there is no driving + force, so the modelled active cooling/heating rate is the intercept only. + Returns None if the profile is not learned yet. + """ + model = self.get_mode_slope_model(fan_mode, hvac_mode) + if model is None: + return None + intercept_a, gain_b = model + return intercept_a + gain_b * max(error, 0.0) + def get_mode_effective_slope(self, fan_mode: str, hvac_mode: str) -> float | None: - """Return the median effective slope for a fan mode in a given HVAC mode. + """Return the representative "working" effective slope for a profile. + + This is the gap-dependent model evaluated at REFERENCE_SLOPE_ERROR — a + representative non-trivial gap — so the reported value reflects the fan's + real cooling/heating power instead of the near-equilibrium median (which is + structurally diluted by the many samples collected close to the setpoint). - Uses median instead of mean for robustness against outlier samples - (e.g. residual inertia from a previous high-speed mode). + For legacy/synthetic constant profiles (gain == 0) this is exactly the old + median estimator, preserving backward compatibility. Effective slope is positive when moving towards target: - In heating: positive raw slope is good @@ -295,11 +419,11 @@ def get_mode_effective_slope(self, fan_mode: str, hvac_mode: str) -> float | Non Returns None if fewer than MIN_MODE_PROFILE_SAMPLES are available. """ - matching_slopes = [sl for (_, fm, sl, hm) in self._slope_samples if fm == fan_mode and hm == hvac_mode] - if len(matching_slopes) < MIN_MODE_PROFILE_SAMPLES: + model = self.get_mode_slope_model(fan_mode, hvac_mode) + if model is None: return None - med = statistics.median(matching_slopes) - return -med if hvac_mode == "cool" else med + intercept_a, gain_b = model + return intercept_a + gain_b * REFERENCE_SLOPE_ERROR def get_profile_spread(self, fan_mode: str, hvac_mode: str) -> float | None: """Return the MAD/median ratio for a profile's absolute slopes. @@ -311,7 +435,7 @@ def get_profile_spread(self, fan_mode: str, hvac_mode: str) -> float | None: Returns None if the profile has fewer than MIN_MODE_PROFILE_SAMPLES samples. """ - abs_slopes = [abs(sl) for (_, fm, sl, hm) in self._slope_samples if fm == fan_mode and hm == hvac_mode] + abs_slopes = [abs(s[2]) for s in self._slope_samples if s[1] == fan_mode and s[3] == hvac_mode] if len(abs_slopes) < MIN_MODE_PROFILE_SAMPLES: return None med = statistics.median(abs_slopes) @@ -334,10 +458,11 @@ def set_mode_effective_slope(self, fan_mode: str, hvac_mode: str, target_slope: self._slope_samples = [s for s in self._slope_samples if not (s[1] == fan_mode and s[3] == hvac_mode)] removed = before - len(self._slope_samples) - # Insert MIN_MODE_PROFILE_SAMPLES synthetic samples at current time + # Insert MIN_MODE_PROFILE_SAMPLES synthetic samples at current time. + # error is None so they produce a constant model (gain 0) at exactly target_slope. now = time.time() for i in range(MIN_MODE_PROFILE_SAMPLES): - self._slope_samples.append((now + i, fan_mode, raw_slope, hvac_mode)) + self._slope_samples.append((now + i, fan_mode, raw_slope, hvac_mode, None)) self.recompute_slope_stats() @@ -348,13 +473,13 @@ def set_mode_effective_slope(self, fan_mode: str, hvac_mode: str, target_slope: def get_mode_sample_count(self, fan_mode: str, hvac_mode: str) -> int: """Return the number of collected samples for one fan/HVAC profile.""" - return sum(1 for (_, fm, _, hm) in self._slope_samples if fm == fan_mode and hm == hvac_mode) + return sum(1 for s in self._slope_samples if s[1] == fan_mode and s[3] == hvac_mode) def get_known_fan_modes(self) -> list[str]: """Return unique fan modes seen in slope samples, preserving first-seen order.""" seen: dict[str, None] = {} - for _, fan_mode, _, _ in self._slope_samples: - seen[fan_mode] = None + for s in self._slope_samples: + seen[s[1]] = None return list(seen.keys()) def get_mode_profiles(self, hvac_mode: str, fan_modes: list[str] | None = None) -> dict[str, dict]: @@ -362,16 +487,25 @@ def get_mode_profiles(self, hvac_mode: str, fan_modes: list[str] | None = None) if fan_modes: ordered_modes = list(dict.fromkeys(fan_modes)) else: - ordered_modes = sorted({fm for (_, fm, _, hm) in self._slope_samples if hm == hvac_mode}) + ordered_modes = sorted({s[1] for s in self._slope_samples if s[3] == hvac_mode}) profiles: dict[str, dict] = {} for fan_mode in ordered_modes: - effective_slope = self.get_mode_effective_slope(fan_mode, hvac_mode) + fit = self._fit_mode_slope(fan_mode, hvac_mode) sample_count = self.get_mode_sample_count(fan_mode, hvac_mode) + if fit is None: + effective_slope = gain = r_squared = time_constant = None + else: + intercept_a, gain, r_squared = fit + effective_slope = intercept_a + gain * REFERENCE_SLOPE_ERROR + time_constant = (1.0 / gain) if gain >= 1e-3 else None profiles[fan_mode] = { "effective_slope": round(effective_slope, 3) if effective_slope is not None else None, + "slope_gain": round(gain, 3) if gain is not None else None, + "r_squared": round(r_squared, 3) if r_squared is not None else None, + "thermal_time_constant_h": round(time_constant, 2) if time_constant is not None else None, "samples": sample_count, - "ready": effective_slope is not None, + "ready": fit is not None, } return profiles @@ -454,9 +588,9 @@ def recompute_slope_stats(self) -> None: self._slope_max = 0.0 self._optimal_cache = None # Invalidate cache when stats are rebuilt self._profile_ready_logged = { - (hm, fm) - for (_, fm, _, hm) in self._slope_samples - if hm != "unknown" and self.get_mode_sample_count(fm, hm) >= MIN_MODE_PROFILE_SAMPLES + (s[3], s[1]) + for s in self._slope_samples + if s[3] != "unknown" and self.get_mode_sample_count(s[1], s[3]) >= MIN_MODE_PROFILE_SAMPLES } for sample in self._slope_samples: @@ -473,18 +607,24 @@ def recompute_slope_stats(self) -> None: def from_dict(cls, data: dict): """Restore from storage. - Handles backward compatibility: old 3-tuple samples (timestamp, fan_mode, slope) - are migrated to 4-tuples by appending hvac_mode="unknown". + Handles backward compatibility for slope_samples across schema versions: + - 3-tuple (timestamp, fan_mode, slope) → hvac_mode="unknown", error=None + - 4-tuple (timestamp, fan_mode, slope, hvac_mode) → error=None + - 5-tuple (timestamp, fan_mode, slope, hvac_mode, temperature_error) → as-is + Samples without a stored error simply don't contribute to the gap-slope + regression (they fall back to the constant median model). Old 2-tuple response_events are migrated to 3-tuples by appending hvac_mode="unknown". """ instance = cls() - # Migrate slope_samples: support both 3-tuple (old) and 4-tuple (new) + # Migrate slope_samples to the canonical 5-tuple shape. raw_samples = data.get("slope_samples", []) instance._slope_samples = [] for sample in raw_samples: if len(sample) == 3: - instance._slope_samples.append((sample[0], sample[1], sample[2], "unknown")) + instance._slope_samples.append((sample[0], sample[1], sample[2], "unknown", None)) + elif len(sample) == 4: + instance._slope_samples.append((sample[0], sample[1], sample[2], sample[3], None)) else: instance._slope_samples.append(tuple(sample)) diff --git a/docs/mpc_mode.md b/docs/mpc_mode.md index 630ea6c..c0e72ed 100644 --- a/docs/mpc_mode.md +++ b/docs/mpc_mode.md @@ -91,19 +91,47 @@ Where: The model reuses the current learning subsystem: - `learning.get_dead_time()` for thermal delay -- `learning.get_mode_effective_slope(fan_mode, hvac_mode)` for reliable per-mode profiles +- `learning.get_mode_slope_model(fan_mode, hvac_mode)` for reliable per-mode profiles If a reliable profile is not yet available for a fan mode, the MPC model falls back to a coarse rank-based estimate derived from the current slope. Confidence is lowered accordingly. +### Gap-Dependent Slope Model + +Effective cooling/heating power is **not constant**: per Newton's law of cooling it scales with +the distance to the setpoint. A single scalar (the historical median) is structurally diluted by the +many samples collected near equilibrium, where the slope is naturally shallow — it under-states the +fan's real working power. + +Each profile therefore learns a linear model by ordinary least squares over its +`(comfort_error, effective_slope)` samples: + +```text +effective_slope(error) = a + b * error (b clamped to >= 0) +``` + +- `learning.get_mode_slope_model()` returns `(a, b)`; `get_mode_slope_gain()` returns `b`. +- `learning.get_mode_effective_slope()` reports the representative **working** slope, i.e. the model + evaluated at `REFERENCE_SLOPE_ERROR` (1 °C). For legacy/synthetic constant profiles (`b == 0`) this + is exactly the previous median estimator, so behaviour is unchanged for those. +- The simulator recomputes the slope **at each step** from the simulated error, so the projection + decelerates realistically as the room approaches the setpoint instead of cooling/heating at a fixed + rate (which produced phantom overshoot past the target), and uses the higher real power when the + room is far from target (faster, more accurate catch-up). + ### Disturbance Handling When the current fan mode has a reliable learned profile and the controller is in `ESTABLISHED` phase, the MPC model estimates a slow disturbance bias from the residual: ```text -residual = observed_effective_power - expected_effective_power +residual = observed_effective_power - expected_effective_power(current_error) bias[k+1] = (1 - beta) * bias[k] + beta * residual ``` +The expected power is the gap-dependent model evaluated **at the current comfort error**, not at the +reference gap. This keeps the bias clean: it captures only genuine external disturbances (solar gain, +occupancy) instead of the systematic variation of power with the distance to setpoint, which the gap +model now explains directly. + This helps the MPC model remain usable when the room is slightly helped or hindered by effects not directly modeled by the fan itself. When a window is detected as open, the MPC model does not trust its own predictions and switches to a `Disturbed` status instead of producing an actionable recommendation. @@ -111,7 +139,8 @@ When a window is detected as open, the MPC model does not trust its own predicti ## MPC-lite Decision Rule The controller simulates every available fan mode on a short fixed horizon of 30 minutes. -The current scaffold keeps the action constant over the horizon to stay simple and fast. +The candidate fan action is held constant over the horizon, but its effective slope is recomputed at +each step from the simulated comfort error (see the gap-dependent slope model above). The simulator supports both `heat` and `cool`; cooling uses the same learned effective-power model with the sign inverted back to room-temperature evolution. Each candidate mode gets a scalar cost: diff --git a/tests/test_learning.py b/tests/test_learning.py index ddcf337..7b9c857 100644 --- a/tests/test_learning.py +++ b/tests/test_learning.py @@ -178,6 +178,117 @@ def test_median_resists_outliers(self): assert slope is not None assert slope == pytest.approx(0.15, abs=0.01) + def test_gap_model_learns_positive_gain(self): + """A profile whose slope scales with the comfort error yields a+b·error.""" + learning = ThermalLearning() + # Perfectly linear: effective_slope = 0.4 + 0.8 * error + for err in [0.5, 1.0, 1.5, 2.0] * 3: + learning.add_slope_sample("superhigh", 0.4 + 0.8 * err, err, hvac_mode="heat") + + a, b = learning.get_mode_slope_model("superhigh", "heat") + assert a == pytest.approx(0.4, abs=0.01) + assert b == pytest.approx(0.8, abs=0.01) + assert learning.get_mode_slope_gain("superhigh", "heat") == pytest.approx(0.8, abs=0.01) + + def test_effective_slope_reports_working_value_at_reference(self): + """get_mode_effective_slope returns the model at REFERENCE_SLOPE_ERROR, not the median.""" + learning = ThermalLearning() + for err in [0.5, 1.0, 1.5, 2.0] * 3: + learning.add_slope_sample("superhigh", 0.4 + 0.8 * err, err, hvac_mode="heat") + + # Working slope at error=1.0 -> 0.4 + 0.8 = 1.2 (the median of the samples is ~1.4) + assert learning.get_mode_effective_slope("superhigh", "heat") == pytest.approx(1.2, abs=0.02) + + def test_get_mode_effective_slope_at_scales_with_error(self): + """The modelled slope scales with the gap and floors the error at 0.""" + learning = ThermalLearning() + for err in [0.5, 1.0, 1.5, 2.0] * 3: + learning.add_slope_sample("superhigh", 0.4 + 0.8 * err, err, hvac_mode="heat") + + assert learning.get_mode_effective_slope_at("superhigh", "heat", 2.0) == pytest.approx(2.0, abs=0.02) + assert learning.get_mode_effective_slope_at("superhigh", "heat", 0.0) == pytest.approx(0.4, abs=0.02) + # Negative error is floored at 0 -> intercept only + assert learning.get_mode_effective_slope_at("superhigh", "heat", -3.0) == pytest.approx(0.4, abs=0.02) + + def test_gap_model_cool_inversion(self): + """In cool mode the raw slope is inverted before fitting the gap model.""" + learning = ThermalLearning() + for err in [0.5, 1.0, 1.5, 2.0] * 3: + # raw cooling slope is negative; effective = 0.4 + 0.8*err + learning.add_slope_sample("superhigh", -(0.4 + 0.8 * err), err, hvac_mode="cool") + + a, b = learning.get_mode_slope_model("superhigh", "cool") + assert a == pytest.approx(0.4, abs=0.02) + assert b == pytest.approx(0.8, abs=0.02) + + def test_constant_error_yields_zero_gain(self): + """When all samples share the same error, the model is the constant median.""" + learning = ThermalLearning() + for _ in range(12): + learning.add_slope_sample("high", 0.9, 0.3, hvac_mode="heat") + + a, b = learning.get_mode_slope_model("high", "heat") + assert b == 0.0 + assert a == pytest.approx(0.9, abs=0.001) + assert learning.get_mode_effective_slope("high", "heat") == pytest.approx(0.9, abs=0.001) + + def test_gain_clamped_non_negative(self): + """A spurious negative correlation (slope falls as gap grows) is clamped to 0.""" + learning = ThermalLearning() + for err in [0.5, 1.0, 1.5, 2.0] * 3: + learning.add_slope_sample("high", 2.0 - 0.5 * err, err, hvac_mode="heat") + + a, b = learning.get_mode_slope_model("high", "heat") + assert b == 0.0 # never model cooling/heating as weaker further from target + + def test_error_persisted_and_restored(self): + """Sample errors survive a to_dict/from_dict round-trip so the gap model is stable.""" + learning = ThermalLearning() + for err in [0.5, 1.0, 1.5, 2.0] * 3: + learning.add_slope_sample("superhigh", 0.4 + 0.8 * err, err, hvac_mode="heat") + + restored = ThermalLearning.from_dict(learning.to_dict()) + a, b = restored.get_mode_slope_model("superhigh", "heat") + assert a == pytest.approx(0.4, abs=0.02) + assert b == pytest.approx(0.8, abs=0.02) + + def test_r_squared_perfect_fit(self): + """A perfectly linear profile yields R² ≈ 1.""" + learning = ThermalLearning() + for err in [0.5, 1.0, 1.5, 2.0] * 3: + learning.add_slope_sample("superhigh", 0.4 + 0.8 * err, err, hvac_mode="heat") + + assert learning.get_mode_slope_r2("superhigh", "heat") == pytest.approx(1.0, abs=0.001) + + def test_r_squared_and_time_constant_none_for_constant_profile(self): + """A constant (no-gain) profile has no regression R² and no time constant.""" + learning = ThermalLearning() + for _ in range(12): + learning.add_slope_sample("high", 0.9, 0.3, hvac_mode="heat") + + assert learning.get_mode_slope_r2("high", "heat") is None + assert learning.get_mode_time_constant("high", "heat") is None + + def test_thermal_time_constant_is_inverse_gain(self): + """The thermal time constant equals 1/gain (hours).""" + learning = ThermalLearning() + for err in [0.5, 1.0, 1.5, 2.0] * 3: + learning.add_slope_sample("superhigh", 0.4 + 0.8 * err, err, hvac_mode="heat") + + # gain b = 0.8 -> tau = 1.25 h + assert learning.get_mode_time_constant("superhigh", "heat") == pytest.approx(1.25, abs=0.02) + + def test_profile_summary_exposes_r2_and_time_constant(self): + """get_mode_profiles surfaces r_squared and thermal_time_constant_h.""" + learning = ThermalLearning() + for err in [0.5, 1.0, 1.5, 2.0] * 3: + learning.add_slope_sample("superhigh", 0.4 + 0.8 * err, err, hvac_mode="heat") + + prof = learning.get_mode_profiles("heat", ["superhigh"])["superhigh"] + assert prof["r_squared"] == pytest.approx(1.0, abs=0.001) + assert prof["thermal_time_constant_h"] == pytest.approx(1.25, abs=0.02) + assert prof["slope_gain"] == pytest.approx(0.8, abs=0.02) + def test_get_dead_time_decoupled_by_hvac_mode(self): """Test that get_dead_time handles separate heating and cooling events correctly.""" learning = ThermalLearning() diff --git a/tests/test_mpc_controller.py b/tests/test_mpc_controller.py index f44ccd6..c0e6bf7 100644 --- a/tests/test_mpc_controller.py +++ b/tests/test_mpc_controller.py @@ -294,6 +294,37 @@ def test_confidence_penalised_by_high_spread() -> None: assert result_tight["mpc_confidence"] > result_noisy["mpc_confidence"] +def test_confidence_high_with_full_coverage_before_global_ready() -> None: + """Full per-mode profile coverage yields a Ready-level confidence even when + the global sample threshold (is_ready) has not been reached. + + Regression guard for the previous behaviour where an HVAC mode used only + part of the year stayed stuck at 'Low confidence' despite every fan-mode + profile being fully learned. + """ + learning = ThermalLearning() + # 12 tight samples per mode -> every profile ready, but well under the + # global MIN_SAMPLES_LEARNING threshold, so is_ready() stays False. + for _ in range(12): + learning.add_slope_sample("low", 0.25, 0.8, "heat") + learning.add_slope_sample("medium", 0.9, 0.8, "heat") + learning.add_slope_sample("high", 1.5, 0.8, "heat") + assert learning.is_ready() is False + + mpc = _build_mpc(learning) + result = mpc.evaluate( + current_temp=19.5, + target_temp=20.0, + vtherm_slope=0.9, + hvac_mode="heat", + current_fan="high", + minutes_since_change=30.0, + ) + assert result["mpc_known_profiles"] == 3 + assert result["mpc_confidence"] >= 50.0 + assert result["mpc_status"] == "Ready" + + @pytest.mark.asyncio async def test_data_collector_records_mpc_columns(tmp_path: Path) -> None: """Data collector CSV includes MPC-specific columns.""" @@ -619,6 +650,68 @@ def test_mpc_handles_long_dead_time_without_blindness() -> None: assert result["mpc_would_change_now"] == "yes" +def _seed_gap_profile(learning: ThermalLearning, fan_mode: str, hvac_mode: str, a: float, b: float) -> None: + """Seed a profile whose effective slope follows a + b·error.""" + import time + now = time.time() + sign = -1.0 if hvac_mode == "cool" else 1.0 + samples = [] + for i, err in enumerate([0.5, 1.0, 1.5, 2.0, 2.5, 3.0] * 2): + samples.append((now + i, fan_mode, sign * (a + b * err), hvac_mode, err)) + learning.slope_samples = list(learning.slope_samples) + samples + + +def test_gap_model_projects_faster_cooling_when_far_from_target() -> None: + """A gap-dependent profile cools faster from a hot room than the same constant slope.""" + gap_learning = ThermalLearning() + _seed_gap_profile(gap_learning, "superhigh", "cool", a=0.5, b=1.0) + gap_mpc = MPCController(learning=gap_learning, deadband=0.3, min_interval=10, fan_modes=["superhigh"]) + + # Equivalent constant-slope profile pinned to the gap model's working value. + working = gap_learning.get_mode_effective_slope("superhigh", "cool") + const_learning = ThermalLearning() + const_learning.set_mode_effective_slope("superhigh", "cool", working) + const_mpc = MPCController(learning=const_learning, deadband=0.3, min_interval=10, fan_modes=["superhigh"]) + + kwargs = dict(current_temp=26.0, target_temp=24.0, vtherm_slope=-1.0, + hvac_mode="cool", current_fan="superhigh", minutes_since_change=20.0) + gap = gap_mpc.evaluate(**kwargs) + const = const_mpc.evaluate(**kwargs) + + # Hot room (error 2.0): gap model uses ~2.5 °C/h, so it cools further than the constant ~1.5. + assert gap["mpc_predicted_temperature_30m"] < const["mpc_predicted_temperature_30m"] + + +def test_gap_model_does_not_plunge_past_target() -> None: + """The gap model decelerates near the setpoint instead of projecting phantom overshoot.""" + learning = ThermalLearning() + _seed_gap_profile(learning, "superhigh", "cool", a=0.5, b=1.0) + mpc = MPCController(learning=learning, deadband=0.3, min_interval=10, fan_modes=["superhigh"]) + + result = mpc.evaluate( + current_temp=24.4, target_temp=24.0, vtherm_slope=-0.5, + hvac_mode="cool", current_fan="superhigh", minutes_since_change=20.0, + ) + # Starting only 0.4°C above target, a 30-min projection must asymptote toward 24.0, + # not dive well below it the way a constant-slope model would. + assert result["mpc_predicted_temperature_30m"] >= 23.7 + + +def test_gap_aware_disturbance_bias_stays_small_without_disturbance() -> None: + """When the observed slope matches the gap model at the current error, bias stays ~0.""" + learning = ThermalLearning() + _seed_gap_profile(learning, "superhigh", "cool", a=0.5, b=1.0) + mpc = MPCController(learning=learning, deadband=0.3, min_interval=10, fan_modes=["superhigh"]) + + # Room 2°C above target -> model expects ~2.5 °C/h effective cooling (raw vtherm_slope ≈ -2.5). + for _ in range(10): + mpc.evaluate( + current_temp=26.0, target_temp=24.0, vtherm_slope=-2.5, + hvac_mode="cool", current_fan="superhigh", minutes_since_change=40.0, + ) + assert abs(mpc.disturbance_bias) < 0.2 + + def test_learning_response_sensor_with_mixed_tuple_lengths() -> None: """SmartFanLearningResponseSensor extra_state_attributes handles mixed lengths in response_events.""" import time diff --git a/tests/test_services.py b/tests/test_services.py index 04d5efb..c34bbe1 100644 --- a/tests/test_services.py +++ b/tests/test_services.py @@ -6,18 +6,28 @@ from homeassistant.exceptions import HomeAssistantError from homeassistant.helpers import entity_registry as er +from types import SimpleNamespace + from custom_components.smart_fan_controller import ( _apply_optimal_parameters, + _async_apply_fan_change, _async_migrate_entity_ids, + _detect_disturbances, + _read_climate_cycle_data, + _resolve_active_force, _resolve_controller_entry, + _should_collect_slope_sample, ) from custom_components.smart_fan_controller.thermal_learning import ThermalLearning from custom_components.smart_fan_controller.const import ( CONF_CLIMATE_ENTITY, CONF_DEADBAND, + CONF_DEFROST_ENTITY, CONF_LIMIT_TIMEOUT, + CONF_OPERATING_ENTITY, DOMAIN, MIN_SAMPLES_LEARNING, + PHASE_ESTABLISHED, build_unique_id, ) @@ -217,3 +227,185 @@ def test_reset_then_relearn(self): assert learning.is_ready() assert learning.slope_sample_count() == MIN_SAMPLES_LEARNING + + +class TestForceFan: + """Tests for the force_fan override resolution logic.""" + + FAN_MODES = ["silent", "low", "med", "high", "superhigh"] + + def test_no_force_returns_none(self): + """No override configured -> MPC stays in control.""" + entry_data = {"force": None} + assert _resolve_active_force(entry_data, self.FAN_MODES, now=1000.0, climate_id="climate.test") is None + + def test_active_force_returns_fan_mode(self): + """An active override returns the forced fan mode.""" + entry_data = {"force": {"fan_mode": "high", "until": 2000.0}} + assert _resolve_active_force(entry_data, self.FAN_MODES, now=1000.0, climate_id="climate.test") == "high" + # Override is preserved while active + assert entry_data["force"] is not None + + def test_expired_force_is_cleared(self): + """An expired override returns None and is cleared so the MPC resumes.""" + entry_data = {"force": {"fan_mode": "high", "until": 999.0}} + assert _resolve_active_force(entry_data, self.FAN_MODES, now=1000.0, climate_id="climate.test") is None + assert entry_data["force"] is None + + def test_invalid_fan_mode_is_cleared(self): + """An override for an unknown fan mode is dropped.""" + entry_data = {"force": {"fan_mode": "turbo", "until": 2000.0}} + assert _resolve_active_force(entry_data, self.FAN_MODES, now=1000.0, climate_id="climate.test") is None + assert entry_data["force"] is None + + def test_unknown_fan_modes_does_not_block_force(self): + """Before fan modes are discovered, the override is still honoured.""" + entry_data = {"force": {"fan_mode": "high", "until": 2000.0}} + assert _resolve_active_force(entry_data, None, now=1000.0, climate_id="climate.test") == "high" + + +class TestReadClimateCycleData: + """Tests for parsing/guarding VTherm climate state (M3 hardening).""" + + @staticmethod + def _state(**overrides): + attrs = { + "specific_states": {"temperature_slope": 0.5}, + "current_temperature": 22.0, + "temperature": 20.0, + "hvac_mode": "heat", + "fan_mode": "low", + } + attrs.update(overrides) + return SimpleNamespace(attributes=attrs) + + def test_valid_state_returns_parsed_tuple(self): + result = _read_climate_cycle_data(self._state(), "climate.test") + assert result == (0.5, 22.0, 20.0, "heat", "low", False) + + def test_non_numeric_slope_skips_cycle(self): + """A transient 'unavailable' slope must skip the cycle, not raise.""" + state = self._state(specific_states={"temperature_slope": "unavailable"}) + assert _read_climate_cycle_data(state, "climate.test") is None + + def test_non_numeric_temperature_skips_cycle(self): + state = self._state(current_temperature="unknown") + assert _read_climate_cycle_data(state, "climate.test") is None + + def test_missing_temperature_skips_cycle(self): + state = self._state(current_temperature=None) + assert _read_climate_cycle_data(state, "climate.test") is None + + def test_missing_slope_defaults_to_zero(self): + """A missing temperature_slope key defaults to 0.0 (not a skip).""" + state = self._state(specific_states={}) + result = _read_climate_cycle_data(state, "climate.test") + assert result is not None and result[0] == 0.0 + + +class TestShouldCollectSlopeSample: + """Tests for slope-sample collection gating (incl. the heat/cool gate, L4).""" + + @staticmethod + def _kwargs(**over): + base = dict( + current_fan="low", + hvac_mode="heat", + is_defrost_active=False, + is_hvac_idle=False, + phase=PHASE_ESTABLISHED, + minutes_since_change=25.0, + learned_dead_time=10.0, # min stable window = 20 min + ctrl_state={"last_setpoint_drop_time": 0.0}, + now=1000.0, + ) + base.update(over) + return base + + def test_happy_path_collects(self): + assert _should_collect_slope_sample(**self._kwargs()) is True + + def test_off_mode_is_skipped(self): + assert _should_collect_slope_sample(**self._kwargs(hvac_mode="off")) is False + + def test_dry_and_fan_only_skipped(self): + assert _should_collect_slope_sample(**self._kwargs(hvac_mode="dry")) is False + assert _should_collect_slope_sample(**self._kwargs(hvac_mode="fan_only")) is False + + def test_cool_mode_collects(self): + assert _should_collect_slope_sample(**self._kwargs(hvac_mode="cool")) is True + + def test_no_fan_skipped(self): + assert _should_collect_slope_sample(**self._kwargs(current_fan=None)) is False + + def test_defrost_or_idle_skipped(self): + assert _should_collect_slope_sample(**self._kwargs(is_defrost_active=True)) is False + assert _should_collect_slope_sample(**self._kwargs(is_hvac_idle=True)) is False + + def test_non_established_phase_skipped(self): + assert _should_collect_slope_sample(**self._kwargs(phase="DEAD_TIME")) is False + + def test_not_stable_long_enough_skipped(self): + # 15 min < 2 * 10 min dead time + assert _should_collect_slope_sample(**self._kwargs(minutes_since_change=15.0)) is False + + def test_recent_setpoint_drop_skipped(self): + # setpoint dropped 1 min ago -> within the learning cooldown + kwargs = self._kwargs(ctrl_state={"last_setpoint_drop_time": 940.0}, now=1000.0) + assert _should_collect_slope_sample(**kwargs) is False + + +class TestDetectDisturbances: + """Tests for external defrost / idle detection.""" + + def test_defrost_entity_on_marks_active(self): + hass = MagicMock() + hass.states.get.return_value = SimpleNamespace(state="on") + defrost_state = {"active": False, "start_time": 0.0} + is_defrost, is_idle = _detect_disturbances(hass, {CONF_DEFROST_ENTITY: "binary_sensor.d"}, defrost_state) + assert is_defrost is True and is_idle is False + assert defrost_state["active"] is True + + def test_defrost_expires_after_cooldown(self): + import time + hass = MagicMock() + defrost_state = {"active": True, "start_time": time.time() - 21 * 60} + is_defrost, _ = _detect_disturbances(hass, {}, defrost_state) + assert is_defrost is False + assert defrost_state["active"] is False + + def test_operating_entity_off_marks_idle(self): + hass = MagicMock() + hass.states.get.return_value = SimpleNamespace(state="off") + _, is_idle = _detect_disturbances(hass, {CONF_OPERATING_ENTITY: "binary_sensor.op"}, {"active": False, "start_time": 0.0}) + assert is_idle is True + + def test_operating_entity_on_not_idle(self): + hass = MagicMock() + hass.states.get.return_value = SimpleNamespace(state="on") + _, is_idle = _detect_disturbances(hass, {CONF_OPERATING_ENTITY: "binary_sensor.op"}, {"active": False, "start_time": 0.0}) + assert is_idle is False + + +class TestApplyFanChange: + """Tests for _async_apply_fan_change confirmation semantics.""" + + @pytest.mark.asyncio + async def test_success_calls_service_and_confirms(self): + hass = MagicMock() + hass.services.async_call = AsyncMock() + confirmed = MagicMock() + await _async_apply_fan_change(hass, "climate.t", "high", "low", "reason", confirmed) + hass.services.async_call.assert_awaited_once() + args = hass.services.async_call.await_args + assert args.args[0] == "climate" and args.args[1] == "set_fan_mode" + assert args.args[2] == {"entity_id": "climate.t", "fan_mode": "high"} + confirmed.assert_called_once() + + @pytest.mark.asyncio + async def test_failure_still_confirms_to_avoid_retry_storm(self): + hass = MagicMock() + hass.services.async_call = AsyncMock(side_effect=RuntimeError("boom")) + confirmed = MagicMock() + await _async_apply_fan_change(hass, "climate.t", "high", "low", "reason", confirmed) + confirmed.assert_called_once()