Fix filament mapping

.github/workflows/release-docker-image.yml

@@ -115,6 +115,13 @@ jobs:
             fi
 
             echo "TAGS=$TAGS" >> "$GITHUB_ENV"
+          elif [[ "${{ github.event_name }}" == "workflow_dispatch" ]]; then
+            # Manual run without tag: use the selected branch name
+            BRANCH="${{ github.ref_name }}"
+            BRANCH_TAG=$(echo "$BRANCH" | tr '[:upper:]' '[:lower:]' | sed 's/[^a-z0-9._-]/-/g')
+            if [[ -n "$BRANCH_TAG" ]]; then
+              echo "TAGS=ghcr.io/${{ env.REPO_NAME }}:${BRANCH_TAG}" >> "$GITHUB_ENV"
+            fi
           elif [[ "${{ github.ref }}" == "refs/heads/main" ]]; then
             # Dev image for pushes to main (and equivalent contexts)
             echo "TAGS=ghcr.io/${{ env.REPO_NAME }}:dev" >> "$GITHUB_ENV"

.github/workflows/tests.yml

@@ -14,6 +14,7 @@ jobs:
     env:
       TRACK_LAYER_USAGE: ${{ matrix.layer_tracking }}
       OPENSPOOLMAN_TEST_DATA: "1"
+      OPENSPOOLMAN_MQTT_LOG_PATH: ${{ github.workspace }}/logs/mqtt.log
     steps:
       - name: Checkout
         uses: actions/checkout@v4

README.md

@@ -181,7 +181,7 @@ SpoolMan can print QR-code stickers for every spool; follow the SpoolMan label g
   - set `DISABLE_MISMATCH_WARNING` to `True` to hide mismatch warnings in the UI (mismatches are still detected and logged to `logs/filament_mismatch.json`, including the detected color difference when applicable).
   - set `CLEAR_ASSIGNMENT_WHEN_EMPTY` to `True` if you want OpenSpoolMan to clear any SpoolMan assignment and reset the AMS tray whenever the printer reports no spool in that slot.
   - set `COLOR_DISTANCE_TOLERANCE` to an integer (default `40`) if you want to make the perceptual ΔE threshold for tray/spool color mismatch warnings stricter or more lenient; when either side (AMS tray or SpoolMan spool) lacks a color the warning is skipped and the UI shows "Color not set".
- - By default, the app reads `data/3d_printer_logs.db` for print history; override it through `OPENSPOOLMAN_PRINT_HISTORY_DB` or via the screenshot helper (which targets `data/demo.db` by default).
+ - By default, the app reads `data/3d_printer_logs.db` for print history; override it through `OPENSPOOLMAN_PRINT_HISTORY_DB` or via the screenshot helper (which targets `data/demo.db` by default). When schema migrations are required, OpenSpoolMan creates a copy with a `.v2` suffix, migrates that copy, and will prefer the versioned file on subsequent starts so you can roll back to the original if needed.
 
  - Run SpoolMan.
  - Add these extra fields in SpoolMan:

agents.md

@@ -52,6 +52,7 @@ Key folders/files you will interact with:
 
 ### 4.1 Local Python run (development)
 1. Configure environment (see §5). Create `config.env` from `config.env.template` or export env vars.
+1. Activate the correct Python environment (virtualenv/conda) for this repo before running or testing.
 2. Start the server:
    - `python wsgi.py`
 

app.py

@@ -562,15 +562,15 @@ def _to_int(value):
   per_page = 50
   offset = max((page - 1) * per_page, 0)
 
-  ams_slot = request.args.get("ams_slot")
+  filament_id = request.args.get("filament_id") or request.args.get("ams_slot")
   print_id = request.args.get("print_id")
   spool_id = request.args.get("spool_id")
   old_spool_id = request.args.get("old_spool_id")
 
   if not old_spool_id:
     old_spool_id = -1
 
-  if READ_ONLY_MODE and all([ams_slot, print_id, spool_id]):
+  if READ_ONLY_MODE and all([filament_id, print_id, spool_id]):
     return render_template('error.html', exception="Live read-only mode: updating print-to-spool assignments is disabled.")
 
   def _consume_for_spool(spool_id_value, grams_value=None, length_value=None):
@@ -581,9 +581,9 @@ def _consume_for_spool(spool_id_value, grams_value=None, length_value=None):
     elif grams_value is not None:
       spoolman_client.consumeSpool(spool_id_value, use_weight=grams_value)
 
