quality: strict typing + full docstring coverage across the package

hueman/bias_control.py

@@ -100,6 +100,7 @@ class TriggerAggregator:
     """
 
     def __init__(self, debounce_ms: int = 0) -> None:
+        """Start with no active sources; edges must hold ``debounce_ms`` to commit."""
         self._debounce_s = debounce_ms / 1000.0
         self._active: set[str] = set()   # sources currently reporting on
         self._committed = False
@@ -108,10 +109,12 @@ def __init__(self, debounce_ms: int = 0) -> None:
         self.last_source: str | None = None   # source that caused the last raw flip
 
     def on(self, now: float, source: str = "default") -> None:
+        """Record an on-edge from ``source`` at ``now`` (idempotent per source)."""
         self._active.add(source)
         self._recompute(now, source)
 
     def off(self, now: float, source: str = "default") -> None:
+        """Record an off-edge from ``source`` at ``now`` (idempotent per source)."""
         self._active.discard(source)
         self._recompute(now, source)
 
@@ -124,13 +127,23 @@ def reset(self, now: float) -> None:
         self._recompute(now, "reset")
 
     def _recompute(self, now: float, source: str) -> None:
+        """Refresh the raw OR-signal after an edge from ``source``.
+
+        On a flip, restart the debounce clock and credit ``source`` as the
+        cause; edges that don't flip the signal leave both alone.
+        """
         raw = bool(self._active)
         if raw != self._raw:
             self._raw = raw
             self._raw_since = now
             self.last_source = source
 
     def tv_on(self, now: float) -> bool:
+        """Return the committed TV-on signal as of ``now``.
+
+        Commits a pending raw flip once it has held for the debounce window;
+        with ``debounce_ms == 0`` this is the raw signal itself.
+        """
         if (
             self._raw != self._committed
             and self._raw_since is not None

hueman/circadian.py

@@ -94,6 +94,7 @@ class CircadianCurve:
     """
 
     def __init__(self, params: CircadianParams = CircadianParams()) -> None:
+        """Store the anchor parameters the curve interpolates between."""
         self._params = params
 
     @property

hueman/circadian_control.py

@@ -19,23 +19,40 @@
 
 @dataclass(frozen=True)
 class DriveTo:
+    """Drive the zone to this curve sample over ``transition_ms``.
+
+    Attributes:
+        brightness: Target brightness percentage (0-100), floor/ceiling applied.
+        mirek: Target colour temperature in mirek.
+        transition_ms: Cross-fade duration for the write, in milliseconds.
+    """
+
     brightness: float
     mirek: int
     transition_ms: int
 
 
 @dataclass(frozen=True)
 class FadeOff:
+    """Fade the zone off over ``transition_ms`` (the hand-off at window close)."""
+
     transition_ms: int
 
 
 @dataclass(frozen=True)
 class Hold:
+    """Do nothing this tick; ``reason`` names the mode holding (for logging)."""
+
     reason: str
 
 
 class CircadianController:
-    """State machine deciding the circadian daemon's per-tick action."""
+    """State machine deciding the circadian daemon's per-tick action.
+
+    Modes: ``DRIVING`` (inside the window, following the curve), ``SUSPENDED``
+    (a manual override was detected; hands off until resumed or the next
+    window open), and ``NIGHT_IDLE`` (outside the window, nothing to do).
+    """
 
     DRIVING = "driving"
     SUSPENDED = "suspended"
@@ -49,6 +66,15 @@ def __init__(
         tz_offset_hours: float,
         tz: str | None = None,
     ) -> None:
+        """Bind the daemon spec, curve, solar calculator and timezone.
+
+        Args:
+            spec: Daemon settings (window anchors, transitions, override policy).
+            circadian: Curve parameters used to build the ``CircadianCurve``.
+            solar: Calculator for the site's sun times and elevation.
+            tz_offset_hours: Fixed UTC offset fallback when ``tz`` is unset.
+            tz: Optional IANA timezone name; when set, local time is DST-correct.
+        """
         self._spec = spec
         self._curve = CircadianCurve(circadian)
         self._solar = solar
