chittios_core/perception/loop.py

"""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)