public_git/bambuddy
1
0
Fork 0
mirror of https://github.com/maziggy/bambuddy.git synced 2026-10-07 23:01:06 +02:00
Code Issues Packages Projects Releases Wiki Activity

fix(notifications): false-positive Print Stopped on reprint after MQTT reconnect (#1807)

Browse Source 
Reprints triggered a bogus "Print Stopped" push notification while the print
      kept running, surfaced by the reconciler synthesising a missed PRINT COMPLETE
      on MQTT reconnect.

      bambu_mqtt:3647 mints a fresh subtask_id per dispatch. On reprint, the
      on_print_start expected-archive promotion only wrote subtask_id when the
      stored value was empty (`not archive.subtask_id`) — so the archive kept the
      FIRST run's id. On the next MQTT reconnect, reconcile_stale_active_prints
      (#1542) compared the stale stored id against the printer's live id, found
      a mismatch, and synthesised a status="aborted" PRINT COMPLETE — which fires
      the "Print Stopped" notification.

      Captured cleanly in the reporter's support bundle:

        [RECONCILE] Printer 1: synthesising missed PRINT COMPLETE for archive 31
          — subtask_id changed ('1844213296' → '2103771517')

      immediately followed by gcode_state: RUNNING on the same wire.

      Fix: update archive.subtask_id whenever the new effective id differs from
      the stored one, not only when the stored one is empty. Inequality check
      preserves the noop-on-stable-push behaviour the original guard provided.

      Two places in main.py (expected-print and duplicate-printing-archive
      branches). 3 new unit tests cover the reprint, first-run, and stable-push
      paths. Reconciler itself unchanged — it was doing the right thing given
      the data it had.
This commit is contained in:
maziggy
2026-06-28 12:45:48 +02:00
parent 8b72b305a3 5c673620f3
commit 5e008744de
7 changed files with 1028 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+3
View File
File diff suppressed because one or more lines are too long
backend/app/core/database.py
+42 -3
View File
@@ -1528,7 +1528,10 @@ async def run_migrations(conn):
except (OperationalError, ProgrammingError):
pass # Already applied
# Create active_print_spoolman table for Spoolman per-filament tracking
# Create active_print_spoolman table for Spoolman per-filament tracking.
# filament_usage is nullable so the no-3MF branch can still create a row
# that carries only tray_remain_start for the remain%-delta fallback
# (#1820 — matches internal-inventory Path 2 in usage_tracker).
await _safe_execute(
conn,
"""
@@ -1536,11 +1539,12 @@ async def run_migrations(conn):
id INTEGER PRIMARY KEY AUTOINCREMENT,
printer_id INTEGER NOT NULL REFERENCES printers(id) ON DELETE CASCADE,
archive_id INTEGER NOT NULL REFERENCES print_archives(id) ON DELETE CASCADE,
filament_usage TEXT NOT NULL,
filament_usage TEXT,
ams_trays TEXT NOT NULL,
slot_to_tray TEXT,
layer_usage TEXT,
filament_properties TEXT,
tray_remain_start TEXT,
UNIQUE(printer_id, archive_id)
)
"""
@@ -1550,15 +1554,50 @@ async def run_migrations(conn):
id SERIAL PRIMARY KEY,
printer_id INTEGER NOT NULL REFERENCES printers(id) ON DELETE CASCADE,
archive_id INTEGER NOT NULL REFERENCES print_archives(id) ON DELETE CASCADE,
filament_usage TEXT NOT NULL,
filament_usage TEXT,
ams_trays TEXT NOT NULL,
slot_to_tray TEXT,
layer_usage TEXT,
filament_properties TEXT,
tray_remain_start TEXT,
UNIQUE(printer_id, archive_id)
)
""",
)
# Migration for installs that already created active_print_spoolman with
# the original schema: add tray_remain_start, and relax filament_usage's
# NOT NULL so the no-3MF branch can persist a remain-only tracking row.
await _safe_execute(conn, "ALTER TABLE active_print_spoolman ADD COLUMN tray_remain_start TEXT")
if is_sqlite():
# SQLite can't ALTER COLUMN; patch sqlite_master directly. Mirrors the
# users.password_hash NULL-relaxation a few hundred lines below — see
# the comment there for the schema_version bump rationale.
try:
result = await conn.execute(
text("SELECT sql FROM sqlite_master WHERE type='table' AND name='active_print_spoolman'")
)
tbl_sql = result.scalar()
if tbl_sql and "filament_usage TEXT NOT NULL" in tbl_sql:
version_result = await conn.execute(text("PRAGMA schema_version"))
schema_version = version_result.scalar() or 0
await conn.execute(text("PRAGMA writable_schema = ON"))
await conn.execute(
text(
"UPDATE sqlite_master "
"SET sql = replace(sql, 'filament_usage TEXT NOT NULL', 'filament_usage TEXT') "
"WHERE type='table' AND name='active_print_spoolman'"
)
)
await conn.execute(text(f"PRAGMA schema_version = {schema_version + 1}"))
await conn.execute(text("PRAGMA writable_schema = OFF"))
except (OperationalError, ProgrammingError) as exc:
logger.warning(
"Could not relax active_print_spoolman.filament_usage NOT NULL via writable_schema: %s — "
"no-3MF Spoolman fallback will be a no-op on this install",
exc,
)
else:
await _safe_execute(conn, "ALTER TABLE active_print_spoolman ALTER COLUMN filament_usage DROP NOT NULL")
# Migration: Add preset_source column to slot_preset_mappings for local preset support
try:
backend/app/main.py
+15 -3
View File
@@ -2561,7 +2561,16 @@ async def on_print_start(printer_id: int, data: dict):
_dispatched = getattr(_client, "last_dispatch_subtask_id", None) if _client else None
if _dispatched:
effective_subtask_id = str(_dispatched).strip() or None
if effective_subtask_id and not archive.subtask_id:
# Update on first-set OR on reprint (the queue dispatcher mints
# a fresh subtask_id per dispatch in bambu_mqtt:3647). Skipping
# the rewrite for reprints leaves the archive holding the FIRST
# run's id; if MQTT then reconnects mid-print, the reconciler
# (#1542) compares the stale stored id against the printer's
# live id, sees a mismatch, and synthesises a bogus PRINT
# COMPLETE — exactly the false-positive "Print Stopped" reported
# in #1807. Inequality check preserves the noop-on-stable-push
# behaviour the earlier `not archive.subtask_id` guard provided.
if effective_subtask_id and archive.subtask_id != effective_subtask_id:
archive.subtask_id = effective_subtask_id
# #1403 follow-up: VP-queue archives are created with
# printer_id=None at queue-add time (we don't know which
@@ -2769,8 +2778,11 @@ async def on_print_start(printer_id: int, data: dict):
)
# Track this as the active print
_active_prints[(printer_id, existing_archive.filename)] = existing_archive.id
# Attach subtask_id retroactively so future restarts can resume
if subtask_id and not existing_archive.subtask_id:
# Attach subtask_id retroactively so future restarts can resume.
# Compare for inequality (not "is empty") to also pick up reprint
# dispatches that mint a fresh id — see #1807 for the bogus
# "Print Stopped" the strict-empty guard caused on reconnect.
if subtask_id and existing_archive.subtask_id != subtask_id:
existing_archive.subtask_id = subtask_id
await db.commit()
# Also set up energy tracking if not already tracked (#941: persisted column)
backend/app/models/active_print_spoolman.py
+12 -1
View File
@@ -24,7 +24,11 @@ class ActivePrintSpoolman(Base):
archive_id: Mapped[int] = mapped_column(ForeignKey("print_archives.id", ondelete="CASCADE"))
# Per-filament usage from 3MF: [{"slot_id": 1, "used_g": 50.5, "type": "PLA"}, ...]
filament_usage: Mapped[list] = mapped_column(JSON)
# Nullable for the no-3MF case ("Untitled" prints where Bambu didn't keep a
# .gcode.3mf on the printer): the row still gets created so the completion
# path can use ``tray_remain_start`` for an AMS remain%-delta write,
# mirroring the internal-inventory Path 2 fallback in usage_tracker (#1820).
filament_usage: Mapped[list | None] = mapped_column(JSON, nullable=True)
# AMS tray state at print start: {0: {"tray_uuid": "...", "tag_uid": "..."}, ...}
ams_trays: Mapped[dict] = mapped_column(JSON)
@@ -40,3 +44,10 @@ class ActivePrintSpoolman(Base):
# Filament properties (density, diameter per filament slot)
# Format: {1: {"density": 1.24, "diameter": 1.75, "type": "PLA"}, ...}
filament_properties: Mapped[dict | None] = mapped_column(JSON, nullable=True)
# AMS tray remain% per slot at print start, captured so the completion
# path can compute a remain-delta when the 3MF didn't cover a slot (or
# there was no 3MF at all — #1820). Matches the internal-inventory
# ``tray_remain_start`` snapshot at usage_tracker.py:301.
# Format: {"<ams_id>-<tray_id>": {"remain": int, "tray_uuid": str}, ...}
tray_remain_start: Mapped[dict | None] = mapped_column(JSON, nullable=True)
backend/app/services/spoolman_tracking.py
+297 -57
View File
@@ -182,6 +182,52 @@ def build_ams_tray_lookup(raw_data: dict) -> dict[int, dict]:
return lookup
def _snapshot_tray_remain(raw_data: dict) -> dict[str, dict]:
"""Capture per-slot ``remain%`` + ``tray_uuid`` at print start so the
completion path can compute a remain-delta when 3MF data doesn't cover
the slot (or there's no 3MF at all — #1820).
Returns ``{"<ams_id>-<tray_id>": {"remain": int, "tray_uuid": str}}``.
Only slots whose ``remain`` is a valid 0..100 int are included; invalid
values mean the AMS hasn't read the spool yet and a delta would be
meaningless. Mirrors the gate in
``usage_tracker.on_print_start:309``.
"""
snapshot: dict[str, dict] = {}
ams_raw = raw_data.get("ams", [])
ams_data = ams_raw.get("ams", []) if isinstance(ams_raw, dict) else ams_raw if isinstance(ams_raw, list) else []
for ams_unit in ams_data:
if not isinstance(ams_unit, dict):
continue
ams_id = int(ams_unit.get("id", 0))
for tray in ams_unit.get("tray", []):
if not isinstance(tray, dict):
continue
tray_id = int(tray.get("id", 0))
remain = tray.get("remain", -1)
if isinstance(remain, int) and 0 <= remain <= 100:
snapshot[f"{ams_id}-{tray_id}"] = {
"remain": remain,
"tray_uuid": tray.get("tray_uuid", "") or "",
}
vt_tray_raw = raw_data.get("vt_tray") or []
if isinstance(vt_tray_raw, dict):
vt_tray_raw = [vt_tray_raw]
for vt in vt_tray_raw:
if not isinstance(vt, dict):
continue
vt_id = int(vt.get("id", 254))
# 254 → (255, 0), 255 → (255, 1) — matches usage_tracker's encoding.
vt_tray_id = vt_id - 254
remain = vt.get("remain", -1)
if isinstance(remain, int) and 0 <= remain <= 100:
snapshot[f"255-{vt_tray_id}"] = {
"remain": remain,
"tray_uuid": vt.get("tray_uuid", "") or "",
}
return snapshot
async def store_print_data(
printer_id: int,
archive_id: int,
@@ -219,38 +265,65 @@ async def store_print_data(
if not spoolman_enabled or spoolman_enabled.lower() != "true":
return
# Get 3MF file path
# Get current AMS tray state up front — needed both for the 3MF path's
# ams_trays field and for the remain%-delta snapshot (#1820 fallback for
# no-3MF "Untitled" prints, mirroring usage_tracker.on_print_start).
state = printer_manager.get_status(printer_id)
ams_trays: dict[int, dict] = {}
tray_remain_start: dict[str, dict] = {}
if state and state.raw_data:
ams_trays = build_ams_tray_lookup(state.raw_data)
tray_remain_start = _snapshot_tray_remain(state.raw_data)
# Try to read per-slot filament estimates from the 3MF. Two paths can
# leave ``filament_usage`` empty: (1) fallback archive (no .gcode.3mf
# was downloadable from the printer — "Untitled" prints, see #1820),
# (2) 3MF present but slice_info missing per-filament estimates.
# Both fall through to the remain%-delta path at completion.
filament_usage: list | None = None
layer_usage_json: dict | None = None
filament_properties: dict | None = None
full_path = (
app_settings.base_dir / file_path
) # SEC-PATH-OK: file_path is archive.file_path / library_file.file_path — DB-stored, internally generated
if not full_path.exists():
logger.debug("[SPOOLMAN] 3MF file not found: %s", full_path)
threemf_available = bool(file_path) and full_path.exists()
queue_item = None
if threemf_available:
# Resolve the queue item once — used both for the plate-scoped 3MF parsing
# fallback (#1697: multi-plate file dispatched for one plate must only count
# that plate's filament) and for the ams_mapping fallback below.
queue_result = await db.execute(
select(PrintQueueItem)
.where(PrintQueueItem.archive_id == archive_id)
.where(PrintQueueItem.status == "printing")
)
queue_item = queue_result.scalar_one_or_none()
# Caller-supplied plate_id wins (direct-Print path); fall back to the queue
# item's plate_id (queue dispatch path).
effective_plate_id = (
plate_id if plate_id is not None else (queue_item.plate_id if queue_item is not None else None)
)
filament_usage = extract_filament_usage_from_3mf(full_path, effective_plate_id) or None
layer_usage = extract_layer_filament_usage_from_3mf(full_path)
if layer_usage:
# Convert int keys to string for JSON serialization
layer_usage_json = {str(k): v for k, v in layer_usage.items()}
logger.debug("[SPOOLMAN] Parsed %s layers from G-code", len(layer_usage))
filament_properties = extract_filament_properties_from_3mf(full_path)
else:
# No 3MF on disk — common for "Untitled" prints whose .gcode.3mf
# was never on the printer's FTP. Logged at debug since the
# fallback path below picks up the slack when remain% is available.
logger.debug("[SPOOLMAN] 3MF file not available: %s", full_path)
# If neither path has anything useful, there's nothing to track.
if not filament_usage and not tray_remain_start:
if threemf_available:
logger.debug("[SPOOLMAN] No filament usage data in 3MF for archive %s", archive_id)
return
# Resolve the queue item once — used both for the plate-scoped 3MF parsing
# fallback (#1697: multi-plate file dispatched for one plate must only count
# that plate's filament) and for the ams_mapping fallback below.
queue_result = await db.execute(
select(PrintQueueItem).where(PrintQueueItem.archive_id == archive_id).where(PrintQueueItem.status == "printing")
)
queue_item = queue_result.scalar_one_or_none()
# Caller-supplied plate_id wins (direct-Print path); fall back to the queue
# item's plate_id (queue dispatch path).
effective_plate_id = plate_id if plate_id is not None else (queue_item.plate_id if queue_item is not None else None)
# Extract per-filament usage from 3MF (total usage for the dispatched plate,
# or the whole file for direct/library prints with no plate_id).
filament_usage = extract_filament_usage_from_3mf(full_path, effective_plate_id)
if not filament_usage:
logger.debug("[SPOOLMAN] No filament usage data in 3MF for archive %s", archive_id)
return
# Get current AMS tray state
state = printer_manager.get_status(printer_id)
ams_trays = {}
if state and state.raw_data:
ams_trays = build_ams_tray_lookup(state.raw_data)
# Prefer the explicit mapping captured from the print command, then fall back
# to any queue mapping stored for scheduled/reprint jobs.
slot_to_tray = ams_mapping if ams_mapping is not None else None
@@ -260,17 +333,6 @@ async def store_print_data(
except json.JSONDecodeError:
pass # Ignore malformed AMS mapping; fall back to default slot assignment
# Parse G-code for per-layer filament usage (for accurate partial usage tracking)
layer_usage = extract_layer_filament_usage_from_3mf(full_path)
layer_usage_json = None
if layer_usage:
# Convert int keys to string for JSON serialization
layer_usage_json = {str(k): v for k, v in layer_usage.items()}
logger.debug("[SPOOLMAN] Parsed %s layers from G-code", len(layer_usage))
# Extract filament properties (density, diameter) for mm -> grams conversion
filament_properties = extract_filament_properties_from_3mf(full_path)
# Delete any existing row for this printer/archive (shouldn't exist, but just in case)
await db.execute(
delete(ActivePrintSpoolman)
@@ -278,7 +340,8 @@ async def store_print_data(
.where(ActivePrintSpoolman.archive_id == archive_id)
)
# Insert new tracking data
# Insert new tracking data. ``filament_usage`` may be None for the
# no-3MF case; report_usage falls back to ``tray_remain_start``.
tracking = ActivePrintSpoolman(
printer_id=printer_id,
archive_id=archive_id,
@@ -287,11 +350,18 @@ async def store_print_data(
slot_to_tray=slot_to_tray,
layer_usage=layer_usage_json,
filament_properties=filament_properties,
tray_remain_start=tray_remain_start or None,
)
db.add(tracking)
await db.commit()
logger.info("[SPOOLMAN] Stored tracking data for print: printer=%s, archive=%s", printer_id, archive_id)
logger.info(
"[SPOOLMAN] Stored tracking data for print: printer=%s, archive=%s (3mf=%s, remain_snapshot=%d slot(s))",
printer_id,
archive_id,
"yes" if filament_usage else "no",
len(tray_remain_start),
)
logger.debug("[SPOOLMAN] Filament usage: %s", filament_usage)
logger.debug("[SPOOLMAN] AMS trays: %s", list(ams_trays.keys()))
if slot_to_tray:
@@ -571,6 +641,7 @@ async def _report_partial_usage(
filament_usage = tracking.filament_usage or []
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 {}
printer_serial = await _get_printer_serial(printer_id)
client = await _get_spoolman_client_with_fallback()
@@ -578,6 +649,24 @@ async def _report_partial_usage(
logger.warning("[SPOOLMAN] Not reachable for partial usage reporting")
return
# No-3MF aborted print (#1820 mirror of the completion path): nothing in
# filament_usage or layer_usage to base partial estimates on, but the
# remain%-delta snapshot we captured at start still describes consumption
# up to the abort moment. Write it the same way report_usage's fallback
# does, then return — there's no 3MF-derived partial to layer on top.
# ``state`` was already fetched at the top of the function for current_layer.
if not filament_usage and not layer_usage and tray_remain_start:
current_lookup = _snapshot_tray_remain(state.raw_data) if state and state.raw_data else {}
await _report_remain_delta_for_slots(
client,
printer_id=printer_id,
tray_remain_start=tray_remain_start,
current_lookup=current_lookup,
handled_global_tray_ids=set(),
archive_id=getattr(tracking, "archive_id", -1),
)
return
# Try to use accurate G-code parsed data
if layer_usage:
layer_usage_int = {
@@ -670,8 +759,15 @@ async def _report_partial_usage(
async def report_usage(printer_id: int, archive_id: int):
"""Report filament usage to Spoolman after print completion.
Uses per-filament usage data captured at print start to report
usage to the correct spools.
Two writers, mirroring the internal-inventory split in usage_tracker:
1. **3MF path (primary)** — per-filament slice estimates captured at
print start drive a precise per-slot ``use_spool`` call.
2. **AMS remain%-delta (fallback)** — for slots the 3MF path didn't
handle (including the no-3MF "Untitled" case from #1820): compute
``start_remain - current_remain``, multiply by the resolved
Spoolman filament's reference weight, and write the delta. Mirrors
``usage_tracker.on_print_complete`` Path 2 (line 517).
"""
async with async_session() as db:
from backend.app.api.routes.settings import get_setting
@@ -692,14 +788,15 @@ async def report_usage(printer_id: int, archive_id: int):
filament_usage = tracking.filament_usage or []
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 {}
printer_serial = await _get_printer_serial(printer_id)
# Delete tracking row (we're done with it)
await db.delete(tracking)
await db.commit()
if not filament_usage:
logger.debug("[SPOOLMAN] No filament usage data for archive %s", archive_id)
if not filament_usage and not tray_remain_start:
logger.debug("[SPOOLMAN] No usage data or remain-snapshot for archive %s", archive_id)
return
# Check if Spoolman is enabled
@@ -712,20 +809,48 @@ async def report_usage(printer_id: int, archive_id: int):
logger.warning("[SPOOLMAN] Not reachable for usage reporting")
return
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]
slot_colors: dict[int, str] = {}
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,
)
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))
# --- Path 2: AMS remain%-delta for slots 3MF didn't cover -------
# Triggered for no-3MF "Untitled" prints (#1820) AND for partial
# 3MF coverage (slots whose filament_id wasn't in slice_info).
if tray_remain_start:
from backend.app.services.printer_manager import printer_manager
current = printer_manager.get_status(printer_id)
current_lookup = _snapshot_tray_remain(current.raw_data) if current and current.raw_data else {}
fallback_updates = await _report_remain_delta_for_slots(
client,
printer_id=printer_id,
tray_remain_start=tray_remain_start,
current_lookup=current_lookup,
handled_global_tray_ids=handled_global_tray_ids,
archive_id=archive_id,
slot_colors_out=slot_colors,
)
spools_updated += fallback_updates
if spools_updated == 0:
logger.info("[SPOOLMAN] Archive %s: no spools updated", archive_id)
@@ -738,6 +863,121 @@ async def report_usage(printer_id: int, archive_id: int):
await _apply_spool_colors_to_archive(db, archive_id, filament_usage, slot_colors)
async def _report_remain_delta_for_slots(
client,
*,
printer_id: int,
tray_remain_start: dict[str, dict],
current_lookup: dict[str, dict],
handled_global_tray_ids: set[int],
archive_id: int,
slot_colors_out: dict[int, str] | None = None,
) -> int:
"""AMS remain%-delta path: write ``(start - current) * filament.weight``
grams to Spoolman for slots the 3MF path didn't cover.
Mirrors ``usage_tracker.on_print_complete`` Path 2: per-slot, gated on a
valid current ``remain%``, skipped on spool swap (``tray_uuid`` changed),
using the resolved spool's filament reference weight rather than MQTT's
unreliable ``tray_weight`` (which is the failure mode #1119 documented).
"""
spools_updated = 0
for slot_key, start in tray_remain_start.items():
try:
ams_id_str, tray_id_str = slot_key.split("-", 1)
ams_id, tray_id = int(ams_id_str), int(tray_id_str)
except (ValueError, AttributeError):
continue
# Skip slots already handled by the 3MF path. Encoding mirrors
# build_ams_tray_lookup: VT trays land at 254/255, AMS-HT keeps
# its native id (>=128), regular AMS slots are ams_id*4+tray_id.
if ams_id == 255:
global_tray_id = 254 + tray_id
elif ams_id >= 128:
global_tray_id = ams_id
else:
global_tray_id = ams_id * 4 + tray_id
if global_tray_id in handled_global_tray_ids:
continue
current = current_lookup.get(slot_key)
if not current:
logger.debug("[SPOOLMAN] AMS%d-T%d: no current remain%% at completion, skipping fallback", ams_id, tray_id)
continue
# Spool swap mid-print — tray_uuid changed. We don't know how much
# of the print went to which spool; skip rather than mis-attribute.
start_uuid = (start.get("tray_uuid") or "").lower()
cur_uuid = (current.get("tray_uuid") or "").lower()
if start_uuid and cur_uuid and start_uuid != cur_uuid:
logger.info(
"[SPOOLMAN] AMS%d-T%d: spool swapped mid-print (uuid changed), skipping remain-delta", ams_id, tray_id
)
continue
delta_pct = start["remain"] - current["remain"]
if delta_pct <= 0:
continue # No consumption captured at AMS granularity, or refilled
spool_id = await _resolve_spool_id_via_slot_assignment(printer_id, ams_id, tray_id)
if spool_id is None:
logger.debug("[SPOOLMAN] AMS%d-T%d: no Spoolman slot assignment, skipping fallback", ams_id, tray_id)
continue
# Look up the spool's filament reference weight. Use a fresh GET so
# we don't depend on a stale cached_spools list. Failure here is
# silent-skip rather than fatal — other slots can still be written.
try:
spool = await client.get_spool(spool_id)
except Exception as exc: # noqa: BLE001
logger.debug("[SPOOLMAN] AMS%d-T%d: get_spool(%s) failed: %s", ams_id, tray_id, spool_id, exc)
continue
filament = spool.get("filament") or {}
ref_weight = filament.get("weight")
if not ref_weight or ref_weight <= 0:
logger.debug(
"[SPOOLMAN] AMS%d-T%d: spool %s has no filament.weight, skipping remain-delta",
ams_id,
tray_id,
spool_id,
)
continue
grams_used = round((delta_pct / 100.0) * ref_weight, 2)
if grams_used <= 0:
continue
try:
await client.use_spool(spool_id, grams_used)
except Exception as exc: # noqa: BLE001
logger.warning(
"[SPOOLMAN] AMS%d-T%d: use_spool(%s, %.2fg) failed: %s", ams_id, tray_id, spool_id, grams_used, exc
)
continue
spools_updated += 1
if slot_colors_out is not None:
color = filament.get("color_hex")
if color:
# No 3MF slot_id for this path — use the AMS slot key so the
# colour map can still be inspected by callers if needed.
# The archive-colour rewrite (#1494) keys on 3MF slot_ids so
# remain-delta-only prints intentionally don't participate
# in that rewrite (matches usage_tracker's slot_id=None).
slot_colors_out[-(global_tray_id + 1)] = color
logger.info(
"[SPOOLMAN] Archive %s AMS%d-T%d: %.2fg via remain-delta (%d%% of %.0fg) -> spool %s",
archive_id,
ams_id,
tray_id,
grams_used,
delta_pct,
ref_weight,
spool_id,
)
return spools_updated
async def _apply_spool_colors_to_archive(
db,
archive_id: int,
backend/tests/unit/test_reprint_updates_subtask_id.py
+223
View File
@@ -0,0 +1,223 @@
"""Regression for #1807: false-positive "Print Stopped" notification on the
expected-archive reprint path.
Bambuddy mints a fresh subtask_id per dispatch (``bambu_mqtt.py:3647``). On a
reprint, the archive row is reused — so the stored ``archive.subtask_id`` is
still the value from the FIRST run. The earlier ``not archive.subtask_id``
guard at ``on_print_start`` skipped the rewrite, so the row kept the stale id.
Then, if MQTT reconnects mid-print (which it routinely does — network blips,
printer reboots, Bambuddy restarts), ``reconcile_stale_active_prints`` (#1542)
compares the printer's live subtask_id against the stored one, sees a
mismatch, and synthesises a "missed PRINT COMPLETE" → bogus Print Stopped
notification while the print keeps running.
The fix: update ``archive.subtask_id`` whenever the new effective id differs
from the stored one, not only when the stored one is empty.
"""
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from backend.app.core.config import settings as app_settings
from backend.app.main import (
_active_prints,
_expected_print_creators,
_expected_print_registered_at,
_expected_prints,
_print_ams_mappings,
_timelapse_baselines,
register_expected_print,
)
@pytest.fixture(autouse=True)
def _clear_dicts():
_expected_prints.clear()
_expected_print_registered_at.clear()
_expected_print_creators.clear()
_print_ams_mappings.clear()
_active_prints.clear()
_timelapse_baselines.clear()
yield
_expected_prints.clear()
_expected_print_registered_at.clear()
_expected_print_creators.clear()
_print_ams_mappings.clear()
_active_prints.clear()
_timelapse_baselines.clear()
def _patches():
return (
patch("backend.app.main.async_session"),
patch("backend.app.main.notification_service"),
patch("backend.app.main.smart_plug_manager"),
patch("backend.app.main.ws_manager"),
patch("backend.app.main.printer_manager"),
patch("backend.app.main.mqtt_relay"),
patch("backend.app.main._record_energy_start", new_callable=AsyncMock),
patch("backend.app.main._load_objects_from_archive"),
patch("backend.app.main._store_spoolman_print_data", new_callable=AsyncMock),
patch("backend.app.main._send_print_start_notification", new_callable=AsyncMock),
patch(
"backend.app.main._list_timelapse_videos",
new=AsyncMock(return_value=([], "/timelapse")),
),
)
def _build_mocks(mock_printer, mock_archive):
def execute_router(stmt, *args, **kwargs):
sql = str(stmt).lower()
if "from printers" in sql or "from printer " in sql:
return MagicMock(
scalar_one_or_none=MagicMock(return_value=mock_printer),
scalars=MagicMock(return_value=MagicMock(all=MagicMock(return_value=[mock_printer]))),
)
if "from print_archives" in sql or "from print_archive" in sql:
return MagicMock(
scalar_one_or_none=MagicMock(return_value=mock_archive),
scalars=MagicMock(return_value=MagicMock(all=MagicMock(return_value=[mock_archive]))),
)
return MagicMock(
scalar_one_or_none=MagicMock(return_value=None),
scalars=MagicMock(return_value=MagicMock(all=MagicMock(return_value=[]))),
)
mock_session = AsyncMock()
mock_session.__aenter__ = AsyncMock(return_value=mock_session)
mock_session.__aexit__ = AsyncMock()
mock_session.execute = AsyncMock(side_effect=execute_router)
mock_session.commit = AsyncMock()
return mock_session
def _make_archive(*, archive_id: int, stored_subtask_id: str | None):
mock_archive = MagicMock()
mock_archive.id = archive_id
mock_archive.filename = "Ikea-drybox_silicabox.3mf"
mock_archive.subtask_id = stored_subtask_id
mock_archive.print_time_seconds = None
mock_archive.created_by_id = None
mock_archive.printer_id = 1
mock_archive.print_name = "Ikea-drybox_silicabox"
mock_archive.status = "archived"
mock_archive.file_path = f"archives/{archive_id}/Ikea-drybox_silicabox.3mf"
mock_archive.energy_start_kwh = None
mock_archive.timelapse_path = None
return mock_archive
def _make_printer():
mock_printer = MagicMock()
mock_printer.id = 1
mock_printer.auto_archive = True
mock_printer.external_camera_enabled = False
mock_printer.external_camera_url = None
mock_printer.name = "TestP1S"
return mock_printer
async def _drive(tmp_path, mock_archive, mqtt_subtask_id: str | None):
"""Drive ``on_print_start`` with a print-start payload carrying the
given ``subtask_id`` (the printer-echoed id at PRINT START — set by the
queue dispatcher's fresh ``submission_id``)."""
mock_printer = _make_printer()
register_expected_print(1, mock_archive.filename, archive_id=mock_archive.id, ams_mapping=None)
mock_session = _build_mocks(mock_printer, mock_archive)
(
async_session_p,
notif_p,
plug_p,
ws_p,
pm_p,
relay_p,
_energy,
_load_obj,
_store_spoolman,
_send_start,
_list_tl,
) = _patches()
with (
async_session_p as mock_session_maker,
notif_p as mock_notif,
plug_p as mock_plug,
ws_p as mock_ws,
pm_p as mock_pm,
relay_p as mock_relay,
_energy,
_load_obj,
_store_spoolman,
_send_start,
_list_tl,
patch.object(app_settings, "base_dir", tmp_path),
):
mock_session_maker.return_value = mock_session
mock_notif.on_print_start = AsyncMock()
mock_plug.on_print_start = AsyncMock()
mock_ws.send_print_start = AsyncMock()
mock_ws.send_archive_updated = AsyncMock()
mock_relay.on_print_start = AsyncMock()
mock_pm.get_printer = MagicMock(return_value=MagicMock(name="Test", serial_number="TEST123"))
# last_dispatch_subtask_id fallback shouldn't fire — MQTT carried one.
mock_pm.get_client = MagicMock(return_value=MagicMock(last_dispatch_subtask_id=None))
from backend.app.main import on_print_start
await on_print_start(
1,
{
"filename": mock_archive.filename,
"subtask_name": mock_archive.print_name,
"raw_data": {"subtask_id": mqtt_subtask_id} if mqtt_subtask_id is not None else {},
},
)
@pytest.mark.asyncio
async def test_reprint_updates_stale_subtask_id(tmp_path):
"""The #1807 case: archive stored an OLD subtask_id from the first run.
On reprint dispatch the printer echoes a fresh one — the archive's
stored id must be rewritten so the reconciler doesn't flag the live
print as stale on next MQTT reconnect."""
archive = _make_archive(archive_id=31, stored_subtask_id="1844213296")
await _drive(tmp_path, archive, mqtt_subtask_id="2103771517")
assert archive.subtask_id == "2103771517", (
"expected-archive reprint promotion must update archive.subtask_id to the "
"new dispatch id; leaving the old value lets reconcile_stale_active_prints "
"synthesise a bogus PRINT COMPLETE on the next MQTT reconnect (#1807)"
)
@pytest.mark.asyncio
async def test_first_run_still_sets_subtask_id(tmp_path):
"""Regression guard for the previously-correct first-run path: an
archive with no stored subtask_id must still have it written on the
first MQTT-echoed PRINT START."""
archive = _make_archive(archive_id=99, stored_subtask_id=None)
await _drive(tmp_path, archive, mqtt_subtask_id="2103771517")
assert archive.subtask_id == "2103771517"
@pytest.mark.asyncio
async def test_stable_push_does_not_rewrite(tmp_path):
"""The original `not archive.subtask_id` guard's intent was to avoid
rewriting on every push that carries the same id. The inequality check
preserves that no-op behaviour: same id in, no rewrite."""
archive = _make_archive(archive_id=15, stored_subtask_id="2103771517")
# Replace the bare attribute with a MagicMock so we can detect any write,
# not just observe the post-call value (which would match even on a
# spurious "store the same value back" rewrite).
initial = archive.subtask_id
await _drive(tmp_path, archive, mqtt_subtask_id="2103771517")
assert archive.subtask_id == initial
backend/tests/unit/test_spoolman_no3mf_remain_fallback.py
+436
View File
@@ -0,0 +1,436 @@
"""AMS remain%-delta fallback for the no-3MF Spoolman path (#1820).
When a Bambu print starts without leaving a retrievable .gcode.3mf on the
printer (subtask_name='名称未設定'/'Untitled'), Bambuddy creates a
fallback archive with no 3MF on disk. Before this fix the Spoolman
tracking row was never created, so the print silently didn't decrement
the spool weight. This is the Spoolman mirror of usage_tracker's Path 2
fallback (already in place for the internal-inventory side).
"""
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from backend.app.services.spoolman_tracking import (
_snapshot_tray_remain,
store_print_data,
)
class TestSnapshotTrayRemain:
def test_captures_valid_remain(self):
raw = {
"ams": [
{
"id": 0,
"tray": [
{"id": 0, "tray_uuid": "AAAA", "remain": 75},
{"id": 1, "tray_uuid": "BBBB", "remain": 30},
],
}
]
}
snap = _snapshot_tray_remain(raw)
assert snap == {
"0-0": {"remain": 75, "tray_uuid": "AAAA"},
"0-1": {"remain": 30, "tray_uuid": "BBBB"},
}
def test_skips_invalid_remain(self):
"""remain=-1 means the AMS hasn't read the spool; a delta would be
meaningless. Skip those slots — usage_tracker does the same
(line 309)."""
raw = {
"ams": [
{
"id": 0,
"tray": [
{"id": 0, "tray_uuid": "AAAA", "remain": 75},
{"id": 1, "tray_uuid": "BBBB", "remain": -1},
{"id": 2, "tray_uuid": "CCCC", "remain": 150},
],
}
]
}
snap = _snapshot_tray_remain(raw)
assert set(snap.keys()) == {"0-0"}
def test_captures_vt_tray(self):
"""External (VT) spool gets ams_id=255, tray_id=vt_id-254."""
raw = {"ams": [], "vt_tray": [{"id": 254, "tray_uuid": "EEEE", "remain": 50}]}
snap = _snapshot_tray_remain(raw)
assert snap == {"255-0": {"remain": 50, "tray_uuid": "EEEE"}}
def test_empty_when_no_ams_data(self):
assert _snapshot_tray_remain({}) == {}
def test_handles_missing_uuid(self):
raw = {"ams": [{"id": 0, "tray": [{"id": 0, "remain": 80}]}]}
snap = _snapshot_tray_remain(raw)
assert snap == {"0-0": {"remain": 80, "tray_uuid": ""}}
class TestStorePrintDataNo3mf:
"""store_print_data must create an ActivePrintSpoolman row even when
no 3MF is available, populating tray_remain_start so report_usage can
write a remain-delta at completion (#1820)."""
@pytest.mark.asyncio
async def test_creates_row_with_snapshot_when_no_3mf(self):
db = AsyncMock()
# No queue lookup for the no-3MF branch — only the DELETE.
delete_result = MagicMock()
db.execute = AsyncMock(side_effect=[delete_result])
db.add = MagicMock()
db.commit = AsyncMock()
printer_manager = MagicMock()
printer_manager.get_status.return_value = SimpleNamespace(
raw_data={
"ams": [
{
"id": 0,
"tray": [
{"id": 0, "tray_uuid": "AAAA", "tag_uid": "11", "tray_type": "PLA", "remain": 80},
{"id": 1, "tray_uuid": "BBBB", "tag_uid": "22", "tray_type": "PLA", "remain": 20},
],
}
]
}
)
mock_settings = MagicMock()
mock_path = MagicMock()
mock_path.exists.return_value = False # no 3MF — the #1820 case
mock_settings.base_dir.__truediv__.return_value = mock_path
with (
patch("backend.app.services.spoolman_tracking.app_settings", mock_settings),
patch("backend.app.api.routes.settings.get_setting", AsyncMock(return_value="true")),
):
await store_print_data(
printer_id=1,
archive_id=42,
file_path="", # fallback-archive file_path
db=db,
printer_manager=printer_manager,
)
db.add.assert_called_once()
tracking = db.add.call_args.args[0]
assert tracking.filament_usage is None
assert tracking.tray_remain_start == {
"0-0": {"remain": 80, "tray_uuid": "AAAA"},
"0-1": {"remain": 20, "tray_uuid": "BBBB"},
}
@pytest.mark.asyncio
async def test_no_row_when_no_3mf_and_no_remain_data(self):
"""If the AMS has no slot with valid remain either (e.g. printer
offline at print start), there's nothing to track. Don't create
a row that contributes no value."""
db = AsyncMock()
db.execute = AsyncMock()
db.add = MagicMock()
db.commit = AsyncMock()
printer_manager = MagicMock()
printer_manager.get_status.return_value = SimpleNamespace(
raw_data={"ams": [{"id": 0, "tray": [{"id": 0, "remain": -1}]}]}
)
mock_settings = MagicMock()
mock_path = MagicMock()
mock_path.exists.return_value = False
mock_settings.base_dir.__truediv__.return_value = mock_path
with (
patch("backend.app.services.spoolman_tracking.app_settings", mock_settings),
patch("backend.app.api.routes.settings.get_setting", AsyncMock(return_value="true")),
):
await store_print_data(
printer_id=1,
archive_id=42,
file_path="",
db=db,
printer_manager=printer_manager,
)
db.add.assert_not_called()
@pytest.mark.asyncio
async def test_3mf_path_also_captures_snapshot(self):
"""The remain snapshot is captured ALWAYS, not just for no-3MF.
That lets report_usage fall back per-slot when 3MF coverage is
partial — same shape as usage_tracker.on_print_complete which
runs Path 1 (3MF) and Path 2 (remain delta) for unhandled slots."""
db = AsyncMock()
queue_item = SimpleNamespace(ams_mapping=None, plate_id=None)
queue_result = MagicMock()
queue_result.scalar_one_or_none.return_value = queue_item
delete_result = MagicMock()
db.execute = AsyncMock(side_effect=[queue_result, delete_result])
db.add = MagicMock()
db.commit = AsyncMock()
printer_manager = MagicMock()
printer_manager.get_status.return_value = SimpleNamespace(
raw_data={"ams": [{"id": 0, "tray": [{"id": 0, "tray_uuid": "AAAA", "tray_type": "PLA", "remain": 90}]}]}
)
mock_settings = MagicMock()
mock_path = MagicMock()
mock_path.exists.return_value = True
mock_settings.base_dir.__truediv__.return_value = mock_path
with (
patch("backend.app.services.spoolman_tracking.app_settings", mock_settings),
patch("backend.app.api.routes.settings.get_setting", AsyncMock(return_value="true")),
patch(
"backend.app.utils.threemf_tools.extract_filament_usage_from_3mf",
return_value=[{"slot_id": 1, "used_g": 10.0, "type": "PLA", "color": "#000000"}],
),
patch("backend.app.utils.threemf_tools.extract_layer_filament_usage_from_3mf", return_value=None),
patch("backend.app.utils.threemf_tools.extract_filament_properties_from_3mf", return_value={}),
):
await store_print_data(
printer_id=1,
archive_id=42,
file_path="archives/test.3mf",
db=db,
printer_manager=printer_manager,
)
db.add.assert_called_once()
tracking = db.add.call_args.args[0]
assert tracking.filament_usage == [{"slot_id": 1, "used_g": 10.0, "type": "PLA", "color": "#000000"}]
assert tracking.tray_remain_start == {"0-0": {"remain": 90, "tray_uuid": "AAAA"}}
class TestReportUsageRemainDelta:
"""report_usage must write a per-slot remain-delta when filament_usage
is missing (no-3MF print), gated on a resolvable Spoolman spool and a
sane current remain%."""
@pytest.mark.asyncio
async def test_remain_delta_writes_to_resolved_spool(self):
"""Print started at remain=80% on a 1000g filament, finished at 60%.
Delta = 20% × 1000g = 200g."""
from backend.app.services.spoolman_tracking import report_usage
tracking = SimpleNamespace(
filament_usage=None,
ams_trays={"0": {"tray_uuid": "AAAA", "tag_uid": "11", "tray_type": "PLA"}},
slot_to_tray=None,
tray_remain_start={"0-0": {"remain": 80, "tray_uuid": "AAAA"}},
)
# Mock db.execute().scalar_one_or_none() -> 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()
client = AsyncMock()
client.get_spool = AsyncMock(return_value={"id": 7, "filament": {"weight": 1000.0, "color_hex": "00FF00"}})
client.use_spool = AsyncMock()
printer_manager = MagicMock()
printer_manager.get_status.return_value = SimpleNamespace(
raw_data={"ams": [{"id": 0, "tray": [{"id": 0, "tray_uuid": "AAAA", "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(
"backend.app.services.spoolman_tracking._resolve_spool_id_via_slot_assignment",
AsyncMock(return_value=7),
),
patch("backend.app.services.printer_manager.printer_manager", printer_manager),
):
await report_usage(printer_id=1, archive_id=42)
client.use_spool.assert_awaited_once_with(7, 200.0)
@pytest.mark.asyncio
async def test_remain_delta_skips_swapped_spool(self):
"""tray_uuid changed between start and completion → user replaced
the spool mid-print. We don't know how much went to each side; skip
rather than mis-charge."""
from backend.app.services.spoolman_tracking import report_usage
tracking = SimpleNamespace(
filament_usage=None,
ams_trays={"0": {"tray_uuid": "AAAA"}},
slot_to_tray=None,
tray_remain_start={"0-0": {"remain": 80, "tray_uuid": "AAAA"}},
)
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()
client = AsyncMock()
client.use_spool = AsyncMock()
printer_manager = MagicMock()
printer_manager.get_status.return_value = SimpleNamespace(
# tray_uuid changed -> swap detected
raw_data={"ams": [{"id": 0, "tray": [{"id": 0, "tray_uuid": "CCCC", "remain": 50}]}]}
)
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._resolve_spool_id_via_slot_assignment",
AsyncMock(return_value=7),
),
patch("backend.app.services.printer_manager.printer_manager", printer_manager),
):
await report_usage(printer_id=1, archive_id=42)
client.use_spool.assert_not_called()
@pytest.mark.asyncio
async def test_remain_delta_skips_slot_handled_by_3mf(self):
"""Mixed coverage: 3MF carried slot 1 (=global tray 0). Remain
delta on the same physical slot must not double-charge it."""
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", "tray_type": "PLA"}},
slot_to_tray=None,
tray_remain_start={"0-0": {"remain": 80, "tray_uuid": "AAAA"}},
)
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()
client = AsyncMock()
client.use_spool = AsyncMock()
client.get_spool = AsyncMock()
printer_manager = MagicMock()
printer_manager.get_status.return_value = SimpleNamespace(
raw_data={"ams": [{"id": 0, "tray": [{"id": 0, "tray_uuid": "AAAA", "remain": 60}]}]}
)
# Make the 3MF path resolve to a spool too, so it actually writes.
async def fake_report_slots(_client, items, *args, **kwargs):
for _slot_id, grams in items:
if grams > 0:
await _client.use_spool(99, grams)
return 1
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._report_spool_usage_for_slots",
AsyncMock(side_effect=fake_report_slots),
),
patch("backend.app.services.printer_manager.printer_manager", printer_manager),
):
await report_usage(printer_id=1, archive_id=42)
# Only the 3MF path called use_spool. get_spool (remain-delta path)
# was never reached because slot 0 was already in the handled set.
client.use_spool.assert_awaited_once_with(99, 50.0)
client.get_spool.assert_not_called()
class TestPartialUsageRemainDelta:
"""cleanup_tracking → _report_partial_usage must also write the
remain-delta for ABORTED no-3MF prints — same shape as the completion
path, otherwise aborts of "Untitled" prints stay silent."""
@pytest.mark.asyncio
async def test_aborted_no_3mf_writes_remain_delta(self):
"""Aborted at remain=70% from start of 90% on a 1000g spool.
Delta = 20% × 1000g = 200g — must be written even though no
3MF / layer data is available."""
from backend.app.services.spoolman_tracking import _report_partial_usage
tracking = SimpleNamespace(
archive_id=99,
filament_usage=None,
layer_usage=None,
filament_properties=None,
ams_trays={"0": {"tray_uuid": "AAAA"}},
slot_to_tray=None,
tray_remain_start={"0-0": {"remain": 90, "tray_uuid": "AAAA"}},
)
client = AsyncMock()
client.get_spool = AsyncMock(return_value={"id": 7, "filament": {"weight": 1000.0}})
client.use_spool = AsyncMock()
printer_manager = MagicMock()
printer_manager.get_status.return_value = SimpleNamespace(
raw_data={"ams": [{"id": 0, "tray": [{"id": 0, "tray_uuid": "AAAA", "remain": 70}]}]},
layer_num=42,
total_layers=100,
)
with (
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._resolve_spool_id_via_slot_assignment",
AsyncMock(return_value=7),
),
patch("backend.app.services.printer_manager.printer_manager", printer_manager),
):
await _report_partial_usage(printer_id=1, tracking=tracking)
client.use_spool.assert_awaited_once_with(7, 200.0)
class _AsyncCtx:
"""Tiny async-context shim returning a pre-built db mock; mirrors
async_session()'s ``async with`` interface."""
def __init__(self, db):
self._db = db
async def __aenter__(self):
return self._db
async def __aexit__(self, *_):
return False