@@ -59,6 +85,7 @@ def __init__(
 
     @property
     def mode(self) -> str:
+        """Return the current daemon mode (``DRIVING``/``SUSPENDED``/``NIGHT_IDLE``)."""
         return self._mode
 
     # -- public accessors (used by the bias subsystem, independent of mode) -- #
@@ -72,27 +99,32 @@ def drive_to(self, now: float) -> DriveTo:
 
     # -- time helpers ------------------------------------------------------- #
     def _local_dt(self, now: float) -> _dt.datetime:
+        """Return ``now`` as a local datetime (DST-correct when a tz name is set)."""
         tz = ZoneInfo(self._tz) if self._tz else _dt.timezone(
             _dt.timedelta(hours=self._tz_offset_hours))
         return _dt.datetime.fromtimestamp(now, tz)
 
     def _minute_of_day(self, now: float) -> float:
+        """Return ``now`` as fractional local minutes after midnight."""
         d = self._local_dt(now)
         return d.hour * 60 + d.minute + d.second / 60.0
 
     def _window(self, date: _dt.date) -> tuple[int, int]:
+        """Return ``(start_min, hand_off_min)`` for ``date``, start sun-resolved."""
         sun = self._solar.sun_times(date)
         start = self._spec.start.resolve(sun.sunrise_min, sun.sunset_min)
         return start, self._spec.hand_off_min
 
     def _in_window(self, now: float) -> bool:
+        """Return whether ``now`` falls in ``[start, hand_off)`` for its local date."""
         date = self._local_dt(now).date()
         start, hand_off = self._window(date)
         minute = self._minute_of_day(now)
         return start <= minute < hand_off
 
     # -- target ------------------------------------------------------------- #
     def _drive_to(self, now: float) -> DriveTo:
+        """Sample the elevation-driven curve at ``now`` and apply floor/ceiling."""
         date = self._local_dt(now).date()
         elev = self._solar.solar_elevation(date, self._minute_of_day(now))
         noon = self._solar.noon_elevation(date)
@@ -106,14 +138,29 @@ def _drive_to(self, now: float) -> DriveTo:
 
     # -- events ------------------------------------------------------------- #
     def on_external_change(self, now: float) -> None:
+        """Suspend on a detected manual override.
+
+        A no-op if ``detect_override`` is disabled, so the daemon keeps
+        driving through external writes.
+        """
         if self._spec.detect_override:
             self._mode = self.SUSPENDED
 
     def on_resume(self, now: float) -> None:
+        """Resume driving (operator/API resume), regardless of the current mode."""
         self._mode = self.DRIVING
 
     # -- per-tick decision -------------------------------------------------- #
     def tick(self, now: float) -> DriveTo | FadeOff | Hold:
+        """Advance the state machine one tick and return the action for ``now``.
+
+        ``SUSPENDED`` holds until an explicit resume, except that a window *open*
+        edge with ``daily_safety_resume`` re-arms driving (so an overnight
+        override never silently disables the daemon forever). ``NIGHT_IDLE``
+        flips to ``DRIVING`` when the window is active. While driving, an
+        in-window tick returns the curve sample; the first out-of-window tick
+        returns a single ``FadeOff`` and drops to ``NIGHT_IDLE``.
+        """
         in_window = self._in_window(now)
         window_opened = in_window and not self._was_in_window
         self._was_in_window = in_window

hueman/circadian_daemon.py

@@ -39,7 +39,8 @@
 import sys
 import threading
 import time
-from typing import Callable
+from collections.abc import Callable
+from typing import cast
 
 import requests
 
@@ -80,6 +81,16 @@ class CircadianDaemon:
             :class:`~hueman.watch.HueEventStream` over ``client``.
     """
 
+    # Attributes assigned on both construction paths (``__init__`` and
+    # ``_setup``); annotated once here so the two assignment sites agree.
+    _bias_sse_on_rid: str | None
+    _bias_sse_off_rid: str | None
+    _bias_rids: dict[str, str]
+    _security_sse_on_rid: str | None
+    _security_sse_off_rid: str | None
+    _security_rids: dict[str, str]
+    _security_light_rids: tuple[str, ...]
+
     def __init__(
         self,
         client: HueClient,
@@ -90,6 +101,7 @@ def __init__(
         sleep: Callable[[float], None] = time.sleep,
         stream_factory: Callable[[], "HueEventStream"] | None = None,
     ) -> None:
+        """Resolve bridge rids from ``state``, validate the config, and wire up."""
         spec = self._require_spec(config)
         rid = state.group(spec.zone).grouped_light_rid
         if rid is None:
@@ -124,7 +136,13 @@ def __init__(
                 )
             if len(sec.groups) < 2:
                 _LOG.warning("security.groups has <2 groups; chaos zone alternation is disabled")
-            self._security_rids = {g: state.group(g).grouped_light_rid for g in sec.groups}
+            # The ``no_gl`` raise above guarantees every group has a rid; the
+            # walrus filter just narrows ``str | None`` -> ``str`` for the checker.
+            self._security_rids = {
+                g: gl_rid
+                for g in sec.groups
+                if (gl_rid := state.group(g).grouped_light_rid) is not None
+            }
             # Member-light rids across the security groups: chaos drives these
             # individually (decorrelated patchwork) instead of two group blobs.
             seen: set[str] = set()
@@ -163,7 +181,7 @@ def _setup(
         self,
         client: HueClient,
         config: Config,
-        rid: str | None,
+        rid: str,
         *,
         clock: Callable[[], float] = time.time,
         sleep: Callable[[float], None] = time.sleep,
@@ -208,9 +226,9 @@ def _setup(
         self._bias_aggregator = TriggerAggregator(
             debounce_ms=(bias.probe_debounce_ms if (bias and bias.probe_enabled) else 0)
         )
-        self._bias_sse_on_rid: str | None = None    # resolved in __init__ (needs BridgeState)
-        self._bias_sse_off_rid: str | None = None
-        self._bias_rids: dict[str, str] = {}        # bias light name -> light_rid
+        self._bias_sse_on_rid = None                # resolved in __init__ (needs BridgeState)
+        self._bias_sse_off_rid = None
+        self._bias_rids = {}                        # bias light name -> light_rid
         self._bias_probe_thread: threading.Thread | None = None
         self._bias_file_thread: threading.Thread | None = None
         # Last committed tv_on we successfully drove the bias set to. Lets the
@@ -232,10 +250,10 @@ def _setup(
         # Read/written without the lock intentionally: benign under the GIL, self-corrects within one tick.
         self._security_active = False
         self._security_last_phase: str | None = None
-        self._security_sse_on_rid: str | None = None
-        self._security_sse_off_rid: str | None = None
-        self._security_rids: dict[str, str] = {}      # group name -> grouped_light rid
-        self._security_light_rids: tuple[str, ...] = ()   # member lights (chaos units)
+        self._security_sse_on_rid = None
+        self._security_sse_off_rid = None
+        self._security_rids = {}                      # group name -> grouped_light rid
+        self._security_light_rids = ()                # member lights (chaos units)
         self._security_thread: threading.Thread | None = None
 
     def _rebuild_security_controller(self) -> None:
@@ -267,7 +285,9 @@ def _resolve_resume_trigger(state: BridgeState, resume_trigger: str | None) -> s
             return None
         scene = state.scene(resume_trigger)
         if scene is not None and "id" in scene:
-            return scene["id"]
+            # CLIP resource ids are always strings; the cast only informs the
+            # checker (the raw payload is dict[str, Any]).
+            return cast(str, scene["id"])
         return resume_trigger
 
     def _configure_logging(self, spec: CircadianDaemonSpec) -> None:
@@ -469,6 +489,9 @@ def _apply_bias(self, now: float) -> None:
         writes are suppressed once the off has landed — re-PUTing off to the
         whole viewing set every tick, all night, is pure queue pressure.
         """
+        bias = self._bias
+        if bias is None:
+            return  # callers guard on spec.bias; narrowing for the type checker
         in_window = self._controller.in_window(now)
         curve = self._controller.drive_to(now) if in_window else None
         tv_on = self._bias_aggregator.tv_on(now)
@@ -479,12 +502,12 @@ def _apply_bias(self, now: float) -> None:
                 "bias: TV %s (%s) -> %d lights, %.1fs edge fade",
                 "on" if tv_on else "off",
                 self._bias_aggregator.last_source or "startup",
-                len(self._bias.lights),
-                self._bias.transition_ms / 1000,
+                len(bias.lights),
+                bias.transition_ms / 1000,
             )
         all_ok = True
         for action in bias_actions(
-            self._bias, tv_on=tv_on, in_window=in_window, curve=curve,
+            bias, tv_on=tv_on, in_window=in_window, curve=curve,
             transition_ms=self._spec.transition_ms, fade_off_ms=self._spec.fade_off_ms,
             edge=edge,
         ):
