diff --git a/CLAUDE.md b/CLAUDE.md index 4fc67e2..01f9d01 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -175,6 +175,42 @@ Triggers (OR-combined, no arming): `hueman security on|off` (writes the adapter (e.g. Home Assistant) watches the cue file / webhook and plays the matching sound. That adapter is out of this repo's scope. +## Night-guide — daemon-native motion path lighting, with a clean hand-back + +`circadian_daemon.night_guide` (optional) is a third actor sharing the driven +zone with the operator and the circadian curve: while the zone is out of the +circadian window (parked at `night_look`, or off), motion on a configured +MotionAware `area` briefly raises it to a soft guide `look` (e.g. dim red, for +a night trip to the bathroom/kitchen), then hands control back once `timeout` +elapses with no further motion — repeated motion extends the episode from the +*latest* event, not the first, so it doesn't blink off mid-trip. + +The hand-back is exact where it has to be and recomputed everywhere else: +- A real manual override showing when the guide engages (`SUSPENDED`) gets + snapshotted (a raw `grouped_light` GET) before the guide look overwrites it, + and restored **verbatim** on hand-back — a manual look is arbitrary, not + derivable from anything else, so it has to be remembered. +- Otherwise (ordinary circadian/`night_look` was showing) there's always a + derivable correct target, so hand-back recomputes it fresh instead of + replaying a snapshot — the current curve sample if the window opened back + up during the episode, otherwise the normal resting state (`night_look` or + off). Same recompute-when-possible split `_restore_after_security` uses. + `Hold` alone would leave the guide look showing forever — it assumes + nothing changed underneath it, which is false here. + +Built daemon-native (pure `nightguide_control.py` timing + I/O in +`circadian_daemon.py`), not as a bridge-native MotionAware `behavior_instance` +or left to the Bridge Pro's own convenience-motion feature: neither has any +concept of "restore whatever was there before" — both only know a *fixed* +configured fallback action, which is exactly what made native motion recalls +fight manual overrides and the daemon's own resume logic before this existed +(2026-08-08 incident, ops-repo memory `resume-race-stale-cmd-reference`). +Consuming the same MotionAware SSE events `rhythm` already reads (both are +independent consumers of the same event — see `_handle_event`), night-guide +never fights the operator (an override suspends and stays suspended straight +through a guide episode) or the curve (circadian reclaims the zone the +instant its window opens, mid-episode or not). + ## Rhythm engine — observe-stage day-phase inference `rhythm:` (optional, alongside `circadian_daemon:`) runs a closed-loop @@ -206,6 +242,7 @@ The codebase is split into a **pure decision layer** (fully unit-tested, no I/O) | `circadian_control.py` | Pure daemon state machine (`CircadianController`): drive window, `DriveTo`/`FadeOff`/`Hold` decisions, settle-and-compare override detection. | | `bias_control.py` | Pure TV-bias core: `bias_actions` (per-light hold/drive/off with edge-aware fades) + `TriggerAggregator` (OR-combined, debounced trigger sources). | | `security_control.py` | Pure security-mode core: `SecurityController` (ALERT breathe → luminance-capped CHAOS frames). | +| `nightguide_control.py` | Pure night-guide timing core: `NightGuideController` (IDLE ↔ GUIDING, timeout measured from the latest motion). | | `engine.py` | Per-area state machine (`PolicyEngine`) for the legacy `watch` runtime. Consumes `MotionPolicy` + explicit `ts` floats; emits `Action` values. Four phases: `STANDBY`, `ACTIVE`, `DIMMING`, `OVERRIDDEN`. | | `payload.py` | Converts engine `TargetState` → CLIP v2 request bodies (sRGB `#rrggbb` → CIE xy). | | `nightmotion.py` | Pure night-motion helpers: builds CLIP `scene` bodies, transforms the MotionAware `behavior_instance`, and `scene_actions_match` (tolerant scene-look diff, reused by the circadian reconciler). | @@ -220,7 +257,7 @@ The codebase is split into a **pure decision layer** (fully unit-tested, no I/O) | `client.py` | CLIP API v2 HTTP client (`HueClient`). | | `pin.py` | Trust-on-first-use TLS pinning (SHA-256 of bridge cert → `.hue-pin.json`). | | `state.py` | Loads live bridge state; resolves resource names → ids (`BridgeState`), including MotionAware areas/services. | -| `circadian_daemon.py` | The resident daemon (`CircadianDaemon`): 60s curve ticks + SSE event loop + TV-bias triggers (probe thread / SSE / control files) + security show + rhythm-engine ticks, all serialised under one lock. | +| `circadian_daemon.py` | The resident daemon (`CircadianDaemon`): 60s curve ticks + SSE event loop + TV-bias triggers (probe thread / SSE / control files) + security show + night-guide motion hold + rhythm-engine ticks, all serialised under one lock. | | `watch.py` | **Legacy** SSE event loop (`MotionController`) for `motion_policies`. Its echo-buffer override detection predates the bridge's periodic re-emission of settled values (the daemon's settle-and-compare replaced it), and it requires legacy PIR sensors. | | `cli.py` | `argparse`-based CLI (`Cli`). Subcommands: `validate`, `auth`, `inventory`, `plan`, `apply`, `preview`, `watch`, `circadian run\|resume`, `rhythm`, `security on\|off\|status`. | diff --git a/hueman/circadian_daemon.py b/hueman/circadian_daemon.py index 0cea291..edf59f9 100644 --- a/hueman/circadian_daemon.py +++ b/hueman/circadian_daemon.py @@ -34,6 +34,17 @@ target. Without both, a tick landing between a manual change and the bridge's re-emission of the settled value would silently fade the change back to the curve and then classify its own landing as "self" — the human loses, with no trace. + +Night-guide (``circadian_daemon.night_guide``, optional) is a third actor sharing +the zone: while circadian isn't actively driving it (out of the window — parked +at ``night_look`` or off), motion on a configured MotionAware area briefly raises +the zone to a soft guide look, then hands back cleanly once motion stops — an +exact snapshot restore if a real manual override was showing (nothing else to +recompute that from), otherwise a recomputed circadian/rest target (see +:meth:`CircadianDaemon._night_guide_restore_locked`). It never fights the +operator or the curve: an override still suspends and stays suspended straight +through a guide episode, and circadian still owns the zone the instant its +window is open. """ from __future__ import annotations @@ -52,7 +63,7 @@ import time from collections.abc import Callable from enum import Enum -from typing import cast +from typing import Any, cast from zoneinfo import ZoneInfo import requests @@ -67,9 +78,10 @@ ) from .circadian_control import CircadianController, DriveTo, FadeOff, Hold from .client import HueClient -from .config import CircadianDaemonSpec, Config, RhythmSpec +from .config import CircadianDaemonSpec, Config, NightGuideSpec, RhythmSpec from .engine import TargetState from .errors import AuthError, BridgeError, ConfigError +from .nightguide_control import NightGuideController from .payload import GroupedLightCommand, LightCommand from .presence import ActivityEvent, PresenceTracker from .rhythm_control import AnchorStore, RhythmEngine, SignalState @@ -123,6 +135,7 @@ class CircadianDaemon: _security_rids: dict[str, str] _security_light_rids: tuple[str, ...] _rhythm_motion_rooms: dict[str, str] + _night_guide_motion_rids: set[str] def __init__( self, @@ -189,6 +202,16 @@ def __init__( self._rebuild_security_controller() self._security_sse_on_rid = self._resolve_resume_trigger(state, sec.sse_on) self._security_sse_off_rid = self._resolve_resume_trigger(state, sec.sse_off) + if spec.night_guide is not None: + area = next( + (a for a in state.motion_areas if a.name == spec.night_guide.area), None + ) + if area is None: + raise ConfigError( + f"circadian_daemon.night_guide.area {spec.night_guide.area!r} not found " + "on the bridge — run `hueman inventory` to list MotionAware areas" + ) + self._night_guide_motion_rids = set(area.service_rids) if config.rhythm is not None: self._rhythm_motion_rooms = { srid: (area.room_name or area.name) @@ -300,6 +323,19 @@ def _setup( self._security_rids = {} # group name -> grouped_light rid self._security_light_rids = () # member lights (chaos units) self._security_thread: threading.Thread | None = None + # --- night-guide: motion-triggered path lighting (optional) --- + self._night_guide: NightGuideSpec | None = spec.night_guide + self._night_guide_controller = ( + NightGuideController(spec.night_guide.timeout_ms) + if spec.night_guide is not None else None + ) + self._night_guide_motion_rids = set() # resolved in __init__ (needs BridgeState) + # Raw grouped_light body captured just before the guide look overwrites + # a real manual override, so the hand-back can restore it verbatim + # instead of guessing. None whenever there is nothing to restore (the + # guide engaged over ordinary circadian/night_look, which is always + # recomputable instead). + self._night_guide_snapshot: dict[str, Any] | None = None # --- rhythm engine (optional, observe stage only) --- self._rhythm: RhythmSpec | None = config.rhythm self._presence: PresenceTracker | None = None @@ -587,6 +623,8 @@ def _tick_once(self, now: float) -> None: self._poll_control_file(now) if self._bias is not None: self._apply_bias(now) + if self._night_guide is not None: + self._night_guide_tick_locked(now) if self._rhythm is not None: try: self._rhythm_tick(now) @@ -703,6 +741,119 @@ def _write(self, target: TargetState, transition_ms: int) -> bool: _LOG.warning("write to %s failed (%s); skipping", self._rid, e) return False + def _rest_target(self) -> TargetState: + """The driven zone's resting target when circadian isn't actively driving it. + + The configured ``night_look`` if set (a static hold, e.g. minimum- + brightness red), otherwise plain off — the same choice + :meth:`_tick_once`'s ``FadeOff`` handling makes at the hand-off edge. + Reused by the night-guide hand-back, which needs the identical + "what should this zone show right now" answer on demand, not just at + the one hand-off edge. + """ + look = self._spec.night_look + if look is not None: + return TargetState( + on=True, brightness=look.brightness, + mirek=(look.color.mirek if look.color else None), + hex=(look.color.hex if look.color else None), + ) + return TargetState.off() + + # -- night-guide: motion-triggered path lighting ------------------------ # + def _night_guide_on_motion_locked(self, now: float) -> None: + """Handle a motion event for the night-guide feature (caller holds the lock). + + Only engages while circadian isn't actively driving the zone + (``not in_window``) — a guide light has no business fighting the day + curve, per the feature's whole point (path lighting for when the zone + is otherwise parked/off). On the IDLE -> GUIDING edge: if a real + manual override is currently showing (``SUSPENDED``), snapshot the + zone's actual bridge state first — there is nothing to recompute a + manual look from, so it has to be remembered — then write the guide + look. ``_cmd_*`` are updated exactly like any other daemon write so + settle-and-compare reads the guide light's own settling as "self", + not a fresh human override. + """ + assert self._night_guide is not None and self._night_guide_controller is not None + if self._controller.in_window(now): + return # circadian owns the zone; stay out of its way + entering = self._night_guide_controller.motion(now) + if not entering: + return # already guiding; the timeout just got pushed out, nothing to write + if self._controller.mode == CircadianController.SUSPENDED: + self._night_guide_snapshot = self._client.get_resource("grouped_light", self._rid) + look = self._night_guide.look + _LOG.info("night-guide: motion -> soft-red guide on") + target = TargetState( + on=True, brightness=look.brightness, + mirek=(look.color.mirek if look.color else None), + hex=(look.color.hex if look.color else None), + ) + if self._write(target, self._night_guide.transition_ms): + self._cmd_on = True + self._cmd_brightness = look.brightness + self._cmd_fade_until = now + self._night_guide.transition_ms / 1000.0 + + def _night_guide_tick_locked(self, now: float) -> None: + """Advance the guide timeout; caller (``_tick_once``) holds the lock.""" + assert self._night_guide_controller is not None + if self._night_guide_controller.tick(now): + self._night_guide_restore_locked(now) + + def _night_guide_restore_locked(self, now: float) -> None: + """Hand the zone back once a guide episode ends (caller holds the lock). + + A snapshot means a real manual override was showing before the guide + look overwrote it: restore it verbatim (a manual look isn't derivable + from anything else). No snapshot means ordinary circadian/night_look + was showing: recompute the current authoritative target fresh — the + curve if the window opened back up during the episode, otherwise the + normal resting state — the same recompute-when-possible split + :meth:`_restore_after_security` uses. Hold alone would leave the guide + look showing forever: it assumes nothing changed underneath it, which + is false here — the guide write is exactly what changed it. + """ + if self._night_guide_snapshot is not None: + body = self._night_guide_snapshot + self._night_guide_snapshot = None + put_body = { + "on": body.get("on", {"on": True}), + "dimming": body.get("dimming", {}), + "color_temperature": body.get("color_temperature", {}), + "color": body.get("color", {}), + "dynamics": {"duration": self._night_guide_transition_ms()}, + } + try: + self._client.update_resource("grouped_light", self._rid, put_body) + except BridgeError as e: + _LOG.warning("night-guide restore write failed (%s); skipping", e) + return + _LOG.info("night-guide: timeout -> restoring the snapshotted manual look") + self._cmd_on = bool(body.get("on", {}).get("on", True)) + dimming = body.get("dimming", {}) + if "brightness" in dimming: + self._cmd_brightness = dimming["brightness"] + self._cmd_fade_until = now + self._night_guide_transition_ms() / 1000.0 + return + _LOG.info("night-guide: timeout -> handing back to circadian") + if self._controller.in_window(now): + action = self._controller.drive_to(now) + target = TargetState(on=True, brightness=action.brightness, mirek=action.mirek, hex=None) + transition_ms = action.transition_ms + else: + target = self._rest_target() + transition_ms = self._night_guide_transition_ms() + if self._write(target, transition_ms): + self._cmd_on = target.on + self._cmd_brightness = target.brightness if target.on else 0.0 + self._cmd_fade_until = now + transition_ms / 1000.0 + + def _night_guide_transition_ms(self) -> int: + """The guide's configured edge fade, for its own hand-back writes.""" + assert self._night_guide is not None + return self._night_guide.transition_ms + def _resume_locked(self, now: float) -> None: """Resume driving and grant a settle-classification grace window. @@ -1235,14 +1386,24 @@ def _handle_event(self, event: BridgeEvent, now: float) -> None: _LOG.info("resume trigger %s -> resumed", event.rid) self._resume_locked(now) return - if (self._presence is not None - and event.rtype in ("convenience_area_motion", "security_area_motion")): - room = self._rhythm_motion_rooms.get(event.rid) - if room is not None and event.data.get("motion", {}).get("motion"): - judgment = self._presence.feed( - ActivityEvent(room=room, kind="motion", ts=now)) - _LOG.debug("rhythm: motion in %r -> human=%s (%s)", - room, judgment.human, judgment.rule) + if event.rtype in ("convenience_area_motion", "security_area_motion"): + # Two independent consumers of the same MotionAware event: rhythm + # (observe-only, feeds presence inference) and night-guide (acts — + # see module docstring). Neither requires the other to be configured. + has_motion = bool(event.data.get("motion", {}).get("motion")) + if self._presence is not None: + room = self._rhythm_motion_rooms.get(event.rid) + if room is not None and has_motion: + judgment = self._presence.feed( + ActivityEvent(room=room, kind="motion", ts=now)) + _LOG.debug("rhythm: motion in %r -> human=%s (%s)", + room, judgment.human, judgment.rule) + if ( + self._night_guide is not None + and has_motion + and event.rid in self._night_guide_motion_rids + ): + self._night_guide_on_motion_locked(now) return if event.rtype != "grouped_light" or event.rid != self._rid: return diff --git a/hueman/config.py b/hueman/config.py index a8a1da4..c0206d3 100644 --- a/hueman/config.py +++ b/hueman/config.py @@ -808,6 +808,60 @@ def parse(cls, value: Any, ctx: str = "circadian_daemon.bias") -> "BiasSpec": ) +@dataclass(frozen=True) +class NightGuideSpec: + """Daemon-native motion-triggered path lighting for when circadian is parked. + + While the driven zone is out of the circadian window (parked at + ``night_look``, or off), motion on the configured MotionAware ``area`` + briefly raises the zone to ``look`` — a soft guide light for a night trip + to the bathroom/kitchen — then hands control back after ``timeout`` with + no further motion. The hand-back is exact where it has to be (a real + manual override showing before the motion gets snapshotted and restored + verbatim — there's nothing else to recompute it from) and recomputed + everywhere else (the current circadian curve if back in-window by then, + otherwise the normal resting state), the same recompute-when-possible + split :meth:`CircadianDaemon._restore_after_security` already uses. + + ``area`` names a ``motion_area_configuration`` (the MotionAware grid, + e.g. "Main Room") — run ``hueman inventory`` to list them. + """ + + area: str + look: LightState + timeout_ms: int + transition_ms: int = 2000 + + @classmethod + def parse(cls, value: Any, ctx: str = "circadian_daemon.night_guide") -> "NightGuideSpec": + """Parse the ``circadian_daemon.night_guide:`` block. + + Requires ``area``, ``look`` (a static colour + brightness, same shape + as ``night_look``) and ``timeout``. ``transition`` (the guide-light + edge fade) defaults to 2s, matching ``bias.transition``. + """ + d = _as_dict(value, ctx) + look_d = _as_dict(_require(d, "look", ctx), f"{ctx}.look") + if "brightness" not in look_d: + raise ConfigError(f"{ctx}.look: 'brightness' is required for a guide look") + try: + bri = float(look_d["brightness"]) + except (TypeError, ValueError): + raise ConfigError( + f"{ctx}.look: brightness must be a number, got {look_d['brightness']!r}") + if not 0 <= bri <= 100: + raise ConfigError(f"{ctx}.look: brightness must be 0-100") + color = Color.parse(look_d, f"{ctx}.look") # requires mirek/kelvin/hex + if color.mode == "circadian": + raise ConfigError(f"{ctx}.look: a guide look must be a static colour, not 'circadian'") + return cls( + area=str(_require(d, "area", ctx)), + look=LightState(on=True, brightness=bri, color=color), + timeout_ms=parse_duration(_require(d, "timeout", ctx), ctx=f"{ctx}.timeout"), + transition_ms=parse_duration(d.get("transition", "2s"), ctx=f"{ctx}.transition"), + ) + + @dataclass(frozen=True) class CircadianDaemonSpec: """Tunables for the persistent circadian daemon (all from YAML, with defaults).""" @@ -846,6 +900,8 @@ class CircadianDaemonSpec: # window-close edge; the daemon then goes night-idle, so overnight manual # changes are never re-driven. night_look: "LightState | None" = None + # Daemon-native motion-triggered path lighting (optional). See NightGuideSpec. + night_guide: "NightGuideSpec | None" = None @classmethod def parse(cls, value: Any, ctx: str = "circadian_daemon") -> "CircadianDaemonSpec": @@ -886,6 +942,8 @@ def parse(cls, value: Any, ctx: str = "circadian_daemon") -> "CircadianDaemonSpe settle_epsilon=float(mo.get("settle_epsilon", 0.75)), bias=BiasSpec.parse(d["bias"], f"{ctx}.bias") if d.get("bias") else None, night_look=cls._parse_night_look(d.get("night_look"), f"{ctx}.night_look"), + night_guide=NightGuideSpec.parse(d["night_guide"], f"{ctx}.night_guide") + if d.get("night_guide") else None, ) @staticmethod diff --git a/hueman/nightguide_control.py b/hueman/nightguide_control.py new file mode 100644 index 0000000..9b3f5ed --- /dev/null +++ b/hueman/nightguide_control.py @@ -0,0 +1,70 @@ +"""Pure decision core for the daemon's night-guide overlay (no clock, no I/O). + +Motion-triggered path lighting for when circadian isn't actively driving the +zone: a brief soft-red "guide" look while someone's up at night, then a clean +hand-back once the motion stops. Everything here is a pure function of +explicit inputs, so it is fully unit-tested without a bridge, mirroring +:mod:`hueman.bias_control` and :mod:`hueman.security_control`. + +This module owns only the timing (when does an episode start/end); the daemon +decides *what* to show and *how* to hand back (an exact snapshot restore for a +real manual override, or a recomputed circadian/rest target otherwise — there +is no bridge state here to make that call from). +""" + +from __future__ import annotations + +IDLE = "idle" +GUIDING = "guiding" + + +class NightGuideController: + """Tracks one guide episode: IDLE until motion, GUIDING until ``timeout`` + with no further motion. + + Repeated motion while GUIDING extends the episode (the timeout is always + measured from the *last* motion, not the first) — the guide light should + stay on for as long as someone's actually moving around, not blink off on + a fixed clock mid-trip. + """ + + def __init__(self, timeout_ms: int) -> None: + """Bind the no-motion timeout (in ms) that ends a guiding episode.""" + self._timeout_s = timeout_ms / 1000.0 + self._state = IDLE + self._last_motion: float | None = None + + @property + def state(self) -> str: + """Return the current state (``IDLE`` or ``GUIDING``).""" + return self._state + + def motion(self, now: float) -> bool: + """Record motion at ``now``; extends or starts a guiding episode. + + Returns: + ``True`` if this motion is a fresh IDLE -> GUIDING edge (the + caller should show the guide look); ``False`` if it just extended + an episode already in progress (nothing new to write). + """ + entering = self._state == IDLE + self._state = GUIDING + self._last_motion = now + return entering + + def tick(self, now: float) -> bool: + """Advance the timeout clock; caller should poll this periodically. + + Returns: + ``True`` exactly on the GUIDING -> IDLE edge (the episode just + ended — the caller should hand control back). ``False`` otherwise + (already IDLE, or still within ``timeout`` of the last motion). + """ + if ( + self._state == GUIDING + and self._last_motion is not None + and now - self._last_motion >= self._timeout_s + ): + self._state = IDLE + return True + return False diff --git a/tests/test_circadian_daemon.py b/tests/test_circadian_daemon.py index a585c91..6e37dce 100644 --- a/tests/test_circadian_daemon.py +++ b/tests/test_circadian_daemon.py @@ -21,10 +21,13 @@ def _restore_daemon_logging(): class _FakeClient: - def __init__(self): + def __init__(self, *, resource_to_return=None): self.writes = [] + self._resource_to_return = resource_to_return # for get_resource (night-guide snapshot) def get_resources(self, rtype): # for BridgeState.load if used return [] + def get_resource(self, rtype, rid): + return self._resource_to_return def update_resource(self, rtype, rid, body): self.writes.append((rtype, rid, body)) @@ -1357,3 +1360,208 @@ def test_security_clear_in_window_reasserts_bias_as_edge(monkeypatch): rt == "light" and rid == "Lcouch" and b.get("on") == {"on": True} for (rt, rid, b) in d._client.writes[-20:] ), "bias not re-driven after the show" + + +# -- night-guide: motion-triggered path lighting ----------------------------- # +def _cfg_night_guide(*, night_look=None, timeout="3m", start="sunrise", hand_off="22:34"): + daemon_cfg = { + "zone": "Night Guide", "interval": "60s", "transition": "75s", + "start": start, "hand_off": hand_off, + "night_guide": { + "area": "Main Room", + "look": {"brightness": 9, "hex": "#ff1400"}, + "timeout": timeout, + }, + } + if night_look is not None: + daemon_cfg["night_look"] = night_look + return Config.parse({ + "bridge": {"host": "x", "application_key": "k"}, + "location": {"lat": 45.5152, "lon": -122.6784, "tz_offset_hours": -7}, + "motion_policies": [], + "circadian_daemon": daemon_cfg, + }) + + +def _motion_event(rid, on): + return BridgeEvent("convenience_area_motion", rid, {"motion": {"motion": on}}) + + +def test_night_guide_motion_writes_the_guide_look_when_parked(): + d = CircadianDaemon.for_test(_FakeClient(), _cfg_night_guide(), grouped_light_rid="GL") + d._night_guide_motion_rids = {"MOTION1"} + t = _epoch(23, 0) # out of window, mode starts NIGHT_IDLE + assert not d._controller.in_window(t) + d._handle_event(_motion_event("MOTION1", True), t) + assert d._client.writes, "expected a guide-look write" + rtype, rid, body = d._client.writes[-1] + assert (rtype, rid) == ("grouped_light", "GL") + assert body["dimming"] == {"brightness": 9.0} + assert d._cmd_brightness == 9.0 + assert d._cmd_on is True + assert d._night_guide_controller.state == "guiding" + + +def test_night_guide_does_not_engage_while_circadian_is_driving(): + d = CircadianDaemon.for_test(_FakeClient(), _cfg_night_guide(), grouped_light_rid="GL") + d._night_guide_motion_rids = {"MOTION1"} + t = _epoch(13, 0) # daytime, well within window + assert d._controller.in_window(t) + d._handle_event(_motion_event("MOTION1", True), t) + assert d._client.writes == [] + assert d._night_guide_controller.state == "idle" + + +def test_night_guide_ignores_motion_from_an_unrelated_area(): + d = CircadianDaemon.for_test(_FakeClient(), _cfg_night_guide(), grouped_light_rid="GL") + d._night_guide_motion_rids = {"MOTION1"} + d._handle_event(_motion_event("OTHER", True), _epoch(23, 0)) + assert d._client.writes == [] + assert d._night_guide_controller.state == "idle" + + +def test_night_guide_ignores_a_motion_false_event(): + d = CircadianDaemon.for_test(_FakeClient(), _cfg_night_guide(), grouped_light_rid="GL") + d._night_guide_motion_rids = {"MOTION1"} + d._handle_event(_motion_event("MOTION1", False), _epoch(23, 0)) + assert d._client.writes == [] + assert d._night_guide_controller.state == "idle" + + +def test_night_guide_repeated_motion_extends_without_rewriting(): + d = CircadianDaemon.for_test(_FakeClient(), _cfg_night_guide(timeout="3m"), grouped_light_rid="GL") + d._night_guide_motion_rids = {"MOTION1"} + t = _epoch(23, 0) + d._handle_event(_motion_event("MOTION1", True), t) + n = len(d._client.writes) + d._handle_event(_motion_event("MOTION1", True), t + 170) # still guiding -> extends, no rewrite + assert len(d._client.writes) == n + d._tick_once(t + 170 + 170) # 170s since the LATEST motion -> still guiding + assert d._night_guide_controller.state == "guiding" + assert len(d._client.writes) == n + d._tick_once(t + 170 + 181) # past 180s since the latest motion -> restores + assert d._night_guide_controller.state == "idle" + + +def test_night_guide_snapshots_a_real_manual_override_before_writing_the_guide_look(): + snapshot_body = { + "on": {"on": True}, + "dimming": {"brightness": 42.0}, + "color_temperature": {}, + "color": {"xy": {"x": 0.4, "y": 0.4}}, + } + d = CircadianDaemon.for_test( + _FakeClient(resource_to_return=snapshot_body), _cfg_night_guide(), grouped_light_rid="GL") + d._night_guide_motion_rids = {"MOTION1"} + t0 = _epoch(21, 0) # daytime: establish driving + a real override + d._tick_once(t0) + d._handle_event(_dim(d._cmd_brightness - 30.0), t0 + 30) + d._handle_event(_dim(d._cmd_brightness - 30.0), t0 + 33) + assert d._controller.mode == "suspended" + t1 = _epoch(23, 0) # later, past hand-off, still suspended + d._handle_event(_motion_event("MOTION1", True), t1) + assert d._night_guide_snapshot == snapshot_body + rtype, rid, body = d._client.writes[-1] + assert body["dimming"] == {"brightness": 9.0} # the guide look, not the snapshot + + +def test_night_guide_restores_the_exact_snapshot_after_timeout(): + snapshot_body = { + "on": {"on": True}, + "dimming": {"brightness": 42.0}, + "color_temperature": {}, + "color": {"xy": {"x": 0.4, "y": 0.4}}, + } + d = CircadianDaemon.for_test( + _FakeClient(resource_to_return=snapshot_body), + _cfg_night_guide(timeout="1m"), grouped_light_rid="GL") + d._night_guide_motion_rids = {"MOTION1"} + t0 = _epoch(21, 0) + d._tick_once(t0) + d._handle_event(_dim(d._cmd_brightness - 30.0), t0 + 30) + d._handle_event(_dim(d._cmd_brightness - 30.0), t0 + 33) + assert d._controller.mode == "suspended" + t1 = _epoch(23, 0) + d._handle_event(_motion_event("MOTION1", True), t1) + n = len(d._client.writes) + d._tick_once(t1 + 61) # past the 1-minute timeout + assert len(d._client.writes) == n + 1 + rtype, rid, body = d._client.writes[-1] + assert body["dimming"] == {"brightness": 42.0} + assert body["color"] == {"xy": {"x": 0.4, "y": 0.4}} + assert d._night_guide_snapshot is None + assert d._controller.mode == "suspended" # the override is still logically in force + + +def test_night_guide_hands_back_to_night_look_when_nothing_was_suspended(): + d = CircadianDaemon.for_test( + _FakeClient(), + _cfg_night_guide(night_look={"brightness": 1, "hex": "#ff0000"}, timeout="1m"), + grouped_light_rid="GL") + d._night_guide_motion_rids = {"MOTION1"} + d._tick_once(_epoch(22, 0)) # driving + d._tick_once(_epoch(23, 0)) # hand-off -> night_look write, NIGHT_IDLE + assert d._client.writes[-1][2]["dimming"] == {"brightness": 1.0} + t1 = _epoch(23, 30) + d._handle_event(_motion_event("MOTION1", True), t1) + assert d._client.writes[-1][2]["dimming"] == {"brightness": 9.0} # guide look + d._tick_once(t1 + 61) + rtype, rid, body = d._client.writes[-1] + assert body["dimming"] == {"brightness": 1.0} # back to the resting night_look + assert d._controller.mode == "night_idle" + + +def test_night_guide_hands_back_to_the_curve_if_window_reopens_during_the_episode(): + # start/hand_off pinned to fixed clock times (not sunrise) so the window + # edge is deterministic in the test, independent of solar computation. + d = CircadianDaemon.for_test( + _FakeClient(), + _cfg_night_guide(timeout="3m", start="06:00", hand_off="22:00"), + grouped_light_rid="GL") + d._night_guide_motion_rids = {"MOTION1"} + t = _epoch(5, 58) # 2 min before the window opens at 06:00 + assert not d._controller.in_window(t) + d._handle_event(_motion_event("MOTION1", True), t) + assert d._client.writes[-1][2]["dimming"] == {"brightness": 9.0} + t1 = t + 3 * 60 + 5 # past the 3-min timeout; window now open + assert d._controller.in_window(t1) + d._tick_once(t1) + rtype, rid, body = d._client.writes[-1] + assert body["dimming"]["brightness"] != 9.0 # replaced by a curve sample, not the guide look + assert d._controller.mode == "driving" + + +def _cfg_rhythm_and_night_guide(): + return Config.parse({ + "bridge": {"host": "x", "application_key": "k"}, + "location": { + "lat": 45.5152, "lon": -122.6784, "tz_offset_hours": -7, + "tz": "America/Los_Angeles", + }, + "motion_policies": [], + "circadian_daemon": { + "zone": "Night Guide", "interval": "60s", "transition": "75s", + "night_guide": { + "area": "Main Room", + "look": {"brightness": 9, "hex": "#ff1400"}, + "timeout": "3m", + }, + }, + "rhythm": {"stage": "observe", "bedroom": "Bedroom"}, + }) + + +def test_night_guide_and_rhythm_both_react_to_the_same_motion_event(caplog): + # Two independent consumers of one SSE event -- neither should block the + # other now that the routing no longer gates night-guide on presence + # being configured. + d = CircadianDaemon.for_test(_FakeClient(), _cfg_rhythm_and_night_guide(), grouped_light_rid="GL") + d._night_guide_motion_rids = {"MOTION1"} + d._rhythm_motion_rooms = {"MOTION1": "Living room"} + caplog.set_level(logging.DEBUG, logger="hueman.circadian_daemon") + d._handle_event(_motion_event("MOTION1", True), _epoch(23, 0)) + assert d._client.writes, "night-guide should have written the guide look" + assert d._night_guide_controller.state == "guiding" + assert any( + "rhythm: motion in 'Living room'" in r.message for r in caplog.records + ), "rhythm should still have recorded the same motion event" diff --git a/tests/test_nightguide_control.py b/tests/test_nightguide_control.py new file mode 100644 index 0000000..df4fabd --- /dev/null +++ b/tests/test_nightguide_control.py @@ -0,0 +1,59 @@ +from __future__ import annotations + +from hueman.nightguide_control import GUIDING, IDLE, NightGuideController + + +def test_starts_idle(): + c = NightGuideController(timeout_ms=180_000) + assert c.state == IDLE + + +def test_motion_enters_guiding_and_reports_the_edge(): + c = NightGuideController(timeout_ms=180_000) + entered = c.motion(100.0) + assert entered is True + assert c.state == GUIDING + + +def test_repeated_motion_while_guiding_is_not_a_fresh_edge(): + c = NightGuideController(timeout_ms=180_000) + assert c.motion(100.0) is True + assert c.motion(105.0) is False # already guiding -> no new write needed + assert c.state == GUIDING + + +def test_tick_before_timeout_does_not_end_the_episode(): + c = NightGuideController(timeout_ms=180_000) # 3 min + c.motion(100.0) + assert c.tick(100.0 + 170) is False + assert c.state == GUIDING + + +def test_tick_past_timeout_ends_the_episode_once(): + c = NightGuideController(timeout_ms=180_000) + c.motion(100.0) + assert c.tick(100.0 + 180) is True # the exit edge + assert c.state == IDLE + assert c.tick(100.0 + 240) is False # already idle -> no repeat edge + + +def test_repeated_motion_extends_the_timeout_from_the_last_event(): + # Someone still moving around at t=170 (just under the 180s timeout) must + # push the deadline out from THAT motion, not the original one -- a guide + # light must not blink off mid-trip just because it's been >3min total. + c = NightGuideController(timeout_ms=180_000) + c.motion(100.0) + c.motion(270.0) # motion again at t=270 (still guiding) + assert c.tick(270.0 + 170) is False # 170s since the LATEST motion -> not yet + assert c.state == GUIDING + assert c.tick(270.0 + 180) is True # 180s since the latest motion -> ends + assert c.state == IDLE + + +def test_motion_after_a_completed_episode_starts_a_fresh_one(): + c = NightGuideController(timeout_ms=180_000) + c.motion(100.0) + c.tick(100.0 + 180) + assert c.state == IDLE + assert c.motion(500.0) is True # fresh edge, independent episode + assert c.state == GUIDING