# SPDX-License-Identifier: Apache-2.0 """The main loop: eye → device → display → ring and motor; meeting end → note → NFC tag; power. python -m goldfish.daemon --sim # on a laptop: type gestures, see the ring as text python -m goldfish.daemon # on the Radxa (untested: the hardware isn't built) `Daemon.step()` is one pass of the loop and takes all its time from the board's clock, so tests and the sim run it on scripted time. Speech reaches the daemon through `hear()`; on the device that is the audio pipeline's job (audio.py's seams, backlog B-35). No radio code anywhere: the only way out is the passive tag. """ from __future__ import annotations import os import sys from dataclasses import dataclass, field from datetime import datetime, timezone from resolution_protocol import parse_duration, unreadable_after from .device import Goldfish from .display import battery_frame, ring_frame, seat_frame from .eye import Eye, EyeTimes, Gesture, press_edges from .handoff import fit, front_matter, ndef_tag_image, section from .haptic import Haptic from .power import Power from .ring import Ring, text from .st25dv import ST25DV, publish READER = "https://npc.here.now/demo-goldfish/r" # the site's reader page; the kit's final reader URL is the owner's call HAPTIC_FOR = {"applied": "lowered", "still-off": "still-off", "provisional": "provisional", "reverted": "reverted"} EVENT_HAPTIC = {"confirmed": "window-closed", "reverted": "reverted"} @dataclass(frozen=True) class Timing: """The loop's own times; the eye's are in eye.EyeTimes, power's in power.PowerTimes.""" fps: float = 30.0 # ring refresh: smooth enough for the 4 s breath speaking: float = 1.5 # profile bindings.indicator.speaking_window battery_at_start: float = 3.0 # display.py: the gauge shows for 3 s at power-on @classmethod def from_profile(cls, p) -> "Timing": ind = p.extra.get("bindings", {}).get("indicator", {}) return cls(speaking=parse_duration(ind.get("speaking_window", "PT1.5S"))) def compose_note(meeting: dict, profile, reader: str = READER, lines: dict | None = None) -> tuple[str, list[str]]: """The meeting's goldfish-note/1 (v01/note-format.md). `lines` = {1: outcomes, 2: gist, 3: attributed}, from the distiller once it exists; until then every section reads "(none)" and the note says why.""" lines = lines or {} t0 = datetime.fromtimestamp(meeting["wall_start"], timezone.utc) until = lambda lv: datetime.fromtimestamp(meeting["wall_start"] + unreadable_after(lv, profile.retention), timezone.utc).strftime("%Y-%m-%d") mins = max(1, round(meeting["duration"] / 60)) pre = front_matter({"recorded": t0.strftime("%Y-%m-%dT%H:%M"), "duration": f"{mins // 60}h {mins % 60:02d}m" if mins >= 60 else f"{mins}m", "voices": meeting["voices"], "protocol": "Resolution Protocol 0.2", "profile": "goldfish", "device": "Goldfish v01"}) pre += f"\n\n# Session note: {t0.day} {t0.strftime('%b %Y, %H:%M')}" if not any(lines.values()): pre += "\n\n> No summaries: the on-device speech models are not installed yet (backlog B-35). This note lists the meeting's facts only." secs = [section("Outcomes", "R1 outcomes", until(1), lines.get(1, [])), section("Gist", "R2 gist", until(2), lines.get(2, [])), section("Attributed", "R3 attributed", until(3), lines.get(3, []))] return fit(secs, reader, preamble=pre) @dataclass class Daemon: board: object # hw.Board g: Goldfish = field(default_factory=Goldfish) reader: str = READER say: object = None # print-like, for what happened; None = quiet flush: list = field(default_factory=list) # callables: audio buffers → store (the pipeline's seam) distil: object = None # callable(meeting) → {1: [...], 2: [...], 3: [...]} (seam) keyring: object = None # keys.Keyring; advanced at each midnight when given bearings: list | None = None # each ring LED's bearing from the ring's centre, degrees (layout.json); # None = display.led_bearings(), evenly spaced def __post_init__(self) -> None: b = self.board self.times = Timing.from_profile(self.g.profile) self.eye = Eye(EyeTimes.from_profile(self.g.profile)) self.ring = Ring(b.spi) self.haptic = Haptic(b.motor) self.power = Power(b.slide, b.keep_alive) self.tag = ST25DV(b.i2c, sleep=b.clock.sleep) self.heard_at: dict[str, float] = {} self.bearing: dict[str, float] = {} # each voice's angle from the head, degrees, when the mics give one self.mics_cut = False self.showing_battery = False self.battery_until = b.clock.now() + self.times.battery_at_start self.last_frame: list | None = None self.last_icon = False self.tag_writes = 0 self.last_segment = None self._seen = len(self.g.session.events) self._day = int(b.clock.wall() // 86400) self.power.start() if not self.power.can_hold: self._say("power: no slide sense or keep-alive (B-44): switching off cuts power at once; only SIGTERM shuts down cleanly") def _say(self, msg: str) -> None: if self.say: self.say(msg) def _buzz(self, name: str | None, now: float | None = None) -> None: if name: self.haptic.play(name, self.board.clock.now()) # from when it is felt, not when the press began self._say(f"buzz: {name}") # --- inputs ------------------------------------------------------------------- def hear(self, label: str, text_: str, now: float | None = None, span: tuple | None = None, bearing: float | None = None) -> list[str]: """A voice spoke (from the recogniser). Updates the speaking light and buzzes what changed. `span` = (t0, t1), the speech the words came from: given, the words are filed in the voice's own record (session.stamp) at the lowest level that applied over them, and a line that lowers its speaker is kept at the level it asked for. Without it nothing is filed (the audio pipeline, B-35, supplies spans). `bearing` = the voice's angle from the head in degrees, from the head and tail microphones (ADR-0015).""" now = self.board.clock.now() if now is None else now self._meeting_gap(now) s, was_off = self.g.session, self.g.room_off lvl = s.applied(label) # a voice heard for the first time starts at the room's level seg = s.stamp(label, span[0], span[1]) if span is not None else None out = self.g.hear(label, text_, now) if seg is not None: seg.level = min(seg.level, s.applied(label)) self.last_segment = seg self.heard_at[label] = now if bearing is not None: self.bearing[label] = bearing new = [e["type"] for e in s.events[self._seen:]] self._seen = len(s.events) if "reverted" in out: self._buzz("reverted", now) elif not was_off and self.g.room_off: self._buzz("lowered", now) elif "provisional" in out: self._buzz("provisional", now) elif "forget" in new or (lvl is not None and s.applied(label) < lvl): self._buzz("own-lowered", now) self._say(f"{label}: {text_!r} → {', '.join(out) or 'nothing'}") return out def speaking(self, label: str, now: float | None = None, bearing: float | None = None) -> None: """The voice-activity detector hears `label` talking, before any words are recognised: a voice already in the session brightens for the speaking window. (Diarised VAD on the device: B-35.)""" now = self.board.clock.now() if now is None else now if label in self.g.session.voices: self.heard_at[label] = now if bearing is not None: self.bearing[label] = bearing def _gesture(self, gs: Gesture) -> None: if gs.kind == "down": self.battery_until = gs.t # a press is never hidden behind the power-on gauge out = self.g.eye_down(gs.t) self._seen = len(self.g.session.events) self._buzz(HAPTIC_FOR.get(out), gs.t) self._say(f"eye down at {gs.t:.3f} s → {out}") elif gs.kind == "hold": self.showing_battery = True self._buzz("battery", gs.t) elif gs.kind == "up": self.g.eye_up(gs.t, gs.held) self.showing_battery = False # --- the meeting ends --------------------------------------------------------- def _meeting_gap(self, now: float) -> None: s = self.g.session if s.last_heard is not None and now >= s.last_heard + s.profile.session_gap: self.end_meeting(now, s.last_heard + s.profile.session_gap) def end_meeting(self, now: float, at: float | None = None) -> str | None: """Close the session (at the gap, or at a clean switch-off), compose the note, put it on the tag.""" s = self.g.session if s.started is None: return None meeting = {"start": s.started, "duration": (s.last_heard or now) - s.started, "voices": len(s.voices), "wall_start": self.board.clock.wall() - (now - s.started), "records": {v.id: list(v.record) for v in s.voices.values()}} if at is None: s.end(now) else: s.advance(now) # the protocol's own gap handling: confirms or undoes an open raise self._seen = len(s.events) self.heard_at.clear() self.bearing.clear() url, dropped = compose_note(meeting, self.g.profile, self.reader, self.distil(meeting) if self.distil else None) publish(self.tag, ndef_tag_image(url)) self.tag_writes += 1 self.g.end_meeting(now, url) self._buzz("note-ready", now) self._say(f"meeting over: note on the tag ({len(url)} chars{', dropped ' + ', '.join(dropped) if dropped else ''})") return url # --- one pass ----------------------------------------------------------------- def frame(self, now: float) -> list: charge = self.board.battery() if charge is not None and (self.showing_battery or now < self.battery_until): return battery_frame(charge) if self.g.room_off or self.mics_cut: return ring_frame([], set(), room_off=True, t=now) s = self.g.session voices = sorted(s.voices.values(), key=lambda v: v.slot) levels = [s.applied(v.label) for v in voices] speaking = {i for i, v in enumerate(voices) if now - self.heard_at.get(v.label, float("-inf")) <= self.times.speaking} pv = self.g.provisional_left(now) seats = [self.bearing.get(v.label) for v in voices] if pv is None and voices and None not in seats: # ADR-0015: each person's lights on the side facing them, as two mirrored arcs (the mics hear only |angle|) return seat_frame([(lv, abs(b)) for lv, b in zip(levels, seats)], speaking, bearings=self.bearings) return ring_frame(levels, speaking, provisional=pv) def step(self) -> bool: """One pass of the loop. False once the power state machine wants to shut down.""" b = self.board now = b.clock.now() for gs in self.eye.step(b.eye.edges(), now): self._gesture(gs) self._meeting_gap(now) s = self.g.session s.advance(now) # closes an expired objection window for e in s.events[self._seen:]: self._buzz(EVENT_HAPTIC.get(e["type"]), now) self._seen = len(s.events) self._midnight() self.last_frame, self.last_icon = self.frame(now), self.g.note_ready(now) self.ring.show(self.last_frame, self.last_icon) self.haptic.tick(now) return not self.power.poll(now, b.battery()) def _midnight(self) -> None: day = int(self.board.clock.wall() // 86400) if day != self._day: self._day = day if self.keyring is not None: from .device import key_curve self.keyring.midnight(today=day, **key_curve(self.g.profile)) def shutdown(self) -> str: """The clean path (B-44, and SIGTERM today): flush, end the meeting onto the tag, sync, ring off.""" c = self.board.clock steps = [(f"flush {i}", fn) for i, fn in enumerate(self.flush)] steps += [("end meeting", lambda: self.end_meeting(c.now())), ("sync", os.sync), ("ring off", self.ring.off), ("motor off", self.haptic.stop)] action = self.power.shutdown(steps, c.now) self._say(f"shutdown ({self.power.reason}): " + "; ".join(f"{n} {r}" for n, r in self.power.log) + f" → {action}") return action def run(self, until=lambda: False) -> str: c = self.board.clock while self.step() and not until(): c.sleep(1 / self.times.fps) return self.shutdown() # --- the laptop demo --------------------------------------------------------------- HELP = """commands (sim time moves only when you say so): p press the eye pp double press h hold the eye 2.5 s s NAME TEXT NAME says TEXT (try: s ana Goldfish, keep my words to just the decisions) w SECONDS wait m toggle the mic slide (mics cut) off power slide off (clean shutdown) q quit""" def sim_command(d: Daemon, line: str) -> bool: """Run one demo command on a sim board; False to stop.""" b, c = d.board, d.board.clock parts = line.strip().split(maxsplit=2) cmd = parts[0] if parts else "" def run_for(s): end = c.now() + s while c.now() < end: if not d.step(): return False c.sleep(min(1 / d.times.fps, end - c.now()) or 1e-3) return d.step() def press(t, held): for e in press_edges(t, held, bounce=3): b.eye.at(*e) if cmd in ("q", "quit"): return False if cmd == "p": press(c.now(), 0.12); return run_for(1.0) # 1 s on, so the next command is a new gesture if cmd == "pp": press(c.now(), 0.1); press(c.now() + 0.3, 0.1); return run_for(1.0) if cmd == "h": press(c.now(), 2.5); return run_for(2.4) and _show(d) and run_for(1.0) if cmd == "s" and len(parts) >= 3: d.hear(parts[1], parts[2]); return run_for(0.1) if cmd == "w" and len(parts) >= 2: return run_for(float(parts[1])) if cmd == "m": d.mics_cut = not d.mics_cut; d._say(f"mics {'cut' if d.mics_cut else 'live'}"); return run_for(0.1) if cmd == "off": if b.slide is not None: b.slide.at(c.now(), False) else: d.power.request("slide") return run_for(0.2) print(HELP) return True def _show(d: Daemon, color: bool = False) -> bool: if d.last_frame is not None: print(f"t={d.board.clock.now():8.2f} s " + text(d.last_frame, d.last_icon, color)) return True def main(argv: list[str] | None = None) -> int: import argparse ap = argparse.ArgumentParser(prog="goldfish.daemon", description=__doc__.split("\n\n")[0]) ap.add_argument("--sim", action="store_true", help="run on a simulated board and read commands from stdin") ap.add_argument("--reader", default=READER, help="the reader page the note's link points to") ap.add_argument("--b44", action="store_true", help="open the power-slide sense and keep-alive lines (not wired in v01)") ap.add_argument("--plain", action="store_true", help="letters instead of coloured dots") a = ap.parse_args(argv) if a.sim: from .hw import sim_board d = Daemon(sim_board(), reader=a.reader, say=lambda m: print(" · " + m)) color = sys.stdout.isatty() and not a.plain print(HELP) d.step(); _show(d, color) for line in sys.stdin: if not sim_command(d, line): break _show(d, color) if d.power.state == "on": return 0 d.shutdown() return 0 import signal from .hw_linux import linux_board d = Daemon(linux_board(b44=a.b44), reader=a.reader, say=print) signal.signal(signal.SIGTERM, lambda *_: d.power.request("signal")) action = d.run() if d.power.reason != "signal": # a signal means systemd is already shutting down os.system(f"systemctl {action}") return 0 if __name__ == "__main__": sys.exit(main())