mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-09-30 03:01:21 +02:00
307 lines
12 KiB
Python
307 lines
12 KiB
Python
"""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
|
||
label_weight: int | None
|
||
core_weight: int | None
|
||
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
|
||
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")
|
||
|
||
|
||
_COLOR_HEX_RE = re.compile(r"^[0-9A-Fa-f]{6}$")
|
||
_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
|
||
|
||
|
||
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 ""
|
||
|
||
|
||
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-char hex; fall back to neutral grey for invalid values
|
||
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 + "FF"
|
||
|
||
label_weight: int = _safe_int(filament.get("weight"), 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.
|
||
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)
|
||
else:
|
||
used_weight = real_used_weight
|
||
weight_used_baseline = 0.0
|
||
|
||
# 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,
|
||
"brand": vendor.get("name") or None,
|
||
"label_weight": label_weight,
|
||
"core_weight": _safe_int(
|
||
spool.get("spool_weight") if spool.get("spool_weight") is not None else filament.get("spool_weight"), 250
|
||
),
|
||
"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,
|
||
"cost_per_kg": _safe_optional_float(spool.get("price")),
|
||
"storage_location": spool.get("location") or None,
|
||
"location_id": None,
|
||
"k_profiles": [],
|
||
}
|