tcom/Lab8DATAPROCESSOR
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Treat InvenTree as filament assignment source

Browse Source
This commit is contained in:
tcomlab
2026-04-15 21:04:50 +03:00
parent 130308a8d9
commit f49dc81ea1
5 changed files with 123 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.env.example
View File

@@ -48,6 +48,7 @@ FILAMENT_PRINTER_LOCATION_MAP=B1:93,B2:94,B3:95,B4:96
FILAMENT_BATCH_SOURCE=tag_uid
FILAMENT_SYNC_SPOOLS=true
FILAMENT_SYNC_ASSIGNMENTS=true
FILAMENT_UNASSIGN_MISSING_ASSIGNMENTS=true
FILAMENT_SYNC_LOCATIONS=true
FILAMENT_SYNC_USAGE=true
FILAMENT_RETURN_UNASSIGNED_TO_STORAGE=false

10
README.md
View File

@@ -21,7 +21,8 @@ For filament, the service can:
- use InvenTree `StockItem.batch` as the spool identity;
- create/update Bambuddy spool records from InvenTree filament stock;
- move InvenTree spool stock between storage and printer locations from Bambuddy assignments;
- create/remove Bambuddy assignments from InvenTree printer locations;
- move InvenTree spool stock between storage and printer locations from Bambuddy assignments when needed;
- subtract Bambuddy filament usage from the matching InvenTree stock item;
- store usage sync state in SQLite to prevent duplicate subtraction.
@@ -102,7 +103,7 @@ The service deliberately starts with `FILAMENT_DRY_RUN=true`. In dry-run mode it
Filament sync has four independent parts:
- spool catalog sync: InvenTree stock items create/update Bambuddy spools;
- assignment sync: InvenTree printer locations create Bambuddy spool assignments;
- assignment sync: InvenTree printer locations create Bambuddy spool assignments and remove assignments that no longer exist in InvenTree printer locations;
- location sync: Bambuddy assignments move InvenTree stock to printer locations; returning unassigned loaded spools to storage is optional;
- usage sync: Bambuddy usage history subtracts grams from the matching InvenTree stock item.
@@ -235,6 +236,7 @@ FILAMENT_PRINTER_LOCATION_MAP=B1:93,B2:94,B3:95,B4:96
FILAMENT_BATCH_SOURCE=tag_uid
FILAMENT_SYNC_SPOOLS=true
FILAMENT_SYNC_ASSIGNMENTS=true
FILAMENT_UNASSIGN_MISSING_ASSIGNMENTS=true
FILAMENT_SYNC_LOCATIONS=true
FILAMENT_SYNC_USAGE=true
FILAMENT_RETURN_UNASSIGNED_TO_STORAGE=false
@@ -314,6 +316,9 @@ Do not commit `.env`. It contains API tokens and is ignored by git.
`FILAMENT_SYNC_ASSIGNMENTS`
: Creates Bambuddy assignments from InvenTree stock items currently stored in printer locations.
`FILAMENT_UNASSIGN_MISSING_ASSIGNMENTS`
: Removes Bambuddy assignments for managed batch codes when the matching InvenTree stock item is no longer in a printer location. This makes InvenTree the source of truth for loaded spools.
`FILAMENT_SYNC_LOCATIONS`
: Moves InvenTree stock items between storage and printer locations from Bambuddy assignments.
@@ -407,6 +412,7 @@ FILAMENT_TRACKING_ENABLED=true
FILAMENT_DRY_RUN=true
FILAMENT_SYNC_SPOOLS=true
FILAMENT_SYNC_ASSIGNMENTS=true
FILAMENT_UNASSIGN_MISSING_ASSIGNMENTS=true
FILAMENT_SYNC_LOCATIONS=true
FILAMENT_SYNC_USAGE=true
FILAMENT_RETURN_UNASSIGNED_TO_STORAGE=false

4
src/bambuddy_inventree_sync/bambuddy.py
View File

@@ -91,6 +91,10 @@ class BambuddyClient:
data = await self._request("POST", "/inventory/assignments", json=payload)
return data if isinstance(data, dict) else {}
async def unassign_spool(self, *, printer_id: int, ams_id: int, tray_id: int) -> dict[str, Any]:
data = await self._request("DELETE", f"/inventory/assignments/{printer_id}/{ams_id}/{tray_id}")
return data if isinstance(data, dict) else {}
async def list_usage(self, *, limit: int, printer_id: int | None = None) -> list[dict[str, Any]]:
params: dict[str, Any] = {"limit": limit}
if printer_id is not None:

1
src/bambuddy_inventree_sync/config.py
View File

@@ -45,6 +45,7 @@ class Settings(BaseSettings):
filament_batch_source: str = "tag_uid"
filament_sync_spools: bool = True
filament_sync_assignments: bool = True
filament_unassign_missing_assignments: bool = True
filament_sync_locations: bool = True
filament_sync_usage: bool = True
filament_return_unassigned_to_storage: bool = False

109
src/bambuddy_inventree_sync/filament.py
View File

