diff --git a/README.md b/README.md index c9103ec..11057ce 100644 --- a/README.md +++ b/README.md @@ -33,11 +33,37 @@ InvenTree imports so it is unit-tested here: The status is read from the retained topic in three places, which between them keep the page honest without polling the printer itself: -| when | hook | -|---|---| -| machine initialisation | `init_machine` | -| a manual restart from the Admin Center | `restart_machine` | -| periodically | `ping_machines`, driven by InvenTree's `MACHINE_PING_ENABLED` (on by default) | +| when | hook | asks the printer? | +|---|---|---| +| machine initialisation | `init_machine` | no — reads the retained topic | +| **Restart Machine** in the Admin Center | `restart_machine` | **yes** — publishes `probe` on the `cmd` topic | +| periodically | `ping_machines`, driven by InvenTree's `MACHINE_PING_ENABLED` (on by default) | no | + +**"Restart Machine" is the refresh button.** InvenTree's machine framework has no generic +custom-action mechanism — `restart_machine` is the one hardcoded action, with its own +endpoint and a fixed place in the menu — so that item is the only lever a driver gets. +It now sends `probe` and the agent goes and looks at the printer, rather than re-reading +the retained topic the page was already showing. The `cmd` topic already existed for +`flush`, and the broker ACL already grants `inventree` write on it, so this needed no +new topic and no credential change. + +The automatic paths deliberately do *not* probe: neither startup nor the periodic ping +should dial a sleeping printer on a schedule nobody asked for. + +**Two schedules, and only one of them is yours.** `MACHINE_PING_ENABLED` ("Enable Machine +Ping", on by default, every 5 minutes) governs everything here — InvenTree's task +short-circuits before any driver is called, so switching it off stops `ping_machines` +entirely. It does *not* reach the print agent: `device.probe_interval_s` on the agent is +its own timer, and that is the one that contacts the printer. Turning off Machine Ping +stops this page **reading**; it does not stop the agent **asking**. Deliberate — the +retained topic is a shared bus and InvenTree is one consumer of it, so a setting here +silently reconfiguring the device agent would be the more surprising behaviour. + +Nothing is lost by leaving the ping on: it reads a retained message and never touches +Bluetooth. And there is little point setting `probe_interval_s` below 5 minutes, since +that is as often as this page re-reads. And a probe the printer +cannot answer falls back to remembered truth, so the page keeps the old reading and its +age rather than going blank. The periodic one is not a nicety. Machine status lives in the Django cache, so clearing that cache drops every `machine:*` key at once — and this deployment runs Redis without diff --git a/inventree_label_dispatch/dispatch.py b/inventree_label_dispatch/dispatch.py index 4e0c236..217fe7c 100644 --- a/inventree_label_dispatch/dispatch.py +++ b/inventree_label_dispatch/dispatch.py @@ -12,9 +12,12 @@ from __future__ import annotations import json +import time import uuid from dataclasses import dataclass +from .status import parse_seen_at + #: Mirrors labelfab's contract v1. Kept as a plain dict builder rather than importing #: labelfab so the InvenTree image does not pull in Pillow/qrcode a second time. CONTRACT_VERSION = 1 @@ -147,6 +150,111 @@ def _on_message(_c, _u, msg): client.disconnect() +def probe_status(printer_id: str, conn: Connection, wait_s: float = 8.0) -> dict | None: + """Ask the agent to look at the printer, and return the first *newer* reading. + + ``read_status`` returns what the agent *remembers*, which is the right default: + the D30 is only reachable while it is being printed to, so remembered truth is + almost always the best truth available and costs nothing. This is the other case -- + a human pressed refresh and wants the printer asked, not the memory read. + + "Newer" is load-bearing and means a ``device_seen_at`` later than the one already + retained, not merely the next message to arrive. The agent publishes on its own + events too, so without that test a print starting, a broker reconnect or somebody + else pressing the same button could be returned as the answer while the real one + was thrown away. + + Subscribes *before* publishing the command, because the answer is a retained + message and a subscription set up afterwards would race the agent. The retained + message already on the topic arrives immediately and is kept as the baseline, so a + probe the agent cannot fulfil (the printer is asleep, which is normal) degrades to + exactly what ``read_status`` would have returned rather than to nothing -- or to + something newer, if one of those other events arrived while we waited. + + ``wait_s`` covers the agent's round trip -- connect, session setup, telemetry, + publish -- which is a few seconds when the printer answers and one connect timeout + when it does not. + + It is not the whole bound, though, and anyone sizing a request timeout around this + should use ``2 * conn.timeout_s + wait_s``: reading the retained message and waiting + for the command's PUBACK can each cost ``conn.timeout_s`` first. Both are sub- + millisecond against a healthy broker, and a broker slow enough for them to matter + has already broken ``read_status`` on every other path. This blocks the request that + triggered it either way, so all three are bounded rather than generous. + """ + import queue + + received: queue.Queue = queue.Queue() + client = _new_client(conn, "probe") + client.on_message = lambda _c, _u, msg: received.put_nowait(msg.payload) + client.connect(conn.host, conn.port, conn.keepalive_s) + client.subscribe(conn.topic(printer_id, "status"), qos=1) + client.loop_start() + try: + latest = None + try: # the retained message: what we already knew, before asking + latest = json.loads(received.get(timeout=conn.timeout_s)) + except (queue.Empty, ValueError): + pass + + # Everything queued before the command goes out is context, not answer: the + # retained message plus anything the agent happened to publish while we were + # connecting. Folding it into the baseline is what makes "newer than what we + # knew" a property of this function rather than a bet on how the agent emits. + # Otherwise, when nothing has ever been observed and the baseline is None, the + # filter weakens to "the first message carrying any timestamp at all", and that + # is only safe because a pre-observation agent publishes null everywhere -- an + # invariant that lives in the other repo and could be relaxed there without + # anyone here noticing. + # + # Drained before publishing rather than after, so a quick answer cannot be + # swallowed into the baseline it is supposed to beat. + while True: + try: + latest = json.loads(received.get_nowait()) + except queue.Empty: + break + except ValueError: + continue + + info = client.publish(conn.topic(printer_id, "cmd"), "probe", qos=1) + info.wait_for_publish(conn.timeout_s) + + # The answer is the first status carrying an observation *newer* than the one we + # already had -- not simply the first status to arrive. The agent publishes on + # its own events too (a print starting and finishing, a broker reconnect, + # somebody else pressing this button), and any of those landing first would + # otherwise be mistaken for the answer while the real one was discarded. The + # symptom is mild and self-correcting, which is exactly why it would have been + # hard to spot: `state=printing` on a page you just refreshed looks plausible. + # + # A status published by a print that *did* reach the printer also carries a + # newer observation, and returning that is right -- it is the same freshness a + # probe would have produced, obtained the same way. + baseline = parse_seen_at((latest or {}).get("device_seen_at")) + deadline = time.monotonic() + wait_s + while time.monotonic() < deadline: + try: + payload = received.get(timeout=max(0.1, deadline - time.monotonic())) + except queue.Empty: + break + try: + message = json.loads(payload) + except ValueError: + continue + seen = parse_seen_at(message.get("device_seen_at")) + if seen is not None and (baseline is None or seen > baseline): + return message + # Not the answer, but still more current than what we started with: keep it + # as the fallback, so a probe that goes unanswered reports the newest state + # rather than the one that happened to be retained when we subscribed. + latest = message + return latest + finally: + client.loop_stop() + client.disconnect() + + def read_status(printer_id: str, conn: Connection) -> dict | None: """One-shot read of the retained status topic. ``None`` if nothing is retained. diff --git a/inventree_label_dispatch/driver.py b/inventree_label_dispatch/driver.py index 00305f5..ea99504 100644 --- a/inventree_label_dispatch/driver.py +++ b/inventree_label_dispatch/driver.py @@ -30,7 +30,14 @@ class only reads settings, turns items into label specs, and publishes. from rest_framework import serializers from .config import connection_from_env -from .dispatch import build_job, build_label, publish, publish_awaiting_result, read_status +from .dispatch import ( + build_job, + build_label, + probe_status, + publish, + publish_awaiting_result, + read_status, +) from .extractors import extract from .status import classify_result, classify_status @@ -111,15 +118,27 @@ def _apply(self, machine, member: str, text: str) -> None: machine.set_status_text(text) def restart_machine(self, machine) -> None: - """Re-read the printer's state after a manual restart. + """Ask the printer what it is, and show the answer. The page's refresh button. A restart clears the machine's shared state, and without this nothing puts it back: the page then shows UNKNOWN with an empty status text -- the defaults `get_shared_state` returns for missing keys -- while the machine itself reports `initialized`, no errors and a driver available. An empty status text is diagnostic on its own, since no path in this driver produces one. + + This is also the only lever there is. InvenTree's machine framework has no + generic custom-action mechanism -- `restart_machine` is the one hardcoded + action, with its own endpoint and a fixed place in the Machines menu -- so + "Restart Machine" *is* the refresh button whether or not it is called that. + Making it re-read the retained topic and nothing else wasted it: the topic + holds what the agent last heard, which is the same thing the page was already + showing. Now it sends `probe` and the agent goes and looks. + + Falls back to remembered truth when the printer does not answer, which is the + normal case for a D30 that has powered itself down: the reading keeps its old + `device_seen_at` and the page says how old it is, instead of going blank. """ - self._refresh_status(machine) + self._refresh_status(machine, probe=True) def ping_machines(self) -> None: """Periodic refresh, driven by InvenTree's MACHINE_PING_ENABLED (default on). @@ -141,10 +160,17 @@ def ping_machines(self) -> None: except Exception as exc: # noqa: BLE001 - one machine must not break the loop machine.handle_error(exc) - def _refresh_status(self, machine) -> None: + def _refresh_status(self, machine, *, probe: bool = False) -> None: + """Read the retained status, or ask the printer for a fresh one. + + ``probe`` is off for the automatic paths -- init and the periodic ping -- so + neither of them dials a sleeping printer on a schedule nobody asked for. It is + on only where a human pressed the button. + """ printer_id = machine.get_setting("PRINTER_ID", "D") try: - status = read_status(printer_id, connection_from_env()) + conn = connection_from_env() + status = probe_status(printer_id, conn) if probe else read_status(printer_id, conn) except Exception as exc: machine.set_status(LabelPrinterStatus.UNKNOWN) machine.set_status_text(_("could not reach the broker: %(err)s") % {"err": str(exc)[:80]}) diff --git a/inventree_label_dispatch/status.py b/inventree_label_dispatch/status.py index e3d6510..826538e 100644 --- a/inventree_label_dispatch/status.py +++ b/inventree_label_dispatch/status.py @@ -29,31 +29,45 @@ _GOOD_JOB_STATES = frozenset({"completed"}) -def _ago(seen: object, now: datetime) -> str: - """How long ago the device fields were true, as a short suffix. ``""`` if unknown. +def parse_seen_at(seen: object) -> datetime | None: + """``device_seen_at`` as an instant, or ``None`` if it is missing or unusable. ``seen`` is typed as ``object`` rather than ``str | None`` on purpose: it comes straight out of ``json.loads`` on a payload this process did not produce, so it can be any JSON type at all, and pretending otherwise would just move the lie into the annotation. - Empty rather than a guess in three cases, all of which mean the same thing: nobody - said. An agent older than the field simply omits it, and the page should look - exactly as it did before rather than sprouting a fake age. + ``None`` in three cases, all of which mean the same thing: nobody said. An agent + older than the field simply omits it. Nothing in here may raise. It runs inside a Django worker on a payload from another process, so a surprise is a 500 on the Machines page, not a test failure -- which is also why the ``Z`` is handled by hand: pydantic emits it and ``fromisoformat`` did not accept it until Python 3.11, while this package supports 3.9. + + Shared with :mod:`dispatch`, which compares two of these rather than rendering one. + Comparing the parsed instants and not the raw strings matters: they sort the same + way only while every producer emits whole seconds, and ``...:42.5Z`` sorts *before* + ``...:42Z`` because ``.`` precedes ``Z``. """ if not seen: - return "" + return None try: stamp = datetime.fromisoformat(str(seen).replace("Z", "+00:00")) except (TypeError, ValueError): + return None + return stamp if stamp.tzinfo is not None else stamp.replace(tzinfo=timezone.utc) + + +def _ago(seen: object, now: datetime) -> str: + """How long ago the device fields were true, as a short suffix. ``""`` if unknown. + + Empty rather than a guess when the agent did not say, so a page served by an agent + older than the field looks exactly as it did before rather than sprouting a fake age. + """ + stamp = parse_seen_at(seen) + if stamp is None: return "" - if stamp.tzinfo is None: - stamp = stamp.replace(tzinfo=timezone.utc) # Clamped at zero: the agent's clock and this one are not the same clock, and a # small skew must read as "just now", never as a negative age. delta = max(0, int((now - stamp).total_seconds())) diff --git a/tests/test_dispatch.py b/tests/test_dispatch.py index 1cbfc28..b6da0e7 100644 --- a/tests/test_dispatch.py +++ b/tests/test_dispatch.py @@ -155,3 +155,206 @@ def publish_with_garbage(topic, payload, qos=0): fake.publish = publish_with_garbage result = dispatch.publish_awaiting_result(job, Connection(host="broker"), timeout_s=1.0) assert result["state"] == "completed" + + +# -- refresh: ask the printer, not the memory ------------------------------- # +# +# read_status returns what the agent remembers, which is the right default. probe_status +# is the other case: a human pressed refresh and wants the printer actually asked. + + +class _ProbeBroker(_FakeClient): + """Retains one status, and optionally publishes more when probed. + + ``fresh`` may be a single status or a list, so a test can reproduce the agent + publishing on its own events -- a print starting, a reconnect -- while a probe is + still in flight. + """ + + def __init__(self, retained, fresh=None): + super().__init__() + self.retained = retained + self.fresh = [] if fresh is None else (fresh if isinstance(fresh, list) else [fresh]) + self.subscribed = [] + self.on_message = None + self.order = [] + + def subscribe(self, topic, qos=0): + self.subscribed.append(topic) + self.order.append("subscribe") + if self.retained is not None: # retained messages arrive on subscribe + self._deliver(self.retained) + + def publish(self, topic, payload, qos=0): + self.order.append("publish") + info = super().publish(topic, payload, qos) + for message in self.fresh: + self._deliver(message) + return info + + def _deliver(self, obj): + payload = obj if isinstance(obj, bytes) else json.dumps(obj).encode() + self.on_message(self, None, type("Msg", (), {"payload": payload})()) + + +def test_probe_asks_the_agent_and_returns_the_fresh_answer(monkeypatch): + fake = _ProbeBroker( + retained={"state": "idle", "media_ok": True, "device_seen_at": "2026-08-01T00:00:00Z"}, + fresh={"state": "idle", "media_ok": True, "device_seen_at": "2026-08-02T12:00:00Z"}, + ) + monkeypatch.setattr(dispatch, "_new_client", lambda conn, suffix: fake) + + got = dispatch.probe_status("d30-workshop", Connection(host="broker"), wait_s=1) + + assert got["device_seen_at"] == "2026-08-02T12:00:00Z" + topic, payload, qos = fake.published[0] + assert topic == "se/v1/print/d30-workshop/cmd" + assert payload == "probe" + assert qos == 1 + + +def test_probe_subscribes_before_asking(monkeypatch): + """The answer is a retained publish; subscribing afterwards races the agent.""" + fake = _ProbeBroker(retained={"state": "idle"}, fresh={"state": "idle"}) + monkeypatch.setattr(dispatch, "_new_client", lambda conn, suffix: fake) + + dispatch.probe_status("d30-workshop", Connection(host="broker"), wait_s=1) + + assert fake.order[:2] == ["subscribe", "publish"] + + +def test_a_printer_that_does_not_answer_falls_back_to_what_we_knew(monkeypatch): + """The normal case for a D30 that has powered itself down. The reading keeps its + old timestamp and the page says how old it is, rather than going blank.""" + fake = _ProbeBroker( + retained={"state": "idle", "media_ok": True, "device_seen_at": "2026-08-01T00:00:00Z"}, + fresh=None, # agent probed, printer was asleep, so it published nothing + ) + monkeypatch.setattr(dispatch, "_new_client", lambda conn, suffix: fake) + + got = dispatch.probe_status("d30-workshop", Connection(host="broker"), wait_s=0.3) + + assert got["device_seen_at"] == "2026-08-01T00:00:00Z" + + +def test_probe_with_nothing_retained_and_no_answer_is_none(monkeypatch): + fake = _ProbeBroker(retained=None, fresh=None) + monkeypatch.setattr(dispatch, "_new_client", lambda conn, suffix: fake) + conn = Connection(host="broker", timeout_s=0.3) + + assert dispatch.probe_status("d30-workshop", conn, wait_s=0.3) is None + + +def test_probe_survives_a_malformed_retained_payload(monkeypatch): + fake = _ProbeBroker(retained=b"{not json", fresh={"state": "idle", "media_ok": True}) + monkeypatch.setattr(dispatch, "_new_client", lambda conn, suffix: fake) + conn = Connection(host="broker", timeout_s=0.3) + + got = dispatch.probe_status("d30-workshop", conn, wait_s=1) + assert got == {"state": "idle", "media_ok": True} + + +def test_a_print_starting_mid_probe_does_not_win(monkeypatch): + """The agent publishes on its own events -- a print starting, a broker reconnect, + another click of the same button. Any of those landing first would otherwise be + returned while the real answer was discarded, and `state=printing` on a page you + just refreshed looks plausible enough that nobody would report it.""" + old = "2026-08-01T00:00:00Z" + fake = _ProbeBroker( + retained={"state": "idle", "media_ok": True, "device_seen_at": old}, + fresh=[ + # the in-flight print announcing itself: same observation, new state + {"state": "printing", "pending_labels": 3, "device_seen_at": old}, + # the probe's actual answer + {"state": "idle", "media_ok": True, "device_seen_at": "2026-08-02T12:00:00Z"}, + ], + ) + monkeypatch.setattr(dispatch, "_new_client", lambda conn, suffix: fake) + + got = dispatch.probe_status("d30-workshop", Connection(host="broker"), wait_s=1) + + assert got["state"] == "idle" + assert got["device_seen_at"] == "2026-08-02T12:00:00Z" + + +def test_a_print_that_reached_the_printer_is_a_valid_answer(monkeypatch): + """It carries a newer observation obtained the same way a probe would have.""" + fake = _ProbeBroker( + retained={"state": "idle", "media_ok": True, "device_seen_at": "2026-08-01T00:00:00Z"}, + fresh=[{"state": "idle", "media_ok": False, "device_seen_at": "2026-08-02T09:00:00Z"}], + ) + monkeypatch.setattr(dispatch, "_new_client", lambda conn, suffix: fake) + + got = dispatch.probe_status("d30-workshop", Connection(host="broker"), wait_s=1) + assert got["media_ok"] is False + + +def test_an_unanswered_probe_reports_the_newest_state_not_the_stale_one(monkeypatch): + """If the printer never answers but the agent said something else meanwhile, that + is still more current than what happened to be retained when we subscribed.""" + old = "2026-08-01T00:00:00Z" + fake = _ProbeBroker( + retained={"state": "idle", "device_seen_at": old}, + fresh=[{"state": "printing", "pending_labels": 2, "device_seen_at": old}], + ) + monkeypatch.setattr(dispatch, "_new_client", lambda conn, suffix: fake) + + got = dispatch.probe_status("d30-workshop", Connection(host="broker"), wait_s=0.3) + assert got["state"] == "printing" + + +def test_an_identical_republish_is_not_an_answer(monkeypatch): + """A broker reconnect republishes the same retained status verbatim.""" + same = {"state": "idle", "media_ok": True, "device_seen_at": "2026-08-01T00:00:00Z"} + fake = _ProbeBroker(retained=same, fresh=[dict(same)]) + monkeypatch.setattr(dispatch, "_new_client", lambda conn, suffix: fake) + + got = dispatch.probe_status("d30-workshop", Connection(host="broker"), wait_s=0.3) + assert got == same # fell through to the fallback rather than claiming a fresh look + + +def test_sub_second_stamps_compare_chronologically_not_lexically(monkeypatch): + """'...:42.5Z' sorts before '...:42Z' as a string, because '.' precedes 'Z'. Parsing + is what keeps a newer reading from being discarded as older.""" + fake = _ProbeBroker( + retained={"state": "idle", "device_seen_at": "2026-08-02T12:00:42Z"}, + fresh=[{"state": "idle", "media_ok": True, "device_seen_at": "2026-08-02T12:00:42.5Z"}], + ) + monkeypatch.setattr(dispatch, "_new_client", lambda conn, suffix: fake) + + got = dispatch.probe_status("d30-workshop", Connection(host="broker"), wait_s=1) + assert got["media_ok"] is True + + +def test_a_status_published_before_the_command_is_context_not_answer(monkeypatch): + """Whatever the agent published while we were connecting is part of what we already + knew. Folding it into the baseline is what stops `probe_status` depending on the + agent's emission contract: without it, a never-observed printer weakens the filter + to "the first message carrying any timestamp at all".""" + + class _EarlyPublisher(_ProbeBroker): + def subscribe(self, topic, qos=0): + super().subscribe(topic, qos) + # arrives after the retained one but before we send the command + self._deliver({"state": "idle", "media_ok": True, "device_seen_at": "2026-08-02T10:00:00Z"}) + + fake = _EarlyPublisher( + retained={"state": "idle", "device_seen_at": None}, + fresh=[{"state": "idle", "media_ok": False, "device_seen_at": "2026-08-02T12:00:00Z"}], + ) + monkeypatch.setattr(dispatch, "_new_client", lambda conn, suffix: fake) + + got = dispatch.probe_status("d30-workshop", Connection(host="broker"), wait_s=1) + + # the 10:00 reading was already on the wire before we asked, so the answer is 12:00 + assert got["device_seen_at"] == "2026-08-02T12:00:00Z" + assert got["media_ok"] is False + + +def test_a_never_observed_printer_still_falls_back_rather_than_inventing(monkeypatch): + """Nothing retained, nothing observed, printer asleep: there is no answer to give.""" + fake = _ProbeBroker(retained={"state": "idle", "device_seen_at": None}, fresh=None) + monkeypatch.setattr(dispatch, "_new_client", lambda conn, suffix: fake) + + got = dispatch.probe_status("d30-workshop", Connection(host="broker"), wait_s=0.3) + assert got == {"state": "idle", "device_seen_at": None}