"""The perception loop — where seeing and hearing become identity (CHT-025…027). The identity runtime answers "who is here?" from a stream of claims, but it makes none itself: on the robot the claims come from a camera and a microphone, and this is the object that turns one into the other. It closes the two paths the runtime's docstring describes only in the abstract:: frame → detect → FaceObservation → face provider → claim → runtime.observe window → segment → SpeechSegment → voice provider → claim → runtime.observe and, on a fixed cadence, calls ``runtime.tick`` so that presence *decays* when observation stops — because leaving is the absence of observations, and without a tick past the staleness horizon the room would never empty (see ``IdentityRuntime.tick``). The control-loop discipline it follows -------------------------------------- The reference conversation app runs one thread that owns its control surface and paces itself off a monotonic clock; the production roadmap holds every subsystem to that pattern (§"Non-negotiables"). This loop obeys it: - **Single owner.** One thread runs :meth:`PerceptionLoop.run`. It is the sole mutator of the runtime — it threads each immutable ``RuntimeStep.runtime`` forward and no one else touches it. Other threads learn what happened through the ``on_step`` callback, which fires on the loop thread. - **Monotonic cadence.** The tick interval is measured against ``time.monotonic``, never wall time, so a clock adjustment cannot make presence decay early or stall. The wall clock is used only for the ``now`` stamped on claims, where a real timestamp is what a replay wants. - **Stop-event driven.** :meth:`run` loops until a ``threading.Event`` is set and exits promptly even mid-wait, so shutting the robot down releases the loop at once rather than after one more poll. Every seam is injected ---------------------- Hardware sits behind :class:`CameraSource` / :class:`MicSource`; detection, segmentation, and identification behind their existing Protocols; and both the wall clock and the monotonic clock are constructor arguments. So the same object a robot drives with real devices and real clocks, a test drives with synthetic frames, a fake runtime, and a scripted clock — the loop's behaviour is a function of its inputs and replays exactly, which is the whole reason the runtime was kept clock-free beneath it. """ from __future__ import annotations import threading import time from collections.abc import Callable from dataclasses import dataclass from datetime import UTC, datetime from typing import Final from chittios_core.identity.face.detector import FaceDetectorProtocol, FaceObservation from chittios_core.identity.runtime import IdentityRuntime, RuntimeStep from chittios_core.identity.types import IdentityProvider from chittios_core.identity.voice.segmenter import SegmenterProtocol, SpeechSegment from chittios_core.perception.sources import CameraSource, MicSource #: How often presence is aged, in seconds. Well below ``fusion.decay``'s 300 s #: staleness horizon so the room empties promptly after someone leaves rather #: than a horizon-length afterward, yet coarse enough that ticking costs nothing #: against the single-digit-Hz claim rate the runtime is built for. DEFAULT_TICK_INTERVAL_S: Final = 1.0 #: How long the loop waits between polling cycles when it is otherwise idle. #: 20 Hz is far faster than frames or utterances arrive, so a present face is #: never missed for want of looking, while the wait keeps a quiet loop off the #: CPU. It is also the upper bound on how long a stop takes to be noticed. DEFAULT_POLL_INTERVAL_S: Final = 0.05 def _wall_clock() -> datetime: """The default source of the ``now`` stamped on claims and ticks.""" return datetime.now(UTC) @dataclass(frozen=True, slots=True) class FacePipeline: """The three seams that turn a frame into a face claim. Grouped so the loop takes one face argument and one voice argument instead of six loose collaborators, and so a caller wires a modality as a unit. Attributes: source: Where frames come from. detector: Locates faces in a frame; returns them largest-first. provider: Identifies one detected face, emitting the §6.2 claim. """ source: CameraSource detector: FaceDetectorProtocol provider: IdentityProvider[FaceObservation] @dataclass(frozen=True, slots=True) class VoicePipeline: """The three seams that turn an audio window into voice claims. Attributes: source: Where audio windows come from. segmenter: Cuts a window into utterances, in the order spoken. provider: Identifies the speaker of one utterance, emitting the claim. """ source: MicSource segmenter: SegmenterProtocol provider: IdentityProvider[SpeechSegment] class PerceptionLoop: """Drives camera and microphone into an :class:`IdentityRuntime`. Construct with the two modality pipelines and a runtime, then hand :meth:`run` a stop event on a dedicated thread:: loop = PerceptionLoop(face_pipeline, voice_pipeline, runtime, on_step=react) stop = threading.Event() threading.Thread(target=loop.run, args=(stop,), daemon=True).start() ... stop.set() # released promptly Args: face: The frame → claim pipeline. voice: The window → claim pipeline. runtime: The identity runtime to drive. Owned exclusively by the loop thread once :meth:`run` is entered; read the live one via :attr:`runtime`. tick_interval_s: How often presence is aged, measured monotonically. poll_interval_s: How long to wait between polling cycles when idle, and the upper bound on stop latency. on_step: Called with every :class:`RuntimeStep` the runtime produces — each observation and each tick — on the loop thread. This is how a caller reacts to greetings, an authorization change, or a re-enrollment prompt. ``None`` discards the steps. clock: The wall clock stamped on claims and ticks. Injected for deterministic replay; defaults to ``datetime.now(UTC)``. monotonic: The monotonic clock the tick cadence is measured against. Injected for deterministic tests; defaults to ``time.monotonic``. """ def __init__( self, face: FacePipeline, voice: VoicePipeline, runtime: IdentityRuntime, *, tick_interval_s: float = DEFAULT_TICK_INTERVAL_S, poll_interval_s: float = DEFAULT_POLL_INTERVAL_S, on_step: Callable[[RuntimeStep], None] | None = None, clock: Callable[[], datetime] = _wall_clock, monotonic: Callable[[], float] = time.monotonic, ) -> None: self._face = face self._voice = voice self._runtime = runtime self._tick_interval_s = tick_interval_s self._poll_interval_s = poll_interval_s self._on_step = on_step self._clock = clock self._monotonic = monotonic self._next_tick_at = 0.0 @property def runtime(self) -> IdentityRuntime: """The current runtime — the latest state the loop has threaded to.""" return self._runtime def run(self, stop: threading.Event) -> None: """Poll both modalities and age presence until ``stop`` is set. Returns when the event is set. Each cycle observes whatever the camera and microphone offer, ages presence if the cadence is due, then waits — interruptibly, so a stop set during the wait returns at once. """ # Anchor the first tick a full interval out rather than firing one # immediately on an empty room. self._next_tick_at = self._monotonic() + self._tick_interval_s while not stop.is_set(): self._observe_faces() self._observe_speech() self._maybe_tick(self._monotonic()) # Event.wait is the interruptible pace: it returns True the moment # the event is set, so shutdown never waits out a full interval. if stop.wait(self._poll_interval_s): break # ── modality paths ─────────────────────────────────────────────────── def _observe_faces(self) -> None: """Fold a claim for every face currently in view. Every detected face is observed, not just the nearest: presence is per-person, and only a claim per face lets the room hold everyone who is actually there. All faces in one frame share its capture instant. """ frame = self._face.source.read() if frame is None: return now = self._clock() for face in self._face.detector.detect(frame): observation = FaceObservation(frame_bgr=frame, face=face) claim = self._face.provider.identify(observation, observed_at=now) self._fold(self._runtime.observe(claim, now=now)) def _observe_speech(self) -> None: """Fold a claim for every utterance in the latest audio window.""" window = self._voice.source.read() if window is None: return now = self._clock() for segment in self._voice.segmenter.segment(window): claim = self._voice.provider.identify(segment, observed_at=now) self._fold(self._runtime.observe(claim, now=now)) def _maybe_tick(self, now_monotonic: float) -> None: """Age presence if a tick interval has elapsed since the last one.""" if now_monotonic < self._next_tick_at: return self._next_tick_at = now_monotonic + self._tick_interval_s self._fold(self._runtime.tick(self._clock())) def _fold(self, step: RuntimeStep) -> None: """Thread the new runtime forward and surface the step to the caller.""" self._runtime = step.runtime if self._on_step is not None: self._on_step(step)