Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 15 additions & 3 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -44,10 +44,22 @@ that cache drops every `machine:*` key at once — and this deployment runs Redi
persistence, so a restart does exactly that. With only `init_machine` the page then sat
at `UNKNOWN` with an empty status text until someone printed or the pods restarted.

Two deliberate distinctions. **Unreported media is not healthy media**: `media_ok` is a
Three deliberate distinctions. **Unreported media is not healthy media**: `media_ok` is a
tri-state, and a printer that has said nothing shows "media unreported" rather than
"media ok". And **battery percentage alone is not enough** — it pins at 100% whenever
the unit is on charge, so the voltage is shown beside it.
"media ok". **Battery percentage alone is not enough** — it pins at 100% whenever
the unit is on charge, so the voltage is shown beside it. And **a memory is not an
observation**: the D30 is only reachable while it is being printed to, so the agent
republishes what it last heard rather than going quiet, and the age of that reading is
shown alongside it.

```
CONNECTED fw 2.1.2 · 100% (4.17V) · media ok · Q223P4C31420105 · seen 6h ago
```

The age comes from `device_seen_at` on the retained status. It annotates the reading
rather than overruling it — a printer that had tape six hours ago is far likelier to
still have tape than to have anything else — so the status mapping is unchanged. An
agent too old to publish the field renders exactly as it did before.

## Print outcomes

Expand Down
52 changes: 48 additions & 4 deletions inventree_label_dispatch/status.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@
this package's test environment -- so the decisions worth testing live here and return
a ``LabelPrinterStatus`` *member name* for the driver to look up.

Two rules the mapping is built around:
Three rules the mapping is built around:

* **Silence is not health.** ``media_ok`` is ``None`` when the printer has not said
anything about its media, which is a different state from "media is fine". Reporting
Expand All @@ -14,15 +14,54 @@
* **A media fault is not a generic error.** InvenTree has NO_MEDIA (301) for exactly
this, and it renders differently from ERROR (500). Using the specific code is the
difference between "go put tape in it" and "something is wrong, go investigate".
* **Memory is not observation.** The D30 is only reachable while it is being printed
to, so the agent remembers what it last heard and republishes it. Nearly everything
on this page is therefore last-known rather than live, and rendering it without its
age is the first rule again one layer along -- "media ok" from three days ago reads
exactly like "media ok" from a minute ago.
"""

from __future__ import annotations

from datetime import datetime, timezone

#: Terminal job states that mean tape came out as intended.
_GOOD_JOB_STATES = frozenset({"completed"})


def _describe(status: dict) -> str:
def _ago(seen, now: datetime) -> str:
"""How long ago the device fields were true, as a short suffix. ``""`` if unknown.

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.

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.
"""
if not seen:
return ""
try:
stamp = datetime.fromisoformat(str(seen).replace("Z", "+00:00"))
except (TypeError, ValueError):
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()))
if delta < 60:
return f"seen {delta}s ago"
if delta < 3600:
return f"seen {delta // 60}m ago"
if delta < 86400:
return f"seen {delta // 3600}h ago"
return f"seen {delta // 86400}d ago"


def _describe(status: dict, now: datetime) -> str:
"""A one-line summary of the device truth on the retained status topic."""
bits: list[str] = []
if firmware := status.get("firmware"):
Expand All @@ -42,20 +81,25 @@ def _describe(status: dict) -> str:

if serial := status.get("serial"):
bits.append(serial)
# Last, because it qualifies everything before it rather than being another reading.
if age := _ago(status.get("device_seen_at"), now):
bits.append(age)
return " · ".join(bits)


def classify_status(status: dict | None) -> tuple[str, str]:
def classify_status(status: dict | None, *, now: datetime | None = None) -> tuple[str, str]:
"""Map a retained ``PrinterStatus`` to ``(LabelPrinterStatus member, text)``.

``status`` is ``None`` when nothing is retained on the topic -- the agent has never
run, or its status was cleared. That is genuinely unknown, not disconnected.

``now`` is injectable so the age rendering is testable without freezing wall time.
"""
if status is None:
return "UNKNOWN", "no retained status; the print agent has not published yet"

state = status.get("state")
detail = _describe(status)
detail = _describe(status, now or datetime.now(timezone.utc))

if state == "disconnected":
# The agent's MQTT will, or its own shutdown notice. The broker is reachable;
Expand Down
70 changes: 70 additions & 0 deletions tests/test_status.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,8 @@

from __future__ import annotations

from datetime import datetime, timedelta, timezone

import pytest

from inventree_label_dispatch.status import classify_result, classify_status
Expand Down Expand Up @@ -121,6 +123,74 @@ def test_missing_optional_fields_do_not_break_the_description():
assert "media unreported" in text


# -- how old the device truth is -------------------------------------------- #
#
# The printer is only reachable while it is being printed to, so the agent republishes
# what it last heard rather than going quiet. That is the right trade, but it means
# "media ok" on this page is usually a memory, and a memory rendered without its age is
# indistinguishable from an observation.

NOW = datetime(2026, 8, 2, 12, 0, 0, tzinfo=timezone.utc)


def _seen(delta: timedelta) -> dict:
return _healthy(device_seen_at=(NOW - delta).isoformat().replace("+00:00", "Z"))


@pytest.mark.parametrize(
("delta", "expected"),
[
(timedelta(seconds=0), "seen 0s ago"),
(timedelta(seconds=42), "seen 42s ago"),
(timedelta(minutes=4), "seen 4m ago"),
(timedelta(hours=6), "seen 6h ago"),
(timedelta(days=3), "seen 3d ago"),
],
)
def test_the_age_of_the_device_truth_is_shown(delta, expected):
_member, text = classify_status(_seen(delta), now=NOW)
assert expected in text


def test_stale_truth_is_still_the_best_truth_available():
"""Age annotates the reading; it does not overrule it. A printer that had tape three
days ago is far more likely to still have tape than to have anything else."""
member, text = classify_status(_seen(timedelta(days=3)), now=NOW)
assert member == "CONNECTED"
assert "media ok" in text
assert "seen 3d ago" in text


def test_an_agent_that_does_not_publish_the_field_renders_as_before():
"""Back-compat with an un-upgraded agent: no field, no invented age."""
_member, text = classify_status(_healthy(), now=NOW)
assert "seen" not in text
assert "fw 2.1.2" in text


@pytest.mark.parametrize("value", [None, "", "not a timestamp", 12345, "2026-13-45T99:99:99Z"])
def test_an_unusable_timestamp_is_dropped_not_raised(value):
"""This runs in a Django worker on a payload from another process, so a surprise
here is a 500 on the Machines page rather than a red test."""
_member, text = classify_status(_healthy(device_seen_at=value), now=NOW)
assert "seen" not in text


def test_a_naive_timestamp_is_read_as_utc():
"""Everything the agent publishes is UTC; a missing offset is not a reason to guess
local time, which would silently shift the age by hours."""
_member, text = classify_status(
_healthy(device_seen_at="2026-08-02T11:00:00"), now=NOW
)
assert "seen 1h ago" in text


def test_a_clock_ahead_of_ours_reads_as_just_now_not_a_negative_age():
"""The agent's clock and InvenTree's are not the same clock."""
_member, text = classify_status(_seen(timedelta(minutes=-5)), now=NOW)
assert "seen 0s ago" in text


# -- job outcomes ----------------------------------------------------------- #


Expand Down
Loading