public_git/bambuddy
1
0
Fork 0
mirror of https://github.com/maziggy/bambuddy.git synced 2026-09-30 03:01:21 +02:00
Code Issues Packages Projects Releases Wiki Activity

fix(spoolman): split mid-print usage across AMS backup switch (#1793)

Browse Source 
usage_tracker's tray-switch split has never had a Spoolman peer.
    An AMS same-material runout switch mid-print charged the whole slot
    to the origin spool via the (via tag) path and double-credited the
    backup via remain-delta — origin exceeded initial_weight.

    Extract the segment-math into utils/tray_split.compute_tray_split_grams
    and call it from both writers so the two inventory backends attribute
    mid-print switches identically. spoolman_tracking gains
    _report_spool_usage_split_by_tray_changes; the Path 2 remain-delta
    fallback now skips trays the split path covered, killing the
    double-count.
This commit is contained in:
maziggy
2026-07-07 10:58:24 +02:00
parent b8b5aaa977 9f4f16e5bd
commit 484ea9c2a4
6 changed files with 871 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+2
View File
File diff suppressed because one or more lines are too long
backend/app/services/spoolman_tracking.py
+228 -17
View File
@@ -569,6 +569,135 @@ async def _report_spool_usage_for_slots(
return spools_updated
async def _report_spool_usage_split_by_tray_changes(
client,
filament_usage: list[dict],
tray_changes: list[tuple[int, int]],
ams_trays: dict[int, dict],
layer_usage: dict[int, dict[int, float]] | None,
filament_properties: dict | None,
total_layers: int,
last_layer_num: int,
method_label: str,
printer_serial: str,
printer_id: int,
slot_colors_out: dict[int, str] | None = None,
) -> tuple[int, set[int]]:
"""Split each slot's grams across ``tray_changes`` and charge per-segment.
Mirrors ``usage_tracker`` Path 1's tray-switch branch so Spoolman and
the internal Spool inventory attribute mid-print AMS-backup switches
identically (#1793 — reporter's origin spool was over-charged the
whole print because this path didn't exist). ``compute_tray_split_grams``
holds the shared segment-math; this function wraps the per-segment
spool resolution + ``use_spool`` sink for the Spoolman side.
Returns ``(spools_updated, handled_global_tray_ids)`` — the caller
passes ``handled_global_tray_ids`` into the remain-delta fallback so
a tray attributed here is not double-charged there.
"""
from backend.app.utils.tray_split import compute_tray_split_grams
spools_updated = 0
handled_global_tray_ids: set[int] = set()
for usage in filament_usage:
slot_id = usage.get("slot_id", 0)
total_weight = usage.get("used_g", 0)
if total_weight <= 0 or slot_id <= 0:
continue
props = (filament_properties or {}).get(str(slot_id)) or (filament_properties or {}).get(slot_id) or {}
segments = compute_tray_split_grams(
tray_changes=tray_changes,
total_weight=float(total_weight),
slot_id=slot_id,
layer_usage=layer_usage,
density=float(props.get("density", 1.24)),
diameter=float(props.get("diameter", 1.75)),
total_layers=total_layers,
last_layer_num=last_layer_num,
)
for seg_idx, tray_global, segment_grams in segments:
if segment_grams <= 0:
continue
# Mark this tray as handled BEFORE the resolution attempt so
# remain-delta doesn't double-charge it, even if we fail to
# find a spool below. Matches usage_tracker behaviour: the
# tray was physically fed from during this print, whether or
# not Spoolman happens to have a matching row.
handled_global_tray_ids.add(tray_global)
tray_info = ams_trays.get(tray_global) or {}
spool_id_to_use: int | None = None
resolution_path = ""
spool_color_hex: str | None = None
spool_tag = _resolve_spool_tag(tray_info, printer_serial, tray_global) if tray_info else ""
if spool_tag:
spool = await client.find_spool_by_tag(spool_tag)
if spool:
spool_id_to_use = spool["id"]
resolution_path = "tag"
spool_color_hex = (spool.get("filament") or {}).get("color_hex")
if spool_id_to_use is None:
seg_ams_id, seg_tray_id = _global_tray_id_to_ams_slot(tray_global)
spool_id_to_use = await _resolve_spool_id_via_slot_assignment(printer_id, seg_ams_id, seg_tray_id)
if spool_id_to_use is not None:
resolution_path = "slot-assignment"
if spool_id_to_use is None:
logger.info(
"[SPOOLMAN] Split slot %s seg %s tray=%d: no spool resolved — %.2fg lost from split accounting",
slot_id,
seg_idx,
tray_global,
segment_grams,
)
continue
# Colour rewrite (#1494) — first segment for a slot wins. The
# UI displays a single colour per slot, so later segments on the
# same slot don't overwrite (a backup swap can be a different
# colour but the archive card stays consistent with the origin).
if slot_colors_out is not None and slot_id not in slot_colors_out:
if spool_color_hex is None:
try:
full_spool = await client.get_spool(spool_id_to_use)
spool_color_hex = (full_spool.get("filament") or {}).get("color_hex")
except Exception as exc: # noqa: BLE001 — colour is non-critical
logger.debug("[SPOOLMAN] Split slot %s: could not fetch spool colour: %s", slot_id, exc)
if spool_color_hex:
slot_colors_out[slot_id] = spool_color_hex
try:
await client.use_spool(spool_id_to_use, round(segment_grams, 2))
logger.info(
"[SPOOLMAN] %s: slot %s seg %s tray=%d: %.2fg -> spool %s (via %s)",
method_label,
slot_id,
seg_idx,
tray_global,
segment_grams,
spool_id_to_use,
resolution_path,
)
spools_updated += 1
except (SpoolmanNotFoundError, SpoolmanClientError, SpoolmanUnavailableError) as exc:
logger.warning(
"[SPOOLMAN] Split slot %s seg %s: failed to record usage for spool %s: %s",
slot_id,
seg_idx,
spool_id_to_use,
exc,
)
return spools_updated, handled_global_tray_ids
async def _report_partial_usage(
printer_id: int,
tracking,
@@ -789,6 +918,13 @@ async def report_usage(printer_id: int, archive_id: int):
ams_trays = {int(k): v for k, v in (tracking.ams_trays or {}).items()}
slot_to_tray = tracking.slot_to_tray
tray_remain_start = tracking.tray_remain_start or {}
# ``layer_usage`` and ``filament_properties`` were added later than
# the base tracking fields; use ``getattr`` so tests that stub
# ``tracking`` as a lightweight SimpleNamespace stay valid, and
# historic ORM rows loaded without these columns can't AttributeError
# on read.
layer_usage_raw = getattr(tracking, "layer_usage", None) or {}
filament_properties = getattr(tracking, "filament_properties", None) or {}
printer_serial = await _get_printer_serial(printer_id)
# Delete tracking row (we're done with it)
@@ -809,29 +945,104 @@ async def report_usage(printer_id: int, archive_id: int):
logger.warning("[SPOOLMAN] Not reachable for usage reporting")
return
# Consult the live printer state for the tray-change log written by
# ``bambu_mqtt.py`` on every mid-print ``tray_now`` change (#957).
# When there's more than one entry, the print traversed >1 AMS tray
# and the split path attributes each segment to the tray that was
# loaded at the time — matches the internal Spool inventory writer
# in ``usage_tracker.py``. Without this, an AMS-backup runout switch
# charges the whole slot to the origin spool and pushes it past
# ``initial_weight`` (#1793).
#
# Split only for SINGLE-slot prints — same gate as
# ``usage_tracker.py:1002``. Multi-slot (multi-colour) prints
# naturally cycle trays for every colour change, so splitting each
# slot's grams across ALL tray_change_log entries would attribute
# slot 1's grams to the segments where slot 2's tray was loaded and
# vice versa. Multi-slot prints fall through to the existing
# single-tray path (which uses the stable ``slot_to_tray`` mapping).
nonzero_slots = [u for u in filament_usage if u.get("used_g", 0) > 0]
tray_changes: list[tuple[int, int]] = []
_state = None
if len(nonzero_slots) == 1:
from backend.app.services.printer_manager import printer_manager as _pm
_state = _pm.get_status(printer_id)
if _state is not None:
tray_changes = list(getattr(_state, "tray_change_log", []) or [])
_total_layers = int(getattr(_state, "total_layers", 0) or 0) if _state else 0
_current_layer = int(getattr(_state, "layer_num", 0) or 0) if _state else 0
# For the linear-fallback denominator when total_layers is 0 (P1S
# firmware resets it at print end). At completion the current layer
# is the print's last valid layer.
_layer_denom_hint = _total_layers or _current_layer
slot_colors: dict[int, str] = {}
handled_global_tray_ids: set[int] = set()
spools_updated = 0
# --- Path 1: 3MF per-slot estimates -----------------------------
if filament_usage:
logger.info("[SPOOLMAN] Reporting per-filament usage for archive %s", archive_id)
usage_items = [(u.get("slot_id", 0), u.get("used_g", 0)) for u in filament_usage]
spools_updated = await _report_spool_usage_for_slots(
client,
usage_items,
ams_trays,
slot_to_tray,
f"Archive {archive_id}",
printer_serial,
printer_id=printer_id,
slot_colors_out=slot_colors,
)
# Track which physical slots the 3MF path already covered so
# Path 2 doesn't double-charge them.
for u in filament_usage:
slot_id = u.get("slot_id", 0)
handled_global_tray_ids.add(_resolve_global_tray_id(slot_id, slot_to_tray, ams_trays))
if len(tray_changes) > 1:
# Tray-split path — attribute per-segment to the tray that
# was loaded at that time.
logger.info(
"[SPOOLMAN] Reporting per-filament usage for archive %s with tray-split "
"(tray_change_log=%s, denom_layers=%d)",
archive_id,
tray_changes,
_layer_denom_hint,
)
# ``tracking.layer_usage`` was serialized to JSON so int keys
# come back as strings. Restore them for the split math.
layer_usage = None
if layer_usage_raw:
try:
layer_usage = {
int(layer): {int(fid): mm for fid, mm in filaments.items()}
for layer, filaments in layer_usage_raw.items()
}
except (TypeError, ValueError, AttributeError):
# AttributeError catches ``inner.items()`` when the
# inner value isn't dict-shaped (corrupt JSON row).
# Missing gcode falls through to the linear-ratio
# branch inside ``compute_tray_split_grams`` — still
# gives a correct split, just less precise.
layer_usage = None
split_updated, split_handled = await _report_spool_usage_split_by_tray_changes(
client,
filament_usage,
tray_changes,
ams_trays,
layer_usage,
filament_properties,
_total_layers,
_layer_denom_hint,
f"Archive {archive_id}",
printer_serial,
printer_id=printer_id,
slot_colors_out=slot_colors,
)
spools_updated += split_updated
handled_global_tray_ids |= split_handled
else:
logger.info("[SPOOLMAN] Reporting per-filament usage for archive %s", archive_id)
usage_items = [(u.get("slot_id", 0), u.get("used_g", 0)) for u in filament_usage]
spools_updated = await _report_spool_usage_for_slots(
client,
usage_items,
ams_trays,
slot_to_tray,
f"Archive {archive_id}",
printer_serial,
printer_id=printer_id,
slot_colors_out=slot_colors,
)
# Track which physical slots the 3MF path already covered so
# Path 2 doesn't double-charge them.
for u in filament_usage:
slot_id = u.get("slot_id", 0)
handled_global_tray_ids.add(_resolve_global_tray_id(slot_id, slot_to_tray, ams_trays))
# --- Path 2: AMS remain%-delta for slots 3MF didn't cover -------
# Triggered for no-3MF "Untitled" prints (#1820) AND for partial
backend/app/services/usage_tracker.py
+18 -40
View File
@@ -1083,6 +1083,10 @@ async def _track_from_3mf(
continue
# --- Mid-print tray switch: split weight across trays ---
# Split math is shared with the Spoolman writer via
# ``utils.tray_split.compute_tray_split_grams`` (#1793) — both
# inventory backends must attribute segments identically or a
# user running dual-mode sees divergent totals.
if len(tray_changes) > 1:
# Compute total weight for this slot (same logic as normal path)
if layer_grams and slot_id in layer_grams:
@@ -1100,8 +1104,6 @@ async def _track_from_3mf(
from backend.app.utils.threemf_tools import (
extract_filament_properties_from_3mf,
extract_layer_filament_usage_from_3mf,
get_cumulative_usage_at_layer,
mm_to_grams,
)
split_layer_usage = extract_layer_filament_usage_from_3mf(file_path)
@@ -1110,46 +1112,20 @@ async def _track_from_3mf(
except Exception:
pass # Fall back to linear splitting
density = split_props.get("density", 1.24)
diameter = split_props.get("diameter", 1.75)
filament_id = slot_id - 1 # 0-based for gcode
from backend.app.utils.tray_split import compute_tray_split_grams
sum_previous = 0.0
for seg_idx, (tray_global, seg_start_layer) in enumerate(tray_changes):
is_last = seg_idx + 1 >= len(tray_changes)
segments = compute_tray_split_grams(
tray_changes=tray_changes,
total_weight=total_weight,
slot_id=slot_id,
layer_usage=split_layer_usage,
density=split_props.get("density", 1.24),
diameter=split_props.get("diameter", 1.75),
total_layers=(state.total_layers if state else 0) or 0,
last_layer_num=last_layer_num,
)
if is_last:
# Last segment: remainder to avoid rounding drift
segment_grams = total_weight - sum_previous
elif split_layer_usage:
seg_end_layer = tray_changes[seg_idx + 1][1]
mm_at_start = get_cumulative_usage_at_layer(split_layer_usage, seg_start_layer).get(filament_id, 0)
mm_at_end = get_cumulative_usage_at_layer(split_layer_usage, seg_end_layer).get(filament_id, 0)
segment_grams = mm_to_grams(mm_at_end - mm_at_start, diameter, density)
else:
# No per-layer data: linear fallback by layer ratio (#1771).
# Cascade denominators because firmware on some models (P1S
# observed) resets `total_layer_num` to 0 at print end —
# `last_layer_num` is the print's last-valid layer captured
# mid-print and survives that reset (same shape as the
# `last_progress` fallback at line 1040). Equal-split is the
# last-resort fence: still wrong, but bounded — never dumps
# the entire print onto the last segment, which was the
# original #1771 symptom for the reporter (P1S, AMS Backup
# fed from spool 1 then spool 2, all 260 g credited to
# spool 2 even though spool 1 had given up its 180 g).
seg_end_layer = tray_changes[seg_idx + 1][1]
denom = (state.total_layers if state else 0) or last_layer_num
if denom > 0:
segment_grams = total_weight * (seg_end_layer - seg_start_layer) / denom
else:
# No layer information available from any source —
# spread evenly across segments. The last segment will
# get the rounding remainder via the `is_last` branch
# above on its own iteration.
segment_grams = total_weight / len(tray_changes)
sum_previous += segment_grams
for seg_idx, tray_global, segment_grams in segments:
if segment_grams <= 0:
continue
@@ -1168,6 +1144,8 @@ async def _track_from_3mf(
if seg_key in handled_trays:
continue
seg_start_layer = tray_changes[seg_idx][1]
is_last = seg_idx + 1 >= len(tray_changes)
logger.info(
"[UsageTracker] 3MF split: segment %d tray=%d (AMS%d-T%d) layers %d-%s -> %.1fg",
seg_idx,
backend/app/utils/tray_split.py
+94
View File
@@ -0,0 +1,94 @@
"""Weight-split math for prints that traversed >1 AMS tray mid-print.
`state.tray_change_log` records `(global_tray_id, layer_num)` tuples every
time `tray_now` changes during a print (see `bambu_mqtt.py:1861`). At
completion, both the internal Spool inventory (`usage_tracker.py`) and the
Spoolman writer (`spoolman_tracking.py`) need to split a slot's total
weight across the segments those changes define — one call site per
inventory backend, one identical splitting algorithm.
The algorithm lives here so the two callers cannot drift: #1793 came from
`spoolman_tracking` never carrying the split at all, while `usage_tracker`
had shipped it since #957 and refined it in #1771. Sharing the helper is
the structural fix; each caller wraps its own "resolve segment tray →
charge N grams" side effect.
"""
from __future__ import annotations
# Qualified-name access (``threemf_tools.mm_to_grams(...)`` rather than
# ``from … import mm_to_grams``) so ``unittest.mock.patch`` on the
# threemf_tools module lands in the helper too — the pre-refactor
# ``usage_tracker`` call site imported at call-time inside a try block,
# which had the same testability property.
from backend.app.utils import threemf_tools
def compute_tray_split_grams(
tray_changes: list[tuple[int, int]],
total_weight: float,
slot_id: int,
layer_usage: dict[int, dict[int, float]] | None,
density: float,
diameter: float,
total_layers: int,
last_layer_num: int,
) -> list[tuple[int, int, float]]:
"""Split ``total_weight`` for a single slice slot across tray segments.
``tray_changes`` is the ordered list ``[(global_tray_id, seg_start_layer), ...]``
exactly as it appears in ``state.tray_change_log``. The last segment
runs to the end of the print; every other segment ends at the next
entry's ``seg_start_layer``.
Preference order for per-segment grams — matches ``usage_tracker`` so
both inventory backends split identically:
1. **G-code cumulative extrusion** (``layer_usage``, indexed by 0-based
filament id). Precise: uses the mm actually consumed between
``seg_start_layer`` and the next segment's start, then converts via
Spoolman-authoritative ``density`` / ``diameter``.
2. **Linear layer-ratio** — ``total_weight * segment_layers / denom``,
with ``denom = total_layers or last_layer_num``. Firmware on P1S
(observed) resets ``total_layer_num`` to 0 at print end, so the
captured ``last_layer_num`` is the durable denominator (see
``usage_tracker.py:1132``). #1771 addressed the pre-fix behaviour
of dumping everything onto the last segment.
3. **Equal-split** — when neither denominator is available (server
restart mid-print, missing state). Wrong but bounded — the last
segment absorbs any rounding drift via the ``is_last`` branch.
Returns ``[(seg_idx, global_tray_id, segment_grams)]``. Empty when
``tray_changes`` is empty; the caller decides whether to fall through
to single-tray attribution (``len(tray_changes) <= 1``).
"""
if not tray_changes:
return []
filament_id = slot_id - 1
n_segments = len(tray_changes)
denom = total_layers or last_layer_num
results: list[tuple[int, int, float]] = []
sum_previous = 0.0
for seg_idx, (tray_global, seg_start_layer) in enumerate(tray_changes):
is_last = seg_idx + 1 >= n_segments
if is_last:
segment_grams = total_weight - sum_previous
elif layer_usage:
seg_end_layer = tray_changes[seg_idx + 1][1]
mm_at_start = threemf_tools.get_cumulative_usage_at_layer(layer_usage, seg_start_layer).get(filament_id, 0)
mm_at_end = threemf_tools.get_cumulative_usage_at_layer(layer_usage, seg_end_layer).get(filament_id, 0)
segment_grams = threemf_tools.mm_to_grams(mm_at_end - mm_at_start, diameter, density)
else:
seg_end_layer = tray_changes[seg_idx + 1][1]
if denom > 0:
segment_grams = total_weight * (seg_end_layer - seg_start_layer) / denom
else:
segment_grams = total_weight / n_segments
sum_previous += segment_grams
results.append((seg_idx, tray_global, segment_grams))
return results
backend/tests/unit/services/test_spoolman_tray_split.py
+350
View File
@@ -0,0 +1,350 @@
"""Spoolman-side mid-print tray-split accounting (#1793).
Reporter (@ojimpo) shipped the OP shape:
- H2S, AMS filament backup ON, two same-material spools loaded
- Single-slot print (72.56g on slot 1)
- Origin ran dry at layer 37, AMS auto-switched to backup, print finished
- Pre-fix: whole 72.56g charged to origin (via tag path) + separate 30g to
backup (via remain-delta) — origin exceeded initial_weight, backup double-count
The fix ports usage_tracker's split path to spoolman_tracking so both
inventory backends attribute segments identically. These tests pin the
OP's shape plus the Path 2 (remain-delta) skip guarantee so it can't
double-charge tray IDs the split path already covered.
"""
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
class _AsyncCtx:
"""async_session() shim — same shape as test_spoolman_no3mf_remain_fallback."""
def __init__(self, db):
self._db = db
async def __aenter__(self):
return self._db
async def __aexit__(self, *_):
return False
def _make_db(tracking):
db = AsyncMock()
select_result = MagicMock()
select_result.scalar_one_or_none.return_value = tracking
db.execute = AsyncMock(return_value=select_result)
db.delete = AsyncMock()
db.commit = AsyncMock()
return db
class TestReportUsageTraySplit:
"""report_usage must consult state.tray_change_log and split per-segment."""
@pytest.mark.asyncio
async def test_op_sample_a_seamless_switch_splits_origin_to_backup(self):
"""Sample A from the reporter, verbatim: 72.56g single-slot print,
AMS runout switch tray 0 → tray 1 at layer 37 of ~100 total.
No gcode layer_usage is provided → linear-by-layer-ratio fallback:
- seg 0 (tray 0, layers 0-37) = 72.56 * 37/100 = 26.85g → spool 8
- seg 1 (tray 1, layers 37-end) = 72.56 - 26.85 = 45.71g → spool 7
Path 2 (remain-delta) must NOT run against either tray — the split
path already covered them.
"""
from backend.app.services.spoolman_tracking import report_usage
tracking = SimpleNamespace(
filament_usage=[{"slot_id": 1, "used_g": 72.56}],
ams_trays={
0: {"tray_uuid": "AAAA", "tag_uid": "T1TAG", "tray_type": "PLA"},
1: {"tray_uuid": "BBBB", "tag_uid": "T2TAG", "tray_type": "PLA"},
},
slot_to_tray=[0],
tray_remain_start={
"0-0": {"remain": 3, "tray_uuid": "AAAA"}, # origin near-empty at start-of-completion snapshot
"0-1": {"remain": 73, "tray_uuid": "BBBB"},
},
layer_usage={},
filament_properties={},
)
db = _make_db(tracking)
client = AsyncMock()
async def _find_spool_by_tag(tag):
return (
{"id": 8, "filament": {"color_hex": "000000"}}
if tag == "AAAA"
else {"id": 7, "filament": {"color_hex": "000000"}}
if tag == "BBBB"
else None
)
client.find_spool_by_tag = AsyncMock(side_effect=_find_spool_by_tag)
client.use_spool = AsyncMock()
printer_manager = MagicMock()
printer_manager.get_status.return_value = SimpleNamespace(
tray_change_log=[(0, 0), (1, 37)],
total_layers=100,
layer_num=100,
raw_data={
"ams": [
{
"id": 0,
"tray": [
{"id": 0, "tray_uuid": "AAAA", "remain": 0},
{"id": 1, "tray_uuid": "BBBB", "remain": 70},
],
}
]
},
)
with (
patch("backend.app.services.spoolman_tracking.async_session", lambda: _AsyncCtx(db)),
patch("backend.app.api.routes.settings.get_setting", AsyncMock(return_value="true")),
patch(
"backend.app.services.spoolman_tracking._get_spoolman_client_with_fallback",
AsyncMock(return_value=client),
),
patch("backend.app.services.spoolman_tracking._get_printer_serial", AsyncMock(return_value="SERIAL")),
patch(
"backend.app.services.spoolman_tracking._apply_spool_colors_to_archive",
AsyncMock(),
),
patch("backend.app.services.printer_manager.printer_manager", printer_manager),
):
await report_usage(printer_id=1, archive_id=143)
# Exactly two use_spool calls — one per segment. Origin (spool 8)
# gets the layers-0-37 slice, backup (spool 7) gets the remainder.
calls = client.use_spool.await_args_list
assert len(calls) == 2, f"expected 2 use_spool calls, got {len(calls)}: {calls}"
by_spool = {c.args[0]: c.args[1] for c in calls}
assert set(by_spool.keys()) == {8, 7}
# Sum must equal the OP's total — no phantom grams created or lost.
assert round(sum(by_spool.values()), 2) == 72.56
# Origin (spool 8) should carry roughly the layers-0-37 fraction.
# Linear: 72.56 * 37/100 = 26.85g. Allow small rounding wiggle.
assert 26.0 < by_spool[8] < 28.0
# Backup (spool 7) carries the remainder.
assert 44.0 < by_spool[7] < 46.6
@pytest.mark.asyncio
async def test_path_2_remain_delta_skips_tray_handled_by_split(self):
"""After the split path attributes segments to tray 0 AND tray 1,
the Path 2 remain-delta iterator must skip BOTH — otherwise backup
would get charged twice (~30g double-count in the OP's Sample A).
"""
from backend.app.services.spoolman_tracking import report_usage
tracking = SimpleNamespace(
filament_usage=[{"slot_id": 1, "used_g": 100.0}],
ams_trays={
0: {"tray_uuid": "AAAA", "tag_uid": "T1TAG", "tray_type": "PLA"},
1: {"tray_uuid": "BBBB", "tag_uid": "T2TAG", "tray_type": "PLA"},
},
slot_to_tray=[0],
tray_remain_start={
"0-0": {"remain": 20, "tray_uuid": "AAAA"},
"0-1": {"remain": 80, "tray_uuid": "BBBB"},
},
layer_usage={},
filament_properties={},
)
db = _make_db(tracking)
client = AsyncMock()
async def _find_spool_by_tag(tag):
return {"id": 8, "filament": {}} if tag == "AAAA" else {"id": 7, "filament": {}}
client.find_spool_by_tag = AsyncMock(side_effect=_find_spool_by_tag)
client.use_spool = AsyncMock()
# If Path 2 ever runs, it needs a filament.weight to compute grams.
# Making it valid means a failure to guard = extra use_spool calls,
# not a silent skip. Combined with a truthy slot-assignment result
# below, this is what actually proves the double-count guard works.
client.get_spool = AsyncMock(return_value={"filament": {"weight": 1000.0}})
printer_manager = MagicMock()
printer_manager.get_status.return_value = SimpleNamespace(
tray_change_log=[(0, 0), (1, 50)],
total_layers=100,
layer_num=100,
raw_data={
"ams": [
{
"id": 0,
"tray": [
{"id": 0, "tray_uuid": "AAAA", "remain": 0},
{"id": 1, "tray_uuid": "BBBB", "remain": 60},
],
}
]
},
)
with (
patch("backend.app.services.spoolman_tracking.async_session", lambda: _AsyncCtx(db)),
patch("backend.app.api.routes.settings.get_setting", AsyncMock(return_value="true")),
patch(
"backend.app.services.spoolman_tracking._get_spoolman_client_with_fallback",
AsyncMock(return_value=client),
),
patch("backend.app.services.spoolman_tracking._get_printer_serial", AsyncMock(return_value="SERIAL")),
patch(
# Path 2 uses this to resolve trays. Return valid IDs so
# the ONLY thing stopping Path 2 from double-charging is
# ``handled_global_tray_ids``. If the guard is broken,
# Path 2 would successfully call ``use_spool`` two more
# times and this test would fail with 4 calls, not 2.
"backend.app.services.spoolman_tracking._resolve_spool_id_via_slot_assignment",
AsyncMock(side_effect=lambda pid, ams, tray: 999 if (ams, tray) == (0, 0) else 888),
),
patch("backend.app.services.printer_manager.printer_manager", printer_manager),
):
await report_usage(printer_id=1, archive_id=200)
# EXACTLY 2 — one per segment; Path 2 must not add a third.
assert client.use_spool.await_count == 2, (
f"Path 2 leaked past the split — expected 2 use_spool calls, got "
f"{client.use_spool.await_count}: {client.use_spool.await_args_list}"
)
@pytest.mark.asyncio
async def test_multi_slot_print_does_not_activate_split_even_with_tray_changes(self):
"""Multi-colour prints normally cycle trays every colour change, so
``tray_change_log`` has many entries — but splitting each slot's
grams across all of them would attribute slot 1's usage to segments
where slot 2's tray was loaded (and vice versa).
Mirrors ``usage_tracker.py:1002``'s gate: split only when there's
exactly one nonzero slot. Multi-slot prints fall through to the
existing single-tray path with its stable ``slot_to_tray`` mapping.
"""
from backend.app.services.spoolman_tracking import report_usage
# Two nonzero slots — regular multi-colour print
tracking = SimpleNamespace(
filament_usage=[
{"slot_id": 1, "used_g": 30.0},
{"slot_id": 2, "used_g": 20.0},
],
ams_trays={
0: {"tray_uuid": "AAAA", "tag_uid": "T1TAG", "tray_type": "PLA"},
1: {"tray_uuid": "BBBB", "tag_uid": "T2TAG", "tray_type": "PLA"},
},
slot_to_tray=[0, 1],
tray_remain_start={
"0-0": {"remain": 90, "tray_uuid": "AAAA"},
"0-1": {"remain": 80, "tray_uuid": "BBBB"},
},
layer_usage={},
filament_properties={},
)
db = _make_db(tracking)
client = AsyncMock()
async def _find_spool_by_tag(tag):
return {"id": 100, "filament": {}} if tag == "AAAA" else {"id": 200, "filament": {}}
client.find_spool_by_tag = AsyncMock(side_effect=_find_spool_by_tag)
client.use_spool = AsyncMock()
printer_manager = MagicMock()
# Multi-colour print naturally cycles between trays many times.
# If we don't gate on single-slot, my split would attribute slot 1's
# grams to every segment — including segments where tray 1 was loaded.
printer_manager.get_status.return_value = SimpleNamespace(
tray_change_log=[(0, 0), (1, 10), (0, 20), (1, 30), (0, 40)],
total_layers=50,
layer_num=50,
raw_data={
"ams": [
{
"id": 0,
"tray": [
{"id": 0, "tray_uuid": "AAAA", "remain": 87},
{"id": 1, "tray_uuid": "BBBB", "remain": 78},
],
}
]
},
)
with (
patch("backend.app.services.spoolman_tracking.async_session", lambda: _AsyncCtx(db)),
patch("backend.app.api.routes.settings.get_setting", AsyncMock(return_value="true")),
patch(
"backend.app.services.spoolman_tracking._get_spoolman_client_with_fallback",
AsyncMock(return_value=client),
),
patch("backend.app.services.spoolman_tracking._get_printer_serial", AsyncMock(return_value="SERIAL")),
patch("backend.app.services.printer_manager.printer_manager", printer_manager),
):
await report_usage(printer_id=1, archive_id=42)
# Split path must NOT engage. The single-tray path charges each
# slot to its stable slot_to_tray mapping: slot 1 → tray 0 → spool
# 100 (30g), slot 2 → tray 1 → spool 200 (20g). Two calls, exact
# weights from the 3MF (not split).
assert client.use_spool.await_count == 2
by_spool = {c.args[0]: c.args[1] for c in client.use_spool.await_args_list}
assert by_spool == {100: 30.0, 200: 20.0}
@pytest.mark.asyncio
async def test_single_tray_change_entry_uses_normal_path(self):
"""Only ONE entry in tray_change_log (start-of-print seed, no
switch) must fall through to the existing single-tray charging
path — not accidentally split when there's nothing to split.
"""
from backend.app.services.spoolman_tracking import report_usage
tracking = SimpleNamespace(
filament_usage=[{"slot_id": 1, "used_g": 50.0}],
ams_trays={0: {"tray_uuid": "AAAA", "tag_uid": "T1TAG", "tray_type": "PLA"}},
slot_to_tray=[0],
tray_remain_start={"0-0": {"remain": 80, "tray_uuid": "AAAA"}},
layer_usage={},
filament_properties={},
)
db = _make_db(tracking)
client = AsyncMock()
client.find_spool_by_tag = AsyncMock(return_value={"id": 8, "filament": {}})
client.use_spool = AsyncMock()
printer_manager = MagicMock()
printer_manager.get_status.return_value = SimpleNamespace(
tray_change_log=[(0, 0)], # just the start-of-print seed
total_layers=100,
layer_num=100,
raw_data={"ams": [{"id": 0, "tray": [{"id": 0, "tray_uuid": "AAAA", "remain": 75}]}]},
)
with (
patch("backend.app.services.spoolman_tracking.async_session", lambda: _AsyncCtx(db)),
patch("backend.app.api.routes.settings.get_setting", AsyncMock(return_value="true")),
patch(
"backend.app.services.spoolman_tracking._get_spoolman_client_with_fallback",
AsyncMock(return_value=client),
),
patch("backend.app.services.spoolman_tracking._get_printer_serial", AsyncMock(return_value="SERIAL")),
patch("backend.app.services.printer_manager.printer_manager", printer_manager),
):
await report_usage(printer_id=1, archive_id=42)
# Single-tray path: exactly one use_spool call, all 50g to spool 8.
client.use_spool.assert_awaited_once_with(8, 50.0)
backend/tests/unit/utils/test_tray_split.py
+179
View File
@@ -0,0 +1,179 @@
"""Pure-logic tests for the mid-print tray-split math (#1793).
The helper lives in ``backend/app/utils/tray_split.py`` and is exercised
by both inventory backends (``usage_tracker`` and ``spoolman_tracking``).
These tests pin the algorithm so a change in one caller can't silently
break the other — cross-inventory parity is a HARD RULE for this project.
"""
from __future__ import annotations
from backend.app.utils.tray_split import compute_tray_split_grams
class TestComputeTraySplitGrams:
"""Segment-attribution algorithm — gcode preferred, linear fallback, equal split."""
def test_empty_tray_changes_returns_empty(self):
assert (
compute_tray_split_grams(
tray_changes=[],
total_weight=100.0,
slot_id=1,
layer_usage=None,
density=1.24,
diameter=1.75,
total_layers=200,
last_layer_num=200,
)
== []
)
def test_single_segment_charges_everything_to_that_tray(self):
segments = compute_tray_split_grams(
tray_changes=[(0, 0)],
total_weight=72.56,
slot_id=1,
layer_usage=None,
density=1.24,
diameter=1.75,
total_layers=100,
last_layer_num=100,
)
assert segments == [(0, 0, 72.56)]
def test_two_segments_linear_split_by_layer_ratio(self):
# Runout at layer 37 of 100 total; no gcode available → linear.
# Segment 0 (tray 0, layers 0-37) = 100 * 37/100 = 37g
# Segment 1 (tray 1, layers 37-end) = 100 - 37 = 63g (remainder)
segments = compute_tray_split_grams(
tray_changes=[(0, 0), (1, 37)],
total_weight=100.0,
slot_id=1,
layer_usage=None,
density=1.24,
diameter=1.75,
total_layers=100,
last_layer_num=100,
)
assert segments == [(0, 0, 37.0), (1, 1, 63.0)]
def test_two_segments_gcode_preferred_over_linear(self):
# layer_usage stores mm of filament extruded per (layer, filament_id).
# Values are cumulative-per-key inside get_cumulative_usage_at_layer.
# 20 layers, filament_id=0 (slot_id=1 → filament_id 0):
# layer 10 → 100mm cumulative
# layer 20 → 300mm cumulative
# tray change at layer 10 → seg 0 spans layers 0-10 (mm 0 → 100),
# seg 1 spans layers 10-end.
# mm_to_grams(100, 1.75, 1.24) ≈ 0.298g; last segment absorbs the rest.
layer_usage = {
5: {0: 50.0},
10: {0: 100.0},
15: {0: 200.0},
20: {0: 300.0},
}
segments = compute_tray_split_grams(
tray_changes=[(0, 0), (1, 10)],
total_weight=1.0, # sentinel — we assert the seg1 remainder
slot_id=1,
layer_usage=layer_usage,
density=1.24,
diameter=1.75,
total_layers=20,
last_layer_num=20,
)
# Seg 0 charged from gcode delta (mm 0 → 100).
# Seg 1 gets total_weight - seg0 as remainder.
assert segments[0][0] == 0
assert segments[0][1] == 0 # tray 0
assert segments[0][2] > 0 # non-zero gcode contribution
assert segments[1][0] == 1
assert segments[1][1] == 1 # tray 1
# Sum equals the input total by construction (last segment absorbs).
assert round(segments[0][2] + segments[1][2], 6) == 1.0
def test_three_segments_last_absorbs_rounding_drift(self):
# 100g over three segments at layers 30 and 60 of 90; linear fallback.
# Seg 0: 100 * 30/90 = 33.3333...
# Seg 1: 100 * 30/90 = 33.3333...
# Seg 2: remainder = 100 - 66.6666... = 33.3333... — exact by construction
segments = compute_tray_split_grams(
tray_changes=[(0, 0), (1, 30), (2, 60)],
total_weight=100.0,
slot_id=1,
layer_usage=None,
density=1.24,
diameter=1.75,
total_layers=90,
last_layer_num=90,
)
assert len(segments) == 3
assert round(sum(g for _, _, g in segments), 6) == 100.0
assert segments[0][1] == 0
assert segments[1][1] == 1
assert segments[2][1] == 2
def test_no_layer_info_at_all_falls_to_equal_split(self):
# Denominator 0 → last-resort equal-split; last segment absorbs remainder.
segments = compute_tray_split_grams(
tray_changes=[(0, 0), (1, 50)],
total_weight=90.0,
slot_id=1,
layer_usage=None,
density=1.24,
diameter=1.75,
total_layers=0,
last_layer_num=0,
)
# 90g / 2 = 45g each; sum still 90 by remainder mechanic.
assert segments == [(0, 0, 45.0), (1, 1, 45.0)]
def test_last_layer_num_used_when_total_layers_zero(self):
# P1S firmware-reset scenario: total_layers=0 at completion, but the
# captured last_layer_num survives. Should give the same linear split
# as if total_layers had held its value (#1771 cascade).
segments_captured = compute_tray_split_grams(
tray_changes=[(0, 0), (1, 30)],
total_weight=100.0,
slot_id=1,
layer_usage=None,
density=1.24,
diameter=1.75,
total_layers=0,
last_layer_num=100,
)
segments_normal = compute_tray_split_grams(
tray_changes=[(0, 0), (1, 30)],
total_weight=100.0,
slot_id=1,
layer_usage=None,
density=1.24,
diameter=1.75,
total_layers=100,
last_layer_num=100,
)
assert segments_captured == segments_normal
def test_slot_id_maps_to_zero_based_filament_id_in_gcode(self):
# slot_id 2 → filament_id 1 in layer_usage. If we mistakenly used
# slot_id as-is, we'd read filament_id 2 which is absent → 0mm delta
# → seg 0 gets 0, seg 1 (remainder) gets the whole total. Guard
# against that regression.
layer_usage = {
5: {0: 0.0, 1: 40.0},
10: {0: 0.0, 1: 80.0},
20: {0: 0.0, 1: 160.0},
}
segments = compute_tray_split_grams(
tray_changes=[(0, 0), (1, 10)],
total_weight=1.0,
slot_id=2,
layer_usage=layer_usage,
density=1.24,
diameter=1.75,
total_layers=20,
last_layer_num=20,
)
# Seg 0 gcode delta on filament_id=1 is non-zero → not 0g.
assert segments[0][2] > 0