hal/reference_hal.py

"""The HAL contract — what ChittiOS expects from a robot body.

Layer 1 of the three-layer architecture (spec §4.1, §6.1). The contract is
deliberately minimal so Core can ignore which body it is running on: audio in
and out, a small set of motion primitives, vision, and optional presence.

The OS sends primitives; the body decides how to express them. A body with no
antennas ignores ``ANTENNA_WIGGLE``. A body with a screen renders a face for
``ATTENTION``. Neither case is an error.

Refinement of spec §6.1
-----------------------
The spec sketches ``motion(primitive: MotionPrimitive, params: dict)``. This
module keeps that shape but replaces the untyped ``dict`` with a union of
per-primitive parameter dataclasses. A ``dict`` forces every adapter into
unchecked key lookups at the one boundary the whole design depends on being
stable — precisely where a type that lies is most expensive. ``motion()``
validates that the parameter object matches the primitive before dispatch.

Fold back into the spec as an erratum to §6.1.
"""

from __future__ import annotations

from collections.abc import AsyncIterator
from dataclasses import dataclass, field
from enum import Enum
from typing import Literal, Protocol, TypeAlias, runtime_checkable

import numpy as np
from numpy.typing import NDArray

# ─────────────────────────────────────────────────────────────────────────
# Media types
# ─────────────────────────────────────────────────────────────────────────

#: Audio is float32 normalized to [-1, 1] throughout.
#:
#: An earlier draft of this contract specified int16 on the reasoning that it
#: was the common native format. Checking the installed SDK showed otherwise:
#: ``MediaManager.get_audio_sample`` returns float32 and ``push_audio_sample``
#: expects it, and Whisper wants normalized float32 as well. int16 is a
#: transport format; float32 is the processing format, and choosing it avoids
#: a quantization round-trip at every stage boundary.
AUDIO_DTYPE = np.float32


@dataclass(frozen=True, slots=True)
class AudioChunk:
    """A block of mono or interleaved audio.

    Attributes:
        samples: float32 normalized to [-1, 1]. Shape ``(n,)`` for mono,
            ``(n, channels)`` interleaved otherwise.
        sample_rate_hz: Sample rate of ``samples``.
        channels: Channel count. Must agree with ``samples``.
        timestamp_s: Monotonic capture or playback time, in seconds.
    """

    samples: NDArray[np.float32]
    sample_rate_hz: int
    channels: int
    timestamp_s: float

    def __post_init__(self) -> None:
        if self.sample_rate_hz <= 0:
            raise ValueError(f"sample_rate_hz must be positive, got {self.sample_rate_hz}")
        if self.channels <= 0:
            raise ValueError(f"channels must be positive, got {self.channels}")
        expected_ndim = 1 if self.channels == 1 else 2
        if self.samples.ndim != expected_ndim:
            raise ValueError(
                f"{self.channels}-channel audio expects a {expected_ndim}-D array, "
                f"got shape {self.samples.shape}"
            )

    @property
    def duration_s(self) -> float:
        """Wall-clock duration of this chunk."""
        return len(self.samples) / self.sample_rate_hz


@dataclass(frozen=True, slots=True)
class ImageFrame:
    """A single camera frame.

    Frames are transient by contract. Spec §5.1 obligation 2 requires that a
    frame be discarded once an embedding has been extracted: no rolling
    buffer, no clip storage, no sightings gallery. Adapters must not retain a
    reference after yielding one, and Core must not persist one.

    Attributes:
        pixels: BGR uint8, shape ``(height, width, 3)``. BGR rather than RGB
            because the upstream YuNet detector expects it.
        timestamp_s: Monotonic capture time, in seconds.
    """

    pixels: NDArray[np.uint8]
    timestamp_s: float

    def __post_init__(self) -> None:
        if self.pixels.ndim != 3 or self.pixels.shape[2] != 3:
            raise ValueError(f"expected BGR frame of shape (h, w, 3), got {self.pixels.shape}")

    @property
    def width(self) -> int:
        return int(self.pixels.shape[1])

    @property
    def height(self) -> int:
        return int(self.pixels.shape[0])


class Presence(Enum):
    """Coarse "is anyone nearby" sensing.

    Deliberately coarse: presence is a cheap trigger, not an identity signal.
    Identity is the Identity subsystem's job and requires biometrics.
    """

    ABSENT = "absent"
    PRESENT = "present"
    UNKNOWN = "unknown"


# ─────────────────────────────────────────────────────────────────────────
# Motion primitives
# ─────────────────────────────────────────────────────────────────────────


class MotionPrimitive(Enum):
    """The vocabulary Core uses to ask a body to move (spec §6.1).

    Extensible: a new primitive is additive, and bodies that do not understand
    it ignore it.
    """

    ORIENT_HEAD = "orient_head"
    ATTENTION = "attention"
    THINKING = "thinking"
    SLEEP = "sleep"
    IDLE = "idle"
    ANTENNA_WIGGLE = "antenna_wiggle"
    EYE_COLOR = "eye_color"
    DISPLAY = "display"


def _require_unit_range(name: str, value: float) -> None:
    if not -1.0 <= value <= 1.0:
        raise ValueError(f"{name} must lie in [-1, 1], got {value}")


def _require_zero_to_one(name: str, value: float) -> None:
    if not 0.0 <= value <= 1.0:
        raise ValueError(f"{name} must lie in [0, 1], got {value}")