@@ -651,6 +674,7 @@ def _run_security_show(self, start: float) -> None:
                         self._security_last_phase = frame.phase
                     futures = []
                     for ft in frame.targets:
+                        rid: str | None
                         if ft.kind == "light":
                             rid, rtype = ft.name, "light"     # chaos units ARE light rids
                         else:

hueman/circadian_scene.py

@@ -14,6 +14,7 @@
 
 from __future__ import annotations
 
+import datetime as _dt
 from dataclasses import dataclass
 
 from .circadian import CircadianCurve, CircadianParams
@@ -41,7 +42,7 @@ class CircadianStep:
 def circadian_timeslots(
     params: CircadianParams,
     solar: SolarCalculator,
-    date,
+    date: _dt.date,
     *,
     hand_off_min: int,
 ) -> list[CircadianStep]:

hueman/cli.py

@@ -17,6 +17,7 @@
 import argparse
 import datetime as _dt
 import sys
+from collections.abc import Callable
 
 from . import __version__
 from .client import HueClient
@@ -45,6 +46,7 @@ class Cli:
     """Parses arguments and dispatches to subcommand handlers."""
 
     def __init__(self) -> None:
+        """Build the argument parser once so ``run`` can be called repeatedly."""
         self._parser = self._build_parser()
 
     def _build_parser(self) -> argparse.ArgumentParser:
@@ -102,7 +104,8 @@ def run(self, argv: list[str] | None = None) -> int:
             A process exit code.
         """
         args = self._parser.parse_args(argv)
-        handler = getattr(self, f"_cmd_{args.command}")
+        # getattr returns Any; the declared type restores checking of the call.
+        handler: Callable[[argparse.Namespace], int] = getattr(self, f"_cmd_{args.command}")
         try:
             return handler(args)
         except HueIacError as error:
@@ -295,11 +298,12 @@ def _print_inventory(state: BridgeState) -> None:
         areas = state.motion_areas
         print(f"\nMotionAware areas — bulb-grid sensing ({len(areas)}):")
         for area in sorted(areas, key=lambda a: a.name):
-            room = area.room_name or "?"
+            # `room` above is bound to a Group by the rooms loop; use a fresh name.
+            area_room = area.room_name or "?"
             sens = f"{area.sensitivity}/{area.sensitivity_max}" if area.sensitivity is not None else "?"
             now = "MOTION" if area.motion else "clear"
             print(
-                f"  - {area.name}  [room: {room}; {area.participant_count} bulbs; "
+                f"  - {area.name}  [room: {area_room}; {area.participant_count} bulbs; "
                 f"sensitivity {sens}; now: {now}]"
             )