jmri_core.beans.fast_clock

  1"""``FastClock`` — model-railroad fast clock.
  2
  3Mirrors JMRI's ``Timebase``. Operators use a fast clock to compress
  4prototypical operating sessions (e.g. an 8-hour timetable runs in
  52 real hours at rate=4). The bean holds:
  6
  7* ``rate`` — clock speedup multiplier (1.0 = real time, 4.0 = 4× faster);
  8* ``running`` — whether the clock is advancing or held at a fixed time;
  9* the *displayed time* at the last anchor (set/start/stop/rate-change)
 10  paired with the monotonic real-clock timestamp at that anchor.
 11
 12Reads are computed on demand from those anchors, so there's no
 13background task: ``time_seconds()`` returns ``base + (now - anchor) *
 14rate`` while running, or ``base`` while stopped. This keeps the bean
 15trivial to reason about and means a server restart that loses the
 16in-memory state simply resets to midnight, stopped — a sensible
 17default for alpha.
 18
 19Notifications: a :class:`Signal` fires whenever the bean is mutated,
 20so listeners (web UI poll, future LogixNG triggers) can react.
 21"""
 22from __future__ import annotations
 23
 24import time
 25from dataclasses import dataclass, field
 26
 27from jmri_core.events import Signal
 28
 29SECONDS_PER_MINUTE = 60
 30SECONDS_PER_HOUR = 3600
 31SECONDS_PER_DAY = 86400
 32
 33# JMRI's Timebase clamps rate to ``[0.1, 100]``; we mirror that.
 34MIN_RATE = 0.1
 35MAX_RATE = 100.0
 36
 37
 38@dataclass
 39class FastClockChange:
 40    """Emitted whenever the fast clock is mutated. Listeners can poll
 41    :meth:`FastClock.time_seconds` directly if they need the current value."""
 42
 43    rate: float
 44    running: bool
 45    seconds_since_midnight: float
 46
 47
 48@dataclass(frozen=True, slots=True)
 49class HardwareClockState:
 50    """Snapshot of a command-station-driven fast clock.
 51
 52    Drivers whose command station has an authoritative fast clock (NCE,
 53    LocoNet's LNCP FastClock, etc.) return this from an async
 54    ``read_hardware_clock()`` method; the JSON ``TimeService`` prefers
 55    the hardware value over the software :class:`FastClock` bean when
 56    the connection exposes one.
 57
 58    * *hours* / *minutes* / *seconds* — the displayed 24-hour time
 59    * *rate* — the CS's fast-clock ratio (typically integer)
 60    * *running* — whether the CS clock is advancing
 61    * *twenty_four_hour* — the CS's display mode (12h with AM/PM vs 24h)
 62    """
 63
 64    hours: int
 65    minutes: int
 66    seconds: int
 67    rate: float
 68    running: bool
 69    twenty_four_hour: bool
 70
 71
 72@dataclass
 73class FastClock:
 74    """Software fast clock.
 75
 76    Default state: midnight, stopped, 1× rate. Real-world time source
 77    is ``time.monotonic`` so the bean is unaffected by wall-clock
 78    adjustments (NTP step, manual changes, DST).
 79    """
 80
 81    rate: float = 1.0
 82    running: bool = False
 83    _base_seconds: float = 0.0       # displayed fast-clock time at the anchor
 84    _real_anchor: float = field(default_factory=time.monotonic)
 85    state_changed: Signal[FastClockChange] = field(default_factory=Signal)
 86
 87    # ----- read --------------------------------------------------------
 88
 89    def time_seconds(self, *, now: float | None = None) -> float:
 90        """Current displayed time, in seconds since midnight (0..86399)."""
 91        if not self.running:
 92            return self._base_seconds % SECONDS_PER_DAY
 93        if now is None:
 94            now = time.monotonic()
 95        elapsed = (now - self._real_anchor) * self.rate
 96        return (self._base_seconds + elapsed) % SECONDS_PER_DAY
 97
 98    def time_hhmm(self, *, now: float | None = None) -> tuple[int, int]:
 99        """Convenience: ``(hours, minutes)`` ready for display."""
100        secs = int(self.time_seconds(now=now))
101        return (secs // SECONDS_PER_HOUR, (secs % SECONDS_PER_HOUR) // SECONDS_PER_MINUTE)
102
103    # ----- mutate ------------------------------------------------------
104
105    async def set_time(self, hours: int, minutes: int) -> None:
106        """Set the displayed time. Does not affect the running flag."""
107        if not 0 <= hours <= 23:
108            raise ValueError(f"hours must be 0..23, got {hours}")
109        if not 0 <= minutes <= 59:
110            raise ValueError(f"minutes must be 0..59, got {minutes}")
111        self._base_seconds = hours * SECONDS_PER_HOUR + minutes * SECONDS_PER_MINUTE
112        self._real_anchor = time.monotonic()
113        await self._notify()
114
115    async def set_rate(self, rate: float) -> None:
116        """Change the speedup multiplier without losing the displayed time."""
117        if not MIN_RATE <= rate <= MAX_RATE:
118            raise ValueError(f"rate must be {MIN_RATE}..{MAX_RATE}, got {rate}")
119        # Capture current displayed time first so the change doesn't jump.
120        current = self.time_seconds()
121        self._base_seconds = current
122        self._real_anchor = time.monotonic()
123        self.rate = rate
124        await self._notify()
125
126    async def start(self) -> None:
127        """Begin advancing the clock. Idempotent."""
128        if self.running:
129            return
130        # Keep the displayed time stable across the start.
131        self._real_anchor = time.monotonic()
132        self.running = True
133        await self._notify()
134
135    async def stop(self) -> None:
136        """Hold the clock at the current displayed time. Idempotent."""
137        if not self.running:
138            return
139        self._base_seconds = self.time_seconds()
140        self.running = False
141        await self._notify()
142
143    # ----- internal ----------------------------------------------------
144
145    async def _notify(self) -> None:
146        await self.state_changed.emit(FastClockChange(
147            rate=self.rate,
148            running=self.running,
149            seconds_since_midnight=self.time_seconds(),
150        ))