@@ -68,6 +68,7 @@ class FilamentTrackingService:
"printer_locations": self.settings.filament_printer_locations,
"printer_ids": self.settings.filament_printer_ids,
"batch_source": self.settings.filament_batch_source,
"unassign_missing_assignments": self.settings.filament_unassign_missing_assignments,
"return_unassigned_to_storage": self.settings.filament_return_unassigned_to_storage,
"assignment_default_ams_id": self.settings.filament_assignment_default_ams_id,
"assignment_start_tray_id": self.settings.filament_assignment_start_tray_id,
@@ -102,6 +103,15 @@ class FilamentTrackingService:
assignments=assignments,
printers=printers,
)[:50],
"stale_assignments": self._stale_assignment_summaries(
assignments=assignments,
spools=spools,
planned=self._planned_assignments(
stock_items=stock_items,
spools_by_batch=spools_by_batch,
printer_ids=self._printer_ids_by_key(printers),
),
)[:50],
"pending_usage": pending_usage[:50],
},
"database": self.database.counts(),
@@ -210,17 +220,62 @@ class FilamentTrackingService:
spools_by_batch=spools_by_batch,
printer_ids=printer_ids,
)
stale_assignments = self._stale_assignment_summaries(
assignments=assignments,
spools=spools,
planned=planned,
)
result: dict[str, Any] = {
"dry_run": dry,
"seen": len(planned),
"created": 0,
"deleted": 0,
"unchanged": 0,
"skipped": 0,
"failed": 0,
"would_create": [],
"would_delete": [],
"missing_printers": self._missing_printer_keys(printer_ids),
}
if self.settings.filament_unassign_missing_assignments:
cleared_stale_spool_ids: set[int] = set()
cleared_stale_slot_keys: set[tuple[int, int, int]] = set()
for item in stale_assignments:
if dry:
result["would_delete"].append(item)
cleared_stale_spool_ids.add(int(item["spool_id"]))
cleared_stale_slot_keys.add((int(item["printer_id"]), int(item["ams_id"]), int(item["tray_id"])))
continue
try:
await self.bambuddy.unassign_spool(
printer_id=int(item["printer_id"]),
ams_id=int(item["ams_id"]),
tray_id=int(item["tray_id"]),
)
result["deleted"] += 1
cleared_stale_spool_ids.add(int(item["spool_id"]))
cleared_stale_slot_keys.add((int(item["printer_id"]), int(item["ams_id"]), int(item["tray_id"])))
except Exception as exc:
logger.exception(
"Failed to unassign Bambuddy spool %s from printer %s",
item.get("spool_id"),
item.get("printer_id"),
)
result["failed"] += 1
result.setdefault("errors", []).append({**item, "error": str(exc)})
assigned_by_spool_id = {
spool_id: assignment
for spool_id, assignment in assigned_by_spool_id.items()
if spool_id not in cleared_stale_spool_ids
}
assigned_by_slot = {
slot_key: assignment
for slot_key, assignment in assigned_by_slot.items()
if slot_key not in cleared_stale_slot_keys
}
for item in planned:
spool_id = int(item["spool_id"])
slot_key = (int(item["printer_id"]), int(item["ams_id"]), int(item["tray_id"]))
@@ -615,6 +670,60 @@ class FilamentTrackingService:
item["already_assigned"] = int(item["spool_id"]) in existing_spool_ids
return planned
def _stale_assignment_summaries(
self,
*,
assignments: list[dict[str, Any]],
spools: list[dict[str, Any]],
planned: list[dict[str, Any]],
) -> list[dict[str, Any]]:
planned_by_spool_id = {int(item["spool_id"]): item for item in planned if item.get("spool_id") is not None}
result: list[dict[str, Any]] = []
for assignment in assignments:
spool_id = assignment.get("spool_id")
slot_key = self._assignment_slot_key(assignment)
if spool_id is None or slot_key is None:
continue
batch = self._batch_for_assignment(assignment, spools)
if not batch:
continue
planned_item = planned_by_spool_id.get(int(spool_id))
if not planned_item:
reason = "not_loaded_in_inventree_printer_location"
elif slot_key == (
int(planned_item["printer_id"]),
int(planned_item["ams_id"]),
int(planned_item["tray_id"]),
):
continue
else:
reason = "slot_differs_from_inventree"
item: dict[str, Any] = {
"assignment_id": assignment.get("id"),
"spool_id": int(spool_id),
"batch": batch,
"printer_id": int(slot_key[0]),
"printer_name": assignment.get("printer_name"),
"ams_id": int(slot_key[1]),
"tray_id": int(slot_key[2]),
"reason": reason,
}
if planned_item:
item["expected"] = {
"printer_key": planned_item.get("printer_key"),
"printer_id": planned_item.get("printer_id"),
"ams_id": planned_item.get("ams_id"),
"tray_id": planned_item.get("tray_id"),
"location_id": planned_item.get("location_id"),
}
result.append(item)
return result
def _planned_assignments(
self,
*,