Feature AMS Info Card (#570)

* AMS Labels addition to PrintersPage

* Added database reinitialization for schema migrations on database restore

* Bug fixes and AMS Label persistence updates

* Update database.py to resolve PR conflicts

* PR Comment Resolution

* Resolve conflicts in database.py for PR#570

* Updates to address PR#570 additional comments

* Optimize visibility handling for popup component

* Improve serial key handling in printers.py

Refactor serial key assignment to handle empty AMS serial gracefully.

* Implement error handling in serial number mapping

Add error handling for serial number mapping.

* Add migration to drop old ams_labels table

---------

Co-authored-by: MartinNYHC <mz@v8w.de>

backend/app/api/routes/inventory.py

@@ -13,6 +13,7 @@ from backend.app.core.auth import RequirePermissionIfAuthEnabled
 from backend.app.core.catalog_defaults import DEFAULT_COLOR_CATALOG, DEFAULT_SPOOL_CATALOG
 from backend.app.core.database import get_db
 from backend.app.core.permissions import Permission
+from backend.app.models.ams_label import AmsLabel
 from backend.app.models.color_catalog import ColorCatalogEntry
 from backend.app.models.spool import Spool
 from backend.app.models.spool_assignment import SpoolAssignment
@@ -638,6 +639,8 @@ async def list_assignments(
     _: User | None = RequirePermissionIfAuthEnabled(Permission.INVENTORY_READ),
 ):
     """List spool assignments, optionally filtered by printer."""
+    from backend.app.services.printer_manager import printer_manager
+
     query = select(SpoolAssignment).options(
         selectinload(SpoolAssignment.spool).selectinload(Spool.k_profiles),
         selectinload(SpoolAssignment.printer),
@@ -645,7 +648,55 @@ async def list_assignments(
     if printer_id is not None:
         query = query.where(SpoolAssignment.printer_id == printer_id)
     result = await db.execute(query)
-    return list(result.scalars().all())
+    assignments = list(result.scalars().all())
+
+    # Build (printer_id, ams_id) -> ams_serial map from live printer states.
+    # Fetch all statuses in one call rather than one get_status() call per printer.
+    serial_map: dict[tuple[int, int], str] = {}
+    seen_printer_ids: set[int] = {a.printer_id for a in assignments}
+    all_statuses = printer_manager.get_all_statuses()
+    for pid in seen_printer_ids:
+        state = all_statuses.get(pid)
+        if state and state.raw_data:
+            for ams_unit in state.raw_data.get("ams", []):
+                sn = str(ams_unit.get("sn") or ams_unit.get("serial_number") or "")
+                if sn:
+                    try:
+                        serial_map[(pid, int(ams_unit.get("id", 0)))] = sn
+                    except (ValueError, TypeError):
+                        continue
+              
+    # Fetch all relevant AMS labels keyed by serial number
+    all_serials = set(serial_map.values())
+    # Also include synthetic fallback keys for assignments without a known serial
+    synthetic_keys: dict[str, tuple[int, int]] = {}
+    for a in assignments:
+        if (a.printer_id, a.ams_id) not in serial_map:
+            synthetic = f"p{a.printer_id}a{a.ams_id}"
+            synthetic_keys[synthetic] = (a.printer_id, a.ams_id)
+            all_serials.add(synthetic)
+
+    label_by_serial: dict[str, str] = {}
+    if all_serials:
+        lbl_result = await db.execute(
+            select(AmsLabel).where(AmsLabel.ams_serial_number.in_(all_serials))
+        )
+        for lbl in lbl_result.scalars().all():
+            label_by_serial[lbl.ams_serial_number] = lbl.label
+
+    # Build response objects, attaching ams_label where available
+    responses: list[SpoolAssignmentResponse] = []
+    for a in assignments:
+        resp = SpoolAssignmentResponse.model_validate(a)
+        sn = serial_map.get((a.printer_id, a.ams_id))
+        if sn and sn in label_by_serial:
+            resp.ams_label = label_by_serial[sn]
+        elif not sn:
+            synthetic = f"p{a.printer_id}a{a.ams_id}"
+            resp.ams_label = label_by_serial.get(synthetic)
+        responses.append(resp)
+
+    return responses
 
 
 @router.post("/assignments", response_model=SpoolAssignmentResponse)

backend/app/api/routes/printers.py

@@ -12,9 +12,11 @@ from backend.app.core.auth import RequirePermissionIfAuthEnabled
 from backend.app.core.config import settings
 from backend.app.core.database import get_db
 from backend.app.core.permissions import Permission
+from backend.app.models.ams_label import AmsLabel
 from backend.app.models.printer import Printer
 from backend.app.models.slot_preset import SlotPresetMapping
 from backend.app.schemas.printer import (
+    AmsLabelBody,
     AMSTray,
     AMSUnit,
     HMSErrorResponse,
@@ -433,6 +435,10 @@ async def get_printer_status(
                     temp=ams_data.get("temp"),
                     is_ams_ht=is_ams_ht,
                     tray=trays,
+                    # Serial number: Bambu MQTT uses "sn" key on AMS unit objects
+                    serial_number=str(ams_data.get("sn") or ams_data.get("serial_number") or ""),
+                    # Firmware version: populated by _handle_version_info from info.module ams/* entries
+                    sw_ver=str(ams_data.get("sw_ver") or ""),
                 )
             )
 
@@ -1956,6 +1962,124 @@ async def reset_ams_slot(
     }
 
 
+@router.get("/{printer_id}/ams-labels")
+async def get_ams_labels(
+    printer_id: int,
+    _=RequirePermissionIfAuthEnabled(Permission.PRINTERS_READ),
+    db: AsyncSession = Depends(get_db),
+):
+    """Get all user-defined AMS labels for a printer, keyed by AMS unit ID.
+
+    Labels are stored by AMS serial number.  This endpoint resolves the current
+    serial-to-ams_id mapping from the live printer state so the response is still
+    keyed by ams_id for UI compatibility.
+    """
+    # Build serial -> ams_id map from live printer state
+    serial_to_ams_id: dict[str, int] = {}
+    state = printer_manager.get_status(printer_id)
+    if state and state.raw_data:
+        for ams_unit in state.raw_data.get("ams", []):
+            sn = str(ams_unit.get("sn") or ams_unit.get("serial_number") or "")
+            if sn:
+                serial_to_ams_id[sn] = int(ams_unit.get("id", 0))
+
+    # Collect all known serials for this printer (live + synthetic fallback keys)
+    serials_to_query = set(serial_to_ams_id.keys())
+
+    # Fetch labels for all known serials
+    labels: dict[int, str] = {}
+    if serials_to_query:
+        result = await db.execute(
+            select(AmsLabel).where(AmsLabel.ams_serial_number.in_(serials_to_query))
+        )
+        for lbl in result.scalars().all():
+            aid = serial_to_ams_id.get(lbl.ams_serial_number)
+            if aid is not None:
+                labels[aid] = lbl.label
+
+    # Also fetch labels stored under synthetic keys for this printer (backward compat)
+    # Collect all synthetic keys first, then query with a single IN clause.
+    if state and state.raw_data:
+        synthetic_key_to_aid: dict[str, int] = {
+            f"p{printer_id}a{int(ams_unit.get('id', 0))}": int(ams_unit.get("id", 0))
+            for ams_unit in state.raw_data.get("ams", [])
+            if int(ams_unit.get("id", 0)) not in labels
+        }
+        if synthetic_key_to_aid:
+            result = await db.execute(
+                select(AmsLabel).where(AmsLabel.ams_serial_number.in_(synthetic_key_to_aid.keys()))
+            )
+            for lbl in result.scalars().all():
+                aid = synthetic_key_to_aid.get(lbl.ams_serial_number)
+                if aid is not None:
+                    labels[aid] = lbl.label
+
+    return labels
+
+
+@router.put("/{printer_id}/ams-labels/{ams_id}")
+async def save_ams_label(
+    printer_id: int,
+    ams_id: int,
+    body: AmsLabelBody,
+    _=RequirePermissionIfAuthEnabled(Permission.PRINTERS_UPDATE),
+    db: AsyncSession = Depends(get_db),
+):
+    """Create or update the friendly name for a specific AMS unit.
+
+    When ``ams_serial`` is provided the label is stored under that serial number so
+    it survives the AMS being moved to a different printer.  When it is absent (e.g.
+    older firmware that does not report a serial) a synthetic key based on the
+    printer_id and ams_id is used as a fallback.
+    """
+    # Verify printer exists
+    result = await db.execute(select(Printer).where(Printer.id == printer_id))
+    if not result.scalar_one_or_none():
+        raise HTTPException(404, "Printer not found")
+
+    # Determine the serial key to store under
+    stripped = body.ams_serial.strip() if body.ams_serial else ""
+    serial_key = stripped if stripped else f"p{printer_id}a{ams_id}"
+
+    result = await db.execute(
+        select(AmsLabel).where(AmsLabel.ams_serial_number == serial_key)
+    )
+    existing = result.scalar_one_or_none()
+
+    if existing:
+        existing.label = body.label
+        existing.ams_id = ams_id
+    else:
+        db.add(AmsLabel(ams_serial_number=serial_key, ams_id=ams_id, label=body.label))
+
+    await db.commit()
+    return {"ams_id": ams_id, "label": body.label}
+
+
+@router.delete("/{printer_id}/ams-labels/{ams_id}")
+async def delete_ams_label(
+    printer_id: int,
+    ams_id: int,
+    ams_serial: str = Query(default="", max_length=50),
+    _=RequirePermissionIfAuthEnabled(Permission.PRINTERS_UPDATE),
+    db: AsyncSession = Depends(get_db),
+):
+    """Delete the friendly name for a specific AMS unit, reverting to the auto label."""
+    stripped = ams_serial.strip() if ams_serial else ""
+    serial_key = stripped if stripped else f"p{printer_id}a{ams_id}"
+
+    result = await db.execute(
+        select(AmsLabel).where(AmsLabel.ams_serial_number == serial_key)
+    )
+    existing = result.scalar_one_or_none()
+
+    if existing:
+        await db.delete(existing)
+        await db.commit()
+
+    return {"success": True}
+
+
 @router.post("/{printer_id}/debug/simulate-print-complete")
 async def debug_simulate_print_complete(
     printer_id: int,

backend/app/api/routes/settings.py

@@ -411,7 +411,7 @@ async def restore_backup(
 
     from fastapi import HTTPException
 
-    from backend.app.core.database import close_all_connections
+    from backend.app.core.database import close_all_connections, init_db, reinitialize_database
     from backend.app.services.virtual_printer import virtual_printer_manager
 
     base_dir = app_settings.base_dir
@@ -498,8 +498,11 @@ async def restore_backup(
                         logger.warning("Could not restore %s directory: %s", name, e)
                         skipped_dirs.append(name)
 
-            # 7. Note: Virtual printer and database will be reinitialized on restart
-            # Do NOT try to restart services here - the database session is closed
+            # 7. Reinitialize the database engine and apply schema migrations so that
+            # tables added after the backup was created (e.g. ams_labels) exist
+            # immediately, without requiring a manual restart.
+            await reinitialize_database()
+            await init_db()
 
             logger.info("Restore complete - restart required")
             message = "Backup restored successfully. Please restart Bambuddy for changes to take effect."