public_git/bambuddy
1
0
Fork 0
mirror of https://github.com/maziggy/bambuddy.git synced 2026-10-22 03:49:18 +02:00
Code Issues Packages Projects Releases Wiki Activity

Use the Spoolman spool's own initial_weight as its label weight (#3194)

Browse Source 
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.
This commit is contained in:
maziggy
2026-09-30 09:12:45 +02:00
parent c145b3ebc1
commit 2f7f17a891
12 changed files with 642 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/api/routes/_spoolman_helpers.py
+63 -2
View File
@@ -176,6 +176,66 @@ def spoolman_tare(spool: dict) -> tuple[float, str]:
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.
@@ -322,7 +382,7 @@ def _map_spoolman_spool(spool: dict) -> MappedSpoolFields:
extra_stops = parse_spoolman_multi_colors(filament)
extra_colors: str | None = ",".join(extra_stops) if extra_stops else None
label_weight: int = _safe_int(filament.get("weight"), 1000)
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
@@ -447,7 +507,8 @@ def _map_spoolman_spool(spool: dict) -> MappedSpoolFields:
"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")),
# 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
backend/app/api/routes/spoolbuddy.py
+2 -3
View File
@@ -891,7 +891,7 @@ async def update_spool_weight(
update while the Spoolman row the user is actually looking at stayed
unchanged (#1530). Mirrors the routing already used by ``nfc/tag-scanned``.
"""
from backend.app.api.routes._spoolman_helpers import _safe_float, spoolman_tare
from backend.app.api.routes._spoolman_helpers import _safe_float, spoolman_net_weight, spoolman_tare
from backend.app.models.spool import Spool
sm_client = await _get_spoolman_client_or_none(db)
@@ -920,7 +920,6 @@ async def update_spool_weight(
async with _translate_spoolbuddy_errors():
sm_spool = await sm_client.get_spool(req.spool_id)
filament = sm_spool.get("filament") or {}
core_weight, tare_source = spoolman_tare(sm_spool)
spool_weight_warning: str | None = None
if tare_source == "fallback":
@@ -932,7 +931,7 @@ async def update_spool_weight(
"spool_weight_not_set: Spoolman spool, filament and vendor have no empty-spool weight configured; "
"weight estimate uses 250g fallback"
)
label_weight = _safe_float(filament.get("weight"), 1000.0)
label_weight = _safe_float(spoolman_net_weight(sm_spool), 1000.0)
remaining_weight = max(0.0, req.weight_grams - core_weight)
async with _translate_spoolbuddy_errors():
backend/app/api/routes/spoolman.py
+4 -3
View File
@@ -10,7 +10,7 @@ from sqlalchemy import delete, select, text
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload
from backend.app.api.routes._spoolman_helpers import _map_spoolman_spool
from backend.app.api.routes._spoolman_helpers import _map_spoolman_spool, spoolman_net_weight
from backend.app.api.routes.spoolman_inventory import _clear_stale_tag_links
from backend.app.core.auth import RequirePermissionIfAuthEnabled
from backend.app.core.database import get_db
@@ -804,11 +804,12 @@ async def get_linked_spools(
# Remove quotes if present (JSON encoded string)
clean_tag = tag.strip('"').upper()
if clean_tag:
filament = spool.get("filament") or {}
linked[clean_tag] = {
"id": spool["id"],
"remaining_weight": spool.get("remaining_weight"),
"filament_weight": filament.get("weight"),
# The spool's own net weight, falling back to the
# filament's; the key predates initial_weight (#3194).
"filament_weight": spoolman_net_weight(spool),
}
return {"linked": linked}
backend/app/api/routes/spoolman_inventory.py
+59 -10
View File
@@ -29,6 +29,10 @@ from backend.app.api.routes._spoolman_helpers import (
_safe_int,
_safe_optional_float,
assert_safe_spoolman_url,
cost_per_kg_to_spoolman_price,
spoolman_cost_per_kg,
spoolman_net_weight,
spoolman_price_weight,
spoolman_tare,
)
from backend.app.core.auth import RequirePermissionIfAuthEnabled
@@ -258,6 +262,9 @@ def _raise_if_partial_failure(spools: list[dict], results: list, operation: str)
async def _apply_price_if_set(client: SpoolmanClient, spool: dict, cost_per_kg: float | None) -> tuple[dict, list[str]]:
"""Patch the spool price; return (updated_spool, warnings).
Spoolman's spool price is what the whole spool cost, so the per-kg rate is
converted at the net weight of the spool Spoolman just created (#3194).
Returns the original spool and a non-empty warnings list when the price
update fails, so the caller can return HTTP 207 instead of silently
discarding the price.
@@ -266,7 +273,9 @@ async def _apply_price_if_set(client: SpoolmanClient, spool: dict, cost_per_kg:
return spool, []
try:
async with _translate_spoolman_errors():
updated = await client.update_spool_full(spool["id"], price=cost_per_kg)
updated = await client.update_spool_full(
spool["id"], price=cost_per_kg_to_spoolman_price(cost_per_kg, spoolman_price_weight(spool))
)
return updated, []
except HTTPException as exc:
if exc.status_code >= 500:
@@ -580,6 +589,10 @@ async def create_spool(
comment=data.note or None,
location=storage_location or None,
spool_weight=(data.core_weight if "core_weight" in data.model_fields_set else None),
# Only a label weight the caller gave: left off, Spoolman uses
# the filament's weight, which is the right size for a caller
# that didn't say (#3194).
initial_weight=(float(data.label_weight) if "label_weight" in data.model_fields_set else None),
)
except HTTPException as exc:
if exc.status_code == 404 and data.spoolman_filament_id is not None:
@@ -670,6 +683,10 @@ async def bulk_create_spools(
comment=data.note or None,
location=storage_location or None,
spool_weight=(data.core_weight if "core_weight" in data.model_fields_set else None),
# Only a label weight the caller gave: left off, Spoolman uses
# the filament's weight, which is the right size for a caller
# that didn't say (#3194).
initial_weight=(float(data.label_weight) if "label_weight" in data.model_fields_set else None),
)
except (SpoolmanUnavailableError, SpoolmanClientError, SpoolmanNotFoundError) as exc:
logger.warning("Bulk spool creation: one spool failed: %s", exc)
@@ -749,7 +766,17 @@ async def update_spool(
# straight back off by `spoolman_color_hex` below, so the two paths landed on
# the same string and the append only obscured which shape was in hand (#2912).
rgba = data.rgba if data.rgba is not None else cur_color
label_weight = data.label_weight if data.label_weight is not None else int(cur_filament.get("weight") or 1000)
# The label weight is the spool's own net weight (initial_weight), not the
# filament's: one filament can have spools of different sizes (#3194).
# The form shows it in whole grams, so it only counts as changed when it
# differs from that; the exact stored value is otherwise left alone. A
# spool with no size at either level takes the one it is given: Spoolman
# refuses a remaining_weight until the spool has one.
cur_net_weight = spoolman_net_weight(current)
cur_has_size = cur_net_weight is not None and cur_net_weight > 0
cur_label_weight = _safe_int(cur_net_weight, 1000) or 1000
label_weight = data.label_weight if data.label_weight is not None else cur_label_weight
label_weight_changed = data.label_weight is not None and (not cur_has_size or data.label_weight != cur_label_weight)
# Default weight_used from the synthetic mapping (label - remaining) so an
# edit that doesn't touch the weight field preserves Spoolman's real
# remaining_weight after a "Reset usage to 0" — the previous code read
@@ -795,14 +822,16 @@ async def update_spool(
# alpha-only edit invisible so the change never lands (#2912).
cur_color_norm = spoolman_color_hex(cur_filament.get("color_hex")) or ""
cur_vendor_name = (cur_vendor.get("name") or "").strip()
cur_weight_int = int(cur_filament.get("weight") or 0)
# The label weight is not part of this: it is written to the spool's
# initial_weight below. Comparing it against the filament's weight PATCHed
# the filament for every spool of another size, or on a shared filament
# relinked the spool to a new duplicate one (#3194).
metadata_unchanged = (
cur_filament_id
and (cur_filament.get("name") or "").strip() == desired_name
and (cur_filament.get("material") or "").upper() == material.upper()
and cur_color_norm == (color_hex or "").upper()
and cur_vendor_name.lower() == ((brand or "").strip().lower())
and cur_weight_int == int(label_weight)
)
if metadata_unchanged:
@@ -820,7 +849,6 @@ async def update_spool(
"name": desired_name,
"material": material,
"color_hex": color_hex,
"weight": float(label_weight),
}
if brand:
vendor_id = await client.find_or_create_vendor(brand)
@@ -846,6 +874,25 @@ async def update_spool(
remaining = max(0.0, label_weight - weight_used)
# The spool's size lives on the spool (initial_weight), never the filament
# (#3194). A relink also sends the current size: Spoolman stamps the new
# filament's weight on a spool without an initial_weight of its own.
new_initial_weight: float | None = None
if label_weight_changed:
new_initial_weight = float(label_weight)
elif filament_id != cur_filament_id and cur_has_size:
new_initial_weight = cur_net_weight
# Spoolman's price is per spool, the form's per kg (#3194). An unchanged
# rate on an unchanged size is not written, so a no-op edit cannot move the
# price by rounding. On a resize the rate is kept, as in internal mode.
cur_cost_per_kg = spoolman_cost_per_kg(current)
cost_per_kg = data.cost_per_kg if data.cost_per_kg is not None else cur_cost_per_kg
new_price: float | None = None
if cost_per_kg is not None and (label_weight_changed or cost_per_kg != cur_cost_per_kg):
price_weight = float(label_weight) if label_weight_changed else spoolman_price_weight(current)
new_price = cost_per_kg_to_spoolman_price(cost_per_kg, price_weight)
# Tag removal: clear only the "tag" key so other custom Spoolman extra fields
# set outside Bambuddy are preserved.
tag_nulled = (
@@ -877,9 +924,12 @@ async def update_spool(
updated = await client.update_spool_full(
spool_id=spool_id,
filament_id=filament_id,
remaining_weight=remaining,
# Without a size Spoolman rejects the whole PATCH (HTTP 400),
# so a spool that still has none gets no remaining_weight.
remaining_weight=(remaining if new_initial_weight is not None or cur_has_size else None),
comment=note or "",
price=data.cost_per_kg,
price=new_price,
initial_weight=new_initial_weight,
extra=extra,
location=storage_location or None,
clear_location=storage_location_changed and not storage_location,
@@ -1202,8 +1252,7 @@ async def sync_spool_weight(
async with _translate_spoolman_errors():
updated = await client.update_spool_full(spool_id=spool_id, remaining_weight=remaining)
upd_filament = updated.get("filament") or {}
label_weight = _safe_int(upd_filament.get("weight"), 1000)
label_weight = _safe_int(spoolman_net_weight(updated), 1000)
weight_used = max(0.0, label_weight - remaining)
await ws_manager.broadcast({"type": "inventory_changed"})
return {"status": "ok", "weight_used": weight_used}
@@ -1409,7 +1458,7 @@ async def sync_spoolman_ams_weights(
skipped += 1
continue
label_weight = _safe_int((spool_dict.get("filament") or {}).get("weight"), 1000)
label_weight = _safe_int(spoolman_net_weight(spool_dict), 1000)
if label_weight <= 0:
logger.debug("Spoolman AMS sync: spool %d has no label_weight, skipping", assignment.spoolman_spool_id)
skipped += 1
backend/app/services/filament_deficit.py
+7 -7
View File
@@ -38,7 +38,7 @@ from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload
from backend.app.api.routes._spoolman_helpers import _map_spoolman_spool
from backend.app.api.routes._spoolman_helpers import _map_spoolman_spool, spoolman_net_weight
from backend.app.core.config import settings as app_settings
from backend.app.models.print_queue import PrintQueueItem
from backend.app.models.spool_assignment import SpoolAssignment
@@ -144,15 +144,15 @@ async def _spoolman_remaining_grams(spoolman_spool_id: int) -> float | None:
return None
# Spoolman exposes either an absolute remaining_weight, or used_weight +
# filament.weight. Either is sufficient — prefer remaining_weight when
# present (the user may have overridden it).
# the spool's net weight (initial_weight, else filament.weight; #3194).
# Either is sufficient — prefer remaining_weight when present (the user
# may have overridden it).
remaining = spool.get("remaining_weight")
if isinstance(remaining, (int, float)) and remaining >= 0:
return float(remaining)
used = spool.get("used_weight")
filament = spool.get("filament") or {}
total = filament.get("weight")
total = spoolman_net_weight(spool)
if isinstance(used, (int, float)) and isinstance(total, (int, float)) and total > 0:
return max(0.0, float(total) - float(used))
@@ -525,7 +525,7 @@ async def build_slot_materials(db: AsyncSession, printer_id: int) -> list[SlotMa
remaining = float(rw)
else:
used = spool_dict.get("used_weight")
total = (spool_dict.get("filament") or {}).get("weight")
total = spoolman_net_weight(spool_dict)
if isinstance(used, (int, float)) and isinstance(total, (int, float)) and total > 0:
remaining = max(0.0, float(total) - float(used))
if remaining is None:
@@ -704,7 +704,7 @@ async def compute_deficit_for_queue_item(
remaining = float(rw)
else:
used = spool_dict.get("used_weight")
total = (spool_dict.get("filament") or {}).get("weight")
total = spoolman_net_weight(spool_dict)
if isinstance(used, (int, float)) and isinstance(total, (int, float)) and total > 0:
remaining = max(0.0, float(total) - float(used))
else:
backend/app/services/spoolman.py
+11
View File
@@ -414,9 +414,14 @@ class SpoolmanClient:
comment: str | None = None,
extra: dict | None = None,
spool_weight: float | None = None,
initial_weight: float | None = None,
) -> dict:
"""Create a new spool in Spoolman."""
data: dict = {"filament_id": filament_id}
# The spool's own net weight. Left off, Spoolman copies the filament's
# weight, which is wrong for a spool of another size (#3194).
if initial_weight is not None:
data["initial_weight"] = initial_weight
if remaining_weight is not None:
data["remaining_weight"] = remaining_weight
if location:
@@ -744,11 +749,17 @@ class SpoolmanClient:
extra: dict | None = None,
spool_weight: float | None = None,
clear_spool_weight: bool = False,
initial_weight: float | None = None,
) -> dict:
"""Update a spool with full field support; unlike update_spool, does not auto-set last_used."""
data: dict = {}
if filament_id is not None:
data["filament_id"] = filament_id
# Spoolman applies the fields in its own schema order, which puts
# initial_weight before remaining_weight, so a remaining_weight in the
# same PATCH is measured against the new net weight.
if initial_weight is not None:
data["initial_weight"] = initial_weight
if remaining_weight is not None:
data["remaining_weight"] = remaining_weight
if comment is not None:
backend/app/services/spoolman_tracking.py
+17 -15
View File
@@ -12,6 +12,7 @@ from dataclasses import dataclass
from sqlalchemy import delete, select
from backend.app.api.routes._spoolman_helpers import spoolman_net_weight
from backend.app.core.config import settings as app_settings
from backend.app.core.database import async_session
from backend.app.services.spoolman import (
@@ -693,10 +694,11 @@ def _spool_cost_per_gram(spool: dict | None) -> float | None:
different vendor, import duty. The spool's own value wins, which is the
order the Spoolman UI presents them in.
The divisor is ``filament.weight``: net filament grams, excluding the core.
That is the same field the remain-delta path already divides by to turn a
remain%% drop into grams, so a spool that can be charged by percentage can
always be priced too.
Each price is divided by the net weight it was paid for, excluding the
core: the spool's price by the spool's own ``initial_weight`` (falling
back to the filament's), the catalogue price by ``filament.weight``. A
250 g spool of a 1000 g filament was otherwise charged at a quarter of
its rate (#3194).
A missing or non-positive price is not a free spool, it is an unpriced one,
and returns None so the caller can fall back to the global default rate
@@ -715,12 +717,12 @@ def _spool_cost_per_gram(spool: dict | None) -> float | None:
# client that writes 0 instead is common enough that reading it as free
# would price a whole print at the default rate while a perfectly good
# catalogue price sat one level down.
raw_price = spool.get("price")
if _as_positive_number(raw_price) is None:
raw_price = filament.get("price")
price = _as_positive_number(raw_price)
weight = _as_positive_number(filament.get("weight"))
price = _as_positive_number(spool.get("price"))
if price is not None:
weight = _as_positive_number(spoolman_net_weight(spool))
else:
price = _as_positive_number(filament.get("price"))
weight = _as_positive_number(filament.get("weight"))
if price is None or weight is None:
return None
# Both operands can be finite and the quotient still overflow. A non-finite
@@ -1657,7 +1659,7 @@ async def _report_remain_delta_for_slots(
)
continue
# Look up the spool's filament reference weight. Use a fresh GET so
# Look up the spool's net weight (#3194). 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:
@@ -1666,10 +1668,10 @@ async def _report_remain_delta_for_slots(
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:
ref_weight = _as_positive_number(spoolman_net_weight(spool))
if ref_weight is None:
logger.debug(
"[SPOOLMAN] AMS%d-T%d: spool %s has no filament.weight, skipping remain-delta",
"[SPOOLMAN] AMS%d-T%d: spool %s has no initial_weight or filament.weight, skipping remain-delta",
ams_id,
tray_id,
spool_id,
@@ -1688,7 +1690,7 @@ async def _report_remain_delta_for_slots(
continue
spools_updated += 1
# ``spool`` here is the full row fetched above for its filament weight,
# ``spool`` here is the full row fetched above for its net weight,
# so the price is already in hand (#2591).
if cost_out is not None:
cost_out.add(grams_used, spool, f"AMS{ams_id}-T{tray_id}")
backend/tests/integration/test_spoolman_initial_weight_3194.py
+334
View File
@@ -0,0 +1,334 @@
"""Spoolman mode reads a spool's size from its own initial_weight (#3194).
Spoolman keeps the net weight of a full spool on the spool (``initial_weight``)
and falls back to the filament's catalogue ``weight``, so one filament can have
spools of different sizes. Bambuddy read only the filament, so a 250 g spool of
a 1000 g filament showed as 1000 g everywhere, and editing its label weight
PATCHed the filament instead of the spool.
"""
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from httpx import AsyncClient
# A 250 g spool of a 1000 g filament with 200 g left.
QUARTER_SPOOL = {
"id": 42,
"initial_weight": 250.0,
"spool_weight": 100.0,
"remaining_weight": 200.0,
"used_weight": 50.0,
"price": 6.25,
"filament": {
"id": 7,
"name": "PLA Basic",
"material": "PLA",
"color_hex": "FF0000",
"weight": 1000.0,
"vendor": {"id": 3, "name": "Bambu Lab"},
},
"location": None,
"comment": None,
"archived": False,
"registered": "2024-01-01T00:00:00+00:00",
"extra": {"tag": '"AABBCCDD"'},
}
@pytest.fixture
async def spoolman_settings(db_session):
from backend.app.models.settings import Settings
db_session.add(Settings(key="spoolman_enabled", value="true"))
db_session.add(Settings(key="spoolman_url", value="http://localhost:7912"))
await db_session.commit()
@pytest.fixture
def client():
mock = MagicMock()
mock.base_url = "http://localhost:7912"
mock.health_check = AsyncMock(return_value=True)
mock.get_all_spools = AsyncMock(return_value=[QUARTER_SPOOL])
mock.get_spools = AsyncMock(return_value=[QUARTER_SPOOL])
mock.get_spool = AsyncMock(return_value=QUARTER_SPOOL)
mock.create_spool = AsyncMock(return_value=QUARTER_SPOOL)
mock.update_spool = AsyncMock(return_value=None)
mock.update_spool_full = AsyncMock(return_value=QUARTER_SPOOL)
mock.merge_spool_extra = AsyncMock(return_value=QUARTER_SPOOL)
mock.find_or_create_filament = AsyncMock(return_value=7)
mock.find_or_create_vendor = AsyncMock(return_value=3)
mock.patch_filament = AsyncMock(return_value={"id": 7})
mock.is_filament_shared = AsyncMock(return_value=True)
mock.ensure_extra_field = AsyncMock(return_value=True)
mock.get_distinct_locations = AsyncMock(return_value=[])
with (
patch("backend.app.api.routes.spoolman_inventory.get_spoolman_client", AsyncMock(return_value=mock)),
patch("backend.app.api.routes.spoolman_inventory.init_spoolman_client", AsyncMock(return_value=mock)),
patch("backend.app.api.routes.spoolman.get_spoolman_client", AsyncMock(return_value=mock)),
patch("backend.app.api.routes.spoolbuddy._get_spoolman_client_or_none", AsyncMock(return_value=mock)),
):
yield mock
class TestReadsUseTheSpoolsWeight:
@pytest.mark.asyncio
@pytest.mark.integration
async def test_inventory_shows_the_spools_own_weight(self, async_client: AsyncClient, spoolman_settings, client):
response = await async_client.get("/api/v1/spoolman/inventory/spools/42")
assert response.status_code == 200
body = response.json()
assert body["label_weight"] == 250
assert body["label_weight"] - body["weight_used"] == pytest.approx(200.0)
# 6.25 for 250 g is 25 per kilo
assert body["cost_per_kg"] == pytest.approx(25.0)
@pytest.mark.asyncio
@pytest.mark.integration
async def test_weigh_reports_weight_used_against_the_spool(
self, async_client: AsyncClient, spoolman_settings, client
):
client.update_spool_full = AsyncMock(return_value={**QUARTER_SPOOL, "remaining_weight": 150.0})
response = await async_client.patch("/api/v1/spoolman/inventory/spools/42/weight", json={"weight_grams": 250.0})
assert response.status_code == 200
assert client.update_spool_full.call_args.kwargs["remaining_weight"] == pytest.approx(150.0)
assert response.json()["weight_used"] == pytest.approx(100.0)
@pytest.mark.asyncio
@pytest.mark.integration
async def test_spoolbuddy_scale_reports_weight_used_against_the_spool(
self, async_client: AsyncClient, spoolman_settings, client
):
response = await async_client.post(
"/api/v1/spoolbuddy/scale/update-spool-weight", json={"spool_id": 42, "weight_grams": 250.0}
)
assert response.status_code == 200
client.update_spool.assert_called_once_with(spool_id=42, remaining_weight=pytest.approx(150.0))
assert response.json()["weight_used"] == pytest.approx(100.0)
@pytest.mark.asyncio
@pytest.mark.integration
async def test_linked_spools_fill_reference_is_the_spools_weight(
self, async_client: AsyncClient, spoolman_settings, client
):
response = await async_client.get("/api/v1/spoolman/spools/linked")
assert response.status_code == 200
assert response.json()["linked"]["AABBCCDD"]["filament_weight"] == pytest.approx(250.0)
class TestCreateWritesTheSpoolsWeight:
@pytest.mark.asyncio
@pytest.mark.integration
async def test_create_sends_initial_weight_and_the_price_of_the_spool(
self, async_client: AsyncClient, spoolman_settings, client
):
"""A 250 g spool picked from a 1000 g catalogue filament is a 250 g spool."""
response = await async_client.post(
"/api/v1/spoolman/inventory/spools",
json={"spoolman_filament_id": 7, "label_weight": 250, "cost_per_kg": 25.0},
)
assert response.status_code == 200
assert client.create_spool.call_args.kwargs["initial_weight"] == 250.0
assert client.create_spool.call_args.kwargs["remaining_weight"] == 250.0
assert client.update_spool_full.call_args.kwargs["price"] == pytest.approx(6.25)
@pytest.mark.asyncio
@pytest.mark.integration
async def test_bulk_create_sends_initial_weight(self, async_client: AsyncClient, spoolman_settings, client):
response = await async_client.post(
"/api/v1/spoolman/inventory/spools/bulk",
json={"spool": {"spoolman_filament_id": 7, "label_weight": 250}, "quantity": 2},
)
assert response.status_code == 200
assert client.create_spool.call_count == 2
for call in client.create_spool.call_args_list:
assert call.kwargs["initial_weight"] == 250.0
@pytest.mark.asyncio
@pytest.mark.integration
async def test_create_without_label_weight_leaves_the_size_to_spoolman(
self, async_client: AsyncClient, spoolman_settings, client
):
"""An API caller that gives no label weight gets the filament's, as
before: Spoolman fills initial_weight from the filament itself."""
response = await async_client.post(
"/api/v1/spoolman/inventory/spools", json={"spoolman_filament_id": 7, "cost_per_kg": 25.0}
)
assert response.status_code == 200
assert client.create_spool.call_args.kwargs["initial_weight"] is None
# priced at the size Spoolman gave the spool (QUARTER_SPOOL: 250 g)
assert client.update_spool_full.call_args.kwargs["price"] == pytest.approx(6.25)
class TestEditWritesTheSpoolNotTheFilament:
@pytest.mark.asyncio
@pytest.mark.integration
@pytest.mark.parametrize("shared", [True, False])
async def test_label_weight_edit_goes_to_initial_weight(
self, async_client: AsyncClient, spoolman_settings, client, shared
):
"""Neither a shared nor a singleton filament is touched, and no
duplicate filament is created for a spool of another size."""
client.is_filament_shared = AsyncMock(return_value=shared)
response = await async_client.patch(
"/api/v1/spoolman/inventory/spools/42",
json={"label_weight": 500, "weight_used": 100.0},
)
assert response.status_code == 200
client.patch_filament.assert_not_called()
client.find_or_create_filament.assert_not_called()
kwargs = client.update_spool_full.call_args.kwargs
assert kwargs["filament_id"] == 7
assert kwargs["initial_weight"] == 500.0
assert kwargs["remaining_weight"] == pytest.approx(400.0)
@pytest.mark.asyncio
@pytest.mark.integration
async def test_untouched_edit_writes_neither_size_nor_price(
self, async_client: AsyncClient, spoolman_settings, client
):
"""The form sends back the label weight and cost per kg it loaded; a
save that changes neither must not write them."""
response = await async_client.patch(
"/api/v1/spoolman/inventory/spools/42",
json={"label_weight": 250, "cost_per_kg": 25.0, "note": "moved"},
)
assert response.status_code == 200
client.patch_filament.assert_not_called()
kwargs = client.update_spool_full.call_args.kwargs
assert kwargs["initial_weight"] is None
assert kwargs["price"] is None
assert kwargs["remaining_weight"] == pytest.approx(200.0)
@pytest.mark.asyncio
@pytest.mark.integration
async def test_fractional_size_survives_an_untouched_edit(
self, async_client: AsyncClient, spoolman_settings, client
):
"""The form shows 250.7 g as 250; saving it must not cut the spool to 250 g
or re-price it."""
spool = {**QUARTER_SPOOL, "initial_weight": 250.7, "price": 6.99}
client.get_spool = AsyncMock(return_value=spool)
loaded = (await async_client.get("/api/v1/spoolman/inventory/spools/42")).json()
assert loaded["label_weight"] == 250
response = await async_client.patch(
"/api/v1/spoolman/inventory/spools/42",
json={"label_weight": loaded["label_weight"], "cost_per_kg": loaded["cost_per_kg"]},
)
assert response.status_code == 200
kwargs = client.update_spool_full.call_args.kwargs
assert kwargs["initial_weight"] is None
assert kwargs["price"] is None
@pytest.mark.asyncio
@pytest.mark.integration
async def test_price_is_converted_at_the_spools_weight(self, async_client: AsyncClient, spoolman_settings, client):
response = await async_client.patch(
"/api/v1/spoolman/inventory/spools/42",
json={"label_weight": 250, "cost_per_kg": 30.0},
)
assert response.status_code == 200
assert client.update_spool_full.call_args.kwargs["price"] == pytest.approx(7.5)
@pytest.mark.asyncio
@pytest.mark.integration
@pytest.mark.parametrize("payload", [{"label_weight": 500}, {"label_weight": 500, "cost_per_kg": 25.0}])
async def test_resize_keeps_the_rate_per_kg(self, async_client: AsyncClient, spoolman_settings, client, payload):
"""25 per kg stays 25 per kg when the spool is resized, as in internal
mode, whether or not the caller (e.g. Bulk Edit) sends the rate."""
response = await async_client.patch("/api/v1/spoolman/inventory/spools/42", json=payload)
assert response.status_code == 200
kwargs = client.update_spool_full.call_args.kwargs
assert kwargs["initial_weight"] == 500.0
assert kwargs["price"] == pytest.approx(12.5)
@pytest.mark.asyncio
@pytest.mark.integration
async def test_colour_change_on_shared_filament_relinks_without_resizing(
self, async_client: AsyncClient, spoolman_settings, client
):
"""A relink to a filament of another weight keeps the spool's size:
Spoolman would otherwise stamp the new filament's weight on a spool
without an initial_weight of its own."""
client.find_or_create_filament = AsyncMock(return_value=8)
spool = {**QUARTER_SPOOL, "initial_weight": None, "remaining_weight": 800.0, "used_weight": 200.0}
client.get_spool = AsyncMock(return_value=spool)
response = await async_client.patch(
"/api/v1/spoolman/inventory/spools/42",
json={"rgba": "00FF00FF"},
)
assert response.status_code == 200
kwargs = client.update_spool_full.call_args.kwargs
assert kwargs["filament_id"] == 8
assert kwargs["initial_weight"] == 1000.0
assert kwargs["remaining_weight"] == pytest.approx(800.0)
assert kwargs["price"] is None
class TestSpoolWithoutAnySize:
"""Spoolman refuses a remaining_weight (HTTP 400, the whole PATCH) while a
spool has neither an initial_weight nor a filament weight."""
BARE = {**QUARTER_SPOOL, "initial_weight": None, "remaining_weight": None, "used_weight": 0.0, "price": None}
@pytest.mark.asyncio
@pytest.mark.integration
async def test_edit_without_label_weight_sends_no_remaining(
self, async_client: AsyncClient, spoolman_settings, client
):
client.get_spool = AsyncMock(
return_value={**self.BARE, "filament": {**QUARTER_SPOOL["filament"], "weight": None}}
)
response = await async_client.patch("/api/v1/spoolman/inventory/spools/42", json={"note": "hi"})
assert response.status_code == 200
kwargs = client.update_spool_full.call_args.kwargs
assert kwargs["remaining_weight"] is None
assert kwargs["initial_weight"] is None
@pytest.mark.asyncio
@pytest.mark.integration
async def test_form_label_weight_becomes_the_size(self, async_client: AsyncClient, spoolman_settings, client):
"""The form shows such a spool as 1000 g and sends that back."""
client.get_spool = AsyncMock(
return_value={**self.BARE, "filament": {**QUARTER_SPOOL["filament"], "weight": None}}
)
response = await async_client.patch(
"/api/v1/spoolman/inventory/spools/42", json={"label_weight": 1000, "weight_used": 100.0}
)
assert response.status_code == 200
kwargs = client.update_spool_full.call_args.kwargs
assert kwargs["initial_weight"] == 1000.0
assert kwargs["remaining_weight"] == pytest.approx(900.0)
class TestAmsSyncUsesTheSpoolsWeight:
@pytest.mark.asyncio
@pytest.mark.integration
async def test_ams_percentage_is_of_the_spool(
self, async_client: AsyncClient, spoolman_settings, client, db_session
):
from backend.app.models.printer import Printer
from backend.app.models.spoolman_slot_assignment import SpoolmanSlotAssignment
printer = Printer(name="P", serial_number="SN3194", ip_address="192.168.1.9", access_code="12345678")
db_session.add(printer)
await db_session.commit()
await db_session.refresh(printer)
db_session.add(SpoolmanSlotAssignment(printer_id=printer.id, ams_id=0, tray_id=0, spoolman_spool_id=42))
await db_session.commit()
state = MagicMock()
state.raw_data = {"ams": [{"id": 0, "tray": [{"id": 0, "remain": 80}]}]}
with (
patch("backend.app.api.routes.spoolman_inventory._get_client", AsyncMock(return_value=client)),
patch("backend.app.api.routes.spoolman_inventory.printer_manager") as pm,
):
pm.get_status = MagicMock(return_value=state)
response = await async_client.post("/api/v1/spoolman/inventory/sync-ams-weights")
assert response.status_code == 200
assert response.json()["synced"] == 1
# 80 % of 250 g, not of the filament's 1000 g
client.update_spool_full.assert_called_once_with(42, remaining_weight=200.0)
backend/tests/unit/services/test_spoolman_print_cost_2591.py
+12
View File
@@ -91,6 +91,18 @@ class TestSpoolCostPerGram:
silently pass through into the archive."""
assert _spool_cost_per_gram(spool) is None
def test_spool_price_is_divided_by_the_spools_own_weight(self):
"""#3194: a spool's price is what that spool cost. 6.25 for a 250 g
spool of a 1000 g filament is 2.5 cents a gram, not 0.625."""
spool = {**_spool(1, price=6.25, weight=1000), "initial_weight": 250.0}
assert _spool_cost_per_gram(spool) == pytest.approx(0.025)
def test_catalogue_price_stays_on_the_filament_weight(self):
"""filament.price is the catalogue price of a filament.weight spool,
whatever size this particular spool is."""
spool = {**_spool(1, filament_price=25.0, weight=1000), "initial_weight": 250.0}
assert _spool_cost_per_gram(spool) == pytest.approx(0.025)
class TestPrintCostAccumulator:
def test_sums_each_slot_at_its_own_rate(self):
backend/tests/unit/test_spoolman_inventory_helpers.py
+85
View File
@@ -8,6 +8,9 @@ from backend.app.api.routes._spoolman_helpers import (
_map_spoolman_spool,
_safe_float,
_safe_int,
cost_per_kg_to_spoolman_price,
spoolman_net_weight,
spoolman_price_to_cost_per_kg,
spoolman_tare,
)
@@ -683,3 +686,85 @@ class TestMapSpoolmanSpoolPrice:
spool = {**MINIMAL_SPOOL, "price": math.inf}
assert _map_spoolman_spool(spool)["cost_per_kg"] is None
class TestSpoolmanNetWeight:
"""The spool's initial_weight wins over the filament's weight, as in Spoolman (#3194)."""
@staticmethod
def _spool(initial=None, filament_weight=1000.0):
return {"id": 1, "initial_weight": initial, "filament": {"weight": filament_weight}}
def test_initial_weight_beats_filament_weight(self):
assert spoolman_net_weight(self._spool(250.0, 1000.0)) == 250.0
def test_falls_back_to_filament_weight(self):
assert spoolman_net_weight(self._spool(None, 1000.0)) == 1000.0
def test_zero_initial_weight_counts_as_unset(self):
"""Spoolman's own /measure treats an initial_weight of 0 as missing."""
assert spoolman_net_weight(self._spool(0, 750.0)) == 750.0
@pytest.mark.parametrize("bad", [float("nan"), float("inf"), "abc", True])
def test_invalid_initial_weight_counts_as_unset(self, bad):
assert spoolman_net_weight(self._spool(bad, 750.0)) == 750.0
def test_filament_weight_returned_as_stored(self):
"""A filament weight of 0 is left to the caller, which keeps its own handling."""
assert spoolman_net_weight(self._spool(None, 0)) == 0.0
@pytest.mark.parametrize("filament", [None, {}, "not a dict", {"weight": True}])
def test_nothing_usable_returns_none(self, filament):
assert spoolman_net_weight({"id": 1, "filament": filament}) is None
class TestSpoolmanPriceConversion:
"""A Spoolman spool price is the price of the whole spool (#3194)."""
def test_price_of_a_quarter_kilo_spool_as_a_rate(self):
assert spoolman_price_to_cost_per_kg(6.25, 250) == pytest.approx(25.0)
def test_rate_to_price_of_a_quarter_kilo_spool(self):
assert cost_per_kg_to_spoolman_price(25.0, 250) == pytest.approx(6.25)
def test_kilo_spool_price_equals_rate(self):
assert spoolman_price_to_cost_per_kg(24.99, 1000) == 24.99
assert cost_per_kg_to_spoolman_price(24.99, 1000) == 24.99
@pytest.mark.parametrize("weight", [250, 333, 750, 1000, 2500])
def test_round_trip_does_not_drift(self, weight):
price = cost_per_kg_to_spoolman_price(19.99, weight)
assert spoolman_price_to_cost_per_kg(price, weight) == pytest.approx(19.99, abs=1e-3)
@pytest.mark.parametrize(("price", "weight"), [(None, 1000), (math.nan, 1000), ("abc", 1000), (10.0, 0)])
def test_unusable_input_returns_none(self, price, weight):
assert spoolman_price_to_cost_per_kg(price, weight) is None
class TestMapperUsesSpoolInitialWeight:
"""#3194: a 250 g spool of a 1000 g filament maps as a 250 g spool."""
SPOOL = {
"id": 9,
"initial_weight": 250.0,
"remaining_weight": 200.0,
"used_weight": 50.0,
"price": 6.25,
"filament": {"id": 1, "material": "PLA", "name": "PLA Basic", "weight": 1000.0},
"extra": {},
}
def test_label_weight_is_the_spools(self):
mapped = _map_spoolman_spool(self.SPOOL)
assert mapped["label_weight"] == 250
# remaining = label - weight_used must match Spoolman's 200 g
assert mapped["label_weight"] - mapped["weight_used"] == pytest.approx(200.0)
def test_cost_per_kg_is_the_spool_price_over_its_weight(self):
assert _map_spoolman_spool(self.SPOOL)["cost_per_kg"] == pytest.approx(25.0)
def test_without_initial_weight_the_filament_weight_is_used(self):
spool = {**self.SPOOL, "initial_weight": None, "remaining_weight": 800.0, "used_weight": 200.0}
mapped = _map_spoolman_spool(spool)
assert mapped["label_weight"] == 1000
assert mapped["cost_per_kg"] == pytest.approx(6.25)
backend/tests/unit/test_spoolman_no3mf_remain_fallback.py
+47
View File
@@ -490,6 +490,53 @@ class TestPartialUsageRemainDelta:
client.use_spool.assert_awaited_once_with(7, 200.0)
@pytest.mark.asyncio
async def test_remain_delta_uses_the_spools_own_weight(self):
"""#3194: the AMS percentage is of this spool, so a 20% drop on a
250 g spool of a 1000 g filament is 50 g, not 200 g."""
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, "initial_weight": 250.0, "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, 50.0)
class _AsyncCtx:
"""Tiny async-context shim returning a pre-built db mock; mirrors