From 58ffd3393ad36b6013106189304b68c8dffcab8c Mon Sep 17 00:00:00 2001 From: Laurens <43173895+GreenGrassBlueOcean@users.noreply.github.com> Date: Sun, 18 Oct 2026 22:09:46 +0200 Subject: [PATCH 1/4] feat(cover,light): pluggable CoverKinematicsStrategy hierarchy and low-priority fade step routing (#730, #739) --- custom_components/myhome/const.py | 2 + custom_components/myhome/cover.py | 344 +++++---- .../myhome/cover_strategies/__init__.py | 124 +++ .../myhome/cover_strategies/base.py | 222 ++++++ .../myhome/cover_strategies/native.py | 100 +++ .../myhome/cover_strategies/simple.py | 72 ++ .../myhome/cover_strategies/timed.py | 192 +++++ .../myhome/cover_strategies/venetian.py | 126 +++ custom_components/myhome/light.py | 10 +- custom_components/myhome/validate.py | 14 +- tests/test_cover_strategies.py | 721 ++++++++++++++++++ tests/test_validate.py | 6 +- 12 files changed, 1790 insertions(+), 143 deletions(-) create mode 100644 custom_components/myhome/cover_strategies/__init__.py create mode 100644 custom_components/myhome/cover_strategies/base.py create mode 100644 custom_components/myhome/cover_strategies/native.py create mode 100644 custom_components/myhome/cover_strategies/simple.py create mode 100644 custom_components/myhome/cover_strategies/timed.py create mode 100644 custom_components/myhome/cover_strategies/venetian.py create mode 100644 tests/test_cover_strategies.py diff --git a/custom_components/myhome/const.py b/custom_components/myhome/const.py index 2e9934bf9..1fe5a4ec1 100644 --- a/custom_components/myhome/const.py +++ b/custom_components/myhome/const.py @@ -95,6 +95,8 @@ def get_ownd_version() -> str: CONF_ADVANCED_SHUTTER = "advanced" CONF_SLAT_TILT = "slat_tilt" CONF_SLAT_TIME = "slat_time" +CONF_STRATEGY = "strategy" +COVER_STRATEGIES = ("native", "timed", "venetian", "simple") DEFAULT_SLAT_TIME = 2.0 MIN_SLAT_TIME = 0.5 MAX_SLAT_TIME = 10.0 diff --git a/custom_components/myhome/cover.py b/custom_components/myhome/cover.py index 974172e6c..89a11f2bc 100644 --- a/custom_components/myhome/cover.py +++ b/custom_components/myhome/cover.py @@ -76,6 +76,7 @@ CONF_SLAT_TILT, CONF_SLAT_TIME, CONF_STATE_SENSOR, + CONF_STRATEGY, CONF_TRAVEL_TIME, CONF_WHO, DEFAULT_MIN_CYCLE_TIME, @@ -110,7 +111,6 @@ from .cover_motion import ( ECHO_WINDOW, END_SETTLE_MARGIN, - MIN_POSITION_DELTA, MOTOR_START_DELAY, MOTOR_STOP_DELAY, WRITE_TIMEOUT, @@ -118,9 +118,20 @@ compute_interpolated_position, compute_run_duration, is_in_echo_window, - quantize_position_delta, travel_for, ) +from .cover_motion import ( + MIN_POSITION_DELTA as MIN_POSITION_DELTA, +) +from .cover_strategies import ( + CoverKinematicsStrategy, + NativePositionStrategy, + SimpleUpDownStopStrategy, + TimedCurtainStrategy, + VenetianTiltStrategy, + clamp_slat_time, + resolve_cover_strategy, +) from .discovery import ( Address, DeviceContext, @@ -141,14 +152,7 @@ OFF_RETRY_DELAY = 1.0 OFF_SETTLE_DELAY = 0.2 - -def _clamp_slat_time(value: Any, fallback: float = DEFAULT_SLAT_TIME) -> float: - """A slat time within MIN_SLAT_TIME..MAX_SLAT_TIME; ``fallback`` when the value is unusable.""" - try: - seconds = float(value) if value is not None else float(fallback) - except (TypeError, ValueError): - seconds = float(fallback) - return max(MIN_SLAT_TIME, min(MAX_SLAT_TIME, seconds)) +_clamp_slat_time = clamp_slat_time async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry, async_add_entities: AddEntitiesCallback) -> None: @@ -209,6 +213,7 @@ def build(ctx: DeviceContext) -> MyHOMECover: advanced=cfg.get(CONF_ADVANCED_SHUTTER, cfg.get("advanced_shutter", False)), slat_tilt=bool(cfg.get(CONF_SLAT_TILT, False)), slat_time=_clamp_slat_time(cfg.get(CONF_SLAT_TIME)), + strategy_name=cfg.get(CONF_STRATEGY), manufacturer=cfg.get(CONF_MANUFACTURER) or "BTicino", model=cfg.get(CONF_DEVICE_MODEL) or "Shutter / Cover", gateway=runtime.gateway, @@ -341,6 +346,7 @@ def __init__( # type: ignore min_motor_pulse: float = DEFAULT_MIN_MOTOR_PULSE, slat_tilt: bool = False, slat_time: float = DEFAULT_SLAT_TIME, + strategy_name: str | None = None, ): super().__init__( hass=hass, @@ -361,8 +367,8 @@ def __init__( # type: ignore self.scope: CoverScope | None = None self._family: CoverFamily | None = None self._advanced = advanced - self._slat_tilt = slat_tilt - self._slat_time = _clamp_slat_time(slat_time) + self._slat_tilt = bool(slat_tilt) + raw_slat_time = _clamp_slat_time(slat_time) self._is_tilting: bool = False self._tilt_start_time: float | None = None self._tilt_initial_position: int = 0 @@ -383,24 +389,24 @@ def __init__( # type: ignore # for compatibility; a stored calibration overrides yaml / default values. base_travel = float(travel_time) if travel_time else float(DEFAULT_TRAVEL_TIME) self._travel_time = travel_time if travel_time else DEFAULT_TRAVEL_TIME - self._travel_time_down = base_travel - self._travel_time_up = base_travel - self._min_motor_pulse = float(min_motor_pulse) + down = base_travel + up = base_travel + raw_min_pulse = float(min_motor_pulse) self._calibration_source = travel_time_source self._calibrated_at: str | None = None self._copied_from: str | None = None if calibration: if calibration.get("slat_time") is not None: - self._slat_time = _clamp_slat_time(calibration["slat_time"], self._slat_time) + raw_slat_time = _clamp_slat_time(calibration["slat_time"], raw_slat_time) if calibration.get("down") and calibration.get("up"): - self._travel_time_down = float(calibration["down"]) - self._travel_time_up = float(calibration["up"]) - self._travel_time = int(round(self._travel_time_down)) + down = float(calibration["down"]) + up = float(calibration["up"]) + self._travel_time = int(round(down)) self._calibration_source = str(calibration.get("source") or "measured") self._calibrated_at = calibration.get("measured_at") self._copied_from = calibration.get("copied_from") or None if calibration.get("min_motor_pulse") is not None: - self._min_motor_pulse = float(calibration["min_motor_pulse"]) + raw_min_pulse = float(calibration["min_motor_pulse"]) self._calibrating = False self._calibration_interrupted: str | None = None self._stopped_event: asyncio.Event = asyncio.Event() @@ -408,13 +414,28 @@ def __init__( # type: ignore self._last_run: dict[str, Any] = {} # the last completed run as measured, see _freeze_position self._motion_started_at: str | None = None - # Both advanced and standard covers support SET_POSITION (standard via travel time estimation) - self._attr_supported_features = ( - CoverEntityFeature.OPEN - | CoverEntityFeature.CLOSE - | CoverEntityFeature.STOP - | CoverEntityFeature.SET_POSITION + effective_travel_time = ( + down + if (calibration and calibration.get("down") and calibration.get("up")) + else travel_time + ) + self._strategy: CoverKinematicsStrategy = resolve_cover_strategy( + strategy_name=strategy_name, + advanced=self._advanced, + travel_time=effective_travel_time, + travel_time_up=up, + travel_time_down=down, + slat_tilt=self._slat_tilt, + slat_time=raw_slat_time, + min_motor_pulse=raw_min_pulse, ) + if isinstance(self._strategy, NativePositionStrategy): + self._advanced = True + if isinstance(self._strategy, VenetianTiltStrategy): + self._slat_tilt = True + + self._attr_supported_features = self._strategy.supported_features + self._attr_assumed_state = self._strategy.assumed_state self._gateway_handler = gateway self._attr_extra_state_attributes = { @@ -428,7 +449,7 @@ def __init__( # type: ignore if not self._advanced: self._refresh_travel_attributes() - self._attr_current_cover_position = 50 + self._attr_current_cover_position = None if self._strategy.assumed_state else 50 self._attr_is_opening = False self._attr_is_closing = False self._attr_is_closed = False @@ -454,6 +475,47 @@ def __init__( # type: ignore self._pending_stop_context: dict[str, Any] | None = None self._auto_stop_duration: float | None = None + @property + def strategy(self) -> CoverKinematicsStrategy: + """Return the active cover kinematics strategy.""" + return self._strategy + + @property + def _travel_time_down(self) -> float: + """Closing travel time in seconds, owned by the active strategy.""" + return self._strategy.travel_time_down + + @_travel_time_down.setter + def _travel_time_down(self, value: float) -> None: + self._strategy.travel_time_down = float(value) + + @property + def _travel_time_up(self) -> float: + """Opening travel time in seconds, owned by the active strategy.""" + return self._strategy.travel_time_up + + @_travel_time_up.setter + def _travel_time_up(self, value: float) -> None: + self._strategy.travel_time_up = float(value) + + @property + def _min_motor_pulse(self) -> float: + """Minimum motor pulse floor in seconds, owned by the active strategy.""" + return self._strategy.min_motor_pulse + + @_min_motor_pulse.setter + def _min_motor_pulse(self, value: float) -> None: + self._strategy.min_motor_pulse = float(value) + + @property + def _slat_time(self) -> float: + """Slat rotation time in seconds, owned by the active strategy.""" + return self._strategy.slat_time + + @_slat_time.setter + def _slat_time(self, value: float) -> None: + self._strategy.slat_time = float(value) + @property def calibration_hub(self) -> CoverCalibrationHub: """Return the CoverCalibrationHub for this cover's gateway.""" @@ -463,14 +525,11 @@ def calibration_hub(self) -> CoverCalibrationHub: def _travel_for(self, opening: bool) -> float: """Full-travel seconds for the given direction (motors are often slower going up).""" - return travel_for(self._travel_time_up, self._travel_time_down, opening) + return self._strategy.travel_for_direction(opening) def _curtain_travel_for(self, opening: bool) -> float: """Curtain-only travel seconds for the given direction (end-to-end travel minus slat rotation).""" - travel = self._travel_for(opening) - if self._slat_tilt and travel > self._slat_time: - return travel - self._slat_time - return travel + return self._strategy.curtain_travel_for(opening) def _refresh_travel_attributes(self) -> None: attrs = self._attr_extra_state_attributes @@ -582,11 +641,7 @@ def _slat_lead_time(self, is_opening: bool, is_closing: bool) -> float: """Seconds a curtain run spends turning the slats before the curtain moves.""" if not self._slat_tilt or self._is_tilting or not (is_opening or is_closing): return 0.0 - curr = self._attr_current_cover_tilt_position - if curr is None: - curr = 0 if is_opening else 100 - remaining = (100 - curr) if is_opening else curr - return max(0.0, remaining / 100.0 * self._slat_time) + return self._strategy.lead_time_for(is_opening, entity=self) def _turn_slats_for_run( self, at: float, run_started_at: float | None, was_opening: bool, was_closing: bool @@ -595,13 +650,10 @@ def _turn_slats_for_run( if not self._slat_tilt or not (was_opening or was_closing): return run_duration = max(0.0, at - run_started_at) if run_started_at is not None else (self._slat_time + 1.0) - turned = (run_duration / self._slat_time) * 100.0 if self._slat_time > 0 else 100.0 - if was_opening: - curr = self._attr_current_cover_tilt_position if self._attr_current_cover_tilt_position is not None else 0 - self._attr_current_cover_tilt_position = min(100, int(round(curr + turned))) - else: - curr = self._attr_current_cover_tilt_position if self._attr_current_cover_tilt_position is not None else 100 - self._attr_current_cover_tilt_position = max(0, int(round(curr - turned))) + direction = "open" if was_opening else "close" + self._attr_current_cover_tilt_position = self._strategy.turn_slats_after( + self._attr_current_cover_tilt_position, run_duration, direction + ) def _freeze_position(self, at: float) -> None: """Turn the running estimate into a fixed position as of ``at``.""" @@ -619,18 +671,31 @@ def _freeze_position(self, at: float) -> None: self._motion_started_at = None self._refresh_travel_attributes() curtain_at = max(self._move_start_time, at - lead) if self._move_start_time is not None else at - frozen = compute_freeze_position( - self._start_position, - self._move_start_time, - curtain_at, - self._curtain_travel_for(is_opening), - is_opening, - is_closing, - self._attr_current_cover_position, - ) + if self._strategy.assumed_state: + frozen: int | None = None + elif self._move_start_time is not None and (is_opening or is_closing): + frozen_pos = self._strategy.position_after( + self._start_position, + max(0.0, at - self._move_start_time), + "open" if is_opening else "close", + entity=self, + lead=lead, + ) + frozen = frozen_pos if frozen_pos is not None else self._start_position + else: + frozen = compute_freeze_position( + self._start_position, + self._move_start_time, + curtain_at, + self._curtain_travel_for(is_opening), + is_opening, + is_closing, + self._attr_current_cover_position, + ) self._attr_current_cover_position = frozen if self._move_start_time is not None: - self._start_position = frozen + if frozen is not None: + self._start_position = frozen self._move_start_time = None self._turn_slats_for_run(at, run_started_at, is_opening, is_closing) self._attr_is_opening = False @@ -738,26 +803,19 @@ def _cancel_stop_task(self) -> None: @property def current_cover_position(self) -> int | None: """Return current cover position (interpolated if moving).""" - if not self._advanced: - is_opening = bool(self._attr_is_opening) - is_closing = bool(self._attr_is_closing) - now = time.monotonic() - lead = self._slat_lead_time(is_opening, is_closing) - curtain_at = max(self._move_start_time, now - lead) if self._move_start_time is not None else now - return compute_interpolated_position( - self._attr_current_cover_position, - self._start_position, - self._move_start_time, - curtain_at, - self._curtain_travel_for(is_opening), - is_opening, - is_closing, - ) - return self._attr_current_cover_position + return self._strategy.current_position(self, time.monotonic()) def _enable_slat_tilt(self) -> None: """Enable Venetian blind slat / louvre tilt capabilities.""" self._slat_tilt = True + if isinstance(self._strategy, TimedCurtainStrategy) and not isinstance(self._strategy, VenetianTiltStrategy): + self._strategy = VenetianTiltStrategy( + slat_time=self._strategy.slat_time, + travel_time=self._strategy.travel_time_down, + travel_time_up=self._strategy.travel_time_up, + travel_time_down=self._strategy.travel_time_down, + min_motor_pulse=self._strategy.min_motor_pulse, + ) self._attr_device_class = CoverDeviceClass.BLIND current_features = self._attr_supported_features or CoverEntityFeature(0) self._attr_supported_features = ( @@ -806,7 +864,12 @@ async def async_added_to_hass(self) -> None: async def async_restore_last_state(self, last_state: Any) -> None: """Restore cover position and closure state.""" - if not self._advanced: + if self._strategy.assumed_state: + if last_state.state in (STATE_CLOSED, "closed"): + self._attr_is_closed = True + elif last_state.state in (STATE_OPEN, "open"): + self._attr_is_closed = False + elif not self._advanced: restored = False last_pos = last_state.attributes.get(ATTR_CURRENT_POSITION) if last_pos is not None: @@ -986,6 +1049,7 @@ async def async_calibrate(self) -> dict: # type: ignore translation_placeholders={"name": self._display_name, "direction": label, "seconds": f"{value:.1f}"}, ) + self._promote_to_timed_strategy() self._travel_time_down = round(down, 2) self._travel_time_up = round(up, 2) self._travel_time = int(round(down)) @@ -1017,6 +1081,33 @@ async def async_calibrate(self) -> dict: # type: ignore ) return result + def _promote_to_timed_strategy(self) -> None: + """Promote a SimpleUpDownStopStrategy cover to Timed/Venetian once calibrated.""" + if isinstance(self._strategy, SimpleUpDownStopStrategy): + if self._slat_tilt: + self._strategy = VenetianTiltStrategy( + slat_time=self._strategy.slat_time, + travel_time=self._strategy.travel_time_down, + travel_time_up=self._strategy.travel_time_up, + travel_time_down=self._strategy.travel_time_down, + min_motor_pulse=self._strategy.min_motor_pulse, + ) + else: + self._strategy = TimedCurtainStrategy( + travel_time=self._strategy.travel_time_down, + travel_time_up=self._strategy.travel_time_up, + travel_time_down=self._strategy.travel_time_down, + min_motor_pulse=self._strategy.min_motor_pulse, + slat_time=self._strategy.slat_time, + ) + self._attr_assumed_state = False + self._attr_supported_features = ( + (self._attr_supported_features or CoverEntityFeature(0)) + | CoverEntityFeature.SET_POSITION + ) + if self._attr_current_cover_position is None: + self._attr_current_cover_position = self._start_position + async def async_stop_calibration(self) -> None: """Stop calibration on this cover's gateway.""" gw_mac = getattr(self._gateway_handler, "mac", None) @@ -1092,6 +1183,7 @@ async def async_set_travel_time( }, ) + self._promote_to_timed_strategy() self._travel_time_down = round(float(down), 2) self._travel_time_up = round(float(up), 2) self._travel_time = int(round(self._travel_time_down)) @@ -1378,6 +1470,15 @@ async def async_set_cover_position(self, **kwargs: Any) -> None: return target_position = kwargs[ATTR_POSITION] + if self._strategy.assumed_state: + if target_position >= 100: + await self.async_open_cover() + elif target_position <= 0: + await self.async_close_cover() + else: + await self.async_stop_cover() + return + if self._advanced: # The level command takes the motor over: a running slat pulse must # not stop it with its own auto-stop. @@ -1394,7 +1495,7 @@ async def async_set_cover_position(self, **kwargs: Any) -> None: # a stale level and drop a retarget (40 -> 80, then 40). Also # skip the early return briefly after our own level write, # before the actuator's first moving status arrives. - target_position = quantize_position_delta(current, target_position, MIN_POSITION_DELTA) + target_position = self._strategy.quantize_target(current, target_position) if target_position == current and not self._in_echo_window(now): return # 0% and 100% are full runs: they end on the motor's limit switch @@ -1453,18 +1554,17 @@ async def async_set_cover_position(self, **kwargs: Any) -> None: self._end_pulse_for_takeover() self._end_curtain_for_takeover(time.monotonic()) opening = diff > 0 + direction = "open" if opening else "close" travel_time = self._curtain_travel_for(opening) lead = self._slat_lead_time(opening, not opening) - curtain_duration = compute_run_duration( - diff, travel_time, min_pulse=max(0.0, self._min_motor_pulse - lead) - ) - run_duration = lead + curtain_duration + run_duration = self._strategy.duration_for(curr_pos, target_position, direction, entity=self) # Where the model puts the cover when the run ends: the target, or # further when min_motor_pulse stretched the run. A run that reaches an # end becomes a full run to that end. - achieved_position = compute_freeze_position( - curr_pos, 0.0, curtain_duration, travel_time, opening, not opening, target_position + achieved_pos = self._strategy.position_after( + curr_pos, run_duration, direction, entity=self, lead=lead ) + achieved_position = achieved_pos if (achieved_pos is not None and travel_time > 0) else target_position if achieved_position >= 100: await self._async_run_to_end(True) return @@ -1472,7 +1572,7 @@ async def async_set_cover_position(self, **kwargs: Any) -> None: await self._async_run_to_end(False) return - written = await self._async_move("open" if opening else "close") + written = await self._async_move(direction) generation = self._run_generation async def _auto_stop() -> None: @@ -1582,8 +1682,12 @@ def _interpolated_tilt(self, at: float) -> int | None: """ if self._tilt_start_time is None or self._slat_time <= 0: return self.current_cover_tilt_position - turned = max(0.0, at - self._tilt_start_time) / self._slat_time * 100.0 - return max(0, min(100, int(round(self._tilt_initial_position + self._tilt_step() * turned)))) + direction = "open" if self._tilt_step() > 0 else "close" + return self._strategy.turn_slats_after( + self._tilt_initial_position, + max(0.0, at - self._tilt_start_time), + direction, + ) def _tilt_step(self) -> int: """+1 while the running pulse opens the slats, -1 while it closes them.""" @@ -1597,19 +1701,21 @@ def _carry_overshoot_into_curtain(self, at: float) -> None: """ if self._advanced or self._tilt_start_time is None or self._attr_current_cover_position is None: return - opening = self._tilt_step() > 0 - to_end = (100 - self._tilt_initial_position if opening else self._tilt_initial_position) / 100.0 * self._slat_time - excess = (at - self._tilt_start_time) - to_end - if excess <= 0: - return - moved = excess / self._curtain_travel_for(opening) * 100.0 - position = max(0, min(100, int(round(self._attr_current_cover_position + (moved if opening else -moved))))) - LOGGER.debug( - "%s Slat pulse of %s ran %.2f s past a full rotation: curtain %s -> %s.", - self._gateway_handler.log_id, self._full_where, excess, self._attr_current_cover_position, position, + direction = "open" if self._tilt_step() > 0 else "close" + elapsed = max(0.0, at - self._tilt_start_time) + position = self._strategy.carry_overshoot_into_curtain( + self._attr_current_cover_position, + self._tilt_initial_position, + elapsed, + direction, ) - self._attr_current_cover_position = self._start_position = position - self._attr_is_closed = position == 0 + if position != self._attr_current_cover_position: + LOGGER.debug( + "%s Slat pulse of %s ran past a full rotation: curtain %s -> %s.", + self._gateway_handler.log_id, self._full_where, self._attr_current_cover_position, position, + ) + self._attr_current_cover_position = self._start_position = position + self._attr_is_closed = position == 0 def _abort_tilt(self) -> None: """Forget a tilt pulse that never reached the bus; the tilt angle stays as it was.""" @@ -1745,8 +1851,8 @@ async def _async_start_tilt(self, target_tilt: int) -> None: if curr_tilt is None: curr_tilt = 0 if self.is_closed else 50 diff = target_tilt - curr_tilt - run_duration = abs(diff) / 100.0 * self._slat_time - if diff == 0 or run_duration < self._min_motor_pulse: + run_duration = self._strategy.tilt_pulse_duration_for(curr_tilt, target_tilt) + if run_duration <= 0: # Nothing to do, or a pulse too short for the motor to answer. if was_tilting: # Already there while the motor still runs: end the pulse here. @@ -1768,7 +1874,9 @@ async def _async_start_tilt(self, target_tilt: int) -> None: # move the pulse's stop and send nothing. self._cancel_stop_task() self._tilt_target_position = target_tilt - self._tilt_duration = abs(target_tilt - self._tilt_initial_position) / 100.0 * self._slat_time + self._tilt_duration = self._strategy.tilt_pulse_duration_for( + self._tilt_initial_position, target_tilt, enforce_min_pulse=False + ) self._start_pulse_auto_stop(self._pulse_written, self._run_generation, self._motor_started) if self.hass is not None: self.async_write_ha_state() @@ -1924,46 +2032,7 @@ def _handle_echo(self, message: OWNMessage, now: float) -> bool: self._last_stop_at = now self._stopped_event.set() if ctx and ctx.get("run_started_at") is not None and ctx.get("run_start_position") is not None: - start_anchor = ctx["run_started_at"] - run_start_pos = ctx["run_start_position"] - is_opening = ctx["is_opening"] - is_closing = ctx["is_closing"] - auto_dur = ctx.get("auto_stop_duration") - lead = float(ctx.get("lead") or 0.0) - raw_elapsed = max(0.0, now - start_anchor) - elapsed = max(raw_elapsed, auto_dur) if auto_dur is not None else raw_elapsed - curtain_elapsed = max(0.0, elapsed - lead) - travel_time = self._curtain_travel_for(is_opening) - if travel_time > 0 and (is_opening or is_closing): - delta = (curtain_elapsed / travel_time) * 100.0 - if is_opening: - reanchored = min(100, int(round(run_start_pos + delta))) - else: - reanchored = max(0, int(round(run_start_pos - delta))) - self._attr_current_cover_position = reanchored - self._start_position = reanchored - self._attr_is_closed = (reanchored == 0) - if self._slat_tilt and (is_opening or is_closing): - raw_tilt = ctx.get("run_start_tilt") - start_tilt = raw_tilt if raw_tilt is not None else (0 if is_opening else 100) - turned = (elapsed / self._slat_time) * 100.0 if self._slat_time > 0 else 100.0 - if is_opening: - self._attr_current_cover_tilt_position = min(100, int(round(start_tilt + turned))) - else: - self._attr_current_cover_tilt_position = max(0, int(round(start_tilt - turned))) - self._last_run = { - "seconds": round(elapsed, 2), - "direction": "open" if is_opening else "close" if is_closing else None, - "ended_at": dt_util.utcnow().isoformat(timespec="milliseconds"), - } - self._refresh_travel_attributes() - LOGGER.debug( - "%s Stop status echo for %s; re-anchored to %s%% (elapsed %.2f s).", - self._gateway_handler.log_id, - self._full_where, - self._attr_current_cover_position, - elapsed, - ) + self._strategy.reanchor_on_stop_echo(self, ctx, now) return True # Opposite or external command during stop echo window: end window and handle externally. self._end_echo_window() @@ -2627,6 +2696,7 @@ def __dir__() -> list[str]: "CalibrationInterrupted", "CoverCalibrationHub", "CoverFamily", + "CoverKinematicsStrategy", "CoverScope", "ECHO_WINDOW", "MOTOR_START_DELAY", @@ -2645,8 +2715,10 @@ def __dir__() -> list[str]: "async_unload_entry", "compute_freeze_position", "compute_interpolated_position", + "compute_run_duration", "get_calibration_hub", "get_last_calibration_trace", "is_in_echo_window", + "resolve_cover_strategy", "travel_for", ] diff --git a/custom_components/myhome/cover_strategies/__init__.py b/custom_components/myhome/cover_strategies/__init__.py new file mode 100644 index 000000000..d1aca7c54 --- /dev/null +++ b/custom_components/myhome/cover_strategies/__init__.py @@ -0,0 +1,124 @@ +"""Pluggable kinematics strategies for MyHOME WHO=2 covers.""" +from __future__ import annotations + +from ..const import DEFAULT_MIN_MOTOR_PULSE, DEFAULT_TRAVEL_TIME +from .base import BaseCoverStrategy, CoverKinematicsStrategy, clamp_slat_time +from .native import NativePositionStrategy +from .simple import SimpleUpDownStopStrategy +from .timed import TimedCurtainStrategy +from .venetian import MIN_TILT_PULSE, VenetianTiltStrategy + +__all__ = [ + "BaseCoverStrategy", + "CoverKinematicsStrategy", + "MIN_TILT_PULSE", + "NativePositionStrategy", + "SimpleUpDownStopStrategy", + "TimedCurtainStrategy", + "VenetianTiltStrategy", + "clamp_slat_time", + "resolve_cover_strategy", +] + + +def resolve_cover_strategy( + *, + advanced: bool = False, + strategy_name: str | None = None, + slat_tilt: bool = False, + slat_time: float | None = None, + travel_time: float | None = DEFAULT_TRAVEL_TIME, + travel_time_up: float | None = None, + travel_time_down: float | None = None, + min_motor_pulse: float = DEFAULT_MIN_MOTOR_PULSE, +) -> CoverKinematicsStrategy: + """Resolve the appropriate ``CoverKinematicsStrategy`` for a cover entity. + + Resolution precedence: + 1. Explicit ``strategy_name`` (``"native"``, ``"simple"``, ``"venetian"``, + ``"timed"``). + 2. ``advanced=True`` -> ``NativePositionStrategy`` (Dimension 10/11 hardware level). + 3. ``slat_tilt=True`` -> ``VenetianTiltStrategy``. + 4. ``travel_time is None`` or ``travel_time <= 0`` -> ``SimpleUpDownStopStrategy`` + (unpositioned UP/DOWN/STOP relay without synthetic 50% startup position). + 5. Default when ``travel_time > 0`` (including ``DEFAULT_TRAVEL_TIME = 25``) -> + ``TimedCurtainStrategy`` so existing WHO=2 installations retain percentage + positioning and interactive UI stopwatch calibration without breaking changes. + """ + try: + effective_travel = ( + float(travel_time) + if travel_time is not None and float(travel_time) > 0 + else float(DEFAULT_TRAVEL_TIME) + ) + has_positive_travel = travel_time is not None and float(travel_time) > 0 + except (TypeError, ValueError): + effective_travel = float(DEFAULT_TRAVEL_TIME) + has_positive_travel = False + + if strategy_name is not None: + normalized = strategy_name.strip().lower() + if normalized == "native": + return NativePositionStrategy( + travel_time=effective_travel, + travel_time_up=travel_time_up, + travel_time_down=travel_time_down, + min_motor_pulse=min_motor_pulse, + slat_time=slat_time, + ) + if normalized == "simple": + return SimpleUpDownStopStrategy( + travel_time=effective_travel, + travel_time_up=travel_time_up, + travel_time_down=travel_time_down, + min_motor_pulse=min_motor_pulse, + slat_time=slat_time, + ) + if normalized == "venetian": + return VenetianTiltStrategy( + slat_time=slat_time, + travel_time=effective_travel, + travel_time_up=travel_time_up, + travel_time_down=travel_time_down, + min_motor_pulse=min_motor_pulse, + ) + if normalized == "timed": + return TimedCurtainStrategy( + travel_time=effective_travel, + travel_time_up=travel_time_up, + travel_time_down=travel_time_down, + min_motor_pulse=min_motor_pulse, + slat_time=slat_time, + ) + + if advanced: + return NativePositionStrategy( + travel_time=effective_travel, + travel_time_up=travel_time_up, + travel_time_down=travel_time_down, + min_motor_pulse=min_motor_pulse, + slat_time=slat_time, + ) + if slat_tilt: + return VenetianTiltStrategy( + slat_time=slat_time, + travel_time=effective_travel, + travel_time_up=travel_time_up, + travel_time_down=travel_time_down, + min_motor_pulse=min_motor_pulse, + ) + if not has_positive_travel: + return SimpleUpDownStopStrategy( + travel_time=effective_travel, + travel_time_up=travel_time_up, + travel_time_down=travel_time_down, + min_motor_pulse=min_motor_pulse, + slat_time=slat_time, + ) + return TimedCurtainStrategy( + travel_time=effective_travel, + travel_time_up=travel_time_up, + travel_time_down=travel_time_down, + min_motor_pulse=min_motor_pulse, + slat_time=slat_time, + ) diff --git a/custom_components/myhome/cover_strategies/base.py b/custom_components/myhome/cover_strategies/base.py new file mode 100644 index 000000000..abd7fda87 --- /dev/null +++ b/custom_components/myhome/cover_strategies/base.py @@ -0,0 +1,222 @@ +"""Base protocol and shared timing state for pluggable MyHOME cover kinematics strategies.""" +from __future__ import annotations + +from typing import Any, Protocol, runtime_checkable + +from homeassistant.components.cover import CoverEntityFeature + +from ..const import ( + DEFAULT_MIN_MOTOR_PULSE, + DEFAULT_SLAT_TIME, + DEFAULT_TRAVEL_TIME, + MAX_SLAT_TIME, + MIN_SLAT_TIME, +) +from ..cover_motion import travel_for + + +def clamp_slat_time(value: Any, fallback: float = DEFAULT_SLAT_TIME) -> float: + """Return a slat rotation time clamped to [MIN_SLAT_TIME, MAX_SLAT_TIME].""" + try: + seconds = float(value) if value is not None else float(fallback) + except (TypeError, ValueError): + seconds = float(fallback) + return max(MIN_SLAT_TIME, min(MAX_SLAT_TIME, seconds)) + + +@runtime_checkable +class CoverKinematicsStrategy(Protocol): + """Protocol implemented by all pluggable WHO=2 cover kinematics strategies. + + Strategies own timing parameters (``travel_time_down``, ``travel_time_up``, + ``min_motor_pulse``, ``slat_time``), the time<->position mapping via + ``position_after`` and ``duration_for``, slat rotation kinematics, and + stop-echo re-anchoring via ``reanchor_on_stop_echo``, allowing linear timed + covers, Venetian slat+roll blinds, and non-linear calibrated geometry + strategies (#374) to plug into ``MyHOMECover`` without modifying the entity class. + """ + + travel_time_down: float + travel_time_up: float + min_motor_pulse: float + slat_time: float + + @property + def name(self) -> str: + """Return strategy identifier.""" + + @property + def supported_features(self) -> CoverEntityFeature: + """Return Home Assistant CoverEntityFeature bitmask for this strategy.""" + + @property + def assumed_state(self) -> bool: + """Return whether cover position/state is assumed rather than bus-reported or calibrated.""" + + def travel_for_direction(self, opening: bool) -> float: + """Return full end-to-end travel seconds for the given direction.""" + + def curtain_travel_for(self, opening: bool) -> float: + """Return vertical curtain travel seconds for the given direction.""" + + def lead_time_for(self, opening: bool, *, entity: Any = None) -> float: + """Return seconds spent rotating slats before vertical curtain motion begins.""" + + def min_pulse_for(self) -> float: + """Return minimum motor pulse floor in seconds.""" + + def slat_time_for(self) -> float: + """Return full slat rotation time in seconds.""" + + def quantize_target(self, current: int | None, target: int) -> int: + """Quantize a target position against the strategy's deadband if applicable.""" + + def turn_slats_after( + self, + start_tilt: int | None, + elapsed: float, + direction: str, + ) -> int: + """Return clamped [0, 100] slat tilt angle after ``elapsed`` seconds in ``direction``.""" + + def carry_overshoot_into_curtain( + self, + start_position: int, + start_tilt: int, + elapsed: float, + direction: str, + ) -> int: + """Convert motor runtime past a full slat rotation into vertical curtain displacement.""" + + def tilt_pulse_duration_for( + self, + start_tilt: int, + target_tilt: int, + *, + enforce_min_pulse: bool = True, + ) -> float: + """Compute motor pulse seconds to rotate slats from ``start_tilt`` to ``target_tilt``.""" + + def position_after( + self, + start: int, + elapsed: float, + direction: str, + *, + entity: Any = None, + lead: float | None = None, + ) -> int | None: + """Return cover position (0..100 or None) after ``elapsed`` seconds in ``direction`` from ``start``.""" + + def duration_for( + self, + start: int, + target: int, + direction: str | None = None, + *, + entity: Any = None, + ) -> float: + """Return motor run duration in seconds required to move from ``start`` to ``target``.""" + + def reanchor_on_stop_echo( + self, + entity: Any, + ctx: dict[str, Any], + now: float, + ) -> int | None: + """Re-anchor position (and slat tilt if applicable) on a relayed stop status echo (#705).""" + + def current_position(self, entity: Any, now: float) -> int | None: + """Return current cover position (0..100) or None if position is unknown.""" + + +class BaseCoverStrategy: + """Shared timing configuration and default slat helpers for WHO=2 cover strategies.""" + + def __init__( + self, + travel_time: float = DEFAULT_TRAVEL_TIME, + *, + travel_time_up: float | None = None, + travel_time_down: float | None = None, + min_motor_pulse: float = DEFAULT_MIN_MOTOR_PULSE, + slat_time: float | None = None, + ) -> None: + """Initialize timing parameters owned by the strategy.""" + base = float(travel_time) if travel_time and travel_time > 0 else float(DEFAULT_TRAVEL_TIME) + self.travel_time_down: float = ( + float(travel_time_down) if travel_time_down is not None and travel_time_down > 0 else base + ) + self.travel_time_up: float = ( + float(travel_time_up) if travel_time_up is not None and travel_time_up > 0 else self.travel_time_down + ) + self.min_motor_pulse: float = max(0.0, float(min_motor_pulse)) + self.slat_time: float = clamp_slat_time(slat_time, DEFAULT_SLAT_TIME) + + def travel_for_direction(self, opening: bool) -> float: + """Return full end-to-end travel seconds for the given direction.""" + return travel_for(self.travel_time_up, self.travel_time_down, opening) + + def curtain_travel_for(self, opening: bool) -> float: + """Return vertical curtain travel seconds for the given direction.""" + return self.travel_for_direction(opening) + + def lead_time_for(self, opening: bool, *, entity: Any = None) -> float: # pylint: disable=unused-argument + """Return seconds spent rotating slats before vertical curtain motion begins.""" + return 0.0 + + def min_pulse_for(self) -> float: + """Return minimum motor pulse floor in seconds (#704, #708).""" + return self.min_motor_pulse + + def slat_time_for(self) -> float: + """Return the effective slat full-rotation time in seconds.""" + return self.slat_time + + def quantize_target(self, current: int | None, target: int) -> int: # pylint: disable=unused-argument + """Return target unchanged by default (overridden by NativePositionStrategy).""" + return int(target) + + def turn_slats_after( + self, + start_tilt: int | None, + elapsed: float, + direction: str, + ) -> int: + """Return clamped [0, 100] slat tilt angle after ``elapsed`` seconds in ``direction``.""" + opening = direction == "open" + curr = int(start_tilt) if start_tilt is not None else (0 if opening else 100) + slat = self.slat_time_for() + turned = (max(0.0, float(elapsed)) / slat) * 100.0 if slat > 0 else 100.0 + if opening: + return min(100, int(round(curr + turned))) + return max(0, int(round(curr - turned))) + + def carry_overshoot_into_curtain( + self, + start_position: int, + start_tilt: int, + elapsed: float, + direction: str, + ) -> int: # pylint: disable=unused-argument + """Return ``start_position`` unchanged by default (overridden by VenetianTiltStrategy).""" + return max(0, min(100, int(round(start_position)))) + + def tilt_pulse_duration_for( + self, + start_tilt: int, + target_tilt: int, + *, + enforce_min_pulse: bool = True, + ) -> float: + """Compute motor pulse seconds to rotate slats from ``start_tilt`` to ``target_tilt``.""" + clamped_start = max(0, min(100, int(round(start_tilt)))) + clamped_target = max(0, min(100, int(round(target_tilt)))) + diff = clamped_target - clamped_start + if diff == 0: + return 0.0 + run_duration = (abs(diff) / 100.0) * self.slat_time_for() + if enforce_min_pulse and run_duration < self.min_pulse_for(): + return 0.0 + return run_duration + diff --git a/custom_components/myhome/cover_strategies/native.py b/custom_components/myhome/cover_strategies/native.py new file mode 100644 index 000000000..6f2fd5120 --- /dev/null +++ b/custom_components/myhome/cover_strategies/native.py @@ -0,0 +1,100 @@ +"""Native hardware position strategy for advanced WHO=2 shutters (Dimension 10/11).""" +from __future__ import annotations + +from typing import Any + +from homeassistant.components.cover import CoverEntityFeature + +from ..const import DEFAULT_MIN_MOTOR_PULSE, DEFAULT_TRAVEL_TIME +from ..cover_motion import MIN_POSITION_DELTA, quantize_position_delta +from .base import BaseCoverStrategy + + +class NativePositionStrategy(BaseCoverStrategy): + """Hardware-calibrated shutter strategy using OpenWebNet Dimension 10/11.""" + + def __init__( + self, + *, + min_position_delta: int = MIN_POSITION_DELTA, + travel_time: float = DEFAULT_TRAVEL_TIME, + travel_time_up: float | None = None, + travel_time_down: float | None = None, + min_motor_pulse: float = DEFAULT_MIN_MOTOR_PULSE, + slat_time: float | None = None, + ) -> None: + """Initialize native position strategy with deadband quantum (#466).""" + super().__init__( + travel_time=travel_time, + travel_time_up=travel_time_up, + travel_time_down=travel_time_down, + min_motor_pulse=min_motor_pulse, + slat_time=slat_time, + ) + self.min_position_delta: int = max(1, int(min_position_delta)) + + @property + def name(self) -> str: + """Return strategy identifier.""" + return "native" + + @property + def supported_features(self) -> CoverEntityFeature: + """Return OPEN | CLOSE | STOP | SET_POSITION.""" + return ( + CoverEntityFeature.OPEN + | CoverEntityFeature.CLOSE + | CoverEntityFeature.STOP + | CoverEntityFeature.SET_POSITION + ) + + @property + def assumed_state(self) -> bool: + """Advanced WHO=2 actuators report authoritative bus position.""" + return False + + def quantize_target(self, current: int | None, target: int) -> int: + """Widen sub-deadband position deltas to ``min_position_delta`` (#466, #702).""" + return quantize_position_delta(current, target, self.min_position_delta) + + def position_after( + self, + start: int, + elapsed: float, + direction: str, + *, + entity: Any = None, + lead: float | None = None, + ) -> int | None: + """Native actuators report position on the bus rather than dead-reckoning.""" + if entity is not None: + pos: int | None = getattr(entity, "_attr_current_cover_position", None) + return pos + return max(0, min(100, int(round(start)))) + + def duration_for( + self, + start: int, + target: int, + direction: str | None = None, + *, + entity: Any = None, + ) -> float: + """Native actuators time their own runs in hardware.""" + return 0.0 + + def reanchor_on_stop_echo( + self, + entity: Any, + ctx: dict[str, Any], + now: float, + ) -> int | None: + """Native actuators rely on Dimension 10 status frames rather than stop-echo timing.""" + pos: int | None = getattr(entity, "_attr_current_cover_position", None) + return pos + + def current_position(self, entity: Any, now: float) -> int | None: + """Return position reported by the actuator.""" + pos: int | None = entity._attr_current_cover_position + return pos + diff --git a/custom_components/myhome/cover_strategies/simple.py b/custom_components/myhome/cover_strategies/simple.py new file mode 100644 index 000000000..3bac22dc6 --- /dev/null +++ b/custom_components/myhome/cover_strategies/simple.py @@ -0,0 +1,72 @@ +"""Simple up/down/stop cover strategy without synthetic stopwatch position.""" +from __future__ import annotations + +from typing import Any + +from homeassistant.components.cover import CoverEntityFeature + +from .base import BaseCoverStrategy + + +class SimpleUpDownStopStrategy(BaseCoverStrategy): + """Stateless UP / DOWN / STOP cover strategy without position estimation.""" + + @property + def name(self) -> str: + """Return strategy identifier.""" + return "simple" + + @property + def supported_features(self) -> CoverEntityFeature: + """Return OPEN | CLOSE | STOP (no synthetic position slider).""" + return ( + CoverEntityFeature.OPEN + | CoverEntityFeature.CLOSE + | CoverEntityFeature.STOP + ) + + @property + def assumed_state(self) -> bool: + """Simple UP/DOWN/STOP relays do not report or estimate intermediate position.""" + return True + + def travel_for_direction(self, opening: bool) -> float: # pylint: disable=unused-argument + """Simple UP/DOWN/STOP covers have no travel duration.""" + return 0.0 + + def position_after( + self, + start: int, + elapsed: float, + direction: str, + *, + entity: Any = None, + lead: float | None = None, + ) -> int | None: + """Simple covers do not estimate percentage position over time.""" + return None + + def duration_for( + self, + start: int, + target: int, + direction: str | None = None, + *, + entity: Any = None, + ) -> float: + """Simple covers do not run timed intermediate moves.""" + return 0.0 + + def reanchor_on_stop_echo( + self, + entity: Any, + ctx: dict[str, Any], + now: float, + ) -> int | None: + """Simple covers have no dead-reckoning anchor to update on stop echoes.""" + return None + + def current_position(self, entity: Any, now: float) -> int | None: + """Simple covers do not fabricate a percentage position.""" + return None + diff --git a/custom_components/myhome/cover_strategies/timed.py b/custom_components/myhome/cover_strategies/timed.py new file mode 100644 index 000000000..8b7486aed --- /dev/null +++ b/custom_components/myhome/cover_strategies/timed.py @@ -0,0 +1,192 @@ +"""Travel-time interpolated curtain strategy for calibrated standard WHO=2 covers.""" +from __future__ import annotations + +from typing import Any + +from homeassistant.components.cover import CoverEntityFeature +from homeassistant.util import dt as dt_util + +from ..const import DEFAULT_MIN_MOTOR_PULSE, DEFAULT_TRAVEL_TIME, LOGGER +from ..cover_motion import compute_run_duration +from .base import BaseCoverStrategy + + +class TimedCurtainStrategy(BaseCoverStrategy): + """Timed curtain kinematics strategy owning time<->position mapping, pulse floor, and stop-echo re-anchoring.""" + + def __init__( + self, + travel_time: float = DEFAULT_TRAVEL_TIME, + *, + travel_time_up: float | None = None, + travel_time_down: float | None = None, + min_motor_pulse: float = DEFAULT_MIN_MOTOR_PULSE, + slat_time: float | None = None, + ) -> None: + """Initialize timed curtain strategy with direction-aware travel times and motor floor.""" + super().__init__( + travel_time=travel_time, + travel_time_up=travel_time_up, + travel_time_down=travel_time_down, + min_motor_pulse=min_motor_pulse, + slat_time=slat_time, + ) + + @property + def name(self) -> str: + """Return strategy identifier.""" + return "timed" + + @property + def supported_features(self) -> CoverEntityFeature: + """Return OPEN | CLOSE | STOP | SET_POSITION.""" + return ( + CoverEntityFeature.OPEN + | CoverEntityFeature.CLOSE + | CoverEntityFeature.STOP + | CoverEntityFeature.SET_POSITION + ) + + @property + def assumed_state(self) -> bool: + """Timed covers expose an interpolated percentage position.""" + return False + + def position_after( + self, + start: int, + elapsed: float, + direction: str, + *, + entity: Any = None, + lead: float | None = None, + ) -> int | None: + """Compute clamped [0, 100] cover position after ``elapsed`` seconds in ``direction`` from ``start``.""" + clamped_start = max(0, min(100, int(round(start)))) + opening = direction == "open" + closing = direction == "close" + if not (opening or closing): + return clamped_start + travel_time = self.curtain_travel_for(opening) + if travel_time <= 0: + if entity is not None and isinstance(getattr(entity, "_attr_current_cover_position", None), int): + return int(entity._attr_current_cover_position) + return clamped_start + effective_lead = ( + self.lead_time_for(opening, entity=entity) + if lead is None + else max(0.0, float(lead)) + ) + curtain_elapsed = max(0.0, float(elapsed) - effective_lead) + delta = (curtain_elapsed / travel_time) * 100.0 + if opening: + return min(100, int(round(clamped_start + delta))) + return max(0, int(round(clamped_start - delta))) + + def duration_for( + self, + start: int, + target: int, + direction: str | None = None, + *, + entity: Any = None, + ) -> float: + """Compute motor run duration in seconds from ``start`` to ``target``, enforcing ``min_motor_pulse``.""" + clamped_start = max(0, min(100, int(round(start)))) + clamped_target = max(0, min(100, int(round(target)))) + diff = clamped_target - clamped_start + if diff == 0: + return 0.0 + opening = (direction == "open") if direction in ("open", "close") else (diff > 0) + if (opening and diff < 0) or (not opening and diff > 0): + return 0.0 + travel_time = self.curtain_travel_for(opening) + if travel_time <= 0: + return 0.0 + lead = self.lead_time_for(opening, entity=entity) + min_pulse = self.min_pulse_for() + curtain_duration = compute_run_duration( + diff, travel_time, min_pulse=max(0.0, min_pulse - lead) + ) + return lead + curtain_duration + + def _reanchor_slats_on_stop_echo( + self, + entity: Any, + ctx: dict[str, Any], + elapsed: float, + is_opening: bool, + is_closing: bool, + ) -> None: + """Hook for subclasses (e.g. VenetianTiltStrategy) to re-anchor slat tilt on stop echo.""" + + def reanchor_on_stop_echo( + self, + entity: Any, + ctx: dict[str, Any], + now: float, + ) -> int | None: + """Re-anchor cover position and travel telemetry on a relayed stop status echo (#705, #707).""" + if not ctx or ctx.get("run_started_at") is None or ctx.get("run_start_position") is None: + pos: int | None = getattr(entity, "_attr_current_cover_position", None) + return pos + start_anchor = float(ctx["run_started_at"]) + run_start_pos = int(ctx["run_start_position"]) + is_opening = bool(ctx.get("is_opening")) + is_closing = bool(ctx.get("is_closing")) + auto_dur = ctx.get("auto_stop_duration") + lead = float(ctx.get("lead") or 0.0) + raw_elapsed = max(0.0, now - start_anchor) + elapsed = max(raw_elapsed, float(auto_dur)) if auto_dur is not None else raw_elapsed + direction = "open" if is_opening else "close" if is_closing else "" + travel_time = self.curtain_travel_for(is_opening) + reanchored: int | None = getattr(entity, "_attr_current_cover_position", None) + if travel_time > 0 and (is_opening or is_closing): + reanchored = self.position_after( + run_start_pos, elapsed, direction, entity=entity, lead=lead + ) + entity._attr_current_cover_position = reanchored + entity._start_position = reanchored + entity._attr_is_closed = reanchored == 0 + self._reanchor_slats_on_stop_echo(entity, ctx, elapsed, is_opening, is_closing) + entity._last_run = { + "seconds": round(elapsed, 2), + "direction": "open" if is_opening else "close" if is_closing else None, + "ended_at": dt_util.utcnow().isoformat(timespec="milliseconds"), + } + if hasattr(entity, "_refresh_travel_attributes"): + entity._refresh_travel_attributes() + log_id = getattr(getattr(entity, "_gateway_handler", None), "log_id", "") + LOGGER.debug( + "%s Stop status echo for %s; re-anchored to %s%% (elapsed %.2f s).", + log_id, + getattr(entity, "_full_where", ""), + entity._attr_current_cover_position, + elapsed, + ) + return reanchored + + def current_position(self, entity: Any, now: float) -> int | None: + """Interpolate current position from start position, motor anchor, and strategy kinematics.""" + if getattr(entity, "_advanced", False) is True: + pos_adv: int | None = getattr(entity, "_attr_current_cover_position", None) + return pos_adv + is_opening = getattr(entity, "_attr_is_opening", False) is True + is_closing = getattr(entity, "_attr_is_closing", False) is True + move_start = getattr(entity, "_move_start_time", None) + if ( + isinstance(move_start, (int, float)) + and (is_opening or is_closing) + and self.curtain_travel_for(is_opening) > 0 + ): + direction = "open" if is_opening else "close" + elapsed = max(0.0, now - float(move_start)) + return self.position_after( + int(getattr(entity, "_start_position", 50)), + elapsed, + direction, + entity=entity, + ) + pos: int | None = getattr(entity, "_attr_current_cover_position", None) + return pos + diff --git a/custom_components/myhome/cover_strategies/venetian.py b/custom_components/myhome/cover_strategies/venetian.py new file mode 100644 index 000000000..37574bf5a --- /dev/null +++ b/custom_components/myhome/cover_strategies/venetian.py @@ -0,0 +1,126 @@ +"""Venetian blind slat-tilt and curtain strategy for WHO=2 shutters (#492, #508).""" +from __future__ import annotations + +from typing import Any + +from homeassistant.components.cover import CoverEntityFeature + +from ..const import ( + DEFAULT_MIN_MOTOR_PULSE, + DEFAULT_TRAVEL_TIME, +) +from .base import clamp_slat_time +from .timed import TimedCurtainStrategy + +MIN_TILT_PULSE: float = DEFAULT_MIN_MOTOR_PULSE + +__all__ = [ + "MIN_TILT_PULSE", + "VenetianTiltStrategy", + "clamp_slat_time", +] + + +class VenetianTiltStrategy(TimedCurtainStrategy): + """Shared slat-phase + curtain-roll Venetian blind strategy aligned with #508. + + End-to-end ``travel_time`` includes one full slat rotation (``slat_time``). + Vertical curtain travel is therefore estimated over ``travel_time - slat_time`` + after the slats finish rotating through their remaining lead angle, avoiding + double-counting between slat tilt and vertical curtain displacement. + """ + + def __init__( + self, + slat_time: float | None = None, + *, + travel_time: float = DEFAULT_TRAVEL_TIME, + travel_time_up: float | None = None, + travel_time_down: float | None = None, + min_motor_pulse: float = DEFAULT_MIN_MOTOR_PULSE, + ) -> None: + """Initialize Venetian strategy with shared ``slat_time``.""" + super().__init__( + travel_time=travel_time, + travel_time_up=travel_time_up, + travel_time_down=travel_time_down, + min_motor_pulse=min_motor_pulse, + slat_time=slat_time, + ) + + @property + def name(self) -> str: + """Return strategy identifier.""" + return "venetian" + + @property + def supported_features(self) -> CoverEntityFeature: + """Return curtain features plus OPEN_TILT | CLOSE_TILT | STOP_TILT | SET_TILT_POSITION.""" + return ( + super().supported_features + | CoverEntityFeature.OPEN_TILT + | CoverEntityFeature.CLOSE_TILT + | CoverEntityFeature.STOP_TILT + | CoverEntityFeature.SET_TILT_POSITION + ) + + def curtain_travel_for(self, opening: bool) -> float: + """Return vertical curtain travel seconds (end-to-end travel minus slat rotation phase).""" + travel = self.travel_for_direction(opening) + slat = self.slat_time_for() + if travel > slat: + return travel - slat + return travel + + def lead_time_for(self, opening: bool, *, entity: Any = None) -> float: + """Return seconds a curtain run spends rotating slats before vertical movement starts.""" + if entity is not None and getattr(entity, "_is_tilting", False) is True: + return 0.0 + curr: int | None = None + if entity is not None: + raw_curr = getattr(entity, "_attr_current_cover_tilt_position", None) + if isinstance(raw_curr, int): + curr = raw_curr + if curr is None: + curr = 0 if opening else 100 + remaining = (100 - curr) if opening else curr + return max(0.0, (remaining / 100.0) * self.slat_time_for()) + + def carry_overshoot_into_curtain( + self, + start_position: int, + start_tilt: int, + elapsed: float, + direction: str, + ) -> int: + """Convert motor runtime past a full slat rotation into vertical curtain displacement.""" + opening = direction == "open" + slat = self.slat_time_for() + to_end = ((100 - start_tilt) if opening else start_tilt) / 100.0 * slat + excess = max(0.0, float(elapsed)) - to_end + if excess <= 0: + return max(0, min(100, int(round(start_position)))) + curtain_travel = self.curtain_travel_for(opening) + if curtain_travel <= 0: + return max(0, min(100, int(round(start_position)))) + moved = (excess / curtain_travel) * 100.0 + delta = moved if opening else -moved + return max(0, min(100, int(round(start_position + delta)))) + + def _reanchor_slats_on_stop_echo( + self, + entity: Any, + ctx: dict[str, Any], + elapsed: float, + is_opening: bool, + is_closing: bool, + ) -> None: + """Re-anchor slat tilt position on a relayed stop status echo.""" + if not (is_opening or is_closing): + return + raw_tilt = ctx.get("run_start_tilt") + start_tilt = int(raw_tilt) if raw_tilt is not None else None + direction = "open" if is_opening else "close" + entity._attr_current_cover_tilt_position = self.turn_slats_after( + start_tilt, elapsed, direction + ) diff --git a/custom_components/myhome/light.py b/custom_components/myhome/light.py index ce540073f..67f1ec4ff 100644 --- a/custom_components/myhome/light.py +++ b/custom_components/myhome/light.py @@ -1,6 +1,7 @@ """Support for MyHome lights.""" import asyncio +import inspect from typing import Any, cast import voluptuous as vol @@ -658,9 +659,12 @@ async def _set_brightness_instant(self, pct: int) -> None: if pct <= 0: await self._gateway_handler.send(OWNLightingCommand.switch_off(self._full_where)) else: - await self._gateway_handler.send( - OWNLightingCommand.set_brightness(self._full_where, pct, 0) - ) + cmd = OWNLightingCommand.set_brightness(self._full_where, pct, 0) + send_fade = getattr(self._gateway_handler, "send_fade_step", None) + if self._fade_engine.is_fading and inspect.iscoroutinefunction(send_fade): + await send_fade(cmd) + else: + await self._gateway_handler.send(cmd) async def _maybe_instant_brightness( self, start_pct: int, target_pct: int, is_on: bool | None = None diff --git a/custom_components/myhome/validate.py b/custom_components/myhome/validate.py index e3de7c839..212e0e93e 100644 --- a/custom_components/myhome/validate.py +++ b/custom_components/myhome/validate.py @@ -97,12 +97,14 @@ CONF_SLAT_TIME, CONF_STANDALONE, CONF_STATE_SENSOR, + CONF_STRATEGY, CONF_TRAVEL_TIME, CONF_TYPE, CONF_WATCHDOG_MARGIN, CONF_WHERE, CONF_WHO, CONF_ZONE, + COVER_STRATEGIES, DEFAULT_MIN_MOTOR_PULSE, DEFAULT_SLAT_TIME, MAX_SLAT_TIME, @@ -509,13 +511,14 @@ def _validate_access(access: dict[str, typing.Any]) -> dict[str, typing.Any]: Optional(CONF_ACCESS): access_schema, Optional(CONF_ADVANCED_SHUTTER, default=False): Boolean(), Optional("advanced_shutter", default=False): Boolean(), + Optional(CONF_STRATEGY): In(list(COVER_STRATEGIES)), Optional(CONF_SLAT_TILT, default=False): Boolean(), Optional(CONF_SLAT_TIME): All( Coerce(float), Range(min=MIN_SLAT_TIME, max=MAX_SLAT_TIME) ), Optional(CONF_TRAVEL_TIME, default=25): All( Coerce(int), - Range(min=int(CALIBRATION_MIN_RUN), max=int(CALIBRATION_MAX_RUN)), + Any(0, Range(min=int(CALIBRATION_MIN_RUN), max=int(CALIBRATION_MAX_RUN))), ), Optional(CONF_MIN_MOTOR_PULSE, default=DEFAULT_MIN_MOTOR_PULSE): All( Coerce(float), Range(min=CALIBRATION_MIN_MOTOR_PULSE, max=CALIBRATION_MAX_MOTOR_PULSE) @@ -536,6 +539,15 @@ def _validate_cover_members(data: dict[str, typing.Any]) -> dict[str, typing.Any where = cfg.get(CONF_WHERE) if not where or not str(where).startswith("#"): raise Invalid("Members can only be defined on a group cover (where must start with #)") + strategy = cfg.get(CONF_STRATEGY) + if strategy == "venetian": + cfg[CONF_SLAT_TILT] = True + elif strategy == "native": + cfg[CONF_ADVANCED_SHUTTER] = True + elif strategy == "simple" and cfg.get(CONF_SLAT_TILT): + raise Invalid("strategy: simple cannot be combined with slat_tilt") + if cfg.get(CONF_SLAT_TILT) and cfg.get(CONF_TRAVEL_TIME) == 0 and not cfg.get(CONF_ADVANCED_SHUTTER): + raise Invalid("slat_tilt requires a positive travel_time or advanced_shutter") if CONF_SLAT_TIME in cfg and not cfg.get(CONF_SLAT_TILT): raise Invalid("slat_time can only be defined on a cover with slat_tilt enabled") cfg.setdefault(CONF_SLAT_TIME, DEFAULT_SLAT_TIME) diff --git a/tests/test_cover_strategies.py b/tests/test_cover_strategies.py new file mode 100644 index 000000000..f9683c73c --- /dev/null +++ b/tests/test_cover_strategies.py @@ -0,0 +1,721 @@ +"""Unit and trace-replay tests for CoverKinematicsStrategy implementations and fade step routing (#730, #739).""" +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from homeassistant.components.cover import CoverEntityFeature +from OWNd.message import OWNEvent +from probatio import Invalid + +from custom_components.myhome.const import ( + CONF_SLAT_TILT, + CONF_SLAT_TIME, + CONF_STRATEGY, + CONF_TRAVEL_TIME, + DEFAULT_SLAT_TIME, + DEFAULT_TRANSITION_MODE, + MAX_SLAT_TIME, + MIN_SLAT_TIME, + TRANSITION_MODE_NATIVE, + TRANSITION_MODE_SOFTWARE, +) +from custom_components.myhome.cover import MyHOMECover +from custom_components.myhome.cover_strategies import ( + BaseCoverStrategy, + CoverKinematicsStrategy, + NativePositionStrategy, + SimpleUpDownStopStrategy, + TimedCurtainStrategy, + VenetianTiltStrategy, + clamp_slat_time, + resolve_cover_strategy, +) +from custom_components.myhome.light_fade import SoftwareFadeEngine +from custom_components.myhome.validate import cover_schema + +TRACES_DIR = Path(__file__).parent / "fixtures" / "traces" + + +def _make_mock_entity(where: str = "21") -> MagicMock: + entity = MagicMock() + entity._full_where = where + entity._gateway_handler = MagicMock() + entity._gateway_handler.log_id = "[TestGW]" + entity._gateway_handler.send = AsyncMock() + entity._publish_state = MagicMock() + entity._refresh_travel_attributes = MagicMock() + entity._attr_is_opening = False + entity._attr_is_closing = False + entity._attr_is_closed = False + entity._attr_current_cover_position = 50 + entity._attr_current_cover_tilt_position = None + entity._start_position = 50 + entity._move_start_time = None + entity._travel_time = 20 + entity._travel_time_up = 20.0 + entity._travel_time_down = 20.0 + entity._min_motor_pulse = 0.4 + entity._slat_tilt = False + entity._slat_time = DEFAULT_SLAT_TIME + entity._is_tilting = False + entity._advanced = False + entity._echo_until = None + entity._in_echo_window = MagicMock(return_value=False) + entity._cancel_stop_task = MagicMock() + entity._end_pulse_for_takeover = MagicMock() + return entity + + +def test_default_transition_mode_remains_software_stepped() -> None: + """Verify DEFAULT_TRANSITION_MODE stays software_stepped so gateways that NACK Dim 1 speed frames still fade (#739).""" + assert DEFAULT_TRANSITION_MODE == TRANSITION_MODE_SOFTWARE + + +def test_software_fade_engine_gating() -> None: + """Verify SoftwareFadeEngine.should_use_software_stepped suppresses stepped mode only when native is selected.""" + mode = TRANSITION_MODE_NATIVE + + engine = SoftwareFadeEngine( + where="11", + create_task_cb=MagicMock(), + send_instant_cb=AsyncMock(), + apply_state_cb=MagicMock(), + update_ha_state_cb=MagicMock(), + get_worker_count_cb=lambda: 1, + get_transition_mode_cb=lambda: mode, + is_on_cb=lambda: True, + ) + + assert engine.should_use_software_stepped(5.0) is False + mode = TRANSITION_MODE_SOFTWARE + assert engine.should_use_software_stepped(None) is False + assert engine.should_use_software_stepped(0.0) is False + assert engine.should_use_software_stepped(5.0) is True + + +def test_clamp_slat_time() -> None: + """Verify clamp_slat_time clamps values and handles invalid inputs.""" + assert clamp_slat_time(None) == DEFAULT_SLAT_TIME + assert clamp_slat_time("invalid", 2.0) == 2.0 + assert clamp_slat_time(0.01) == MIN_SLAT_TIME + assert clamp_slat_time(99.0) == MAX_SLAT_TIME + assert clamp_slat_time(2.5) == 2.5 + + +def test_cover_schema_accepts_strategy_and_zero_travel_time() -> None: + """Verify cover_schema accepts CONF_STRATEGY and travel_time: 0 while rejecting incompatible combinations.""" + # Simple strategy via travel_time: 0 + cfg_zero = cover_schema({"shutter_a": {"where": "21", "name": "A", CONF_TRAVEL_TIME: 0}}) + assert cfg_zero["2-21"][CONF_TRAVEL_TIME] == 0 + + # Explicit strategy names + for strat_name, extra in ( + ("simple", {CONF_TRAVEL_TIME: 0}), + ("timed", {CONF_TRAVEL_TIME: 20}), + ("venetian", {CONF_TRAVEL_TIME: 20, CONF_SLAT_TILT: True, CONF_SLAT_TIME: 1.5}), + ("native", {"advanced": True}), + ): + payload: dict[str, Any] = {"where": "21", "name": "A", CONF_STRATEGY: strat_name, **extra} + validated = cover_schema({"shutter_a": payload}) + assert validated["2-21"][CONF_STRATEGY] == strat_name + + # Reject unknown strategy name + with pytest.raises(Invalid): + cover_schema({"shutter_a": {"where": "21", "name": "A", CONF_STRATEGY: "impulse"}}) + + # Reject travel_time: 0 when slat_tilt is enabled without advanced + with pytest.raises(Invalid): + cover_schema( + { + "shutter_a": { + "where": "21", + "name": "A", + CONF_TRAVEL_TIME: 0, + CONF_SLAT_TILT: True, + } + } + ) + + # Reject strategy: simple when slat_tilt is enabled + with pytest.raises(Invalid): + cover_schema( + { + "shutter_a": { + "where": "21", + "name": "A", + CONF_STRATEGY: "simple", + CONF_SLAT_TILT: True, + } + } + ) + + +def test_resolve_cover_strategy_variants() -> None: + """Verify resolve_cover_strategy selects the right strategy for all configurations.""" + # Explicit strategy_name + assert isinstance(resolve_cover_strategy(strategy_name="simple"), SimpleUpDownStopStrategy) + assert isinstance(resolve_cover_strategy(strategy_name="native"), NativePositionStrategy) + assert isinstance( + resolve_cover_strategy(strategy_name="venetian", slat_time=2.0), + VenetianTiltStrategy, + ) + assert isinstance( + resolve_cover_strategy(strategy_name="timed", travel_time=30), + TimedCurtainStrategy, + ) + assert isinstance( + resolve_cover_strategy(strategy_name="unknown", travel_time=25), + TimedCurtainStrategy, + ) + + # Implicit resolution + assert isinstance(resolve_cover_strategy(advanced=True), NativePositionStrategy) + venetian = resolve_cover_strategy(slat_tilt=True, slat_time=1.8, travel_time=20) + assert isinstance(venetian, VenetianTiltStrategy) + assert venetian.slat_time == 1.8 + + # Uncalibrated / unconfigured (travel_time <= 0 or None) -> SimpleUpDownStopStrategy + assert isinstance(resolve_cover_strategy(travel_time=0), SimpleUpDownStopStrategy) + assert isinstance(resolve_cover_strategy(travel_time=-5), SimpleUpDownStopStrategy) + assert isinstance(resolve_cover_strategy(travel_time=None), SimpleUpDownStopStrategy) + assert isinstance(resolve_cover_strategy(travel_time="invalid"), SimpleUpDownStopStrategy) # type: ignore[arg-type] + + # Positive travel_time -> TimedCurtainStrategy + timed = resolve_cover_strategy(travel_time=25) + assert isinstance(timed, TimedCurtainStrategy) + assert isinstance(timed, CoverKinematicsStrategy) + assert isinstance(timed, BaseCoverStrategy) + + +def test_simple_up_down_stop_strategy() -> None: + """Verify SimpleUpDownStopStrategy properties and protocol methods.""" + strategy = SimpleUpDownStopStrategy() + entity = _make_mock_entity() + + assert strategy.name == "simple" + assert strategy.assumed_state is True + assert strategy.supported_features == ( + CoverEntityFeature.OPEN | CoverEntityFeature.CLOSE | CoverEntityFeature.STOP + ) + assert strategy.current_position(entity, 100.0) is None + assert strategy.position_after(50, 5.0, "open") is None + assert strategy.duration_for(0, 100, "open") == 0.0 + assert strategy.reanchor_on_stop_echo(entity, {}, 100.0) is None + assert strategy.travel_for_direction(True) == 0.0 + assert strategy.curtain_travel_for(True) == 0.0 + assert strategy.lead_time_for(True) == 0.0 + assert strategy.carry_overshoot_into_curtain(50, 50, 1.0, "open") == 50 + + +def test_native_position_strategy() -> None: + """Verify NativePositionStrategy quantization and protocol methods.""" + strategy = NativePositionStrategy() + entity = _make_mock_entity() + + assert strategy.name == "native" + assert strategy.assumed_state is False + assert strategy.supported_features == ( + CoverEntityFeature.OPEN + | CoverEntityFeature.CLOSE + | CoverEntityFeature.STOP + | CoverEntityFeature.SET_POSITION + ) + assert strategy.position_after(40, 2.0, "open") == 40 + assert strategy.position_after(40, 2.0, "open", entity=entity) == 50 + assert strategy.duration_for(40, 80, "open") == 0.0 + assert strategy.reanchor_on_stop_echo(entity, {}, 100.0) == 50 + assert strategy.current_position(entity, 100.0) == 50 + + # quantize_target + assert strategy.quantize_target(None, 55) == 55 + assert strategy.quantize_target(50, 100) == 100 + assert strategy.quantize_target(50, 0) == 0 + assert strategy.quantize_target(50, 51) == 52 + assert strategy.quantize_target(50, 49) == 48 + + +def test_timed_curtain_strategy() -> None: + """Verify TimedCurtainStrategy owns travel_time_up/down and min_motor_pulse without reading back from entity.""" + strategy = TimedCurtainStrategy( + travel_time=20.0, + travel_time_up=25.0, + travel_time_down=20.0, + min_motor_pulse=0.5, + ) + entity = _make_mock_entity() + + assert strategy.name == "timed" + assert strategy.assumed_state is False + assert strategy.travel_for_direction(True) == 25.0 + assert strategy.travel_for_direction(False) == 20.0 + assert strategy.curtain_travel_for(True) == 25.0 + assert strategy.lead_time_for(True) == 0.0 + assert strategy.min_pulse_for() == 0.5 + assert strategy.quantize_target(50, 51) == 51 + + # position_after + assert strategy.position_after(50, 5.0, "stop") == 50 + assert strategy.position_after(50, 5.0, "open") == 70 # 5 / 25 * 100 = 20% -> 70% + assert strategy.position_after(50, 5.0, "close") == 25 # 5 / 20 * 100 = 25% -> 25% + assert strategy.position_after(50, 6.0, "open", lead=1.0) == 70 + + # travel_time <= 0 edge cases + strategy.travel_time_up = 0.0 + assert strategy.position_after(40, 5.0, "open", entity=entity) == 50 + entity._attr_current_cover_position = None + assert strategy.position_after(40, 5.0, "open", entity=entity) == 40 + assert strategy.duration_for(40, 80, "open", entity=entity) == 0.0 + strategy.travel_time_up = 25.0 + entity._attr_current_cover_position = 50 + + # duration_for + assert strategy.duration_for(50, 50) == 0.0 + assert strategy.duration_for(50, 60, "close") == 0.0 + assert strategy.duration_for(50, 40, "open") == 0.0 + # 1% on 25s up = 0.25s, clamped to min_motor_pulse 0.5s + assert strategy.duration_for(50, 51, "open") == 0.5 + assert strategy.duration_for(50, 70) == 5.0 # 20% of 25s = 5.0s + + # reanchor_on_stop_echo + assert strategy.reanchor_on_stop_echo(entity, {}, 100.0) == 50 + ctx_open = { + "run_started_at": 100.0, + "run_start_position": 50, + "is_opening": True, + "is_closing": False, + "auto_stop_duration": 4.0, + "lead": 0.0, + } + assert strategy.reanchor_on_stop_echo(entity, ctx_open, 105.0) == 70 + assert entity._attr_current_cover_position == 70 + assert entity._last_run["seconds"] == 5.0 + + ctx_close = { + "run_started_at": 100.0, + "run_start_position": 20, + "is_opening": False, + "is_closing": True, + "auto_stop_duration": None, + "lead": 0.0, + } + assert strategy.reanchor_on_stop_echo(entity, ctx_close, 104.0) == 0 + assert entity._attr_is_closed is True + + # current_position + entity._advanced = True + assert strategy.current_position(entity, 100.0) == 0 + entity._advanced = False + assert strategy.current_position(entity, 100.0) == 0 + entity._start_position = 20 + entity._move_start_time = 100.0 + entity._attr_is_opening = True + assert strategy.current_position(entity, 105.0) == 40 + + +def test_venetian_tilt_strategy() -> None: + """Verify VenetianTiltStrategy shared slat+curtain geometry and pulse helpers.""" + strategy = VenetianTiltStrategy(slat_time=2.0, travel_time=22.0, min_motor_pulse=0.4) + entity = _make_mock_entity() + entity._slat_tilt = True + + assert strategy.name == "venetian" + assert strategy.slat_time == 2.0 + assert strategy.supported_features & CoverEntityFeature.SET_TILT_POSITION + + # Shared geometry: curtain travel = 22.0 - 2.0 = 20.0s + assert strategy.slat_time_for() == 2.0 + assert strategy.curtain_travel_for(True) == 20.0 + # When travel <= slat_time, curtain_travel_for returns travel + short_strategy = VenetianTiltStrategy(slat_time=5.0, travel_time=3.0) + assert short_strategy.curtain_travel_for(True) == 3.0 + + # lead_time_for + entity._is_tilting = True + assert strategy.lead_time_for(True, entity=entity) == 0.0 + entity._is_tilting = False + assert strategy.lead_time_for(True) == 2.0 # curr=None -> 0 when opening -> 100% of 2.0s + assert strategy.lead_time_for(False) == 2.0 # curr=None -> 100 when closing -> 100% of 2.0s + entity._attr_current_cover_tilt_position = 50 + assert strategy.lead_time_for(True, entity=entity) == 1.0 + + # turn_slats_after + assert strategy.turn_slats_after(0, 1.0, "open") == 50 + assert strategy.turn_slats_after(100, 1.0, "close") == 50 + strategy.slat_time = 0.0 + assert strategy.turn_slats_after(0, 0.5, "open") == 100 + strategy.slat_time = 2.0 + + # carry_overshoot_into_curtain + assert strategy.carry_overshoot_into_curtain(50, 50, 0.5, "open") == 50 + # 1.0s to finish slats + 2.0s overshoot on 20.0s curtain = +10% -> 60% + assert strategy.carry_overshoot_into_curtain(50, 50, 3.0, "open") == 60 + assert strategy.carry_overshoot_into_curtain(50, 50, 3.0, "close") == 40 + strategy.travel_time_up = 0.0 + assert strategy.carry_overshoot_into_curtain(50, 50, 3.0, "open") == 50 + strategy.travel_time_up = 22.0 + + # tilt_pulse_duration_for + assert strategy.tilt_pulse_duration_for(50, 50) == 0.0 + assert strategy.tilt_pulse_duration_for(50, 55) == 0.0 # 0.1s < 0.4s min_motor_pulse + assert strategy.tilt_pulse_duration_for(0, 50) == 1.0 + + # _reanchor_slats_on_stop_echo + strategy._reanchor_slats_on_stop_echo(entity, {}, 1.0, False, False) + ctx_venetian = { + "run_started_at": 100.0, + "run_start_position": 50, + "run_start_tilt": 0, + "is_opening": True, + "is_closing": False, + "auto_stop_duration": None, + "lead": 2.0, + } + # 4.0s elapsed: 2.0s slat rotation (0->100%) + 2.0s curtain travel (2/20 = +10% -> 60%) + assert strategy.reanchor_on_stop_echo(entity, ctx_venetian, 104.0) == 60 + assert entity._attr_current_cover_tilt_position == 100 + + +@pytest.mark.asyncio +async def test_calibrated_geometry_strategy_subclass_override_is_not_bypassed(hass) -> None: + """Verify a TimedCurtainStrategy subclass overriding travel_for_direction is used by MyHOMECover.""" + + class CalibratedGeometryStrategy(TimedCurtainStrategy): + def travel_for_direction(self, opening: bool) -> float: + return 10.0 if opening else 40.0 + + gw = MagicMock() + gw.mac = "00:03:50:00:12:34" + gw.log_id = "[TestGW]" + gw.send = AsyncMock() + + with patch("custom_components.myhome.myhome_device.Entity.__init__", return_value=None): + cover = MyHOMECover( + hass=hass, + name="Calibrated Cover", + entity_name="Calibrated Cover", + device_id="21", + who="2", + where="21", + interface=None, + advanced=False, + manufacturer="BTicino", + model="Shutter", + gateway=gw, + travel_time=20, + ) + cover._strategy = CalibratedGeometryStrategy(travel_time=20.0) + cover._attr_current_cover_position = 50 + # 20% upward move (50 -> 70) uses overridden 10.0s -> 2.0s + assert cover._strategy.duration_for(50, 70, "open", entity=cover) == 2.0 + # 20% downward move (50 -> 30) uses overridden 40.0s -> 8.0s + assert cover._strategy.duration_for(50, 30, "close", entity=cover) == 8.0 + + +@pytest.mark.asyncio +async def test_myhome_cover_simple_strategy_and_calibration_promotion(hass) -> None: + """Test MyHOMECover with SimpleUpDownStopStrategy and promotion to TimedCurtainStrategy on calibration.""" + gw = MagicMock() + gw.mac = "00:03:50:00:12:34" + gw.log_id = "[TestGW]" + gw.send = AsyncMock() + + with patch("custom_components.myhome.myhome_device.Entity.__init__", return_value=None): + simple_cover = MyHOMECover( + hass=hass, + name="Simple Cover", + entity_name="Simple Cover", + device_id="21", + who="2", + where="21", + interface=None, + advanced=False, + manufacturer="BTicino", + model="Shutter", + gateway=gw, + travel_time=0, + strategy_name="simple", + ) + simple_cover.entity_id = "cover.simple" + simple_cover.async_schedule_update_ha_state = MagicMock() + + assert simple_cover.strategy.name == "simple" + assert simple_cover.assumed_state is True + assert simple_cover.current_cover_position is None + + await simple_cover.async_restore_last_state(MagicMock(state="closed")) + assert simple_cover.is_closed is True + await simple_cover.async_restore_last_state(MagicMock(state="open")) + assert simple_cover.is_closed is False + + await simple_cover.async_set_cover_position(position=100) + assert simple_cover.is_opening is True + await simple_cover.async_set_cover_position(position=0) + assert simple_cover.is_closing is True + await simple_cover.async_set_cover_position(position=50) + assert simple_cover.is_closing is False + + await simple_cover.async_open_cover() + assert simple_cover.is_opening is True + assert simple_cover.is_closed is False + await simple_cover.async_close_cover() + assert simple_cover.is_closing is True + await simple_cover.async_stop_cover() + assert simple_cover.is_closing is False + + simple_cover.handle_event(OWNEvent.parse("*2*1*21##")) + assert simple_cover.is_opening is True + simple_cover.handle_event(OWNEvent.parse("*2*0*21##")) + assert simple_cover.is_opening is False + + # Promoting to timed strategy when calibration travel_time is applied + simple_cover._travel_time = 15 + simple_cover._travel_time_down = 15.0 + simple_cover._travel_time_up = 15.0 + simple_cover._promote_to_timed_strategy() + assert simple_cover.strategy.name == "timed" + assert simple_cover.assumed_state is False + assert bool(simple_cover._attr_supported_features & CoverEntityFeature.SET_POSITION) + + # Calling _enable_slat_tilt() upgrades TimedCurtainStrategy to VenetianTiltStrategy + simple_cover._enable_slat_tilt() + assert simple_cover.strategy.name == "venetian" + + # Also test promotion to VenetianTiltStrategy when _slat_tilt is True + simple_venetian = MyHOMECover( + hass=hass, + name="Simple Venetian", + entity_name="Simple Venetian", + device_id="22", + who="2", + where="22", + interface=None, + advanced=False, + manufacturer="BTicino", + model="Shutter", + gateway=gw, + travel_time=0, + ) + simple_venetian._slat_tilt = True + simple_venetian._travel_time_down = 20.0 + simple_venetian._travel_time_up = 20.0 + simple_venetian._promote_to_timed_strategy() + assert simple_venetian.strategy.name == "venetian" + + +@pytest.mark.asyncio +async def test_replay_f461_live_shutter_05_trace_on_timed_strategy(hass) -> None: + """Replay live BTicino F461 run2_shutter_05_travel_and_reversal trace through TimedCurtainStrategy.""" + trace_path = TRACES_DIR / "f461_live_2026-10-18" / "myhome_trace_F461_live_2026-10-18.json" + trace = json.loads(trace_path.read_text(encoding="utf-8")) + run2_frames = [ + f + for f in trace["frames"] + if f.get("run") == "run2_shutter_05_travel_and_reversal" and f.get("direction") == "rx" + ] + assert len(run2_frames) == 18 + + gw = MagicMock() + gw.mac = trace["gateway"]["mac"] + gw.log_id = "[F461]" + gw.send = AsyncMock() + + with patch("custom_components.myhome.myhome_device.Entity.__init__", return_value=None): + shutter = MyHOMECover( + hass=hass, + name="Shutter 05", + entity_name="Shutter 05", + device_id="05", + who="2", + where="05", + interface=None, + advanced=False, + manufacturer="BTicino", + model="BT-F411/2", + gateway=gw, + travel_time=5, + ) + shutter.entity_id = "cover.shutter_05" + shutter.async_schedule_update_ha_state = MagicMock() + # Start closed (0%) before run2 begins + shutter._attr_current_cover_position = 0 + shutter._attr_is_closed = True + + positions_on_stop: list[int | None] = [] + for frame in run2_frames: + ts = float(frame["timestamp"]) + raw = frame["raw"] + msg = OWNEvent.parse(raw) + assert msg is not None + with patch("custom_components.myhome.cover.time.monotonic", return_value=ts): + shutter.handle_event(msg) + if raw == "*2*0*05##": + positions_on_stop.append(shutter.current_cover_position) + + # After first open pulse (22.72 -> 24.852 = 2.132s of 5.0s -> 43%), + # second open pulse (26.421 -> 28.014 = 1.593s of 5.0s -> +32% = 75%), + # and full close reversal (28.119 -> 34.341 = 6.222s -> 0% closed): + assert 43 in positions_on_stop + assert 75 in positions_on_stop + assert positions_on_stop[-1] == 0 + assert shutter.is_closed is True + + +@pytest.mark.asyncio +async def test_replay_issue_466_f454_trace_on_native_strategy(hass) -> None: + """Replay F454 Dim 10 / WHAT 0 frames from issue_466 trace fixture through NativePositionStrategy.""" + trace_path = TRACES_DIR / "issue_466" / "config_entry-myhome_F454_step_commands.json" + diag = json.loads(trace_path.read_text(encoding="utf-8")) + recent_frames = diag["data"]["bus_monitor"]["recent_frames"] + where13_frames = [f for f in recent_frames if f.get("where") == "13" and f.get("direction") == "rx"] + assert len(where13_frames) >= 4 + + gw = MagicMock() + gw.mac = "00:03:50:00:04:54" + gw.log_id = "[F454]" + gw.send = AsyncMock() + + with patch("custom_components.myhome.myhome_device.Entity.__init__", return_value=None): + cover = MyHOMECover( + hass=hass, + name="Cover 13", + entity_name="Cover 13", + device_id="13", + who="2", + where="13", + interface=None, + advanced=True, + manufacturer="BTicino", + model="F411U2", + gateway=gw, + strategy_name="native", + ) + cover.entity_id = "cover.cover_13" + cover.async_schedule_update_ha_state = MagicMock() + assert isinstance(cover.strategy, NativePositionStrategy) + + for frame in where13_frames[:4]: + msg = OWNEvent.parse(frame["raw"]) + assert msg is not None + with patch( + "custom_components.myhome.cover.time.monotonic", + return_value=float(frame["timestamp"]), + ): + cover.handle_event(msg) + + assert cover.current_cover_position == 100 + assert cover.is_closed is False + assert cover.is_opening is False + assert cover.is_closing is False + + +@pytest.mark.asyncio +async def test_replay_oracle_who2_venetian_tilt_trace_on_venetian_strategy(hass) -> None: + """Replay oracle_who2_venetian_tilt trace fixture through VenetianTiltStrategy.""" + trace_path = TRACES_DIR / "oracle_who2_venetian_tilt" / "myhome_trace_oracle_who2_venetian_tilt.json" + trace = json.loads(trace_path.read_text(encoding="utf-8")) + frames = trace["frames"] + assert len(frames) == 10 + + gw = MagicMock() + gw.mac = "00:03:50:00:04:54" + gw.log_id = "[F454]" + gw.send = AsyncMock() + + with patch("custom_components.myhome.myhome_device.Entity.__init__", return_value=None): + venetian = MyHOMECover( + hass=hass, + name="Venetian 31", + entity_name="Venetian 31", + device_id="31", + who="2", + where="31", + interface=None, + advanced=False, + manufacturer="BTicino", + model="F411/2", + gateway=gw, + travel_time=22, + slat_tilt=True, + slat_time=2.0, + strategy_name="venetian", + ) + venetian.entity_id = "cover.venetian_31" + venetian.async_schedule_update_ha_state = MagicMock() + assert isinstance(venetian.strategy, VenetianTiltStrategy) + venetian._attr_current_cover_position = 0 + venetian._attr_current_cover_tilt_position = 0 + venetian._attr_is_closed = True + + # Replay the 1.0s open pulse (frames[1] *2*1*31## -> frames[2] *2*0*31##): + # 1.0s of 2.0s slat_time rotates slats 0% -> 50% while curtain stays at 0% + for frame in frames[1:3]: + msg = OWNEvent.parse(frame["raw"]) + assert msg is not None + with patch( + "custom_components.myhome.cover.time.monotonic", + return_value=float(frame["timestamp"]), + ): + venetian.handle_event(msg) + assert venetian.current_cover_tilt_position == 50 + assert venetian.current_cover_position == 0 + + # Replay the 1.0s close pulse (frames[3] *2*2*31## -> frames[4] *2*0*31##): + # 1.0s of 2.0s slat_time rotates slats 50% -> 0% + for frame in frames[3:5]: + msg = OWNEvent.parse(frame["raw"]) + assert msg is not None + with patch( + "custom_components.myhome.cover.time.monotonic", + return_value=float(frame["timestamp"]), + ): + venetian.handle_event(msg) + assert venetian.current_cover_tilt_position == 0 + assert venetian.current_cover_position == 0 + + +@pytest.mark.asyncio +async def test_light_set_brightness_instant_routes_fade_steps(hass) -> None: + """Verify MyHOMELight._set_brightness_instant uses gateway.send_fade_step during active fade.""" + from custom_components.myhome.light import MyHOMELight + + gw = MagicMock() + gw.mac = "00:03:50:00:12:34" + gw.log_id = "[TestGW]" + gw.send = AsyncMock() + gw.send_fade_step = AsyncMock() + + with patch("custom_components.myhome.myhome_device.Entity.__init__", return_value=None): + dimmer = MyHOMELight( + hass=hass, + name="Dimmer 11", + entity_name="Dimmer 11", + icon=None, + icon_on=None, + device_id="11", + who="1", + where="11", + interface=None, + dimmable=True, + manufacturer="BTicino", + model="Dimmer", + gateway=gw, + ) + dimmer.entity_id = "light.dimmer_11" + + # When not fading, routes via gateway.send + await dimmer._set_brightness_instant(50) + gw.send.assert_awaited_once() + gw.send_fade_step.assert_not_awaited() + + # When fading and send_fade_step is an async coroutine function, routes via send_fade_step + gw.send.reset_mock() + fake_task = MagicMock() + fake_task.done.return_value = False + dimmer._fade_engine.fade_task = fake_task + await dimmer._set_brightness_instant(60) + gw.send_fade_step.assert_awaited_once() + gw.send.assert_not_awaited() diff --git a/tests/test_validate.py b/tests/test_validate.py index b37d4dc16..97e1ba063 100644 --- a/tests/test_validate.py +++ b/tests/test_validate.py @@ -749,14 +749,14 @@ def test_device_class_remapping_and_defaults(self): res_cov = cover_schema(cov_data) assert res_cov["2-22"][CONF_ADVANCED_SHUTTER] is True - # Cover travel_time bounds (1..300) - for valid_tt in (1, 25, 300): + # Cover travel_time bounds (0..300; 0 selects SimpleUpDownStopStrategy) + for valid_tt in (0, 1, 25, 300): res_tt = cover_schema( {"cov1": {CONF_WHERE: "22", CONF_NAME: "Living Cover", "travel_time": valid_tt}} ) assert res_tt["2-22"]["travel_time"] == valid_tt - for invalid_tt in (0, -5, 301): + for invalid_tt in (-1, -5, 301): with pytest.raises(Invalid): cover_schema( {"cov1": {CONF_WHERE: "22", CONF_NAME: "Living Cover", "travel_time": invalid_tt}} From d1a6bdeb90cd92f6de64d90d7cf30e20f4abd1ff Mon Sep 17 00:00:00 2001 From: Laurens <43173895+GreenGrassBlueOcean@users.noreply.github.com> Date: Tue, 20 Oct 2026 12:12:38 +0200 Subject: [PATCH 2/4] fix(cover): reject contradictory cover strategies, report simple-cover closure, demote on reset (#739) - validate: an explicit strategy cannot be combined with advanced (other than native), strategy: timed with slat_tilt or with travel_time: 0. - cover: _advanced / _slat_tilt follow the resolved strategy, so the advanced code paths never run under a timed or simple strategy (the #702 deadband was lost and Dimension 10 levels were hidden); drop the matching guard in TimedCurtainStrategy.current_position. - cover: an assumed-state (simple) cover is closed after a closing run and open after an opening one instead of staying open. - cover: reset_cover_travel_time returns a promoted travel_time: 0 cover to SimpleUpDownStopStrategy; an explicit strategy: simple is never promoted, matching what a restart resolves. - cover: travel attributes come from the strategy, so a simple cover shows no travel time. - docs: document strategy and travel_time: 0 in covers.md. Co-Authored-By: Claude Opus 5.5 --- custom_components/myhome/cover.py | 47 +++++- .../myhome/cover_strategies/timed.py | 3 - custom_components/myhome/validate.py | 7 + docs/configuration/covers.md | 13 ++ tests/test_cover_strategies.py | 153 +++++++++++++++++- 5 files changed, 212 insertions(+), 11 deletions(-) diff --git a/custom_components/myhome/cover.py b/custom_components/myhome/cover.py index 89a11f2bc..0cc2a467f 100644 --- a/custom_components/myhome/cover.py +++ b/custom_components/myhome/cover.py @@ -419,6 +419,7 @@ def __init__( # type: ignore if (calibration and calibration.get("down") and calibration.get("up")) else travel_time ) + self._strategy_name = strategy_name self._strategy: CoverKinematicsStrategy = resolve_cover_strategy( strategy_name=strategy_name, advanced=self._advanced, @@ -429,10 +430,13 @@ def __init__( # type: ignore slat_time=raw_slat_time, min_motor_pulse=raw_min_pulse, ) - if isinstance(self._strategy, NativePositionStrategy): - self._advanced = True + # The resolved strategy decides; the entity's code paths follow it, so an + # explicit strategy next to a contradicting flag cannot mix the two. + self._advanced = isinstance(self._strategy, NativePositionStrategy) if isinstance(self._strategy, VenetianTiltStrategy): self._slat_tilt = True + elif isinstance(self._strategy, SimpleUpDownStopStrategy): + self._slat_tilt = False self._attr_supported_features = self._strategy.supported_features self._attr_assumed_state = self._strategy.assumed_state @@ -533,9 +537,11 @@ def _curtain_travel_for(self, opening: bool) -> float: def _refresh_travel_attributes(self) -> None: attrs = self._attr_extra_state_attributes - attrs["travel_time"] = int(round(self._travel_time_down)) - attrs["travel_time_down"] = round(self._travel_time_down, 2) - attrs["travel_time_up"] = round(self._travel_time_up, 2) + # Through the strategy: a simple cover has no travel time to show. + down = self._strategy.travel_for_direction(False) + attrs["travel_time"] = int(round(down)) + attrs["travel_time_down"] = round(down, 2) + attrs["travel_time_up"] = round(self._strategy.travel_for_direction(True), 2) attrs["min_motor_pulse"] = round(self._min_motor_pulse, 2) attrs["calibration_source"] = self._calibration_source attrs["calibrated_at"] = self._calibrated_at @@ -706,6 +712,10 @@ def _freeze_position(self, at: float) -> None: self._auto_stop_duration = None if self._attr_current_cover_position is not None: self._attr_is_closed = (self._attr_current_cover_position == 0) + elif self._strategy.assumed_state and (is_opening or is_closing): + # No position to derive it from: an assumed-state cover is closed after + # a closing run and open after an opening one, as HA's optimistic covers. + self._attr_is_closed = is_closing def _abort_motion(self) -> None: """Abort motion state when a command frame failed to reach the bus.""" @@ -1082,8 +1092,11 @@ async def async_calibrate(self) -> dict: # type: ignore return result def _promote_to_timed_strategy(self) -> None: - """Promote a SimpleUpDownStopStrategy cover to Timed/Venetian once calibrated.""" - if isinstance(self._strategy, SimpleUpDownStopStrategy): + """Promote a SimpleUpDownStopStrategy cover to Timed/Venetian once calibrated. + + An explicit ``strategy: simple`` stays simple, as it does after a restart. + """ + if isinstance(self._strategy, SimpleUpDownStopStrategy) and self._strategy_name != "simple": if self._slat_tilt: self._strategy = VenetianTiltStrategy( slat_time=self._strategy.slat_time, @@ -1108,6 +1121,23 @@ def _promote_to_timed_strategy(self) -> None: if self._attr_current_cover_position is None: self._attr_current_cover_position = self._start_position + def _demote_to_simple_strategy(self) -> None: + """Return a promoted cover to SimpleUpDownStopStrategy when its calibration is reset.""" + if isinstance(self._strategy, (SimpleUpDownStopStrategy, NativePositionStrategy)): + return + self._strategy = SimpleUpDownStopStrategy( + travel_time=self._strategy.travel_time_down, + travel_time_up=self._strategy.travel_time_up, + travel_time_down=self._strategy.travel_time_down, + min_motor_pulse=self._strategy.min_motor_pulse, + slat_time=self._strategy.slat_time, + ) + self._attr_assumed_state = True + self._attr_supported_features = ( + self._attr_supported_features or CoverEntityFeature(0) + ) & ~CoverEntityFeature.SET_POSITION + self._attr_current_cover_position = None + async def async_stop_calibration(self) -> None: """Stop calibration on this cover's gateway.""" gw_mac = getattr(self._gateway_handler, "mac", None) @@ -1286,6 +1316,9 @@ async def async_reset_travel_time(self) -> None: else: base_travel = float(DEFAULT_TRAVEL_TIME) source = "default" + if base_travel <= 0 or cfg.get(CONF_STRATEGY) == "simple": + # Back to the uncalibrated relay the yaml describes. + self._demote_to_simple_strategy() self._min_motor_pulse = float(cfg.get(CONF_MIN_MOTOR_PULSE, DEFAULT_MIN_MOTOR_PULSE)) self._travel_time_down = base_travel diff --git a/custom_components/myhome/cover_strategies/timed.py b/custom_components/myhome/cover_strategies/timed.py index 8b7486aed..3a7ece330 100644 --- a/custom_components/myhome/cover_strategies/timed.py +++ b/custom_components/myhome/cover_strategies/timed.py @@ -168,9 +168,6 @@ def reanchor_on_stop_echo( def current_position(self, entity: Any, now: float) -> int | None: """Interpolate current position from start position, motor anchor, and strategy kinematics.""" - if getattr(entity, "_advanced", False) is True: - pos_adv: int | None = getattr(entity, "_attr_current_cover_position", None) - return pos_adv is_opening = getattr(entity, "_attr_is_opening", False) is True is_closing = getattr(entity, "_attr_is_closing", False) is True move_start = getattr(entity, "_move_start_time", None) diff --git a/custom_components/myhome/validate.py b/custom_components/myhome/validate.py index 212e0e93e..0921c5056 100644 --- a/custom_components/myhome/validate.py +++ b/custom_components/myhome/validate.py @@ -540,6 +540,13 @@ def _validate_cover_members(data: dict[str, typing.Any]) -> dict[str, typing.Any if not where or not str(where).startswith("#"): raise Invalid("Members can only be defined on a group cover (where must start with #)") strategy = cfg.get(CONF_STRATEGY) + advanced = bool(cfg.get(CONF_ADVANCED_SHUTTER) or cfg.get("advanced_shutter")) + if strategy not in (None, "native") and advanced: + raise Invalid(f"strategy: {strategy} cannot be combined with advanced (use strategy: native)") + if strategy == "timed" and cfg.get(CONF_SLAT_TILT): + raise Invalid("strategy: timed cannot be combined with slat_tilt (use strategy: venetian)") + if strategy == "timed" and cfg.get(CONF_TRAVEL_TIME) == 0: + raise Invalid("strategy: timed requires a positive travel_time") if strategy == "venetian": cfg[CONF_SLAT_TILT] = True elif strategy == "native": diff --git a/docs/configuration/covers.md b/docs/configuration/covers.md index fb4f3fd2a..bd6eeb825 100644 --- a/docs/configuration/covers.md +++ b/docs/configuration/covers.md @@ -28,6 +28,19 @@ The integration distinguishes between two types of MyHOME covers: > [!NOTE] > Every discovered cover starts as a **timed** cover. Position-reporting hardware is not detected automatically: set `advanced_shutter: true` for that cover in `/config/myhome.yaml` (see [Troubleshooting → Cover position is wrong](troubleshooting.md#cover-position-is-wrong)). Advanced covers refuse `myhome.calibrate_cover` and travel times in `myhome.set_cover_travel_time`, as they do not need a travel time (a Venetian blind with `slat_tilt` still accepts `slat_time` there). +### Choosing the cover model (`strategy`) + +Each WHO=2 cover runs one of four position models. Without `strategy`, the flags pick it: `advanced_shutter: true` gives `native`, `slat_tilt: true` gives `venetian`, `travel_time: 0` gives `simple`, and anything else gives `timed`. Set `strategy` in `/config/myhome.yaml` to name it explicitly: + +| `strategy` | Position | Use for | +| :--- | :--- | :--- | +| `timed` | Estimated from the travel time | Standard relay actuators (the default). Needs a positive `travel_time`. | +| `venetian` | Estimated, plus slat tilt | Venetian blinds; implies `slat_tilt: true` and needs a positive `travel_time` (or `advanced_shutter`). | +| `native` | Reported by the actuator (`Dimension = 10`) | Advanced actuators; implies `advanced_shutter: true`. Can be combined with `slat_tilt`. | +| `simple` | None (open / close / stop only) | Relays with no measured travel time. The entity has no position slider and an assumed state: closed after a closing run, open after an opening one. | + +An explicit `strategy` cannot be combined with a flag that selects another model (`advanced_shutter` with anything but `native`, `slat_tilt` with `timed` or `simple`, `travel_time: 0` with `timed`): the configuration is rejected. A cover set up with `travel_time: 0` becomes `timed` once it is calibrated (`myhome.calibrate_cover` or `myhome.set_cover_travel_time`), and `simple` again after `myhome.reset_cover_travel_time`. An explicit `strategy: simple` stays simple even when travel times are stored. + --- ## ⏱️ Timed Cover Position Engine diff --git a/tests/test_cover_strategies.py b/tests/test_cover_strategies.py index f9683c73c..1e8e0672b 100644 --- a/tests/test_cover_strategies.py +++ b/tests/test_cover_strategies.py @@ -437,7 +437,6 @@ async def test_myhome_cover_simple_strategy_and_calibration_promotion(hass) -> N model="Shutter", gateway=gw, travel_time=0, - strategy_name="simple", ) simple_cover.entity_id = "cover.simple" simple_cover.async_schedule_update_ha_state = MagicMock() @@ -719,3 +718,155 @@ async def test_light_set_brightness_instant_routes_fade_steps(hass) -> None: await dimmer._set_brightness_instant(60) gw.send_fade_step.assert_awaited_once() gw.send.assert_not_awaited() + + +def _make_cover(hass, **kwargs: Any) -> MyHOMECover: + gw = MagicMock() + gw.mac = "00:03:50:00:12:34" + gw.log_id = "[TestGW]" + gw.send = AsyncMock() + args: dict[str, Any] = { + "hass": hass, + "name": "Cover 21", + "entity_name": "Cover 21", + "device_id": "21", + "who": "2", + "where": "21", + "interface": None, + "advanced": False, + "manufacturer": "BTicino", + "model": "Shutter", + "gateway": gw, + "travel_time": 20, + **kwargs, + } + with patch("custom_components.myhome.myhome_device.Entity.__init__", return_value=None): + cover = MyHOMECover(**args) + cover.entity_id = "cover.cover_21" + cover.async_write_ha_state = MagicMock() + cover.async_schedule_update_ha_state = MagicMock() + return cover + + +@pytest.mark.parametrize( + "extra", + [ + {CONF_STRATEGY: "timed", "advanced": True}, + {CONF_STRATEGY: "simple", "advanced": True}, + {CONF_STRATEGY: "venetian", "advanced": True}, + {CONF_STRATEGY: "timed", "advanced_shutter": True}, + {CONF_STRATEGY: "timed", CONF_SLAT_TILT: True}, + {CONF_STRATEGY: "timed", CONF_TRAVEL_TIME: 0}, + ], +) +def test_cover_schema_rejects_strategy_contradicting_flags(extra: dict[str, Any]) -> None: + """An explicit strategy cannot be combined with a flag that selects another one (#739).""" + with pytest.raises(Invalid): + cover_schema({"shutter_a": {"where": "21", "name": "A", **extra}}) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("strategy_name", ["timed", "simple", "venetian"]) +async def test_entity_flags_follow_explicit_strategy(hass, strategy_name: str) -> None: + """A non-native strategy never runs the advanced code paths, whatever the flags say (#739).""" + cover = _make_cover(hass, advanced=True, strategy_name=strategy_name) + assert cover.strategy.name == strategy_name + assert cover._advanced is False + if strategy_name == "simple": + assert cover._slat_tilt is False + + +@pytest.mark.asyncio +async def test_native_strategy_keeps_slat_tilt(hass) -> None: + """An advanced blind with slats stays native and keeps its tilt features.""" + cover = _make_cover(hass, advanced=True, slat_tilt=True) + assert cover.strategy.name == "native" + assert cover._advanced is True + assert cover._slat_tilt is True + assert cover.supported_features & CoverEntityFeature.SET_TILT_POSITION + + +@pytest.mark.asyncio +async def test_simple_cover_reports_closed_after_closing_run(hass) -> None: + """Without a position, a closing run leaves the cover closed and an opening run open (#739).""" + cover = _make_cover(hass, travel_time=0) + assert cover.strategy.name == "simple" + + cover.handle_event(OWNEvent.parse("*2*2*21##")) + cover.handle_event(OWNEvent.parse("*2*0*21##")) + assert cover.is_closing is False + assert cover.is_closed is True + assert cover.current_cover_position is None + + cover.handle_event(OWNEvent.parse("*2*1*21##")) + cover.handle_event(OWNEvent.parse("*2*0*21##")) + assert cover.is_opening is False + assert cover.is_closed is False + + # A stop with no run behind it leaves the state alone. + cover.handle_event(OWNEvent.parse("*2*2*21##")) + cover.handle_event(OWNEvent.parse("*2*0*21##")) + cover.handle_event(OWNEvent.parse("*2*0*21##")) + assert cover.is_closed is True + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("device_config", "strategy_name"), + [({CONF_TRAVEL_TIME: 0}, None), ({CONF_TRAVEL_TIME: 25, CONF_STRATEGY: "simple"}, "simple")], +) +async def test_reset_travel_time_demotes_promoted_cover( + hass, device_config: dict[str, Any], strategy_name: str | None +) -> None: + """Resetting the calibration of a promoted simple cover makes it simple again (#739).""" + cover = _make_cover(hass, travel_time=device_config[CONF_TRAVEL_TIME], strategy_name=strategy_name) + cover._device_config = MagicMock(return_value=device_config) + cover._fire_calibration_event = MagicMock() + # Promotion is what calibration does on a travel_time: 0 cover. + cover._strategy_name = None + cover._promote_to_timed_strategy() + cover._travel_time_down = cover._travel_time_up = 15.0 + assert cover.strategy.name == "timed" + + await cover.async_reset_travel_time() + + assert cover.strategy.name == "simple" + assert cover.assumed_state is True + assert not cover.supported_features & CoverEntityFeature.SET_POSITION + assert cover.current_cover_position is None + + +@pytest.mark.asyncio +async def test_reset_travel_time_keeps_timed_cover_timed(hass) -> None: + """A cover with a positive yaml travel_time stays timed after a reset.""" + cover = _make_cover(hass, travel_time=20) + cover._device_config = MagicMock(return_value={CONF_TRAVEL_TIME: 20}) + cover._fire_calibration_event = MagicMock() + await cover.async_reset_travel_time() + assert cover.strategy.name == "timed" + assert cover._travel_time_down == 20.0 + + +@pytest.mark.asyncio +async def test_explicit_simple_strategy_is_not_promoted(hass) -> None: + """strategy: simple stays simple when calibrated, as it does after a restart (#739).""" + cover = _make_cover(hass, travel_time=25, strategy_name="simple") + cover._promote_to_timed_strategy() + assert cover.strategy.name == "simple" + restarted = _make_cover( + hass, travel_time=25, strategy_name="simple", calibration={"down": 15.0, "up": 16.0} + ) + assert restarted.strategy.name == "simple" + + +@pytest.mark.asyncio +async def test_simple_cover_publishes_no_travel_time(hass) -> None: + """A simple cover shows zero travel time, a calibrated one its measured times (#739).""" + cover = _make_cover(hass, travel_time=0) + attrs = cover._attr_extra_state_attributes + assert (attrs["travel_time"], attrs["travel_time_down"], attrs["travel_time_up"]) == (0, 0.0, 0.0) + cover._promote_to_timed_strategy() + cover._travel_time_down = 15.0 + cover._travel_time_up = 16.5 + cover._refresh_travel_attributes() + assert (attrs["travel_time"], attrs["travel_time_down"], attrs["travel_time_up"]) == (15, 15.0, 16.5) From 082f723b511f52ef6acd73ade2317b13b4791c87 Mon Sep 17 00:00:00 2001 From: Laurens <43173895+GreenGrassBlueOcean@users.noreply.github.com> Date: Tue, 20 Oct 2026 12:29:05 +0200 Subject: [PATCH 3/4] test(cover): cover reset on an uncalibrated simple cover (#739) _demote_to_simple_strategy's early return was never exercised, failing the 100% coverage gate. Drop the unreachable NativePositionStrategy check (reset returns before demotion on advanced covers) and test a reset on a simple cover that was never promoted. Co-Authored-By: Claude Opus 5.5 --- custom_components/myhome/cover.py | 2 +- tests/test_cover_strategies.py | 13 +++++++++++++ 2 files changed, 14 insertions(+), 1 deletion(-) diff --git a/custom_components/myhome/cover.py b/custom_components/myhome/cover.py index 0cc2a467f..c098a3269 100644 --- a/custom_components/myhome/cover.py +++ b/custom_components/myhome/cover.py @@ -1123,7 +1123,7 @@ def _promote_to_timed_strategy(self) -> None: def _demote_to_simple_strategy(self) -> None: """Return a promoted cover to SimpleUpDownStopStrategy when its calibration is reset.""" - if isinstance(self._strategy, (SimpleUpDownStopStrategy, NativePositionStrategy)): + if isinstance(self._strategy, SimpleUpDownStopStrategy): return self._strategy = SimpleUpDownStopStrategy( travel_time=self._strategy.travel_time_down, diff --git a/tests/test_cover_strategies.py b/tests/test_cover_strategies.py index 1e8e0672b..54a8fe758 100644 --- a/tests/test_cover_strategies.py +++ b/tests/test_cover_strategies.py @@ -870,3 +870,16 @@ async def test_simple_cover_publishes_no_travel_time(hass) -> None: cover._travel_time_up = 16.5 cover._refresh_travel_attributes() assert (attrs["travel_time"], attrs["travel_time_down"], attrs["travel_time_up"]) == (15, 15.0, 16.5) + + +@pytest.mark.asyncio +async def test_reset_travel_time_on_uncalibrated_simple_cover(hass) -> None: + """Resetting a simple cover that was never calibrated leaves it simple.""" + cover = _make_cover(hass, travel_time=0) + cover._device_config = MagicMock(return_value={CONF_TRAVEL_TIME: 0}) + cover._fire_calibration_event = MagicMock() + strategy = cover.strategy + await cover.async_reset_travel_time() + assert cover.strategy is strategy + assert cover.assumed_state is True + assert cover.current_cover_position is None From 5882b89debeec14b73ceb79ae2de8ffdb4824e7d Mon Sep 17 00:00:00 2001 From: Laurens <43173895+GreenGrassBlueOcean@users.noreply.github.com> Date: Tue, 20 Oct 2026 13:50:48 +0200 Subject: [PATCH 4/4] feat(cover): strategy-owned lead, transitions and run lifecycle hooks (#739) Addresses the #739 review: a calibrated strategy (#374) now plugs into MyHOMECover without touching cover.py. - Lead: every curtain run takes its lead from strategy.lead_time_for, not only covers with slat_tilt; the same lead drives the freeze, set-position and stop-echo paths. An auto-stop re-anchor clears it for the remaining segment (_move_lead), as a Venetian blind's turned slats did before. - Transitions: with_slat_tilt / promoted / demoted let the strategy pick its successor. Only the built-in classes swap, carrying their timing parameters exactly; any other strategy keeps itself. - Lifecycle: on_run_started, on_run_ended (stop, end_stop, takeover, reversal, abort) and on_position_confirmed (end stop, calibration); position_after may return None while the position is unknown; export_state / restore_state ride on the HA restore data, keyed by strategy name. Built-in strategies are stateless no-ops. - tests/test_cover_strategy_contract.py drives a piecewise, stateful strategy (3 s slat phase without tilt control) through keypad runs, set-position, the stop echo, the 0/100 settle, every run end, unknown position and a restart. Co-Authored-By: Claude Opus 5.5 --- custom_components/myhome/cover.py | 137 ++++--- .../myhome/cover_strategies/__init__.py | 3 +- .../myhome/cover_strategies/base.py | 82 +++- .../myhome/cover_strategies/simple.py | 9 +- .../myhome/cover_strategies/timed.py | 20 +- tests/test_cover_strategy_contract.py | 360 ++++++++++++++++++ 6 files changed, 551 insertions(+), 60 deletions(-) create mode 100644 tests/test_cover_strategy_contract.py diff --git a/custom_components/myhome/cover.py b/custom_components/myhome/cover.py index c098a3269..520a5e65d 100644 --- a/custom_components/myhome/cover.py +++ b/custom_components/myhome/cover.py @@ -37,6 +37,7 @@ from homeassistant.helpers import entity_registry as er from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.event import async_call_later, async_track_state_change_event +from homeassistant.helpers.restore_state import ExtraStoredData, RestoredExtraData from homeassistant.util import dt as dt_util from OWNd.message import ( OWNAutomationCommand, @@ -127,7 +128,6 @@ CoverKinematicsStrategy, NativePositionStrategy, SimpleUpDownStopStrategy, - TimedCurtainStrategy, VenetianTiltStrategy, clamp_slat_time, resolve_cover_strategy, @@ -476,6 +476,9 @@ def __init__( # type: ignore self._run_start_position: int | None = None self._run_start_tilt: int | None = None self._run_lead: float = 0.0 + # Lead still ahead of the segment timed from _move_start_time (0 once an + # auto-stop re-anchors the run on its target). + self._move_lead: float = 0.0 self._pending_stop_context: dict[str, Any] | None = None self._auto_stop_duration: float | None = None @@ -630,22 +633,25 @@ def _anchor_run(self, at: float) -> None: self._run_started_at = at if self._run_start_position is None: self._run_start_position = self._start_position + is_opening = bool(self._attr_is_opening) + is_closing = bool(self._attr_is_closing) if self._slat_tilt and self._run_start_tilt is None: - is_opening = bool(self._attr_is_opening) - is_closing = bool(self._attr_is_closing) self._run_start_tilt = ( self._attr_current_cover_tilt_position if self._attr_current_cover_tilt_position is not None else (0 if is_opening else 100) ) - self._run_lead = self._slat_lead_time(is_opening, is_closing) + # The strategy owns the lead of every run, with or without slat tilt. + self._run_lead = self._move_lead = self._slat_lead_time(is_opening, is_closing) + if is_opening or is_closing: + self._strategy.on_run_started(self, "open" if is_opening else "close", self._run_start_position, at) started = dt_util.utcnow() - timedelta(seconds=max(0.0, time.monotonic() - at)) self._motion_started_at = started.isoformat(timespec="milliseconds") self._refresh_travel_attributes() def _slat_lead_time(self, is_opening: bool, is_closing: bool) -> float: """Seconds a curtain run spends turning the slats before the curtain moves.""" - if not self._slat_tilt or self._is_tilting or not (is_opening or is_closing): + if self._is_tilting or not (is_opening or is_closing): return 0.0 return self._strategy.lead_time_for(is_opening, entity=self) @@ -662,11 +668,11 @@ def _turn_slats_for_run( ) def _freeze_position(self, at: float) -> None: - """Turn the running estimate into a fixed position as of ``at``.""" + """Turn the running estimate into a fixed position as of ``at`` (a stop ended the run).""" is_opening = bool(self._attr_is_opening) is_closing = bool(self._attr_is_closing) run_started_at = self._run_started_at - lead = self._slat_lead_time(is_opening, is_closing) + lead = self._move_lead if self._move_start_time is not None else 0.0 if self._run_started_at is not None: self._last_run = { "seconds": round(max(0.0, at - self._run_started_at), 2), @@ -687,7 +693,8 @@ def _freeze_position(self, at: float) -> None: entity=self, lead=lead, ) - frozen = frozen_pos if frozen_pos is not None else self._start_position + # None: the strategy does not know the position (yet). + frozen = frozen_pos else: frozen = compute_freeze_position( self._start_position, @@ -704,11 +711,13 @@ def _freeze_position(self, at: float) -> None: self._start_position = frozen self._move_start_time = None self._turn_slats_for_run(at, run_started_at, is_opening, is_closing) + if run_started_at is not None and (is_opening or is_closing): + self._strategy.on_run_ended(self, "stop", frozen, at) self._attr_is_opening = False self._attr_is_closing = False self._run_start_position = None self._run_start_tilt = None - self._run_lead = 0.0 + self._run_lead = self._move_lead = 0.0 self._auto_stop_duration = None if self._attr_current_cover_position is not None: self._attr_is_closed = (self._attr_current_cover_position == 0) @@ -721,9 +730,11 @@ def _abort_motion(self) -> None: """Abort motion state when a command frame failed to reach the bus.""" self._cancel_stop_task() self._end_echo_window() + if self._run_started_at is not None: + self._strategy.on_run_ended(self, "abort", self._attr_current_cover_position, time.monotonic()) self._run_start_position = None self._run_start_tilt = None - self._run_lead = 0.0 + self._run_lead = self._move_lead = 0.0 self._pending_stop_context = None self._auto_stop_duration = None self._move_start_time = None @@ -818,14 +829,8 @@ def current_cover_position(self) -> int | None: def _enable_slat_tilt(self) -> None: """Enable Venetian blind slat / louvre tilt capabilities.""" self._slat_tilt = True - if isinstance(self._strategy, TimedCurtainStrategy) and not isinstance(self._strategy, VenetianTiltStrategy): - self._strategy = VenetianTiltStrategy( - slat_time=self._strategy.slat_time, - travel_time=self._strategy.travel_time_down, - travel_time_up=self._strategy.travel_time_up, - travel_time_down=self._strategy.travel_time_down, - min_motor_pulse=self._strategy.min_motor_pulse, - ) + # The strategy decides: built-in timed becomes Venetian, others stay. + self._strategy = self._strategy.with_slat_tilt() self._attr_device_class = CoverDeviceClass.BLIND current_features = self._attr_supported_features or CoverEntityFeature(0) self._attr_supported_features = ( @@ -907,6 +912,25 @@ async def async_restore_last_state(self, last_state: Any) -> None: self._attr_current_cover_tilt_position = max(0, min(100, int(round(float(last_tilt))))) except (ValueError, TypeError): pass + await self._async_restore_strategy_state() + + @property + def extra_restore_state_data(self) -> ExtraStoredData | None: + """The strategy's model state, kept across restarts when it has any.""" + state = self._strategy.export_state() + if state is None: + return None + return RestoredExtraData({"strategy": self._strategy.name, "state": state}) + + async def _async_restore_strategy_state(self) -> None: + """Hand the strategy the state it exported before the restart (same strategy only).""" + try: + extra = await self.async_get_last_extra_data() + except Exception: # pylint: disable=broad-except # not registered with the restore cache + return + data = extra.as_dict() if extra is not None else {} + if data.get("strategy") == self._strategy.name and isinstance(data.get("state"), dict): + self._strategy.restore_state(data["state"]) async def async_will_remove_from_hass(self) -> None: """Run when entity will be removed from hass.""" @@ -1069,6 +1093,7 @@ async def async_calibrate(self) -> dict: # type: ignore # The sequence ends with the cover fully open. self._attr_current_cover_position = 100 self._start_position = 100 + self._strategy.on_position_confirmed(self, 100, "calibration") if self._slat_tilt: self._attr_current_cover_tilt_position = 100 self._attr_is_closed = False @@ -1092,52 +1117,46 @@ async def async_calibrate(self) -> dict: # type: ignore return result def _promote_to_timed_strategy(self) -> None: - """Promote a SimpleUpDownStopStrategy cover to Timed/Venetian once calibrated. + """Let the strategy pick its successor once travel times are calibrated. - An explicit ``strategy: simple`` stays simple, as it does after a restart. + The built-in simple strategy becomes timed (or Venetian); any other + strategy, a custom one included, keeps itself and its parameters. An + explicit ``strategy: simple`` stays simple, as it does after a restart. """ - if isinstance(self._strategy, SimpleUpDownStopStrategy) and self._strategy_name != "simple": - if self._slat_tilt: - self._strategy = VenetianTiltStrategy( - slat_time=self._strategy.slat_time, - travel_time=self._strategy.travel_time_down, - travel_time_up=self._strategy.travel_time_up, - travel_time_down=self._strategy.travel_time_down, - min_motor_pulse=self._strategy.min_motor_pulse, - ) - else: - self._strategy = TimedCurtainStrategy( - travel_time=self._strategy.travel_time_down, - travel_time_up=self._strategy.travel_time_up, - travel_time_down=self._strategy.travel_time_down, - min_motor_pulse=self._strategy.min_motor_pulse, - slat_time=self._strategy.slat_time, - ) - self._attr_assumed_state = False - self._attr_supported_features = ( - (self._attr_supported_features or CoverEntityFeature(0)) - | CoverEntityFeature.SET_POSITION - ) - if self._attr_current_cover_position is None: - self._attr_current_cover_position = self._start_position + if self._strategy_name == "simple": + return + promoted = self._strategy.promoted(slat_tilt=self._slat_tilt) + if promoted is self._strategy: + return + self._strategy = promoted + self._attr_assumed_state = False + self._attr_supported_features = ( + (self._attr_supported_features or CoverEntityFeature(0)) + | CoverEntityFeature.SET_POSITION + ) + if self._attr_current_cover_position is None: + self._attr_current_cover_position = self._start_position def _demote_to_simple_strategy(self) -> None: - """Return a promoted cover to SimpleUpDownStopStrategy when its calibration is reset.""" - if isinstance(self._strategy, SimpleUpDownStopStrategy): + """Let the strategy pick its successor when the calibration is reset to no travel time. + + The built-in timed strategy becomes simple again; any other strategy keeps itself. + """ + demoted = self._strategy.demoted() + if demoted is self._strategy: return - self._strategy = SimpleUpDownStopStrategy( - travel_time=self._strategy.travel_time_down, - travel_time_up=self._strategy.travel_time_up, - travel_time_down=self._strategy.travel_time_down, - min_motor_pulse=self._strategy.min_motor_pulse, - slat_time=self._strategy.slat_time, - ) + self._strategy = demoted self._attr_assumed_state = True self._attr_supported_features = ( self._attr_supported_features or CoverEntityFeature(0) ) & ~CoverEntityFeature.SET_POSITION self._attr_current_cover_position = None + def _end_run_on_reversal(self, now: float) -> None: + """Tell the strategy the opposite direction on the bus ended the running run.""" + if not self._advanced and self._run_started_at is not None: + self._strategy.on_run_ended(self, "reversal", self._start_position, now) + async def async_stop_calibration(self) -> None: """Stop calibration on this cover's gateway.""" gw_mac = getattr(self._gateway_handler, "mac", None) @@ -1412,7 +1431,7 @@ async def _async_move(self, direction: str, is_tilting: bool = False) -> asyncio self._start_position = self.current_cover_position if self.current_cover_position is not None else (0 if direction == "open" else 100) self._run_start_position = self._start_position self._run_start_tilt = None - self._run_lead = 0.0 + self._run_lead = self._move_lead = 0.0 self._pending_stop_context = None self._auto_stop_duration = None # The clock starts when the frame is written (see _track_write), @@ -1477,6 +1496,8 @@ async def _settle() -> None: self._start_position = self._attr_current_cover_position if self._slat_tilt: self._attr_current_cover_tilt_position = 100 if opening else 0 + self._strategy.on_run_ended(self, "end_stop", self._start_position, time.monotonic()) + self._strategy.on_position_confirmed(self, self._start_position, "end_stop") self._move_start_time = None self._run_started_at = None self._motion_started_at = None @@ -1485,7 +1506,7 @@ async def _settle() -> None: self._attr_is_closed = not opening self._run_start_position = None self._run_start_tilt = None - self._run_lead = 0.0 + self._run_lead = self._move_lead = 0.0 self._pending_stop_context = None self._auto_stop_duration = None self._refresh_travel_attributes() @@ -1626,6 +1647,7 @@ async def _auto_stop() -> None: if self._slat_tilt: self._attr_current_cover_tilt_position = 100 if opening else 0 self._move_start_time = time.monotonic() + self._move_lead = 0.0 # the lead was spent before the target await self.async_stop_cover() if self.hass is not None: self.async_write_ha_state() @@ -1653,7 +1675,7 @@ async def async_stop_cover(self, **kwargs: Any) -> None: # pylint: disable=unus "is_opening": is_opening, "is_closing": is_closing, "auto_stop_duration": self._auto_stop_duration, - "lead": self._run_lead if self._run_start_tilt is not None else self._slat_lead_time(is_opening, is_closing), + "lead": self._run_lead if self._run_started_at is not None else self._slat_lead_time(is_opening, is_closing), "run_start_tilt": self._run_start_tilt if self._run_start_tilt is not None else self._attr_current_cover_tilt_position, } self._begin_command("stop") @@ -1845,6 +1867,7 @@ def _end_curtain_for_takeover(self, now: float) -> None: self._start_position = self.current_cover_position if self.current_cover_position is not None else self._start_position self._attr_current_cover_position = self._start_position self._turn_slats_for_run(now, self._run_started_at, bool(self._attr_is_opening), bool(self._attr_is_closing)) + self._strategy.on_run_ended(self, "takeover", self._start_position, now) self._move_start_time = None self._run_started_at = None @@ -2194,6 +2217,7 @@ def handle_event(self, message: OWNMessage) -> None: self._start_position = self.current_cover_position if self.current_cover_position is not None else 0 if was_closing: self._turn_slats_for_run(now, self._run_started_at, False, True) + self._end_run_on_reversal(now) was_already_opening = bool(self._attr_is_opening) self._attr_is_opening = True self._attr_is_closing = False @@ -2223,6 +2247,7 @@ def handle_event(self, message: OWNMessage) -> None: self._start_position = self.current_cover_position if self.current_cover_position is not None else 100 if was_opening: self._turn_slats_for_run(now, self._run_started_at, True, False) + self._end_run_on_reversal(now) was_already_closing = bool(self._attr_is_closing) self._attr_is_opening = False self._attr_is_closing = True diff --git a/custom_components/myhome/cover_strategies/__init__.py b/custom_components/myhome/cover_strategies/__init__.py index d1aca7c54..becf9f0ee 100644 --- a/custom_components/myhome/cover_strategies/__init__.py +++ b/custom_components/myhome/cover_strategies/__init__.py @@ -2,7 +2,7 @@ from __future__ import annotations from ..const import DEFAULT_MIN_MOTOR_PULSE, DEFAULT_TRAVEL_TIME -from .base import BaseCoverStrategy, CoverKinematicsStrategy, clamp_slat_time +from .base import RUN_END_REASONS, BaseCoverStrategy, CoverKinematicsStrategy, clamp_slat_time from .native import NativePositionStrategy from .simple import SimpleUpDownStopStrategy from .timed import TimedCurtainStrategy @@ -13,6 +13,7 @@ "CoverKinematicsStrategy", "MIN_TILT_PULSE", "NativePositionStrategy", + "RUN_END_REASONS", "SimpleUpDownStopStrategy", "TimedCurtainStrategy", "VenetianTiltStrategy", diff --git a/custom_components/myhome/cover_strategies/base.py b/custom_components/myhome/cover_strategies/base.py index abd7fda87..8dcf3687e 100644 --- a/custom_components/myhome/cover_strategies/base.py +++ b/custom_components/myhome/cover_strategies/base.py @@ -1,7 +1,7 @@ """Base protocol and shared timing state for pluggable MyHOME cover kinematics strategies.""" from __future__ import annotations -from typing import Any, Protocol, runtime_checkable +from typing import Any, Protocol, TypeVar, cast, runtime_checkable from homeassistant.components.cover import CoverEntityFeature @@ -34,6 +34,13 @@ class CoverKinematicsStrategy(Protocol): stop-echo re-anchoring via ``reanchor_on_stop_echo``, allowing linear timed covers, Venetian slat+roll blinds, and non-linear calibrated geometry strategies (#374) to plug into ``MyHOMECover`` without modifying the entity class. + + The entity asks the strategy for the lead of every curtain run (not only on + slat blinds), lets it pick its successor when slat tilt is enabled or the + calibration changes (``with_slat_tilt``, ``promoted``, ``demoted``: a custom + strategy keeps itself), reports each run's start and end and every confirmed + position, accepts ``None`` from ``position_after`` while the position is + unknown, and keeps ``export_state`` across restarts. """ travel_time_down: float @@ -129,6 +136,39 @@ def reanchor_on_stop_echo( def current_position(self, entity: Any, now: float) -> int | None: """Return current cover position (0..100) or None if position is unknown.""" + def with_slat_tilt(self) -> CoverKinematicsStrategy: + """Return the strategy to use once slat tilt is enabled (``self`` to keep this one).""" + + def promoted(self, *, slat_tilt: bool) -> CoverKinematicsStrategy: + """Return the strategy to use once travel times are calibrated (``self`` to keep this one).""" + + def demoted(self) -> CoverKinematicsStrategy: + """Return the strategy to use once the calibration is reset to no travel time.""" + + def on_run_started(self, entity: Any, direction: str, start_position: int | None, at: float) -> None: + """A curtain run started at monotonic ``at`` (called again if the run is re-anchored).""" + + def on_run_ended(self, entity: Any, reason: str, position: int | None, at: float) -> None: + """A curtain run ended at monotonic ``at``; ``reason`` is one of ``RUN_END_REASONS``.""" + + def on_position_confirmed(self, entity: Any, position: int, source: str) -> None: + """The cover is known to be at ``position`` (``source``: ``end_stop``, ``calibration``).""" + + def export_state(self) -> dict[str, Any] | None: + """Return JSON-serialisable model state to keep across restarts, or None.""" + + def restore_state(self, state: dict[str, Any]) -> None: + """Restore the state ``export_state`` returned before a restart.""" + + +#: Why a curtain run ended: a stop frame or stop status, the motor reaching its +#: limit switch (settled on a timer), a new command taking the motor over, the +#: opposite direction seen on the bus, or a command that never reached the bus. +RUN_END_REASONS = ("stop", "end_stop", "takeover", "reversal", "abort") + + +_StrategyT = TypeVar("_StrategyT", bound="BaseCoverStrategy") + class BaseCoverStrategy: """Shared timing configuration and default slat helpers for WHO=2 cover strategies.""" @@ -220,3 +260,43 @@ def tilt_pulse_duration_for( return 0.0 return run_duration + def _rebuilt_as(self, cls: type[_StrategyT]) -> _StrategyT: + """A ``cls`` instance carrying this strategy's timing parameters unchanged.""" + other = cls() + other.travel_time_down = self.travel_time_down + other.travel_time_up = self.travel_time_up + other.min_motor_pulse = self.min_motor_pulse + other.slat_time = self.slat_time + return other + + def _kept(self) -> CoverKinematicsStrategy: + """This strategy as its own successor (every concrete subclass is a full strategy).""" + return cast(CoverKinematicsStrategy, self) + + def with_slat_tilt(self) -> CoverKinematicsStrategy: + """Keep this strategy when slat tilt is enabled (built-in timed covers override).""" + return self._kept() + + def promoted(self, *, slat_tilt: bool) -> CoverKinematicsStrategy: # pylint: disable=unused-argument + """Keep this strategy once calibrated (the built-in simple strategy overrides).""" + return self._kept() + + def demoted(self) -> CoverKinematicsStrategy: + """Keep this strategy when its calibration is reset (built-in timed covers override).""" + return self._kept() + + def on_run_started(self, entity: Any, direction: str, start_position: int | None, at: float) -> None: + """No run tracking by default.""" + + def on_run_ended(self, entity: Any, reason: str, position: int | None, at: float) -> None: + """No run tracking by default.""" + + def on_position_confirmed(self, entity: Any, position: int, source: str) -> None: + """No position tracking by default.""" + + def export_state(self) -> dict[str, Any] | None: + """Stateless by default: nothing to keep across restarts.""" + return None + + def restore_state(self, state: dict[str, Any]) -> None: + """Stateless by default: nothing to restore.""" diff --git a/custom_components/myhome/cover_strategies/simple.py b/custom_components/myhome/cover_strategies/simple.py index 3bac22dc6..7dd974fad 100644 --- a/custom_components/myhome/cover_strategies/simple.py +++ b/custom_components/myhome/cover_strategies/simple.py @@ -5,7 +5,9 @@ from homeassistant.components.cover import CoverEntityFeature -from .base import BaseCoverStrategy +from .base import BaseCoverStrategy, CoverKinematicsStrategy +from .timed import TimedCurtainStrategy +from .venetian import VenetianTiltStrategy class SimpleUpDownStopStrategy(BaseCoverStrategy): @@ -70,3 +72,8 @@ def current_position(self, entity: Any, now: float) -> int | None: """Simple covers do not fabricate a percentage position.""" return None + def promoted(self, *, slat_tilt: bool) -> CoverKinematicsStrategy: + """Become timed (or Venetian) once travel times are calibrated; a subclass stays itself.""" + if type(self) is not SimpleUpDownStopStrategy: + return self._kept() + return self._rebuilt_as(VenetianTiltStrategy if slat_tilt else TimedCurtainStrategy) diff --git a/custom_components/myhome/cover_strategies/timed.py b/custom_components/myhome/cover_strategies/timed.py index 3a7ece330..9115f26ed 100644 --- a/custom_components/myhome/cover_strategies/timed.py +++ b/custom_components/myhome/cover_strategies/timed.py @@ -8,7 +8,7 @@ from ..const import DEFAULT_MIN_MOTOR_PULSE, DEFAULT_TRAVEL_TIME, LOGGER from ..cover_motion import compute_run_duration -from .base import BaseCoverStrategy +from .base import BaseCoverStrategy, CoverKinematicsStrategy class TimedCurtainStrategy(BaseCoverStrategy): @@ -178,12 +178,30 @@ def current_position(self, entity: Any, now: float) -> int | None: ): direction = "open" if is_opening else "close" elapsed = max(0.0, now - float(move_start)) + # The lead still ahead of this segment, as the entity anchored it. + move_lead = getattr(entity, "_move_lead", None) return self.position_after( int(getattr(entity, "_start_position", 50)), elapsed, direction, entity=entity, + lead=float(move_lead) if isinstance(move_lead, (int, float)) else None, ) pos: int | None = getattr(entity, "_attr_current_cover_position", None) return pos + def with_slat_tilt(self) -> CoverKinematicsStrategy: + """The built-in timed strategy becomes Venetian; a subclass stays itself.""" + if type(self) is not TimedCurtainStrategy: + return self._kept() + from .venetian import VenetianTiltStrategy # pylint: disable=import-outside-toplevel + + return self._rebuilt_as(VenetianTiltStrategy) + + def demoted(self) -> CoverKinematicsStrategy: + """The built-in timed strategy becomes simple again; a subclass stays itself.""" + if type(self) is not TimedCurtainStrategy: + return self._kept() + from .simple import SimpleUpDownStopStrategy # pylint: disable=import-outside-toplevel + + return self._rebuilt_as(SimpleUpDownStopStrategy) diff --git a/tests/test_cover_strategy_contract.py b/tests/test_cover_strategy_contract.py new file mode 100644 index 000000000..07032c4a7 --- /dev/null +++ b/tests/test_cover_strategy_contract.py @@ -0,0 +1,360 @@ +"""The CoverKinematicsStrategy contract driven through MyHOMECover (#374, #739). + +A piecewise, stateful strategy (a slat phase the cover cannot tilt, a slow +lift-off and a fast roll, a position unknown until a full run) is plugged +into an unmodified MyHOMECover and driven through set-position, keypad runs, +the stop echo, the 0/100 settle, run lifecycle hooks and a restart. +""" +from __future__ import annotations + +from typing import Any +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from homeassistant.components.cover import CoverEntityFeature +from homeassistant.helpers.restore_state import RestoredExtraData +from OWNd.message import OWNEvent + +from custom_components.myhome.cover import MyHOMECover +from custom_components.myhome.cover_motion import END_SETTLE_MARGIN +from custom_components.myhome.cover_strategies import ( + RUN_END_REASONS, + BaseCoverStrategy, + SimpleUpDownStopStrategy, + TimedCurtainStrategy, +) + +SLAT_PHASE = 3.0 # seconds the motor turns slats before the curtain lifts +CURTAIN = 10.0 # curtain seconds from closed to open after the slat phase + + +def _time_at(position: float) -> float: + """Curtain seconds from closed to ``position``: 0-20 % in 5 s, 20-100 % in 5 s.""" + position = max(0.0, min(100.0, position)) + return position / 20.0 * 5.0 if position <= 20 else 5.0 + (position - 20.0) / 80.0 * 5.0 + + +def _position_at(seconds: float) -> float: + seconds = max(0.0, min(CURTAIN, seconds)) + return 20.0 * seconds / 5.0 if seconds <= 5.0 else 20.0 + (seconds - 5.0) * 80.0 / 5.0 + + +class PiecewiseGeometryStrategy(TimedCurtainStrategy): + """A non-linear, stateful model of the kind #374 calibrates.""" + + def __init__(self, *, known: bool = True) -> None: + super().__init__(travel_time=SLAT_PHASE + CURTAIN) + self.known = known + self.events: list[tuple[Any, ...]] = [] + + @property + def name(self) -> str: + return "piecewise" + + def curtain_travel_for(self, opening: bool) -> float: + return CURTAIN + + def lead_time_for(self, opening: bool, *, entity: Any = None) -> float: + return SLAT_PHASE + + def position_after( + self, start: int, elapsed: float, direction: str, *, entity: Any = None, lead: float | None = None + ) -> int | None: + if not self.known: + return None + curtain = max(0.0, elapsed - (SLAT_PHASE if lead is None else lead)) + moved = _time_at(start) + (curtain if direction == "open" else -curtain) + return int(round(_position_at(moved))) + + def duration_for( + self, start: int, target: int, direction: str | None = None, *, entity: Any = None + ) -> float: + if start == target: + return 0.0 + return SLAT_PHASE + abs(_time_at(target) - _time_at(start)) + + def on_run_started(self, entity: Any, direction: str, start_position: int | None, at: float) -> None: + self.events.append(("started", direction, start_position, at)) + + def on_run_ended(self, entity: Any, reason: str, position: int | None, at: float) -> None: + assert reason in RUN_END_REASONS + self.events.append(("ended", reason, position, at)) + + def on_position_confirmed(self, entity: Any, position: int, source: str) -> None: + self.known = True + self.events.append(("confirmed", position, source)) + + def export_state(self) -> dict[str, Any] | None: + return {"known": self.known} + + def restore_state(self, state: dict[str, Any]) -> None: + self.known = bool(state.get("known")) + + +def _cover(hass, strategy: Any = None, **kwargs: Any) -> MyHOMECover: + gw = MagicMock() + gw.mac = "00:03:50:00:12:34" + gw.log_id = "[TestGW]" + gw.send = AsyncMock() + args: dict[str, Any] = { + "hass": hass, + "name": "Cover 21", + "entity_name": "Cover 21", + "device_id": "21", + "who": "2", + "where": "21", + "interface": None, + "advanced": False, + "manufacturer": "BTicino", + "model": "Shutter", + "gateway": gw, + "travel_time": 13, + **kwargs, + } + with patch("custom_components.myhome.myhome_device.Entity.__init__", return_value=None): + cover = MyHOMECover(**args) + cover.entity_id = "cover.cover_21" + cover.async_write_ha_state = MagicMock() + cover.async_schedule_update_ha_state = MagicMock() + if strategy is not None: + cover._strategy = strategy + return cover + + +class _Clock: + """Monotonic time the patched asyncio.sleep advances.""" + + def __init__(self, start: float = 1000.0) -> None: + self.now = start + self.slept: list[float] = [] + + def monotonic(self) -> float: + return self.now + + async def sleep(self, seconds: float) -> None: + self.slept.append(seconds) + self.now += max(0.0, seconds) + + +def _event(cover: MyHOMECover, frame: str) -> None: + cover.handle_event(OWNEvent.parse(frame)) + + +@pytest.mark.asyncio +async def test_piecewise_strategy_drives_cover_without_cover_py_changes(hass) -> None: + """Set-position, keypad runs, the stop echo and the 0/100 settle all follow the strategy.""" + strategy = PiecewiseGeometryStrategy() + cover = _cover(hass, strategy) + clock = _Clock() + cover._attr_current_cover_position = cover._start_position = 0 + cover._attr_is_closed = True + + with ( + patch("custom_components.myhome.cover.time.monotonic", side_effect=clock.monotonic), + patch("asyncio.sleep", side_effect=clock.sleep), + ): + # A 3 s keypad run from closed only turns the slats: still 0, not 30. + _event(cover, "*2*1*21##") + clock.now += 3.0 + _event(cover, "*2*0*21##") + assert cover.current_cover_position == 0 + assert cover.is_closed is True + + # 3 s of slats plus 5 s of slow lift-off reach 20 %. + clock.now += 10.0 + _event(cover, "*2*1*21##") + clock.now += 8.0 + _event(cover, "*2*0*21##") + assert cover.current_cover_position == 20 + + # Back to closed, then set-position 0 -> 20: runs the strategy's 8 s + # and reports 20, not 50. + clock.now += 10.0 + _event(cover, "*2*2*21##") + clock.now += 13.0 + _event(cover, "*2*0*21##") + assert cover.current_cover_position == 0 + clock.now += 10.0 + await cover.async_set_cover_position(position=20) + _event(cover, "*2*1*21##") # the gateway's direction echo anchors the run + await cover._stop_task + assert 8.0 in clock.slept + assert str(cover._gateway_handler.send.await_args_list[-1].args[0]) == "*2*0*21##" + assert cover.current_cover_position == 20 + + # Stop echo: the relayed stop re-anchors with the strategy's lead. + clock.now += 10.0 + await cover.async_open_cover() + _event(cover, "*2*1*21##") + clock.now += 4.0 + await cover.async_stop_cover() + clock.now += 0.3 + _event(cover, "*2*0*21##") + assert cover.current_cover_position == strategy.position_after(20, 4.3, "open", lead=SLAT_PHASE) == 41 + + # 0/100: a full run settles on travel + END_SETTLE_MARGIN and confirms the end. + clock.now += 10.0 + await cover.async_set_cover_position(position=100) + _event(cover, "*2*1*21##") + await cover._stop_task + assert SLAT_PHASE + CURTAIN + END_SETTLE_MARGIN in clock.slept + assert cover.current_cover_position == 100 + + started = [e for e in strategy.events if e[0] == "started"] + ended = [e[1:3] for e in strategy.events if e[0] == "ended"] + assert started[0][1:3] == ("open", 0) + assert ended == [ + ("stop", 0), + ("stop", 20), + ("stop", 0), + ("stop", 20), + ("stop", 36), # frozen at the stop write; the relayed echo then re-anchors to 41 + ("end_stop", 100), + ] + assert strategy.events[-1] == ("confirmed", 100, "end_stop") + + +@pytest.mark.asyncio +async def test_run_end_hooks_for_takeover_reversal_and_abort(hass) -> None: + """Every way a run can end reaches the strategy.""" + strategy = PiecewiseGeometryStrategy() + cover = _cover(hass, strategy) + clock = _Clock() + cover._attr_current_cover_position = cover._start_position = 0 + + with patch("custom_components.myhome.cover.time.monotonic", side_effect=clock.monotonic): + # Opposite direction on the bus. + _event(cover, "*2*1*21##") + clock.now += 8.0 + _event(cover, "*2*2*21##") + # A new command takes the motor over. + clock.now += 1.0 + await cover.async_open_cover() + cover._cancel_stop_task() + _event(cover, "*2*1*21##") + # A command that never reached the bus. + clock.now += 1.0 + cover._abort_motion() + + reasons = [e[1] for e in strategy.events if e[0] == "ended"] + assert reasons == ["reversal", "takeover", "abort"] + assert ("ended", "reversal", 20, 1008.0) in strategy.events + + +@pytest.mark.asyncio +async def test_position_unknown_until_a_full_run(hass) -> None: + """A strategy may leave the position unknown until a run confirms an end.""" + strategy = PiecewiseGeometryStrategy(known=False) + cover = _cover(hass, strategy) + clock = _Clock() + cover._attr_current_cover_position = None + + with ( + patch("custom_components.myhome.cover.time.monotonic", side_effect=clock.monotonic), + patch("asyncio.sleep", side_effect=clock.sleep), + ): + _event(cover, "*2*1*21##") + clock.now += 6.0 + _event(cover, "*2*0*21##") + assert cover.current_cover_position is None + + await cover.async_set_cover_position(position=100) + _event(cover, "*2*1*21##") + await cover._stop_task + + assert strategy.known is True + assert cover.current_cover_position == 100 + assert ("confirmed", 100, "end_stop") in strategy.events + + +@pytest.mark.asyncio +async def test_strategy_state_survives_a_restart(hass) -> None: + """export_state goes into the restore data and comes back to the same strategy only.""" + strategy = PiecewiseGeometryStrategy(known=True) + cover = _cover(hass, strategy) + stored = cover.extra_restore_state_data + assert stored is not None + assert stored.as_dict() == {"strategy": "piecewise", "state": {"known": True}} + + last_state = MagicMock(state="open", attributes={}) + + restarted = _cover(hass, PiecewiseGeometryStrategy(known=False)) + restarted.async_get_last_extra_data = AsyncMock(return_value=RestoredExtraData(stored.as_dict())) + await restarted.async_restore_last_state(last_state) + assert restarted.strategy.known is True + + # Another strategy's state is ignored, and so is a missing record. + other = _cover(hass, PiecewiseGeometryStrategy(known=False)) + other.async_get_last_extra_data = AsyncMock( + return_value=RestoredExtraData({"strategy": "timed", "state": {"known": True}}) + ) + await other.async_restore_last_state(last_state) + assert other.strategy.known is False + other.async_get_last_extra_data = AsyncMock(return_value=None) + await other.async_restore_last_state(last_state) + assert other.strategy.known is False + + # Not registered with the restore cache: nothing to restore, no error. + other.async_get_last_extra_data = AsyncMock(side_effect=RuntimeError("no restore cache")) + await other.async_restore_last_state(last_state) + assert other.strategy.known is False + + # Built-in strategies are stateless: no restore data at all. + assert _cover(hass).extra_restore_state_data is None + + +@pytest.mark.asyncio +async def test_custom_strategies_survive_strategy_transitions(hass) -> None: + """Slat tilt, calibration and its reset keep a custom strategy and its parameters.""" + strategy = PiecewiseGeometryStrategy() + strategy.min_motor_pulse = 0.7 + cover = _cover(hass, strategy) + + cover._enable_slat_tilt() + assert cover.strategy is strategy + assert cover.supported_features & CoverEntityFeature.SET_TILT_POSITION + + cover._device_config = MagicMock(return_value={"travel_time": 0}) + cover._fire_calibration_event = MagicMock() + await cover.async_reset_travel_time() + assert cover.strategy is strategy + assert cover.supported_features & CoverEntityFeature.SET_POSITION + + class CustomSimple(SimpleUpDownStopStrategy): + pass + + simple = CustomSimple() + uncalibrated = _cover(hass, simple, travel_time=0) + uncalibrated._promote_to_timed_strategy() + assert uncalibrated.strategy is simple + + +def test_builtin_transitions_carry_the_timing_parameters() -> None: + """Built-in successors keep every timing parameter exactly, a zero travel time included.""" + simple = SimpleUpDownStopStrategy() + simple.travel_time_down, simple.travel_time_up = 21.5, 23.0 + simple.min_motor_pulse, simple.slat_time = 0.6, 1.5 + + timed = simple.promoted(slat_tilt=False) + venetian = simple.promoted(slat_tilt=True) + assert (type(timed).__name__, type(venetian).__name__) == ("TimedCurtainStrategy", "VenetianTiltStrategy") + assert type(timed.with_slat_tilt()).__name__ == "VenetianTiltStrategy" + assert venetian.with_slat_tilt() is venetian + assert timed.promoted(slat_tilt=True) is timed # already calibrated: nothing to promote + for successor in (timed, venetian, timed.with_slat_tilt()): + assert (successor.travel_time_down, successor.travel_time_up) == (21.5, 23.0) + assert (successor.min_motor_pulse, successor.slat_time) == (0.6, 1.5) + + timed.travel_time_down = timed.travel_time_up = 0.0 + demoted = timed.demoted() + assert isinstance(demoted, SimpleUpDownStopStrategy) + assert (demoted.travel_time_down, demoted.travel_time_up) == (0.0, 0.0) + assert venetian.demoted() is venetian + assert demoted.demoted() is demoted + + base = BaseCoverStrategy() + assert base.export_state() is None + base.restore_state({}) + base.on_run_started(None, "open", 0, 0.0) + base.on_run_ended(None, "stop", 0, 0.0) + base.on_position_confirmed(None, 0, "end_stop")