-  if all([ams_slot, print_id, spool_id]):
-    filament = print_history_service.get_filament_for_slot(print_id, ams_slot)
-    print_history_service.update_filament_spool(print_id, ams_slot, spool_id)
+  if all([filament_id, print_id, spool_id]):
+    filament = print_history_service.get_filament_for_filament_id(print_id, filament_id)
+    print_history_service.update_filament_spool(print_id, filament_id, spool_id)
 
     if(filament["spool_id"] != int(spool_id) and (not old_spool_id or (old_spool_id and filament["spool_id"] == int(old_spool_id)))):
       grams_used = _to_float(filament.get("grams_used"))
@@ -676,7 +676,7 @@ def _consume_for_spool(spool_id_value, grams_value=None, length_value=None):
 def print_select_spool():
 
   try:
-    ams_slot = request.args.get("ams_slot")
+    filament_id = request.args.get("filament_id") or request.args.get("ams_slot")
     print_id = request.args.get("print_id")
     old_spool_id = request.args.get("old_spool_id")
 
@@ -685,15 +685,15 @@ def print_select_spool():
     if not old_spool_id:
       old_spool_id = -1
 
-    if not all([ams_slot, print_id]):
+    if not all([filament_id, print_id]):
       return render_template('error.html', exception="Missing spool ID or print ID.")
 
     spools = mqtt_bambulab.fetchSpools()
 
     materials = extract_materials(spools)
     selected_materials = []
 
-    filament = print_history_service.get_filament_for_slot(print_id, ams_slot)
+    filament = print_history_service.get_filament_for_filament_id(print_id, filament_id)
 
     try:
       filament_material = filament["filament_type"] if filament else None