@dataclass(frozen=True, slots=True)
class OrientHead:
    """Point the head, in normalized units.

    Normalized rather than degrees so that Core never encodes one body's
    joint limits. Each adapter maps [-1, 1] onto its own safe range — for
    Reachy Mini, pitch and roll to ±40°, yaw to ±180° (spec §"Safety Limits").
    """

    yaw: float = 0.0
    pitch: float = 0.0
    roll: float = 0.0

    primitive = MotionPrimitive.ORIENT_HEAD

    def __post_init__(self) -> None:
        _require_unit_range("yaw", self.yaw)
        _require_unit_range("pitch", self.pitch)
        _require_unit_range("roll", self.roll)


@dataclass(frozen=True, slots=True)
class Attention:
    """Attend to a point in the body frame, in metres, or relax if ``None``."""

    target: tuple[float, float, float] | None = None

    primitive = MotionPrimitive.ATTENTION


@dataclass(frozen=True, slots=True)
class Thinking:
    """Signal deliberation. ``intensity`` scales how pronounced it is."""

    intensity: float = 0.5

    primitive = MotionPrimitive.THINKING

    def __post_init__(self) -> None:
        _require_zero_to_one("intensity", self.intensity)


@dataclass(frozen=True, slots=True)
class Sleep:
    """Adopt the sleep posture."""

    primitive = MotionPrimitive.SLEEP


@dataclass(frozen=True, slots=True)
class Idle:
    """Resume idle micro-motion — the difference between present and dormant."""

    primitive = MotionPrimitive.IDLE


AntennaSide: TypeAlias = Literal["left", "right", "both"]


@dataclass(frozen=True, slots=True)
class AntennaWiggle:
    """Wiggle one or both antennas. Ignored by bodies without antennas."""

    speed: float = 0.5
    side: AntennaSide = "both"

    primitive = MotionPrimitive.ANTENNA_WIGGLE

    def __post_init__(self) -> None:
        _require_zero_to_one("speed", self.speed)


EyeMode: TypeAlias = Literal["solid", "pulse", "blink"]


@dataclass(frozen=True, slots=True)
class EyeColor:
    """Set expressive eye colour. Ignored by bodies without eye rings."""

    rgb: tuple[int, int, int] = (255, 255, 255)
    mode: EyeMode = "solid"

    primitive = MotionPrimitive.EYE_COLOR

    def __post_init__(self) -> None:
        for channel, value in zip("rgb", self.rgb, strict=True):
            if not 0 <= value <= 255:
                raise ValueError(f"rgb channel {channel} must lie in [0, 255], got {value}")


@dataclass(frozen=True, slots=True)
class Display:
    """Render to a status surface. Ignored by bodies without a display.

    Spec §5.1 obligation 5 leans on this: identity state must be legible to
    the household, and on Reachy Mini this is where it becomes visible.
    """

    content: dict[str, str] = field(default_factory=dict)

    primitive = MotionPrimitive.DISPLAY


#: Every motion parameter object. Each carries the primitive it belongs to, so
#: an adapter can dispatch on ``command.primitive`` without a separate argument.
MotionCommand: TypeAlias = (
    OrientHead | Attention | Thinking | Sleep | Idle | AntennaWiggle | EyeColor | Display
)


# ─────────────────────────────────────────────────────────────────────────
# The contract
# ─────────────────────────────────────────────────────────────────────────


class VisionUnsupportedError(NotImplementedError):
    """Raised by ``vision_frame`` on a body with no camera."""


class PresenceUnsupportedError(NotImplementedError):
    """Raised by ``presence`` on a body with no presence sensor."""


@runtime_checkable
class HALAdapter(Protocol):
    """What a body must provide for ChittiOS Core to run on it.

    Four capability groups. Audio and motion are required; vision and presence
    may raise their respective ``*UnsupportedError``. An identity-enabled
    deployment requires vision — Core checks this at startup rather than
    discovering it at first recognition.

    All methods are non-blocking in the sense that motion returns once the
    motion is *queued*, not once it completes. Expression depends on this:
    speech must never block on motion, nor motion on speech (spec §4.2).
    """

    async def audio_in_stream(self) -> AsyncIterator[AudioChunk]:
        """Yield captured audio until the stream is closed."""
        ...

    async def audio_out(self, chunks: AsyncIterator[AudioChunk]) -> None:
        """Play a stream of audio, returning when it has been consumed."""
        ...

    async def motion(self, command: MotionCommand) -> None:
        """Queue one motion primitive. Returns once queued, not once complete.

        Unsupported primitives are ignored silently — that is the contract,
        not a failure. A body without antennas is still a valid body.
        """
        ...

    async def vision_frame(self) -> ImageFrame | None:
        """Return the most recent camera frame, or ``None`` if none is ready.

        Raises:
            VisionUnsupportedError: if the body has no camera.
        """
        ...

    async def presence(self) -> Presence:
        """Return coarse presence.

        Raises:
            PresenceUnsupportedError: if the body has no presence sensor.
        """
        ...


def validate_command(command: MotionCommand) -> MotionPrimitive:
    """Return the primitive a command belongs to, checking it is well-formed.

    Adapters call this before dispatch so that a malformed command fails at
    the boundary with a clear message, rather than partway through a motion.
    """
    primitive = getattr(command, "primitive", None)
    if not isinstance(primitive, MotionPrimitive):
        raise TypeError(
            f"{type(command).__name__} is not a motion command: it declares no MotionPrimitive."
        )
    return primitive