gps-denied-onboard/src/gps_denied_onboard/clock/tlog_derived.py

"""``TlogDerivedClock`` strategy (AZ-398) — replay-only.

Advances ``monotonic_ns`` on each call to the next timestamp emitted by
the wrapped tlog-timestamp source. Out-of-order timestamps raise
:class:`ClockOrderingError` (AC-6) — replay determinism is hard-failed,
never silently smoothed.

The strategy is constructed by the replay composition root (AZ-401)
with a callable that yields tlog timestamps as the parser advances.
For unit tests, an iterator of pre-known timestamps suffices.
"""

from __future__ import annotations

from collections.abc import Callable, Iterable, Iterator


class ClockOrderingError(RuntimeError):
    """Raised when the tlog-timestamp source emits a non-monotonic value.

    Replay must be deterministic; a strategy that silently smooths
    backward jumps would mask a genuine recording corruption. The error
    names the offending pair so the operator can correlate against the
    tlog message stream.
    """


class TlogDerivedClock:
    """Replay :class:`Clock` strategy driven by the tlog timestamp stream.

    The source can be either a callable returning ``int`` ns (typical
    when wired against the live tlog parser, AZ-399) or an iterable of
    pre-known ``int`` ns values (typical in unit tests). Both are normalised
    to an internal :class:`Iterator` lazily.

    Semantics:

    - :meth:`monotonic_ns` pulls the next value from the source on every
      call and returns it (advance-on-call). The most-recently-returned
      value is cached for :meth:`time_ns` so the two methods stay aligned.
    - :meth:`time_ns` returns the latest cached value; if no call to
      :meth:`monotonic_ns` has happened yet, it returns 0 (the replay
      composition root must pump at least one frame before any consumer
      asks for wall-clock time).
    - :meth:`sleep_until_ns` is a no-op (``pace=ASAP``).
    """

    __slots__ = ("_source", "_last_ns")

    def __init__(
        self,
        source: Callable[[], int] | Iterable[int],
    ) -> None:
        if callable(source):
            self._source: Iterator[int] = _iter_from_callable(source)
        else:
            self._source = iter(source)
        self._last_ns = 0

    def monotonic_ns(self) -> int:
        try:
            next_ns = next(self._source)
        except StopIteration:
            return self._last_ns
        if next_ns < self._last_ns:
            raise ClockOrderingError(
                f"TlogDerivedClock: non-monotonic timestamp "
                f"{next_ns} ns followed {self._last_ns} ns"
            )
        self._last_ns = next_ns
        return next_ns

    def time_ns(self) -> int:
        return self._last_ns

    def sleep_until_ns(self, target_ns: int) -> None:
        """No-op in ASAP pace (Invariant 6)."""
        return None


def _iter_from_callable(source: Callable[[], int]) -> Iterator[int]:
    """Wrap a callable as an iterator that calls it on each ``next()``.

    Used when the tlog parser exposes a ``next_timestamp_ns()``-style
    hook; consumers should NOT pass a side-effectful callable that
    blocks — the source is expected to be cheap (microsecond-class).
    """
    while True:
        yield source()


__all__ = ["ClockOrderingError", "TlogDerivedClock"]