Files
maziggy 2f7f17a891 Use the Spoolman spool's own initial_weight as its label weight (#3194)
Spoolman keeps a full spool's net weight on the spool (initial_weight)
and falls back to the filament's weight, but Bambuddy read only the
filament, so a 250 g spool of a 1000 g filament showed and synced as
1000 g. The list, weigh, AMS sync, SpoolBuddy scale, remain-% tracking,
fill bar and cost now share one lookup. Create writes initial_weight,
and a label-weight edit writes it instead of patching or duplicating the
filament. The spool form's cost per kg is converted at the spool's size
to and from Spoolman's per-spool price.
2026-09-30 09:12:45 +02:00

522 lines
22 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""Pure helper functions for Spoolman spool mapping.
No heavy dependencies — importable in unit tests without the full backend stack.
"""
from __future__ import annotations
import json
import logging
import math
import re
from typing import Any
from typing_extensions import TypedDict
from backend.app.api.routes._url_safety import assert_safe_lan_service_url
logger = logging.getLogger(__name__)
class MappedSpoolFields(TypedDict):
"""Full shape of the dict returned by _map_spoolman_spool (InventorySpool-compatible)."""
id: int
material: str | None
subtype: str | None
brand: str | None
color_name: str | None
color_name_is_synthesized: bool
rgba: str | None
extra_colors: str | None
effect_type: None
label_weight: int | None
core_weight: int | None
core_weight_is_inherited: bool
core_weight_catalog_id: None
weight_used: float | None
weight_used_baseline: float | None
weight_locked: bool
last_scale_weight: None
last_weighed_at: None
slicer_filament: None
slicer_filament_name: str | None
nozzle_temp_min: int | None
nozzle_temp_max: None
note: str | None
added_full: None
last_used: str | None
encode_time: str | None
tag_uid: str | None
tray_uuid: str | None
data_origin: str | None
tag_type: str | None
archived_at: str | None
created_at: str | None # None when Spoolman spool has no registered timestamp
updated_at: str | None
cost_per_kg: float | None
# Spoolman's native filament.article_number, surfaced as the internal
# material number (#2870). Read-only in Spoolman mode — the number is
# filament-level there and maintained in Spoolman itself.
material_number: str | None
storage_location: str | None
location_id: int | None
k_profiles: list[Any]
class NormalizedVendorRef(TypedDict):
"""Vendor reference embedded in a NormalizedFilament."""
id: int
name: str
class NormalizedFilament(TypedDict):
"""Normalised Spoolman filament dict returned by the /filaments catalog endpoint."""
id: int
name: str
material: str | None
color_hex: str | None
color_name: str | None
weight: int | None
spool_weight: float | None
vendor: NormalizedVendorRef | None
def assert_safe_spoolman_url(url: str) -> None:
"""Raise ValueError if the Spoolman *url* should be blocked as an SSRF risk.
Thin wrapper over the shared LAN-service policy — see
``_url_safety.assert_safe_lan_service_url`` for what is and isn't
rejected, and why loopback/RFC-1918 are deliberately permitted (running
Spoolman on the same host or home LAN is THE normal topology).
Kept as a named function because the "Spoolman URL …" wording in its
errors is user-facing and asserted by existing tests.
"""
assert_safe_lan_service_url(url, label="Spoolman URL")
# Six characters, or eight when the filament carries an alpha byte. The write
# side stores eight only for genuinely translucent spools (#2912); rejecting
# them here turned every clear spool into neutral grey on read.
_COLOR_HEX_RE = re.compile(r"^[0-9A-Fa-f]{6}(?:[0-9A-Fa-f]{2})?$")
_TAG_HEX_RE = re.compile(r"^[0-9A-F]+$")
def _safe_int(value: object, fallback: int) -> int:
"""Convert value to int, returning fallback for None/NaN/Inf/non-numeric."""
try:
f = float(value) # type: ignore[arg-type]
if math.isfinite(f):
return int(f)
except (TypeError, ValueError):
pass
return fallback
def _safe_float(value: object, fallback: float) -> float:
"""Convert value to float, returning fallback for None/NaN/Inf/non-numeric."""
try:
f = float(value) # type: ignore[arg-type]
if math.isfinite(f):
return f
except (TypeError, ValueError):
pass
return fallback
def _safe_optional_float(value: object) -> float | None:
"""Convert value to finite float, or None if missing/NaN/Infinite/non-numeric.
Used for optional monetary fields (price) to prevent Infinity/NaN from
reaching JSON serialisation, which raises ValueError with allow_nan=False.
"""
if value is None:
return None
try:
f = float(value) # type: ignore[arg-type]
if math.isfinite(f):
return f
except (TypeError, ValueError):
pass
return None
# Used only when Spoolman has no empty-spool weight at any level.
SPOOLMAN_FALLBACK_TARE = 250.0
def spoolman_tare(spool: dict) -> tuple[float, str]:
"""The empty-spool weight of a Spoolman spool, and where it came from.
Spoolman resolves the tare as the spool's own ``spool_weight``, then the
filament's ``spool_weight``, then the vendor's ``empty_spool_weight``, and
its own ``/measure`` endpoint follows that order. Skipping the vendor
level made a spool whose tare lives only on its vendor weigh against
250 g instead (#3195). Every Spoolman-mode tare in Bambuddy goes through
here so the weigh endpoints and the displayed core weight cannot drift
apart again.
Returns ``(grams, source)`` with source one of ``"spool"``,
``"filament"``, ``"vendor"`` or ``"fallback"``. 0 is a real tare, not a
missing one; a value that is not a finite number counts as missing.
"""
filament = spool.get("filament") or {}
vendor = filament.get("vendor") or {}
for source, raw in (
("spool", spool.get("spool_weight")),
("filament", filament.get("spool_weight")),
("vendor", vendor.get("empty_spool_weight")),
):
value = _safe_optional_float(raw)
if value is not None:
return value, source
return SPOOLMAN_FALLBACK_TARE, "fallback"
def spoolman_net_weight(spool: dict) -> float | None:
"""The net filament weight of a full Spoolman spool (its label weight).
Spoolman keeps it on the spool as ``initial_weight`` and falls back to the
filament's catalogue ``weight``, so one filament can have spools of
different sizes. Reading only the filament made a 250 g spool of a 1000 g
filament look four times its size (#3194). Like Spoolman, an
``initial_weight`` of 0 counts as unset.
Returns None when neither level has a finite number (a bool is not
one); the filament's value is otherwise returned as stored, 0 included,
so callers keep their own handling of a filament without a weight.
"""
initial = _weight_or_none(spool.get("initial_weight"))
if initial is not None and initial > 0:
return initial
filament = spool.get("filament")
return _weight_or_none(filament.get("weight")) if isinstance(filament, dict) else None
def _weight_or_none(value: object) -> float | None:
# A bool is an int in Python; True must not read as a 1 g spool.
return None if isinstance(value, bool) else _safe_optional_float(value)
def spoolman_price_to_cost_per_kg(price: object, net_weight: float) -> float | None:
"""A Spoolman spool price as a price per kilogram.
Spoolman's ``price`` on a spool is what that spool cost, not a rate; the
spool form edits a rate, so the two are converted at the spool's net
weight. They agree only for 1000 g spools (#3194). Rounded to 4 places so
a round trip through the form doesn't drift by float noise.
"""
value = _safe_optional_float(price)
if value is None or net_weight <= 0:
return None
return round(value * 1000.0 / net_weight, 4)
def cost_per_kg_to_spoolman_price(cost_per_kg: float, net_weight: float) -> float:
"""Inverse of spoolman_price_to_cost_per_kg: the price of a spool of ``net_weight`` grams."""
return round(cost_per_kg * net_weight / 1000.0, 4)
def spoolman_price_weight(spool: dict) -> float:
"""The weight a Spoolman spool's price is converted at.
The spool's exact net weight, not the whole-gram label weight, so a
250.7 g spool is not re-priced by a no-op edit. A spool with no usable
weight keeps the old 1:1 reading of its price as a per-kg rate.
"""
net = spoolman_net_weight(spool)
return net if net is not None and net > 0 else 1000.0
def spoolman_cost_per_kg(spool: dict) -> float | None:
"""The spool form's cost per kg for a Spoolman spool (#3194)."""
return spoolman_price_to_cost_per_kg(spool.get("price"), spoolman_price_weight(spool))
def _extract_extra_str(extra: dict, key: str) -> str:
"""Extract a JSON-encoded string from a Spoolman extra dict.
Spoolman stores extra values as JSON-stringified text — a stored string
"GFL05" appears as `'"GFL05"'` (six chars including the quotes). This
unwraps that, returning the bare string. Returns "" for missing keys,
non-strings, or invalid JSON.
"""
raw = extra.get(key)
if not isinstance(raw, str):
return ""
try:
decoded = json.loads(raw)
except (json.JSONDecodeError, ValueError):
# Tolerate bare-string values written without JSON encoding.
return raw
return decoded if isinstance(decoded, str) else ""
# Spool.extra key holding the consumed-counter baseline: the value Spoolman's
# used_weight had when the user last pressed "Reset usage to 0". Displayed
# consumed is used_weight minus this, which is what internal mode has always
# done with its own weight_used_baseline column (#1644). See #2906.
BAMBU_WEIGHT_USED_BASELINE_KEY = "bambu_weight_used_baseline"
def _extract_extra_float(extra: dict, key: str) -> float | None:
"""Extract a JSON-encoded number from a Spoolman extra dict.
Same storage convention as :func:`_extract_extra_str` — Spoolman keeps
extra values as JSON text, so 250.0 is stored as ``"250.0"``. Returns
``None`` for missing keys, non-finite values, or anything that does not
decode to a number, so the caller can tell "no baseline recorded" from
"a baseline of zero".
"""
raw = extra.get(key)
if raw is None:
return None
if isinstance(raw, (int, float)) and not isinstance(raw, bool):
value = float(raw)
elif isinstance(raw, str):
try:
decoded = json.loads(raw)
except (json.JSONDecodeError, ValueError):
return None
if isinstance(decoded, str):
# Spoolman's extra fields default to field_type "text", which only
# accepts values that decode to a str, so numbers written through
# it arrive as '"263.0"' rather than '263.0'. Both spellings have
# to read back the same, or a baseline written by the reset would
# be invisible to the code that subtracts it.
try:
value = float(decoded)
except ValueError:
return None
elif isinstance(decoded, (int, float)) and not isinstance(decoded, bool):
value = float(decoded)
else:
return None
else:
return None
return value if math.isfinite(value) else None
def parse_spoolman_multi_colors(filament: dict) -> list[str]:
"""Spoolman's ``multi_color_hexes`` as a list of bare 6/8-char hex tokens.
Spoolman stores the extra stops of a gradient / dual / multi-colour
filament here, and writes the field as a comma-separated string in some
releases and a list in others -- both shapes are accepted. Tokens keep the
case they arrived in and lose any leading ``#``, which is the form
``Spool.extra_colors`` stores and ``parseStops`` on the client expects.
Shared with the label renderer rather than parsed twice: the two read the
same field for the same purpose, and a swatch on a printer card that
disagreed with the swatch on the printed label would be worse than either
being wrong on its own.
"""
raw = filament.get("multi_color_hexes")
if isinstance(raw, str):
tokens = raw.split(",")
elif isinstance(raw, list):
tokens = [str(token) for token in raw]
else:
return []
return [cleaned for token in tokens if (cleaned := token.strip().lstrip("#"))]
def _map_spoolman_spool(spool: dict) -> MappedSpoolFields:
"""Convert a raw Spoolman spool dict to the InventorySpool-compatible format.
Fields not supported by Spoolman (k_profiles, slicer_filament, …) are
returned as None / empty so the frontend can still render them without
errors. The ``data_origin`` field is set to ``"spoolman"`` so UI code can
distinguish these spools from local ones.
"""
raw_id = spool.get("id")
if raw_id is None:
raise ValueError("Spoolman spool is missing required 'id' field")
try:
spool_id: int = int(raw_id)
except (TypeError, ValueError):
raise ValueError(f"Spoolman spool 'id' is not a valid integer: {raw_id!r}")
if spool_id <= 0:
raise ValueError(f"Spoolman spool 'id' must be a positive integer, got {spool_id}")
filament: dict = spool.get("filament") or {}
if not filament:
logger.warning(
"Spoolman spool %s has no filament data — all filament fields will use defaults",
spool_id,
)
vendor: dict = filament.get("vendor") or {}
extra: dict = spool.get("extra") or {}
# RFID tag stored as JSON-encoded string in Spoolman extra.tag.
# 32-char hex → Bambu Lab tray UUID; 8–30-char hex → NFC tag UID.
# Accepting the full realistic UID range (4-byte = 8 chars, 7-byte = 14 chars,
# 10-byte = 20 chars) avoids silently dropping valid SpoolBuddy-written tags.
raw_tag: str = (extra.get("tag") or "").strip('"').upper()
_raw_is_hex = bool(_TAG_HEX_RE.match(raw_tag))
tag_uid = raw_tag if _raw_is_hex and 8 <= len(raw_tag) <= 30 else None
tray_uuid = raw_tag if _raw_is_hex and len(raw_tag) == 32 else None
# Subtype = filament name with material prefix stripped
material: str = (filament.get("material") or "").strip()
filament_name: str = (filament.get("name") or "").strip()
if material and filament_name.upper().startswith(material.upper()):
subtype: str | None = filament_name[len(material) :].strip() or None
else:
subtype = filament_name or None
# Colour: validate as 6- or 8-char hex; fall back to neutral grey for invalid
# values. An 8-char value already carries its alpha, so appending the opaque
# byte would push it to ten and lose the translucency it was stored to keep.
raw_color = (filament.get("color_hex") or "").upper().removeprefix("#")
color_hex: str = raw_color if _COLOR_HEX_RE.match(raw_color) else "808080"
rgba: str = color_hex if len(color_hex) == 8 else color_hex + "FF"
# Spoolman carries the extra stops but has no concept of a surface effect
# -- its only neighbouring field is `multi_color_direction`, which says how
# the stops are laid out, not that the filament is silk or glitter. So a
# Spoolman spool can render its gradient and never an effect overlay, and
# `effect_type` is pinned to None rather than guessed at.
extra_stops = parse_spoolman_multi_colors(filament)
extra_colors: str | None = ",".join(extra_stops) if extra_stops else None
label_weight: int = _safe_int(spoolman_net_weight(spool), 1000)
real_used_weight: float = _safe_float(spool.get("used_weight"), 0.0)
# Parity with internal mode (#1390): the InventorySpool shape lets the
# frontend compute `remaining = label_weight - weight_used` and
# `consumed = weight_used - weight_used_baseline`. Map Spoolman's two
# independent fields (used_weight, remaining_weight) onto that shape:
# weight_used = label_weight - remaining_weight (so remaining matches)
# baseline = weight_used - used_weight (so consumed matches)
# When remaining_weight is unset (legacy spools, or filament linked but
# never primed), fall back to the old behaviour: weight_used =
# used_weight, baseline = 0.
#
# A "Reset usage to 0" records the then-current used_weight under
# BAMBU_WEIGHT_USED_BASELINE_KEY in spool.extra and touches nothing else,
# so displayed consumed is used_weight minus that. It is folded into the
# baseline here because the frontend's arithmetic is fixed: it always shows
# `weight_used - weight_used_baseline`. The reset used to PATCH Spoolman's
# used_weight to 0 instead, and Spoolman recomputes remaining from initial
# minus used, so the spool jumped back to full (#2906).
# `or 0.0` would collapse None and 0.0, which is the one distinction
# _extract_extra_float exists to preserve; spell the absent case out.
# Clamp the low end too: the write side already refuses a negative
# baseline, and without the same rule here a negative value stored by hand
# would be subtracted, inflating the displayed counter. min() below only
# bounds the other end.
stored = _extract_extra_float(extra, BAMBU_WEIGHT_USED_BASELINE_KEY)
stored_baseline = 0.0 if stored is None else max(0.0, stored)
remaining_raw = spool.get("remaining_weight")
if remaining_raw is not None:
remaining_weight: float = _safe_float(remaining_raw, 0.0)
used_weight: float = max(0.0, float(label_weight) - remaining_weight)
weight_used_baseline: float = max(0.0, used_weight - real_used_weight + stored_baseline)
else:
used_weight = real_used_weight
weight_used_baseline = stored_baseline # already clamped to >= 0 above
# Never let the displayed counter read negative. used_weight can fall below
# the baseline after a reset without anyone touching it: the AMS sync
# writes remaining_weight from the tray percentage, and Spoolman derives
# used_weight from that. The counter then reads 0 until consumption climbs
# past the baseline again. Internal mode behaves the same way, and it is
# far better than before #2906, when the next sync undid the reset outright.
weight_used_baseline = min(weight_used_baseline, used_weight)
# Archived state – Spoolman uses a boolean ``archived`` field
archived: bool = spool.get("archived", False)
archived_at: str | None = None
if archived:
archived_at = spool.get("last_used") or spool.get("registered") or None
created_at: str | None = spool.get("registered") or None
# Spoolman has no `color_name` field on Filament — confirmed against the
# FilamentUpdateParameters schema in 0.23.1: name/vendor_id/material/price/
# density/diameter/weight/spool_weight/article_number/comment/extruder_temp/
# bed_temp/color_hex/multi_color_hexes/multi_color_direction/external_id/
# extra, no color_name (#1357). The previous attempt (b8e350c3) was
# PATCHing a key Spoolman silently discards, which is why color_name
# never actually persisted from the user's edits.
#
# We persist it ourselves under spool.extra.bambu_color_name (JSON-encoded
# string, same pattern as bambu_slicer_filament). Read order:
# 1. spool.extra.bambu_color_name (the canonical store)
# 2. filament.color_name (forward-compat — picks up the value if a
# future Spoolman release adds the field, or if an admin populated
# it via a custom extra-field they registered themselves)
# 3. subtype (synth fallback so the inventory list isn't a sea of
# "Unknown color" entries on installs with neither field set)
#
# color_name_is_synthesized = True only when we fell back to subtype.
# The edit form uses it to leave the input blank, so the user doesn't
# round-trip the synth value back as if they had set it.
extra_color_name = _extract_extra_str(extra, "bambu_color_name") or None
stored_color_name = extra_color_name or (filament.get("color_name") or None)
color_name: str | None = stored_color_name or subtype or None
color_name_is_synthesized: bool = stored_color_name is None and color_name is not None
nozzle_temp_raw = filament.get("settings_extruder_temp")
nozzle_temp_min: int | None = _safe_int(nozzle_temp_raw, 0) or None
return {
"id": spool_id,
"material": material,
"subtype": subtype,
"color_name": color_name,
"color_name_is_synthesized": color_name_is_synthesized,
"rgba": rgba,
"extra_colors": extra_colors,
"effect_type": None,
"brand": vendor.get("name") or None,
"label_weight": label_weight,
"core_weight": _safe_int(spoolman_tare(spool)[0], 250),
# True when the spool has no spool_weight of its own and core_weight is
# the filament type's, the vendor's or the 250 g fallback. The spool form needs it
# to copy a spool without dropping an own tare or stamping an
# inherited one (#2908).
"core_weight_is_inherited": spool.get("spool_weight") is None,
"core_weight_catalog_id": None,
"weight_used": used_weight,
"weight_used_baseline": weight_used_baseline,
"weight_locked": False,
"last_scale_weight": None,
"last_weighed_at": None,
# BambuStudio slicer preset — Spoolman has no native field, so the
# update endpoint persists these under bambu_slicer_filament[_name]
# in the spool's extra dict. Values are JSON-encoded strings; an
# empty string ("") means cleared. Falls back to Spoolman's
# filament_name for slicer_filament_name when nothing is stored.
"slicer_filament": (_extract_extra_str(extra, "bambu_slicer_filament") or None),
"slicer_filament_name": (_extract_extra_str(extra, "bambu_slicer_filament_name") or (filament_name or None)),
"nozzle_temp_min": nozzle_temp_min,
"nozzle_temp_max": None,
"note": spool.get("comment") or None,
"added_full": None,
"last_used": spool.get("last_used"),
# encode_time semantics differ: local records NFC write time; Spoolman first_used
# records first print use — different events; using first_used as best available proxy.
"encode_time": spool.get("first_used"),
"tag_uid": tag_uid,
"tray_uuid": tray_uuid,
"data_origin": "spoolman",
"tag_type": "spoolman",
"archived_at": archived_at,
"created_at": created_at,
# Spoolman has no updated_at field; use registered timestamp as best available proxy
"updated_at": created_at,
# Spoolman's spool price is what the whole spool cost (#3194).
"cost_per_kg": spoolman_cost_per_kg(spool),
# Spoolman's filament.article_number maps 1:1 onto the internal
# material number (#2870): both identify the purchasable product.
# Trimmed for the same reason the schema validator trims the internal
# one — the filter chip builds its options from trimmed values and
# matches exactly, so a padded number would list and match nothing.
"material_number": ((filament.get("article_number") or "").strip() or None),
"storage_location": spool.get("location") or None,
"location_id": None,
"k_profiles": [],
}