diff --git a/README.md b/README.md index 307bd8c..4accd47 100644 --- a/README.md +++ b/README.md @@ -82,9 +82,20 @@ Protocol constants, tape geometry and the day-one bring-up checklist live in [`HARDWARE-NOTES.md`](HARDWARE-NOTES.md). Read it before changing anything in `labelfab.device`. -**There is no read channel.** The D30 never reports back, so "printed" means "the bytes -were accepted and we waited the physical print duration". Out-of-tape, jams and low -battery are undetectable. +**There is a read channel**, contrary to the received wisdom: the D30 answers queries on +both transports and pushes media-state changes unsolicited. The retained status topic +carries what it said — serial, firmware, battery percentage and terminal voltage, and +whether media is loaded — not just whether the agent is up. + +It says so *and when it said so*. The printer is only reachable while it is being printed +to (it auto-powers-off, and a held-open socket just relocates that), so the agent +remembers the last thing it heard and republishes it, stamped with `device_seen_at`. +A consumer can then tell live truth from remembered truth instead of guessing, and no +printer is woken to keep a status page tidy. `device.probe_on_start` surveys the printer +once at startup when it happens to be awake. + +Jams remain undetectable — the media bit distinguishes loaded from not, and nothing +observed so far distinguishes a jam from either. ## License diff --git a/deploy/agent.toml.example b/deploy/agent.toml.example index 92cde54..0acedb4 100644 --- a/deploy/agent.toml.example +++ b/deploy/agent.toml.example @@ -42,6 +42,10 @@ raster_width_px = 96 # 96 = 12mm head (verified); 120 = 15mm (day-1 hypo pace_factor = 1.2 # lower until a long strip garbles, then +50% density = 1 # 1 light | 2 medium | 4 heavy; raise if codes misscan wake_dummy_feed = false +# Survey the printer once at startup so the status topic carries live truth instead of +# what was remembered from the last print. Cannot wake a sleeping unit, so a miss costs +# one connect timeout and the stored status is published unchanged. +probe_on_start = true [tape] width_mm = 15.0 diff --git a/src/labelfab/agent/__init__.py b/src/labelfab/agent/__init__.py index 041c5dd..339f205 100644 --- a/src/labelfab/agent/__init__.py +++ b/src/labelfab/agent/__init__.py @@ -11,6 +11,7 @@ from labelfab.agent.coalescer import Batch, Coalescer, PendingLabel from labelfab.agent.config import Config, load +from labelfab.agent.device_state import DeviceSnapshot from labelfab.agent.publisher import NullPublisher, Publisher, RecordingPublisher from labelfab.agent.spool import InsertResult, Outcome, Spool from labelfab.agent.worker import PrintWorker @@ -19,6 +20,7 @@ "Batch", "Coalescer", "Config", + "DeviceSnapshot", "InsertResult", "NullPublisher", "Outcome", diff --git a/src/labelfab/agent/__main__.py b/src/labelfab/agent/__main__.py index 0fcde99..be69b70 100644 --- a/src/labelfab/agent/__main__.py +++ b/src/labelfab/agent/__main__.py @@ -93,6 +93,11 @@ def run(self) -> int: signal.signal(signal.SIGINT, self.stop) signal.signal(signal.SIGTERM, self.stop) + # Before the source starts, so whatever these two learn is what the first + # retained status carries: the source republishes the last thing published, + # and nothing has been published yet. + if self.config.device.probe_on_start: + self.worker.probe_device() self.worker.recover() if self.dir_source: self.dir_source.poll() diff --git a/src/labelfab/agent/config.py b/src/labelfab/agent/config.py index c10dfa8..6903999 100644 --- a/src/labelfab/agent/config.py +++ b/src/labelfab/agent/config.py @@ -95,6 +95,13 @@ class DeviceSection(BaseModel): #: Drop the socket between batches: the D30 auto-sleeps, so a held-open socket #: just relocates the failure. Reconnect-per-batch is cheaper to reason about. idle_disconnect: bool = True + #: Ask the printer what it is once at startup, so the status topic carries live + #: truth rather than remembered truth from the moment the agent comes up. This + #: cannot wake a sleeping unit -- it has powered its radio off and only the button + #: brings it back -- so the cost of a miss is one connect timeout, after which the + #: stored snapshot is published unchanged. On by default; turn it off if that + #: timeout is in the way of a fast start. Never runs on a broker reconnect. + probe_on_start: bool = True @field_validator("density") @classmethod diff --git a/src/labelfab/agent/device_state.py b/src/labelfab/agent/device_state.py new file mode 100644 index 0000000..5ce7bd1 --- /dev/null +++ b/src/labelfab/agent/device_state.py @@ -0,0 +1,123 @@ +"""What the printer last said about itself, and when it said it. + +The D30 only talks while it is being printed to. It auto-powers-off, and +``device.idle_disconnect`` defaults to true precisely because a held-open socket just +relocates the failure -- so between jobs there is nobody to ask. Every consumer of the +status topic therefore reads *remembered* device truth most of the time, and the honest +way to serve that is to remember it deliberately and say how old it is, rather than +either forgetting it or waking the printer to refresh it. + +``seen_at`` is the whole point. Without it a consumer cannot distinguish "media is +loaded" from "media was loaded at some unknown time in the past", and rendering the +second as the first is exactly the failure the tri-state ``media_ok`` was introduced to +avoid, one layer along. + +This also collects the six loose ``_device_*`` fields the worker used to carry and the +two places that built a ``PrinterStatus`` out of them. That duplication is why the +worker and the MQTT source disagreed: one published everything the printer had said, +the other published a hardcoded blank. +""" + +from __future__ import annotations + +from datetime import datetime, timezone + +from pydantic import BaseModel, ConfigDict + +from labelfab.contract import PrinterStatus +from labelfab.device.d30 import MODEL +from labelfab.device.feedback import DeviceFeedback + + +class DeviceSnapshot(BaseModel): + """Last-known device truth, persisted in the spool and republished on connect.""" + + # ``extra="ignore"`` is load-bearing rather than a default: this is stored in the + # spool DB, which outlives package upgrades, so a row written by a different agent + # version has to degrade to the fields it has in common. Failing to decode it would + # take the agent down at boot, and a status field is not worth an outage. + model_config = ConfigDict(extra="ignore", frozen=True) + + serial: str | None = None + firmware: str | None = None + battery_pct: int | None = None + voltage_v: float | None = None + media_ok: bool | None = None + #: What the printer was complaining about the last time it was asked. Never latched + #: -- see ``merge``. + fault: str | None = None + #: Epoch seconds when the printer last reported anything, or ``None`` if it never + #: has. Stored as a float to match the rest of the spool; converted to a UTC + #: instant at the wire boundary. + seen_at: float | None = None + + # -- accumulation ------------------------------------------------------- # + + def merge(self, fb: DeviceFeedback, *, now: float) -> DeviceSnapshot: + """Fold one connection's feedback in, keeping whatever it did not report. + + A connection that answered nothing must not blank out a serial we already + know. ``fault`` is the deliberate exception: it is recomputed from scratch + every connection, so a cleared media error clears the status instead of + latching an error forever. + + ``seen_at`` only moves when the printer actually said something. Advancing it + on a silent connection would claim a freshness we do not have, and this + timestamp is the only thing standing between a consumer and treating + three-day-old truth as live. + """ + reported = (fb.serial, fb.firmware, fb.battery_pct, fb.voltage_v, fb.paper_ok) + return DeviceSnapshot( + serial=fb.serial or self.serial, + firmware=fb.firmware or self.firmware, + battery_pct=self.battery_pct if fb.battery_pct is None else fb.battery_pct, + voltage_v=self.voltage_v if fb.voltage_v is None else fb.voltage_v, + media_ok=self.media_ok if fb.paper_ok is None else fb.paper_ok, + fault=fb.fault(), + seen_at=now if any(v is not None for v in reported) else self.seen_at, + ) + + # -- publishing --------------------------------------------------------- # + + def settled_state(self) -> str: + """What to publish once nothing is in flight. + + "idle" unless the printer is complaining. Kept apart from the status builder so + that "printing" and "disconnected" -- facts about the link, not about the media + -- are never silently rewritten into it. + """ + return "error" if self.fault else "idle" + + def to_status( + self, + printer_id: str, + *, + state: str, + tape_width_mm: float | None = None, + pending_labels: int = 0, + ) -> PrinterStatus: + """Build the retained status message from this snapshot.""" + return PrinterStatus( + printer_id=printer_id, + state=state, # type: ignore[arg-type] + model=MODEL, + serial=self.serial, + firmware=self.firmware, + battery_pct=self.battery_pct, + voltage_v=self.voltage_v, + media_ok=self.media_ok, + tape_width_mm=tape_width_mm, + pending_labels=pending_labels, + error=self.fault, + device_seen_at=self.seen_at_utc(), + ) + + def seen_at_utc(self) -> datetime | None: + """``seen_at`` as a UTC instant, whole seconds. + + Truncated because sub-second precision on "when did the printer last speak" is + noise in a payload whose primary reader is a human running ``mosquitto_sub -v``. + """ + if self.seen_at is None: + return None + return datetime.fromtimestamp(self.seen_at, tz=timezone.utc).replace(microsecond=0) diff --git a/src/labelfab/agent/source_mqtt.py b/src/labelfab/agent/source_mqtt.py index 7821552..3c6f40b 100644 --- a/src/labelfab/agent/source_mqtt.py +++ b/src/labelfab/agent/source_mqtt.py @@ -55,10 +55,16 @@ def __init__(self, config: Config, spool: Spool, enqueue: Callable[[str], None]) if config.mqtt.username: self.client.username_pw_set(config.mqtt.username, config.mqtt.password) - # Last will: if the agent drops, the retained status flips to disconnected so - # a producer sees the printer is unreachable before it ever clicks print. - will = PrinterStatus(printer_id=config.agent.printer_id, state="disconnected") - self.client.will_set(config.topic("status"), will.model_dump_json(), qos=1, retain=True) + # What the retained topic currently says. Seeded from the spool so a fresh + # process starts out holding the previous one's device truth, then updated by + # every retained publish -- which is what lets a reconnect restore the topic + # instead of flattening it. + snapshot = spool.device_snapshot() + self._last_status = snapshot.to_status( + config.agent.printer_id, + state=snapshot.settled_state(), + tape_width_mm=config.tape.width_mm, + ) self.client.on_connect = self._on_connect self.client.on_message = self._on_message @@ -66,24 +72,58 @@ def __init__(self, config: Config, spool: Spool, enqueue: Callable[[str], None]) # -- lifecycle ---------------------------------------------------------- # def start(self) -> None: + # Armed here rather than in __init__ so it reflects whatever the startup probe + # and crash recovery learned, which happen after this object is built. + self._arm_will() self.client.connect(self.config.mqtt.host, self.config.mqtt.port, self.config.mqtt.keepalive_s) self.client.loop_start() + def _arm_will(self) -> None: + """Point the will at the current last-known truth. + + The will lives in the CONNECT packet, so the broker holds a fixed payload for + the whole session and no amount of calling this changes what fires if *this* + connection drops. paho rebuilds CONNECT from these fields on every reconnect + though, so re-arming still moves the staleness window from "whenever the + process started" -- potentially weeks -- down to "the last reconnect", which + here is a few hours at most. Called from ``_on_connect``, i.e. from paho's own + network thread, which is also the thread that builds the packet, so there is + no window where the payload and the topic disagree. + """ + self.client.will_set( + self.config.topic("status"), self._disconnected().model_dump_json(), qos=1, retain=True + ) + def stop(self) -> None: try: - self.publish_status( - PrinterStatus(printer_id=self.config.agent.printer_id, state="disconnected") - ) + self.publish_status(self._disconnected()) finally: self.client.loop_stop() self.client.disconnect() + def _disconnected(self) -> PrinterStatus: + """Last-known truth with the link marked down. + + Backs both the will and the shutdown notice: if the agent drops, a producer + must see that the printer is unreachable before it ever clicks print. Losing + the link is news about reachability, though, not grounds for forgetting the + serial, the firmware and the media state -- ``device_seen_at`` rides along to + say how old they are. + """ + return self._last_status.model_copy(update={"state": "disconnected", "pending_labels": 0}) + # -- Publisher protocol ------------------------------------------------- # def publish_result(self, result: JobResult) -> None: self.client.publish(self.config.topic("results"), result.model_dump_json(), qos=1) def publish_status(self, status: PrinterStatus, *, retain: bool = True) -> None: + if retain: + # Recorded before the publish and regardless of whether it succeeds: the + # startup probe and crash recovery both publish before the client is + # connected, so those go nowhere and the connect below is what lands them. + # A non-retained publish is not what the topic holds, so it is not recorded. + self._last_status = status self.client.publish(self.config.topic("status"), status.model_dump_json(), qos=1, retain=retain) def publish_progress(self, job_id: str, printed: int, total: int) -> None: @@ -98,13 +138,15 @@ def _on_connect(self, client, userdata, flags, reason_code, properties=None) -> return client.subscribe(self.config.topic("jobs"), qos=self.config.mqtt.qos) client.subscribe(self.config.topic("cmd"), qos=1) - self.publish_status( - PrinterStatus( - printer_id=self.config.agent.printer_id, - state="idle", - tape_width_mm=self.config.tape.width_mm, - ) - ) + # Republish what the topic already said, rather than a freshly synthesised + # blank. This fires on every reconnect, not just the first connect, and paho + # reconnects silently -- the Istio route for the broker caps at a 24h timeout + # and flaps besides, so it runs several times a day. Publishing a hardcoded + # bare status here was not merely uninformative, it was lossy: it overwrote a + # retained message that was correct a moment earlier, and the printer is asleep + # and cannot be asked again. + self.publish_status(self._last_status) + self._arm_will() # for the connect after this one; see _arm_will def _on_message(self, client, userdata, msg) -> None: if msg.topic.endswith("/cmd"): diff --git a/src/labelfab/agent/spool.py b/src/labelfab/agent/spool.py index 5b8bffb..c3dc028 100644 --- a/src/labelfab/agent/spool.py +++ b/src/labelfab/agent/spool.py @@ -14,10 +14,16 @@ its cached result and prints nothing. * ``dedupe_key`` deduplicates individual *labels* across different jobs, so a "print bin A4" that fires twice an hour apart does not waste tape. + +One thing here is not about jobs at all: the last :class:`DeviceSnapshot`. It is stored +alongside them because this file is already the agent's only durable place, and because +the printer is only reachable during a print -- so the alternative to remembering what +it said is waking it up to ask again. """ from __future__ import annotations +import logging import sqlite3 import time from collections.abc import Callable @@ -25,8 +31,13 @@ from enum import Enum from pathlib import Path +from pydantic import ValidationError + +from labelfab.agent.device_state import DeviceSnapshot from labelfab.contract import JobResult, PrintJob +log = logging.getLogger("labelfab.spool") + SCHEMA = """ CREATE TABLE IF NOT EXISTS jobs ( job_id TEXT PRIMARY KEY, @@ -51,6 +62,14 @@ printed_at REAL NOT NULL, PRIMARY KEY (job_id, label_index) ); + +-- One row, holding the last DeviceSnapshot. The printer only talks while it is being +-- printed to, so without this the agent starts every process knowing nothing about the +-- hardware it drives until the next job happens to come along. +CREATE TABLE IF NOT EXISTS device_state ( + id INTEGER PRIMARY KEY CHECK (id = 1), + payload TEXT NOT NULL +); """ @@ -186,3 +205,34 @@ def printed_labels(self, job_id: str) -> set[int]: "SELECT label_index FROM printed_labels WHERE job_id = ?", (job_id,) ).fetchall() return {r["label_index"] for r in rows} + + # -- last-known device truth -------------------------------------------- # + # + # Written after every print, read once at startup. This is the only state here that + # is not about jobs, and it lives here because the alternative -- asking the printer + # on boot -- means waking a device whose whole operating model is "asleep until + # needed", every time the agent restarts. + + def save_device_snapshot(self, snapshot: DeviceSnapshot) -> None: + self._db.execute( + "INSERT INTO device_state (id, payload) VALUES (1, ?) " + "ON CONFLICT(id) DO UPDATE SET payload = excluded.payload", + (snapshot.model_dump_json(),), + ) + + def device_snapshot(self) -> DeviceSnapshot: + """The last snapshot written, or an empty one. + + Empty covers both "this agent has never printed" and "the stored row is no + longer readable". The second degrades rather than raising on purpose: an + unusable status field is a cosmetic problem, and taking the print agent down at + boot over one would be a much larger one. + """ + row = self._db.execute("SELECT payload FROM device_state WHERE id = 1").fetchone() + if row is None: + return DeviceSnapshot() + try: + return DeviceSnapshot.model_validate_json(row["payload"]) + except ValidationError: + log.warning("stored device snapshot is unreadable; starting without one") + return DeviceSnapshot() diff --git a/src/labelfab/agent/worker.py b/src/labelfab/agent/worker.py index da32ed4..89895cc 100644 --- a/src/labelfab/agent/worker.py +++ b/src/labelfab/agent/worker.py @@ -30,7 +30,6 @@ PX_PER_MM, JobResult, LabelResult, - PrinterStatus, PrintJob, TapeSpec, ) @@ -44,6 +43,10 @@ #: Backoff between reconnect attempts within one send, in seconds. _BACKOFF_S = (1.0, 3.0, 9.0) +#: How long the startup probe waits for the printer's replies to arrive on the +#: transport's reader thread before reading them. +_PROBE_SETTLE_S = 1.0 + @dataclass class _JobAcc: @@ -95,14 +98,10 @@ def __init__( self.retry_interval_s = retry_interval_s #: job_id -> last stall time, so a napping printer is retried, not busy-looped. self._stalled: dict[str, float] = {} - #: Latest device serial reported by the printer, on either transport. - self._device_serial: str | None = None - self._device_firmware: str | None = None - self._device_battery_pct: int | None = None - self._device_voltage_v: float | None = None - self._device_media_ok: bool | None = None - #: Last fault the printer reported, or None. Drives the "error" status state. - self._device_fault: str | None = None + #: What the printer last said about itself. Seeded from the spool, so a restart + #: starts out knowing what the previous process learned instead of publishing a + #: row of nulls until something happens to print. + self._device = spool.device_snapshot() self.coalescer = Coalescer( max_wait_s=config.strip.max_wait_s, max_length_mm=config.strip.max_length_mm, @@ -252,6 +251,48 @@ def recover(self) -> None: except KeyError: pass + def probe_device(self) -> bool: + """Ask the printer what it is, if it happens to be awake. Returns whether it answered. + + This is not a wake attempt and cannot be one: a slept D30 has powered its radio + down (``AUTO_POWER_TIME``, HARDWARE-NOTES §7) and only the button brings it + back, so this either finds it already awake and learns the truth, or fails + inside the transport's connect timeout and leaves the remembered snapshot + exactly as it was. + + Deliberately called from the run loop at startup, and only there. Not from the + MQTT ``on_connect`` callback: that is paho's network thread, so probing from it + would touch the printer concurrently with a print -- the one thing the + single-threaded loop exists to make impossible -- and it fires on every broker + reconnect, which happens several times a day and says nothing about the printer. + + One attempt, no backoff. The retry ladder in ``_send`` exists to get a *job* + onto tape; there is no job here and nothing is lost by giving up immediately. + """ + printer = self.printer_factory() + try: + printer.connect() + printer.refresh_telemetry() + except D30Error as exc: + printer.close() + log.info("startup probe: printer did not answer (%s); keeping the stored status", exc) + return False + except Exception: # a status nicety must never be what stops the agent starting + printer.close() + log.exception("startup probe failed; keeping the stored status") + return False + # Replies land asynchronously on the transport's reader thread, and unlike the + # print path there is no print duration to cover the wait. Settling short is + # safe by construction: an empty merge keeps every stored field and does not + # advance seen_at, so the worst case is learning nothing, never losing anything. + self.sleep(_PROBE_SETTLE_S) + self._capture_feedback(printer) + printer.close() + # Publish it: the MQTT source has not connected yet, so this does not reach the + # broker -- it updates what the source will republish the moment it does. + self._publish_status(self._settled_state()) + return True + # -- flushing and printing ---------------------------------------------- # def _flush(self) -> None: @@ -375,26 +416,17 @@ def _capture_feedback(self, printer: PhomemoD30) -> None: changes just as BLE does -- so this is unconditional. A transport that somehow has no ``feedback`` is a programming error worth surfacing, not a case to silently skip; the previous ``None`` guard would have hidden exactly that. + + Persisted immediately. The printer is only reachable while it is being printed + to, so this instant is the *only* chance to learn any of it, and dropping it on + process exit is what left the status topic full of nulls. """ fb = printer.feedback - # Only overwrite what it actually reported this time: a connection that - # answered nothing must not blank out a serial we already know. The fault is - # the exception -- it is recomputed every time, so a cleared media error - # clears the status rather than latching an error forever. - if fb.serial: - self._device_serial = fb.serial - if fb.firmware: - self._device_firmware = fb.firmware - if fb.battery_pct is not None: - self._device_battery_pct = fb.battery_pct - if fb.voltage_v is not None: - self._device_voltage_v = fb.voltage_v - if fb.paper_ok is not None: - self._device_media_ok = fb.paper_ok - self._device_fault = fb.fault() + self._device = self._device.merge(fb, now=self.clock()) + self.spool.save_device_snapshot(self._device) log.info("device feedback: %s", fb.summary()) - if self._device_fault: - log.warning("printer reports a fault: %s", self._device_fault) + if self._device.fault: + log.warning("printer reports a fault: %s", self._device.fault) # -- result accounting -------------------------------------------------- # @@ -470,27 +502,15 @@ def _reject(self, job: PrintJob, reason: str) -> None: # -- status ------------------------------------------------------------- # def _settled_state(self) -> str: - """What to publish once a batch is done. - - "idle" unless the printer is complaining. Kept explicit rather than folded into - ``_publish_status`` so "printing" and "disconnected" -- which are facts about - the link, not the media -- are never silently rewritten. - """ - return "error" if self._device_fault else "idle" + """What to publish once a batch is done. See ``DeviceSnapshot.settled_state``.""" + return self._device.settled_state() def _publish_status(self, state: str, *, pending: int = 0) -> None: self.publisher.publish_status( - PrinterStatus( - printer_id=self.config.agent.printer_id, - state=state, # type: ignore[arg-type] - model=MODEL, - serial=self._device_serial, - firmware=self._device_firmware, - battery_pct=self._device_battery_pct, - voltage_v=self._device_voltage_v, - media_ok=self._device_media_ok, + self._device.to_status( + self.config.agent.printer_id, + state=state, tape_width_mm=self.config.tape.width_mm, pending_labels=pending, - error=self._device_fault, ) ) diff --git a/src/labelfab/contract/models.py b/src/labelfab/contract/models.py index 8b5b0ee..69104bb 100644 --- a/src/labelfab/contract/models.py +++ b/src/labelfab/contract/models.py @@ -7,6 +7,7 @@ from __future__ import annotations +from datetime import datetime from typing import Annotated, Any, Literal from pydantic import BaseModel, ConfigDict, Field, model_validator @@ -260,6 +261,16 @@ class PrinterStatus(Base): tape_width_mm: float | None = None pending_labels: int = 0 error: str | None = None + #: When the device fields above were last heard from the printer; ``None`` if it + #: has never said anything. The D30 is only reachable while it is being printed to, + #: so most reads of this topic are reads of *remembered* truth: the agent stores + #: what it last heard and republishes it across restarts and broker reconnects + #: rather than going quiet. This is how a consumer tells the two apart. + #: + #: An absolute instant rather than an age because this message is *retained*: an + #: age is computed at publish time and then sits on the broker getting wronger, + #: while a timestamp stays true no matter how long nobody republishes it. + device_seen_at: datetime | None = None def job_schema() -> dict[str, Any]: diff --git a/tests/test_agent_config.py b/tests/test_agent_config.py index 640e6dc..8195462 100644 --- a/tests/test_agent_config.py +++ b/tests/test_agent_config.py @@ -46,6 +46,16 @@ def test_density_defaults_to_light_and_reaches_the_driver(): assert printer.config.density == 1 +def test_the_startup_probe_is_on_and_can_be_turned_off(tmp_path): + """On by default because it cannot wake a sleeping printer, so the only cost of a + miss is one connect timeout. The knob exists for hosts where that is in the way.""" + assert Config().device.probe_on_start is True + + toml = tmp_path / "agent.toml" + toml.write_text("[device]\nprobe_on_start = false\n") + assert load(toml).device.probe_on_start is False + + def test_an_unknown_density_is_rejected_at_load(): import pytest from pydantic import ValidationError diff --git a/tests/test_device_state.py b/tests/test_device_state.py new file mode 100644 index 0000000..bc07660 --- /dev/null +++ b/tests/test_device_state.py @@ -0,0 +1,117 @@ +"""Last-known device truth: what is remembered, and how old it is allowed to look. + +The rules here all serve one property. The printer is only reachable while it is being +printed to, so almost every read of the status topic is a read of *remembered* truth. +That is fine as long as nobody can mistake it for live truth — which is what ``seen_at`` +is for, and why it is not allowed to drift forward on its own. +""" + +from __future__ import annotations + +from datetime import timezone + +from labelfab.agent import DeviceSnapshot +from labelfab.device import DeviceFeedback + +SERIAL = "Q223P4C31420105" + +#: Frames as the printer actually sends them; see tests/test_feedback.py. +FULL = ("1a08" + SERIAL.encode().hex(), "1a07020102", "1a0464", "1a2f01a1", "1a0689") +MEDIA_BAD = "1a0688" + + +def _feedback(*frames_hex: str) -> DeviceFeedback: + fb = DeviceFeedback() + for frame in frames_hex: + fb.ingest(bytes.fromhex(frame)) + return fb + + +def test_merge_decodes_a_whole_connection(): + snap = DeviceSnapshot().merge(_feedback(*FULL), now=1000.0) + assert snap.serial == SERIAL + assert snap.firmware == "2.1.2" + assert snap.battery_pct == 100 + assert snap.voltage_v == 4.17 + assert snap.media_ok is True + assert snap.fault is None + assert snap.seen_at == 1000.0 + + +def test_a_quiet_connection_keeps_what_we_already_knew(): + """Otherwise one uncommunicative connect erases a serial learned an hour ago.""" + known = DeviceSnapshot().merge(_feedback(*FULL), now=1000.0) + still = known.merge(_feedback(), now=2000.0) + assert still.serial == SERIAL + assert still.firmware == "2.1.2" + assert still.battery_pct == 100 + assert still.media_ok is True + + +def test_a_quiet_connection_does_not_advance_seen_at(): + """The timestamp is the only thing separating remembered truth from live truth. + + Moving it because we merely *connected* would launder three-day-old media state + into something a consumer renders as current. + """ + known = DeviceSnapshot().merge(_feedback(*FULL), now=1000.0) + assert known.merge(_feedback(), now=9999.0).seen_at == 1000.0 + + +def test_a_partial_report_advances_seen_at_and_leaves_the_rest(): + later = DeviceSnapshot().merge(_feedback(*FULL), now=1000.0).merge(_feedback("1a0458"), now=2000.0) + assert later.battery_pct == 88 # the one thing it reported + assert later.serial == SERIAL # everything it did not + assert later.seen_at == 2000.0 + + +def test_a_fault_does_not_latch(): + """A media error that has cleared must stop being reported, or the printer looks + broken until someone restarts the agent.""" + faulted = DeviceSnapshot().merge(_feedback(MEDIA_BAD), now=1000.0) + assert faulted.fault and faulted.settled_state() == "error" + + cleared = faulted.merge(_feedback(*FULL), now=2000.0) + assert cleared.fault is None + assert cleared.settled_state() == "idle" + assert cleared.media_ok is True + + +def test_status_carries_the_snapshot_and_its_age(): + snap = DeviceSnapshot().merge(_feedback(*FULL), now=1000.0) + status = snap.to_status("d30-workshop", state="idle", tape_width_mm=15.0, pending_labels=3) + + assert status.serial == SERIAL + assert status.media_ok is True + assert status.pending_labels == 3 + assert status.device_seen_at is not None + assert status.device_seen_at.tzinfo is not None # a bare instant is unusable + assert status.device_seen_at.timestamp() == 1000.0 + + +def test_never_observed_publishes_null_not_a_guess(): + status = DeviceSnapshot().to_status("d30-workshop", state="idle") + assert status.device_seen_at is None + assert status.media_ok is None # silence is still not health + + +def test_seen_at_is_whole_seconds_utc(): + """Sub-second precision on "when did the printer last speak" is noise in a payload + whose main reader is a human running mosquitto_sub.""" + seen = DeviceSnapshot(seen_at=1000.75).seen_at_utc() + assert seen is not None + assert seen.microsecond == 0 + assert seen.tzinfo is timezone.utc + + +def test_a_snapshot_survives_a_json_round_trip(): + snap = DeviceSnapshot().merge(_feedback(*FULL), now=1000.0) + assert DeviceSnapshot.model_validate_json(snap.model_dump_json()) == snap + + +def test_a_snapshot_from_another_version_degrades_rather_than_raising(): + """This row outlives package upgrades. Refusing to decode one written by a different + agent version would turn a cosmetic status field into a boot failure.""" + snap = DeviceSnapshot.model_validate_json(f'{{"serial": "{SERIAL}", "future_field": 42}}') + assert snap.serial == SERIAL + assert snap.seen_at is None diff --git a/tests/test_source_mqtt.py b/tests/test_source_mqtt.py new file mode 100644 index 0000000..39ffede --- /dev/null +++ b/tests/test_source_mqtt.py @@ -0,0 +1,273 @@ +"""What the retained status topic holds, across connects and reconnects. + +Driven through a fake paho client rather than a broker: everything worth asserting +here is about *which payload* is published on which callback, and none of it needs a +network. The one thing a broker would add — that paho reconnects on its own, silently, +several times a day — is the premise, not something under test. +""" + +from __future__ import annotations + +import json + +import pytest +from conftest import make_config + +from labelfab.agent import DeviceSnapshot, Spool +from labelfab.agent.source_mqtt import FLUSH_COMMAND, MqttSource +from labelfab.contract import PrinterStatus + +SERIAL = "Q223P4C31420105" + + +class FakeClient: + """Stands in for ``paho.mqtt.client.Client``, recording publishes and subscriptions.""" + + def __init__(self, *_args, **_kwargs) -> None: + self.published: list[tuple[str, str, bool]] = [] + self.subscribed: list[str] = [] + self.will: tuple[str, str] | None = None + self.on_connect = None + self.on_message = None + self.acked: list[int] = [] + + # -- setup the source performs ------------------------------------------ # + def ws_set_options(self, **_kw) -> None: ... + def tls_set(self, *_a, **_kw) -> None: ... + def username_pw_set(self, *_a, **_kw) -> None: ... + def connect(self, *_a, **_kw) -> None: ... + def loop_start(self) -> None: ... + def loop_stop(self) -> None: ... + def disconnect(self) -> None: ... + + def will_set(self, topic, payload, qos=0, retain=False) -> None: + self.will = (topic, payload) + + def subscribe(self, topic, qos=0) -> None: + self.subscribed.append(topic) + + def publish(self, topic, payload, qos=0, retain=False) -> None: + self.published.append((topic, payload, retain)) + + def ack(self, mid, qos) -> None: + self.acked.append(mid) + + # -- what tests read ---------------------------------------------------- # + @property + def statuses(self) -> list[dict]: + return [json.loads(p) for t, p, _ in self.published if t.endswith("/status")] + + +@pytest.fixture +def source(tmp_path, monkeypatch): + """Build an MqttSource over a fake client, on a spool the test can pre-load.""" + import paho.mqtt.client as mqtt + + monkeypatch.setattr(mqtt, "Client", FakeClient) + + def build(snapshot: DeviceSnapshot | None = None) -> MqttSource: + spool = Spool(tmp_path / "spool.db") + if snapshot is not None: + spool.save_device_snapshot(snapshot) + config = make_config() + config.mqtt.host = "broker.invalid" + return MqttSource(config, spool, lambda _job_id: None) + + return build + + +def _connect(src: MqttSource) -> None: + """Fire the callback paho fires on connect — and on every silent reconnect.""" + src._on_connect(src.client, None, {}, 0) + + +KNOWN = DeviceSnapshot( + serial=SERIAL, firmware="2.1.2", battery_pct=100, voltage_v=4.18, media_ok=True, seen_at=1000.0 +) + + +def test_a_reconnect_republishes_device_truth_instead_of_erasing_it(source): + """The reported defect. A reconnect used to publish a hardcoded bare status, + retained, over a message that was correct a moment earlier — and the printer is + asleep, so nothing could put it back until the next print.""" + src = source() + src.publish_status( + PrinterStatus(printer_id="d30-workshop", state="idle", serial=SERIAL, media_ok=True) + ) + + _connect(src) # paho reconnects on its own, several times a day + + restored = src.client.statuses[-1] + assert restored["serial"] == SERIAL + assert restored["media_ok"] is True + + +def test_a_fresh_process_republishes_what_the_last_one_learned(source): + """The original issue: a restarted agent had nothing to say about the printer until + something happened to print.""" + src = source(KNOWN) + _connect(src) + + published = src.client.statuses[-1] + assert published["serial"] == SERIAL + assert published["firmware"] == "2.1.2" + assert published["media_ok"] is True + assert published["device_seen_at"] is not None + assert published["state"] == "idle" + + +def test_a_remembered_fault_is_not_downgraded_to_healthy_on_restart(source): + src = source(KNOWN.model_copy(update={"media_ok": False, "fault": "media not ready"})) + _connect(src) + + published = src.client.statuses[-1] + assert published["state"] == "error" + assert published["media_ok"] is False + assert published["error"] == "media not ready" + + +def test_an_agent_that_has_never_printed_claims_nothing(source): + src = source() + _connect(src) + + published = src.client.statuses[-1] + assert published["state"] == "idle" + assert published["serial"] is None + assert published["media_ok"] is None # not "fine", just unsaid + assert published["device_seen_at"] is None + + +def test_the_status_is_retained_and_the_topics_are_subscribed(source): + src = source(KNOWN) + _connect(src) + + assert all(retain for topic, _, retain in src.client.published if topic.endswith("/status")) + assert src.client.subscribed == [ + "se/v1/print/d30-workshop/jobs", + "se/v1/print/d30-workshop/cmd", + ] + + +def test_a_failed_connect_publishes_nothing(source): + src = source(KNOWN) + src._on_connect(src.client, None, {}, 5) # not authorised + assert src.client.published == [] + + +def test_the_will_carries_last_known_truth(source): + """An ungraceful drop should say the printer is unreachable, not forget what it is.""" + src = source(KNOWN) + src.start() + + assert src.client.will is not None + will = json.loads(src.client.will[1]) + assert will["state"] == "disconnected" + assert will["serial"] == SERIAL + assert will["device_seen_at"] is not None + + +def test_the_will_is_re_armed_so_it_does_not_describe_startup_forever(source): + """The will is fixed for the life of a connection -- it rides in the CONNECT packet + and the broker holds it. paho rebuilds that packet on every reconnect, though, so + re-arming bounds how stale the will can be by the reconnect interval rather than by + the process lifetime.""" + src = source() + src.start() + assert json.loads(src.client.will[1])["serial"] is None # nothing known at boot + + _connect(src) + src.publish_status(KNOWN.to_status("d30-workshop", state="idle")) + _connect(src) # a later reconnect: this is where the newer will gets armed + + will = json.loads(src.client.will[1]) + assert will["state"] == "disconnected" + assert will["serial"] == SERIAL + assert will["device_seen_at"] is not None + + +def test_a_stale_will_is_corrected_by_the_restart_that_follows_it(tmp_path, monkeypatch): + """The will is frozen at connect, so a fault learned mid-session cannot reach it: if + the agent is killed after that fault, the broker publishes the older, healthier + reading over the newer one. Nothing can change what the broker holds for a live + session -- what stops it mattering is that the next process seeds from the spool, + which does have the fault, and republishes on connect. The unit is Restart=always + with RestartSec=5, so that correction is seconds behind the will.""" + import paho.mqtt.client as mqtt + + monkeypatch.setattr(mqtt, "Client", FakeClient) + config = make_config() + config.mqtt.host = "broker.invalid" + + spool = Spool(tmp_path / "spool.db") + spool.save_device_snapshot(KNOWN) + healthy = MqttSource(config, spool, lambda _j: None) + healthy.start() + assert json.loads(healthy.client.will[1])["media_ok"] is True # what a crash will say + + # Mid-session: a print finds the tape gone. The topic is correct; the will is not. + faulted = KNOWN.model_copy(update={"media_ok": False, "fault": "media not ready", "seen_at": 5000.0}) + spool.save_device_snapshot(faulted) + healthy.publish_status(faulted.to_status("d30-workshop", state="error")) + spool.close() + + # SIGKILL: the broker publishes the stale will, then systemd brings the agent back. + restarted = MqttSource(config, Spool(tmp_path / "spool.db"), lambda _j: None) + _connect(restarted) + + corrected = restarted.client.statuses[-1] + assert corrected["media_ok"] is False + assert corrected["error"] == "media not ready" + assert corrected["device_seen_at"] > json.loads(healthy.client.will[1])["device_seen_at"] + + +def test_shutdown_says_disconnected_without_forgetting(source): + src = source(KNOWN) + src.stop() + + final = src.client.statuses[-1] + assert final["state"] == "disconnected" + assert final["pending_labels"] == 0 + assert final["serial"] == SERIAL + + +def test_a_reconnect_mid_print_republishes_printing_not_idle(source): + """`state` is a fact about the link and the batch. Synthesising "idle" here would + tell a producer the strip it is waiting on had finished.""" + src = source(KNOWN) + src.publish_status(KNOWN.to_status("d30-workshop", state="printing", pending_labels=12)) + + _connect(src) + + republished = src.client.statuses[-1] + assert republished["state"] == "printing" + assert republished["pending_labels"] == 12 + + +def test_a_non_retained_status_does_not_become_what_the_topic_holds(source): + src = source(KNOWN) + src.publish_status(PrinterStatus(printer_id="d30-workshop", state="printing"), retain=False) + + _connect(src) + + assert src.client.statuses[-1]["serial"] == SERIAL + + +def test_a_flush_command_reaches_the_print_loop(tmp_path, monkeypatch): + """The cmd topic is handled on paho's thread, so it must only ever enqueue.""" + import paho.mqtt.client as mqtt + + monkeypatch.setattr(mqtt, "Client", FakeClient) + enqueued: list[str] = [] + config = make_config() + config.mqtt.host = "broker.invalid" + src = MqttSource(config, Spool(tmp_path / "spool.db"), enqueued.append) + + class _Msg: + topic = "se/v1/print/d30-workshop/cmd" + payload = b"flush" + qos = 1 + mid = 7 + + src._on_message(src.client, None, _Msg()) + assert enqueued == [FLUSH_COMMAND] + assert src.client.acked == [7] diff --git a/tests/test_spool.py b/tests/test_spool.py index 6f02d46..2b5961f 100644 --- a/tests/test_spool.py +++ b/tests/test_spool.py @@ -4,7 +4,7 @@ from conftest import Clock, make_job -from labelfab.agent import Outcome, Spool +from labelfab.agent import DeviceSnapshot, Outcome, Spool from labelfab.contract import JobResult, LabelResult @@ -45,6 +45,35 @@ def test_label_printed_survives_and_dedups(tmp_path): assert spool.printed_labels("j") == {0, 2} +def test_device_snapshot_survives_the_process(tmp_path): + """The printer only talks during a print, so this is the only way a restarted agent + knows anything about the hardware it drives.""" + path = tmp_path / "s.db" + snap = DeviceSnapshot(serial="Q223P4C31420105", firmware="2.1.2", media_ok=True, seen_at=1000.0) + + first = Spool(path) + assert first.device_snapshot() == DeviceSnapshot() # nothing learned yet + first.save_device_snapshot(snap) + first.close() + + assert Spool(path).device_snapshot() == snap + + +def test_the_latest_device_snapshot_replaces_the_last(tmp_path): + spool = Spool(tmp_path / "s.db") + spool.save_device_snapshot(DeviceSnapshot(battery_pct=100, seen_at=1000.0)) + spool.save_device_snapshot(DeviceSnapshot(battery_pct=88, seen_at=2000.0)) + assert spool.device_snapshot().battery_pct == 88 + + +def test_an_unreadable_device_snapshot_does_not_take_the_agent_down(tmp_path): + """A status field is cosmetic; refusing to boot over one would not be.""" + spool = Spool(tmp_path / "s.db") + spool.save_device_snapshot(DeviceSnapshot(serial="X")) + spool._db.execute("UPDATE device_state SET payload = 'not json' WHERE id = 1") + assert spool.device_snapshot() == DeviceSnapshot() + + def test_dedupe_recorded_and_purged(tmp_path): clock = Clock() spool = Spool(tmp_path / "s.db", clock=clock) diff --git a/tests/test_worker.py b/tests/test_worker.py index 2bb4047..8765035 100644 --- a/tests/test_worker.py +++ b/tests/test_worker.py @@ -228,6 +228,81 @@ def test_a_fault_clears_on_the_next_good_batch(harness): assert settled.media_ok is True +def test_status_says_when_the_printer_was_last_heard_from(harness, clock): + h = harness() + _reporting_factory(h) + h.submit(make_job("j", n_labels=1, flush=True)) + + seen = h.publisher.statuses[-1].device_seen_at + assert seen is not None and seen.timestamp() == clock.now + + +def test_device_truth_is_persisted_for_the_next_process(harness): + """Everything the printer says is said during a print. Not writing it down here is + what left the status topic full of nulls after a restart.""" + h = harness() + _reporting_factory(h) + h.submit(make_job("j", n_labels=1, flush=True)) + + stored = h.spool.device_snapshot() + assert stored.serial == "Q223P4C31420105" + assert stored.firmware == "2.1.2" + assert stored.media_ok is True + assert stored.seen_at is not None + + +def test_a_restarted_worker_does_not_start_blind(tmp_path, clock): + """A fresh process on the same spool publishes what the previous one learned — + including on the disconnected status it emits when the printer is asleep.""" + from conftest import WorkerHarness + + first = WorkerHarness(tmp_path, clock) + _reporting_factory(first) + first.submit(make_job("j", n_labels=1, flush=True)) + first.spool.close() + + restarted = WorkerHarness(tmp_path, clock) # same spool.db, new worker + restarted.offline = True # printer has since gone to sleep + restarted.submit(make_job("k", n_labels=1, flush=True)) + + status = restarted.publisher.statuses[-1] + assert status.state == "disconnected" + assert status.serial == "Q223P4C31420105" + assert status.media_ok is True + assert status.device_seen_at is not None + + +def test_the_startup_probe_surveys_a_printer_that_is_awake(harness, clock): + h = harness() + _reporting_factory(h) + + assert h.worker.probe_device() is True + + status = h.publisher.statuses[-1] + assert status.state == "idle" + assert status.serial == "Q223P4C31420105" + assert status.device_seen_at is not None and status.device_seen_at.timestamp() == clock.now + assert h.frames == 0 # surveyed, not printed + + +def test_a_sleeping_printer_leaves_the_stored_status_alone(tmp_path, clock): + """A slept D30 has its radio off and cannot be woken over Bluetooth, so the probe + is expected to miss. Missing must cost nothing beyond a connect timeout.""" + from conftest import WorkerHarness + + first = WorkerHarness(tmp_path, clock) + _reporting_factory(first) + first.submit(make_job("j", n_labels=1, flush=True)) + first.spool.close() + + restarted = WorkerHarness(tmp_path, clock) + restarted.offline = True + + assert restarted.worker.probe_device() is False + assert restarted.spool.device_snapshot().serial == "Q223P4C31420105" + assert restarted.publisher.statuses == [] # nothing learned, nothing to say + + def test_a_fault_is_captured_even_when_the_print_fails(harness): """The fault is usually *why* it failed, so closing without reading it loses it.""" h = harness()