@@ -706,7 +706,7 @@ def print_select_spool():
     return render_template(
       'print_select_spool.html',
       spools=spools,
-      ams_slot=ams_slot,
+      filament_id=filament_id,
       print_id=print_id,
       old_spool_id=old_spool_id,
       change_spool=change_spool,

data/print_history_db.md

@@ -1,6 +1,6 @@
 Demo print history database
 ===========================
 
-- Default location: `data/3d_printer_logs.db`.
+- Default location: `data/3d_printer_logs.db`. If a schema migration is needed, OpenSpoolMan copies the database to a `.migrated` filename, migrates that copy, and will prefer the migrated file afterward so the original remains available for rollback.
 - For screenshots or demos, override with `OPENSPOOLMAN_PRINT_HISTORY_DB=/path/to/data/demo.db` or pass `--print-history-db` to `scripts/generate_screenshots.py`/`pytest -m screenshots`.
 - Schema matches `print_history.py` (`prints` + `filament_usage`). Populate it from your own data or via a live snapshot export; demo rows are no longer bundled by default.

docs/ams_mapping_print_history_fix.md

@@ -0,0 +1,24 @@
+# AMS Mapping / Print History Fix
+
+## Root cause
+- Print history stored `filament_id` from the 3MF, but the filament usage tracker wrote updates using tool index + 1.
+- AMS mapping indices are tool indices, not 3MF filament IDs. The mapping between tool index and filament ID was inconsistent.
+- Negative AMS mapping values (e.g., `-1` placeholders) were parsed as real AMS targets, which can produce wrong tray/spool resolution.
+
+## Changes
+- `tools_3mf.py`
+  - Build `tool_index_to_filament_id` / `filament_id_to_tool_index` mappings based on tool indices (contiguous) or usage order only when gaps exist.
+- `spoolman_service.py`
+  - Resolve AMS mapping by tool index using `filament_id_to_tool_index`.
+  - Treat negative AMS mapping values as `unused`.
+- `filament_usage_tracker.py`
+  - Resolve the 3MF `filament_id` from the tool index before writing history.
+  - Skip history updates when a filament ID cannot be resolved (avoid wrong assignments).
+- `mqtt_bambulab.py`
+  - Local tray mapping now stores AMS mappings by tool index consistently.
+- `tests/test_printhistory_spool_assignment_from_replay.py`
+  - Expected spool assignment now derives the tool index mapping from 3MF metadata.
+
+## Tests
+- `python -m pytest -s tests/test_printhistory_spool_assignment_from_replay.py`
+- `python -m pytest -s tests/test_printhistory_accounting.py tests/test_ams_mapping_from_replay.py`

docs/pytest_best_practices_audit.md

@@ -0,0 +1,53 @@
+# Pytest Best-Practices Audit – OpenSpoolMan
+
+## Inventory (Tests and Entry Points)
+- `tests/test_ams_model_detection.py` -> `mqtt_bambulab.identify_ams_model_from_module`, `identify_ams_models_from_modules`, `identify_ams_models_by_id`
+- `tests/test_ams_mapping_parser.py` -> `spoolman_service.parse_ams_mapping`, `parse_ams_mapping_value`
+- `tests/test_ams_mapping_from_replay.py` -> `mqtt_bambulab` replay path via `mqtt_replay`, `print_history` spool assignment
+- `tests/test_printhistory_replay_e2e.py` -> `print_history` inserts via `mqtt_bambulab` replay
+- `tests/test_printhistory_accounting.py` -> `print_history`, `filament_usage_tracker` consumption via `mqtt_replay`
+- `tests/test_printhistory_spool_assignment_from_replay.py` -> `print_history` spool assignment via `mqtt_replay`
+- `tests/test_mqtt_replay.py` -> `mqtt_bambulab.processMessage`, `FilamentUsageTracker.on_message` (replay of payloads)
+- `tests/test_print_run_logic.py` -> `mqtt_bambulab.processMessage`, `bambu_state` run tracking helpers
+- `tests/test_filament_mismatch.py` -> `spoolman_service.augmentTrayDataWithSpoolMan`
+- `tests/test_tray_remaining.py` -> `templates/fragments/tray.html` (rendering behavior)
+- `tests/test_screenshots.py` -> `scripts/generate_screenshots.py` (Playwright, optional)
+- `tests/test_policy_no_test_logic.py` -> guard against test-owned log parsing
+
+## Changes Applied
+- `pytest.ini`
+  - Added `addopts = -ra --strict-markers --strict-config` for stricter pytest defaults.
+  - Added `testpaths = tests` and `python_files = test_*.py` to avoid collecting non-test helper modules.
+- `requirements.txt`
+  - Added `pytest-asyncio` to make the asyncio config explicit and avoid implicit plugin availability.
+- `tests/test_ams_model_detection.py`
+  - Converted repeated tests into parameterized cases for clarity and maintainability.
+- `tests/test_ams_mapping_parser.py`
+  - Split monolithic test into focused tests and added parameterized value parsing.
+- `tests/test_tray_remaining.py`
+  - Parameterized AMS model visibility check to reduce duplication.
+- `tests/test_print_run_logic.py`
+  - Added `print_run_env` fixture to centralize setup, reset global state, and ensure deterministic feature flags.
+  - Renamed tests to clearer `test_<expected>_when_<condition>` style.
+- `tests/test_filament_mismatch.py`
+  - Renamed tests for clearer intent.
+  - Replaced aggregated failure loop with parameterized cases for isolated, deterministic failures.
+- `tests/test_mqtt_replay.py`
+  - Wraps production `map_filament` and `_resolve_tray_mapping` to observe assignments before state is cleared (observation only, no logic duplication).
+- `tests/test_printhistory_replay_e2e.py`
+  - Simplified assertions to direct dict comparisons for clarity.
+- `tests/test_printhistory_spool_assignment_from_replay.py`
+  - Simplified expected-vs-actual comparisons using direct dict equality.
+
+## Best-Practice Goals Covered
+- Clear Arrange/Act/Assert flow with focused tests and fixtures.
+- Extensive parameterization to reduce duplicate tests and improve failure granularity.
+- Deterministic setup via fixture-managed global state and feature flags.
+- Assertions compare direct outputs instead of re-implementing business rules.
+
+## Anti-Patterns Removed
+- Monolithic tests with multiple independent behaviors bundled together.
+
+## New Fixtures and Parameterization
+- `print_run_env` fixture in `tests/test_print_run_logic.py` for state reset and shared mocks.
+- Parameterized test cases in `tests/test_ams_model_detection.py`, `tests/test_ams_mapping_parser.py`, `tests/test_tray_remaining.py`, and `tests/test_filament_mismatch.py`.

docs/test_audit.md

@@ -0,0 +1,43 @@
+# Test Audit – OpenSpoolMan
+
+## Inventory
+- `tests/test_ams_model_detection.py` -> `mqtt_bambulab.identify_ams_model_from_module`, `identify_ams_models_from_modules`, `identify_ams_models_by_id`
+- `tests/test_ams_mapping_parser.py` -> `spoolman_service.parse_ams_mapping`, `parse_ams_mapping_value`
+- `tests/test_ams_mapping_from_replay.py` -> `mqtt_bambulab` replay path via `mqtt_replay` fixture + spool assignment persisted in `print_history`
+- `tests/test_printhistory_replay_e2e.py` -> `print_history` inserts via `mqtt_bambulab` replay
+- `tests/test_printhistory_accounting.py` -> `print_history` + `filament_usage_tracker` consumption via `mqtt_replay`
+- `tests/test_printhistory_spool_assignment_from_replay.py` -> `print_history` spool assignment via `mqtt_replay`
+- `tests/test_mqtt_replay.py` -> `mqtt_bambulab.processMessage`, `FilamentUsageTracker.on_message` (replay of real payloads)
+- `tests/test_print_run_logic.py` -> `mqtt_bambulab.processMessage`, `bambu_state` run tracking helpers
+- `tests/test_filament_mismatch.py` -> `spoolman_service.augmentTrayDataWithSpoolMan` mismatch detection
+- `tests/test_tray_remaining.py` -> Jinja template `templates/fragments/tray.html`
+- `tests/test_screenshots.py` -> `scripts/generate_screenshots.py` (Playwright, optional)
+- `tests/test_policy_no_test_logic.py` -> guard test to prevent log parsing in tests
+- `test.py` (script) -> `mqtt_bambulab.processMessage` (manual replay helper, not pytest)
+
+## Changes Applied
+- Replaced test-owned MQTT log parsing with production helpers.
+  - `tests/conftest.py`: now uses `mqtt_bambulab.replay_mqtt_log`.
+  - `tests/test_mqtt_replay.py`: now uses `mqtt_bambulab.iter_mqtt_payloads_from_log`.
+  - `test.py`: now uses `mqtt_bambulab.iter_mqtt_payloads_from_log`.
+- Removed manual AMS mapping derivation in `tests/test_printhistory_spool_assignment_from_replay.py`.
+  - The test now asserts against expected spool IDs from fixture JSON, using production replay output only.
+- Added a guard test to catch direct MQTT log parsing in tests.
+
+## Anti-Patterns Removed
+- Direct parsing of MQTT log lines in tests (`split("::")`, manual JSON extraction).
+- Manual reconstruction of AMS-to-spool mappings inside tests.
+
+## New Production Entry Points
+Added to `mqtt_bambulab.py`:
+- `iter_mqtt_payloads_from_lines(lines)`
+- `iter_mqtt_payloads_from_log(log_path)`
+- `replay_mqtt_payloads(payloads)`
+- `replay_mqtt_log(log_path)`
+
+These are thin wrappers around the existing ingestion path and do not re-implement business rules.
+
+## Rationale
+- Tests now feed raw fixture data into production code and assert on production outputs or side effects.
+- Mapping and parsing logic lives in production modules, eliminating test-owned business logic.
+- The guard test prevents regressions toward log parsing in tests.

filament_usage_tracker.py

@@ -286,7 +286,11 @@ def on_message(self, message: dict) -> None:
         self._mc_remaining_time_minutes = None
 
     if command == "project_file":
-      self._pending_project_file = print_obj
+      if TRACK_LAYER_USAGE and self.active_model is None:
+        self._handle_print_start(print_obj)
+        self._pending_project_file = None
+      else:
+        self._pending_project_file = print_obj
 
     if command == "push_status":
       if "layer_num" in print_obj:
@@ -604,21 +608,24 @@ def _apply_usage_for_filament(self, filament: int, usage_mm: float) -> bool:
     usage_grams = self._mm_to_grams(total_mm, diameter_mm, density)
 
     usage_rounded = round(total_mm, 5)
-    filament_key = filament + 1
-    previous_grams = self.cumulative_grams_used.get(filament_key, 0.0)
-    self.cumulative_grams_used[filament_key] = previous_grams + usage_grams
-    previous_length = self.cumulative_length_used.get(filament_key, 0.0)
+    filament_id = self._resolve_filament_id(filament)
+    tracking_key = filament_id if filament_id is not None else f"tool_{filament}"
+    previous_grams = self.cumulative_grams_used.get(tracking_key, 0.0)
+    self.cumulative_grams_used[tracking_key] = previous_grams + usage_grams
+    previous_length = self.cumulative_length_used.get(tracking_key, 0.0)
     cumulative_length = previous_length + usage_rounded
-    self.cumulative_length_used[filament_key] = cumulative_length
+    self.cumulative_length_used[tracking_key] = cumulative_length
 
-    grams_rounded = round(self.cumulative_grams_used[filament_key], 2)
+    grams_rounded = round(self.cumulative_grams_used[tracking_key], 2)
     log(f"[filament-tracker] Consume spool {spool_id} for filament {filament} with {usage_rounded}mm ({grams_rounded}g cumulative) (tray_uid={tray_uid})")
 
     consumeSpool(spool_id, use_length=usage_rounded)
 
-    if self.print_id:
-      update_filament_spool(self.print_id, filament_key, spool_id)
-      update_filament_grams_used(self.print_id, filament_key, grams_rounded, length_used=cumulative_length)
+    if self.print_id and filament_id is not None:
+      update_filament_spool(self.print_id, filament_id, spool_id)
+      update_filament_grams_used(self.print_id, filament_id, grams_rounded, length_used=cumulative_length)
+    elif self.print_id and filament_id is None:
+      log(f"[filament-tracker] Skipping print history update for tool {filament}: missing filament_id mapping")
 
     self._filament_spool_id_map[filament] = spool_id
     self._spool_data_cache[spool_id] = spool_data
@@ -737,7 +744,11 @@ def _bind_initial_spools(self) -> None:
       if spool_id is None:
         continue
 
-      update_filament_spool(self.print_id, filament_index + 1, spool_id)
+      filament_id = self._resolve_filament_id(filament_index)
+      if filament_id is not None:
+        update_filament_spool(self.print_id, filament_id, spool_id)
+      else:
+        log(f"[filament-tracker] Skipping initial spool bind for tool {filament_index}: missing filament_id mapping")
       self._filament_spool_id_map[filament_index] = spool_id
 
       if spool_id not in self._spool_data_cache:
@@ -806,6 +817,33 @@ def _resolve_tray_mapping(self, filament_index: int) -> int | None:
       return self.ams_mapping[filament_index]
     return EXTERNAL_SPOOL_ID
 
+  def _resolve_filament_id(self, filament_index: int) -> int | None:
+    metadata = self.print_metadata or {}
+    tool_to_filament = metadata.get("tool_index_to_filament_id") or {}
+    if tool_to_filament:
+      direct = tool_to_filament.get(filament_index)
+      if direct is None:
+        direct = tool_to_filament.get(str(filament_index))
+      try:
+        return int(direct) if direct is not None else None
+      except (TypeError, ValueError):
+        return None
+
+    filament_id_order = metadata.get("filament_id_order") or []
+    if filament_id_order:
+      try:
+        ordered_ids = [int(fid) for fid in filament_id_order]
+      except (TypeError, ValueError):
+        ordered_ids = []
+      if ordered_ids == sorted(ordered_ids):
+        start = ordered_ids[0] if ordered_ids else 0
+        if ordered_ids == list(range(start, start + len(ordered_ids))):
+          if 0 <= filament_index < len(ordered_ids):
+            return ordered_ids[filament_index]
+      return None
+
+    return filament_index + 1
+
   def _lookup_spool_for_tray(self, tray_uid: str):
     for spool in fetchSpools():
       extra = spool.get("extra") or {}
@@ -852,11 +890,11 @@ def _attempt_print_resume(self, task_id, subtask_id) -> None:
     self.cumulative_length_used = {}
     if self.print_id:
       existing_usage = get_all_filament_usage_for_print(self.print_id)
-      for ams_slot, usage in existing_usage.items():
+      for filament_id, usage in existing_usage.items():
         grams_value = usage.get("grams_used") if isinstance(usage, dict) else usage
         length_value = usage.get("length_used") if isinstance(usage, dict) else None
         if grams_value is not None:
-          self.cumulative_grams_used[ams_slot] = grams_value
+          self.cumulative_grams_used[filament_id] = grams_value
         if length_value is not None:
-          self.cumulative_length_used[ams_slot] = length_value
-        log(f"[filament-tracker] Resumed cumulative usage for filament {ams_slot}: {grams_value}g, {length_value or 0}mm")
+          self.cumulative_length_used[filament_id] = length_value
+        log(f"[filament-tracker] Resumed cumulative usage for filament {filament_id}: {grams_value}g, {length_value or 0}mm")