"""Maintenance tracking API routes.""" import logging from datetime import datetime, timezone from fastapi import APIRouter, Depends, HTTPException from sqlalchemy import select from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import selectinload from backend.app.core.auth import ( RequestPrinterScope, RequirePermissionIfAuthEnabled, RequirePrinterPermissionIfAuthEnabled, ) from backend.app.core.database import get_db from backend.app.core.permissions import Permission from backend.app.core.printer_scope import PrinterScope from backend.app.models.maintenance import MaintenanceHistory, MaintenanceType, PrinterMaintenance from backend.app.models.printer import Printer from backend.app.models.user import User from backend.app.schemas.maintenance import ( MaintenanceHistoryResponse, MaintenanceStatus, MaintenanceTypeCreate, MaintenanceTypeResponse, MaintenanceTypeUpdate, PerformMaintenanceRequest, PrinterMaintenanceOverview, PrinterMaintenanceResponse, PrinterMaintenanceUpdate, ) from backend.app.services.notification_service import notification_service from backend.app.utils.printer_models import get_rod_type, normalize_printer_model_id logger = logging.getLogger(__name__) router = APIRouter(prefix="/maintenance", tags=["maintenance"]) # Default maintenance types DEFAULT_MAINTENANCE_TYPES = [ # Carbon rod models only (X1/P1) # Note: carbon rods must NOT be lubricated — they use plain bearings # and lubrication degrades print quality. Only cleaning is offered. { "name": "Clean Carbon Rods", "description": "Wipe carbon rods with a dry cloth", "default_interval_hours": 100.0, "icon": "Sparkles", }, # Steel rod models only (P2S) { "name": "Lubricate Steel Rods", "description": "Apply lubricant to steel rods for smooth motion", "default_interval_hours": 50.0, "icon": "Droplet", }, { "name": "Clean Steel Rods", "description": "Wipe steel rods with a dry cloth", "default_interval_hours": 100.0, "icon": "Sparkles", }, # Linear rail models only (A1/H2) { "name": "Lubricate Linear Rails", "description": "Apply lubricant to linear rails for smooth motion", "default_interval_hours": 50.0, "icon": "Droplet", }, { "name": "Clean Linear Rails", "description": "Wipe linear rails with a dry cloth to remove dust and debris", "default_interval_hours": 100.0, "icon": "Sparkles", }, # Universal (all models) { "name": "Clean Nozzle/Hotend", "description": "Clean nozzle exterior and perform cold pull if needed", "default_interval_hours": 100.0, "icon": "Flame", }, { "name": "Check Belt Tension", "description": "Verify and adjust belt tension for X/Y axes", "default_interval_hours": 200.0, "icon": "Ruler", }, { "name": "Clean Build Plate", "description": "Deep clean build plate with IPA or soap", "default_interval_hours": 25.0, "icon": "Square", }, { "name": "Check PTFE Tube", "description": "Inspect PTFE tube for wear or discoloration", "default_interval_hours": 500.0, "icon": "Cable", }, ] # H3C tasks, from Bambu's H3C and AMS 3H maintenance guides. Those give an # interval only for the air filter (1440 h of printing); the others follow # Bambu's H2C table (/en/h2c/maintenance/period-maintenance), and the AMS 3H # ones, which have none published, use conservative defaults. All intervals are # editable per printer. _H3C_WIKI = "https://wiki.bambulab.com/en/h3c" H3C_MAINTENANCE_TYPES = [ { "name": "Lubricate X/Y Rails (H3C)", "description": "Wipe old oil and dust off the X and Y guide rails, put a drop of lubricant oil along " "each (oil only, no grease), then run Vibration Compensation", "default_interval_hours": 30.0, "interval_type": "days", "icon": "Droplet", "wiki_url": f"{_H3C_WIKI}/maintenance/lubricate-x-y-z-axis", }, { "name": "Grease Z Screws (H3C)", "description": "Wipe and lightly grease all three Z lead screws, wipe and oil the Z linear rods, " "then run Vibration Compensation", "default_interval_hours": 90.0, "interval_type": "days", "icon": "Droplet", "wiki_url": f"{_H3C_WIKI}/maintenance/lubricate-x-y-z-axis", }, { "name": "Oil XY Idler Pulleys (H3C)", "description": "Wipe the 12 XY idler pulleys and oil their bearings; sooner if the toolhead gets noisy", "default_interval_hours": 90.0, "interval_type": "days", "icon": "Cog", "wiki_url": f"{_H3C_WIKI}/maintenance/idle-gear", }, { "name": "Grease Toolhead (H3C)", "description": "Take off the toolhead front-left cover, grease the pitch change cam and the ToolLock pin, " "and turn both motors with the release socket to spread it", "default_interval_hours": 90.0, "interval_type": "days", "icon": "Wrench", "wiki_url": f"{_H3C_WIKI}/maintenance/periodic-maintenance", }, { "name": "Oil SwapTek Rack (H3C)", "description": "Wipe both ends of the SwapTek rack hanger, add a drop of lubricating oil to each bearing " "and push the hanger up and down", "default_interval_hours": 30.0, "interval_type": "days", "icon": "Layers", "wiki_url": f"{_H3C_WIKI}/maintenance/periodic-maintenance", }, { "name": "Clean Hotends (H3C)", "description": "Fetch each hotend, heat it above the filament's softening point and wipe it " "with the nozzle cleaning sponge", "default_interval_hours": 100.0, "interval_type": "hours", "icon": "Flame", "wiki_url": f"{_H3C_WIKI}/maintenance/periodic-maintenance", }, { "name": "Clean Extruder Wheel (H3C)", "description": "Check the toolhead extruder wheel for filament debris and brush it clean", "default_interval_hours": 7.0, "interval_type": "days", "icon": "CircleDot", "wiki_url": f"{_H3C_WIKI}/troubleshooting/extruder-cleaning-guide", }, { "name": "Clean Cameras (H3C)", "description": "Wipe the live view, toolhead and BirdsEye camera lenses with a cleaning wipe", "default_interval_hours": 90.0, "interval_type": "days", "icon": "Target", "wiki_url": f"{_H3C_WIKI}/maintenance/periodic-maintenance", }, { "name": "Clean Fans (H3C)", "description": "Brush the dust off the part cooling, hotend cooling and chamber cooling fans", "default_interval_hours": 90.0, "interval_type": "days", "icon": "Fan", "wiki_url": f"{_H3C_WIKI}/maintenance/periodic-maintenance", }, { "name": "Replace Air Filter (H3C)", "description": "Bambu rates the filter for 1440 hours of printing (about 60 days)", "default_interval_hours": 1440.0, "interval_type": "hours", "icon": "Filter", "wiki_url": f"{_H3C_WIKI}/maintenance/replace-air-filter", }, { "name": "Check Filament Tubes (H3C)", "description": "Look for tubes bent out of shape behind the purge wiper and for worn AMS 3H tubes; " "straighten or replace them", "default_interval_hours": 500.0, "interval_type": "hours", "icon": "Cable", "wiki_url": f"{_H3C_WIKI}/maintenance/periodic-maintenance", }, { "name": "AMS 3H: Check Desiccant", "description": "Check the desiccant on the drying chamber cover; dry or replace it when it is spent", "default_interval_hours": 30.0, "interval_type": "days", "icon": "Wind", "wiki_url": f"{_H3C_WIKI}/maintenance/ams_3h", }, { "name": "AMS 3H: Inlets & Gears", "description": "Clear filament debris from the inlet funnels, grease the loading gears and the " "air inlet/outlet flap gears", "default_interval_hours": 90.0, "interval_type": "days", "icon": "Box", "wiki_url": f"{_H3C_WIKI}/maintenance/ams_3h", }, { "name": "AMS 3H: Clean Dryer Fans", "description": "Unload all filament, lift the AMS 3H off and brush the dust off the heating unit fans", "default_interval_hours": 180.0, "interval_type": "days", "icon": "Thermometer", "wiki_url": f"{_H3C_WIKI}/maintenance/ams_3h", }, ] # System types of one model (short name): seeded and listed only while a # printer of that model exists, so other installs see just the defaults above. MODEL_MAINTENANCE_TYPES: dict[str, list[dict]] = {"H3C": H3C_MAINTENANCE_TYPES} _MODEL_ONLY_TYPES: dict[str, str] = {t["name"]: m for m, types in MODEL_MAINTENANCE_TYPES.items() for t in types} _H3C_MODELS = frozenset(["H3C"]) # Generic system types a model's own one replaces there. _REPLACED_ON_MODELS: dict[str, frozenset[str]] = { "Lubricate Linear Rails": _H3C_MODELS, # -> X/Y rails + Z lead screws "Clean Linear Rails": _H3C_MODELS, # -> X/Y rails (wiped before oiling) "Clean Nozzle/Hotend": _H3C_MODELS, # -> Clean Hotends "Check PTFE Tube": _H3C_MODELS, # -> Check Filament Tubes } # System types that only apply to printers with a specific rod/rail type. # "carbon" = X1/P1 series (carbon rods), "steel_rod" = P2S (steel rods), # "linear_rail" = A1/H2 series. Types not listed here apply to all printers. _ROD_TYPE_REQUIREMENTS: dict[str, str] = { "Clean Carbon Rods": "carbon", "Lubricate Steel Rods": "steel_rod", "Clean Steel Rods": "steel_rod", "Lubricate Linear Rails": "linear_rail", "Clean Linear Rails": "linear_rail", } def _short_model(printer_model: str | None) -> str: """ "N10" or "H3C" -> "H3C": the short name, uppercase, no spaces or dashes.""" if not printer_model: return "" short = normalize_printer_model_id(printer_model.strip()) or "" return short.upper().replace(" ", "").replace("-", "") def _should_apply_to_printer(type_name: str, printer_model: str | None) -> bool: """Check if a system maintenance type should apply to a given printer model.""" model = _short_model(printer_model) only_on = _MODEL_ONLY_TYPES.get(type_name) if only_on is not None: return model == only_on if model in _REPLACED_ON_MODELS.get(type_name, frozenset()): return False rod_requirement = _ROD_TYPE_REQUIREMENTS.get(type_name) if rod_requirement is None: return True # Not model-specific, applies to all rod_type = get_rod_type(printer_model) if rod_type is None: # Unknown model — default to carbon rods (legacy behavior) return rod_requirement == "carbon" return rod_type == rod_requirement async def get_printer_total_hours(db: AsyncSession, printer_id: int) -> float: """Calculate total active hours for a printer from runtime counter plus offset. Uses the runtime_seconds counter which tracks actual machine active time (RUNNING state only — paused time is excluded since maintenance intervals measure mechanical wear, not wall-clock active time, see #1521). """ # Get printer runtime and offset result = await db.execute( select(Printer.runtime_seconds, Printer.print_hours_offset).where(Printer.id == printer_id) ) row = result.one_or_none() if not row: return 0.0 runtime_seconds = row[0] or 0 offset = row[1] or 0.0 runtime_hours = runtime_seconds / 3600.0 return runtime_hours + offset async def _printer_models_present(db: AsyncSession) -> set[str]: """The short model names of the printers this install has.""" models = (await db.execute(select(Printer.model).distinct())).scalars().all() return {_short_model(m) for m in models if m} def _hidden_model_type(maint_type: MaintenanceType, models_present: set[str]) -> bool: """A model's own system type while no printer of that model exists.""" model = _MODEL_ONLY_TYPES.get(maint_type.name) return bool(maint_type.is_system) and model is not None and model not in models_present async def ensure_default_types(db: AsyncSession) -> None: """Ensure default maintenance types exist, remove stale/duplicate ones.""" result = await db.execute( select(MaintenanceType).where(MaintenanceType.is_system.is_(True)).order_by(MaintenanceType.id) ) existing = result.scalars().all() models_present = await _printer_models_present(db) wanted = DEFAULT_MAINTENANCE_TYPES + [ type_def for model, types in MODEL_MAINTENANCE_TYPES.items() if model in models_present for type_def in types ] # A model's types are kept (hidden) while no printer of it exists. default_names = {t["name"] for t in DEFAULT_MAINTENANCE_TYPES} | set(_MODEL_ONLY_TYPES) # Remove stale system types no longer in defaults (e.g. renamed types) # and deduplicate: if concurrent requests created the same type twice, # keep only the first (lowest id) and delete the rest. seen_names: set[str] = set() for t in existing: if t.name not in default_names or t.name in seen_names: await db.delete(t) else: seen_names.add(t.name) # Create any missing default types for type_def in wanted: if type_def["name"] not in seen_names: new_type = MaintenanceType( name=type_def["name"], description=type_def["description"], default_interval_hours=type_def["default_interval_hours"], interval_type=type_def.get("interval_type", "hours"), icon=type_def["icon"], wiki_url=type_def.get("wiki_url"), is_system=True, ) db.add(new_type) await db.commit() # ============== Maintenance Types ============== @router.get("/types", response_model=list[MaintenanceTypeResponse]) async def get_maintenance_types( db: AsyncSession = Depends(get_db, scope="function"), _: User | None = RequirePermissionIfAuthEnabled(Permission.MAINTENANCE_READ), ): """Get all maintenance types.""" await ensure_default_types(db) result = await db.execute( select(MaintenanceType) .where(MaintenanceType.is_deleted.is_(False)) .order_by(MaintenanceType.is_system.desc(), MaintenanceType.name) ) models_present = await _printer_models_present(db) return [t for t in result.scalars().all() if not _hidden_model_type(t, models_present)] @router.post("/types", response_model=MaintenanceTypeResponse) async def create_maintenance_type( data: MaintenanceTypeCreate, db: AsyncSession = Depends(get_db, scope="function"), _: User | None = RequirePermissionIfAuthEnabled(Permission.MAINTENANCE_CREATE), ): """Create a custom maintenance type.""" new_type = MaintenanceType( name=data.name, description=data.description, default_interval_hours=data.default_interval_hours, interval_type=data.interval_type, icon=data.icon, wiki_url=data.wiki_url, is_system=False, ) db.add(new_type) await db.commit() await db.refresh(new_type) return new_type @router.patch("/types/{type_id}", response_model=MaintenanceTypeResponse) async def update_maintenance_type( type_id: int, data: MaintenanceTypeUpdate, db: AsyncSession = Depends(get_db, scope="function"), _: User | None = RequirePermissionIfAuthEnabled(Permission.MAINTENANCE_UPDATE), ): """Update a maintenance type.""" result = await db.execute(select(MaintenanceType).where(MaintenanceType.id == type_id)) maint_type = result.scalar_one_or_none() if not maint_type: raise HTTPException(status_code=404, detail="Maintenance type not found") update_data = data.model_dump(exclude_unset=True) for key, value in update_data.items(): setattr(maint_type, key, value) await db.commit() await db.refresh(maint_type) return maint_type @router.delete("/types/{type_id}") async def delete_maintenance_type( type_id: int, db: AsyncSession = Depends(get_db, scope="function"), _: User | None = RequirePermissionIfAuthEnabled(Permission.MAINTENANCE_DELETE), ): """Delete a maintenance type.""" result = await db.execute(select(MaintenanceType).where(MaintenanceType.id == type_id)) maint_type = result.scalar_one_or_none() if not maint_type: raise HTTPException(status_code=404, detail="Maintenance type not found") if maint_type.is_system: maint_type.is_deleted = True await db.commit() return {"status": "deleted"} await db.delete(maint_type) await db.commit() return {"status": "deleted"} @router.post("/types/restore-defaults") async def restore_default_maintenance_types( db: AsyncSession = Depends(get_db, scope="function"), _: User | None = RequirePermissionIfAuthEnabled(Permission.MAINTENANCE_DELETE), ): """Restore deleted default maintenance types.""" await ensure_default_types(db) result = await db.execute( select(MaintenanceType).where(MaintenanceType.is_system.is_(True)).where(MaintenanceType.is_deleted.is_(True)) ) deleted_types = result.scalars().all() for maint_type in deleted_types: maint_type.is_deleted = False await db.commit() return {"restored": len(deleted_types)} # ============== Printer Maintenance ============== def maintenance_alert_values(item: MaintenanceStatus) -> tuple[float, float, str]: """What the MQTT relay's maintenance/alert reports: (value, threshold, unit). Print-hour items count hours since the last maintenance against the interval; time-based ones count days, and their interval_hours holds days. """ if item.interval_type == "days": return (item.days_since_maintenance or 0.0, item.interval_hours, "days") return (item.hours_since_maintenance, item.interval_hours, "hours") async def _get_or_create_maintenance_item(db: AsyncSession, printer_id: int, type_id: int) -> PrinterMaintenance: """The printer's row for this type, created with defaults if missing. Insert first, in a savepoint: two reads of a new printer (two dashboards, or a dashboard and the background check) each used to create the row. The unique index now refuses the second, which takes the first's (FO-C14). """ item = PrinterMaintenance( printer_id=printer_id, maintenance_type_id=type_id, enabled=True, last_performed_hours=0.0, ) try: async with db.begin_nested(): db.add(item) await db.flush() return item except IntegrityError: existing = await db.scalar( select(PrinterMaintenance).where( PrinterMaintenance.printer_id == printer_id, PrinterMaintenance.maintenance_type_id == type_id, ) ) if existing is None: raise return existing def _start_model_task_clock(item: PrinterMaintenance, maint_type: MaintenanceType, printer: Printer, now) -> None: """Count a model's own task from when the printer was added, not from "never". Otherwise a new printer's calendar tasks all open as overdue. Fills only an empty last_performed_at and writes no history row. """ if maint_type.name in _MODEL_ONLY_TYPES and item.last_performed_at is None: item.last_performed_at = printer.created_at or now.replace(tzinfo=None) async def _get_printer_maintenance_internal( printer_id: int, db: AsyncSession, commit: bool = True, ) -> PrinterMaintenanceOverview: """Internal helper to get maintenance overview for a specific printer.""" await ensure_default_types(db) # Get printer result = await db.execute(select(Printer).where(Printer.id == printer_id)) printer = result.scalar_one_or_none() if not printer: raise HTTPException(status_code=404, detail="Printer not found") total_hours = await get_printer_total_hours(db, printer_id) # Get all maintenance types result = await db.execute(select(MaintenanceType).where(MaintenanceType.is_deleted.is_(False))) all_types = result.scalars().all() # Get printer's maintenance items result = await db.execute( select(PrinterMaintenance) .where(PrinterMaintenance.printer_id == printer_id) .options(selectinload(PrinterMaintenance.maintenance_type)) ) existing_items = {item.maintenance_type_id: item for item in result.scalars().all()} maintenance_items = [] due_count = 0 warning_count = 0 now = datetime.now(timezone.utc) for maint_type in all_types: # Skip system types that don't apply to this printer model # (e.g., "Clean Carbon Rods" for H2D which has steel rods) if maint_type.is_system and not _should_apply_to_printer(maint_type.name, printer.model): continue item = existing_items.get(maint_type.id) default_interval_type = getattr(maint_type, "interval_type", "hours") or "hours" if item: _start_model_task_clock(item, maint_type, printer, now) interval = item.custom_interval_hours or maint_type.default_interval_hours # Use custom interval type if set, otherwise use type's default interval_type = getattr(item, "custom_interval_type", None) or default_interval_type enabled = item.enabled last_performed_hours = item.last_performed_hours last_performed_at = item.last_performed_at item_id = item.id else: # Only auto-create maintenance items for system types # Custom types need to be manually assigned per printer if not maint_type.is_system: continue # Create default entry for this printer/type, or take the one a # concurrent read just created. item = await _get_or_create_maintenance_item(db, printer_id, maint_type.id) _start_model_task_clock(item, maint_type, printer, now) interval = item.custom_interval_hours or maint_type.default_interval_hours interval_type = getattr(item, "custom_interval_type", None) or default_interval_type enabled = item.enabled last_performed_hours = item.last_performed_hours last_performed_at = item.last_performed_at item_id = item.id # Calculate status based on interval type if interval_type == "days": # Time-based: calculate days since last performed if last_performed_at: # DB stores naive datetimes; treat as UTC for comparison if last_performed_at.tzinfo is None: last_performed_at = last_performed_at.replace(tzinfo=timezone.utc) days_since = (now - last_performed_at).total_seconds() / 86400.0 else: # Never performed - consider it due days_since = interval + 1 days_until = interval - days_since is_due = days_until <= 0 is_warning = days_until <= (interval * 0.1) and not is_due # For compatibility, also set hours values (but they won't be primary) hours_since = total_hours - last_performed_hours hours_until = 0 # Not applicable for time-based else: # Print-hours based (default) hours_since = total_hours - last_performed_hours hours_until = interval - hours_since is_due = hours_until <= 0 is_warning = hours_until <= (interval * 0.1) and not is_due # Calculate days for reference if last_performed_at: if last_performed_at.tzinfo is None: last_performed_at = last_performed_at.replace(tzinfo=timezone.utc) days_since = (now - last_performed_at).total_seconds() / 86400.0 else: days_since = None days_until = None if enabled: if is_due: due_count += 1 elif is_warning: warning_count += 1 maintenance_items.append( MaintenanceStatus( id=item_id, printer_id=printer_id, printer_name=printer.name, printer_model=printer.model, maintenance_type_id=maint_type.id, maintenance_type_name=maint_type.name, maintenance_type_icon=maint_type.icon, maintenance_type_wiki_url=getattr(maint_type, "wiki_url", None), enabled=enabled, interval_hours=interval, interval_type=interval_type, current_hours=total_hours, hours_since_maintenance=hours_since, hours_until_due=hours_until, days_since_maintenance=days_since if interval_type == "days" else None, days_until_due=days_until if interval_type == "days" else None, is_due=is_due, is_warning=is_warning, last_performed_at=last_performed_at, ) ) if commit: await db.commit() return PrinterMaintenanceOverview( printer_id=printer_id, printer_name=printer.name, printer_model=printer.model, total_print_hours=total_hours, maintenance_items=maintenance_items, due_count=due_count, warning_count=warning_count, ) @router.get("/printers/{printer_id}", response_model=PrinterMaintenanceOverview) async def get_printer_maintenance( printer_id: int, db: AsyncSession = Depends(get_db, scope="function"), _: User | None = RequirePrinterPermissionIfAuthEnabled(Permission.MAINTENANCE_READ), ): """Get maintenance overview for a specific printer.""" return await _get_printer_maintenance_internal(printer_id, db, commit=True) @router.get("/overview", response_model=list[PrinterMaintenanceOverview]) async def get_all_maintenance_overview( db: AsyncSession = Depends(get_db, scope="function"), _: User | None = RequirePermissionIfAuthEnabled(Permission.MAINTENANCE_READ), printer_scope: PrinterScope = RequestPrinterScope, ): """Get maintenance overview for all active printers.""" await ensure_default_types(db) result = await db.execute(select(Printer).where(Printer.is_active.is_(True))) printers = [p for p in result.scalars().all() if printer_scope.allows(p.id)] overviews = [] for printer in printers: # Don't commit after each printer, commit once at the end overview = await _get_printer_maintenance_internal(printer.id, db, commit=False) overviews.append(overview) # Commit any new maintenance items created await db.commit() return overviews @router.patch("/items/{item_id}", response_model=PrinterMaintenanceResponse) async def update_printer_maintenance( item_id: int, data: PrinterMaintenanceUpdate, db: AsyncSession = Depends(get_db, scope="function"), _: User | None = RequirePermissionIfAuthEnabled(Permission.MAINTENANCE_UPDATE), printer_scope: PrinterScope = RequestPrinterScope, ): """Update a printer maintenance item (e.g., custom interval, enabled).""" result = await db.execute( select(PrinterMaintenance) .where(PrinterMaintenance.id == item_id) .options(selectinload(PrinterMaintenance.maintenance_type)) ) item = result.scalar_one_or_none() if not item or not printer_scope.allows(item.printer_id): raise HTTPException(status_code=404, detail="Maintenance item not found") update_data = data.model_dump(exclude_unset=True) for key, value in update_data.items(): setattr(item, key, value) await db.commit() await db.refresh(item) return item @router.post("/printers/{printer_id}/assign/{type_id}", response_model=PrinterMaintenanceResponse) async def assign_maintenance_type( printer_id: int, type_id: int, db: AsyncSession = Depends(get_db, scope="function"), _: User | None = RequirePrinterPermissionIfAuthEnabled(Permission.MAINTENANCE_CREATE), ): """Assign a maintenance type to a specific printer (for custom types).""" # Verify printer exists result = await db.execute(select(Printer).where(Printer.id == printer_id)) printer = result.scalar_one_or_none() if not printer: raise HTTPException(status_code=404, detail="Printer not found") # Verify maintenance type exists result = await db.execute(select(MaintenanceType).where(MaintenanceType.id == type_id)) maint_type = result.scalar_one_or_none() if not maint_type: raise HTTPException(status_code=404, detail="Maintenance type not found") # Check if already assigned result = await db.execute( select(PrinterMaintenance).where( PrinterMaintenance.printer_id == printer_id, PrinterMaintenance.maintenance_type_id == type_id, ) ) existing = result.scalar_one_or_none() if existing: raise HTTPException(status_code=400, detail="Maintenance type already assigned to this printer") # Create the assignment item = PrinterMaintenance( printer_id=printer_id, maintenance_type_id=type_id, enabled=True, last_performed_hours=0.0, ) db.add(item) try: await db.commit() except IntegrityError: # A concurrent assign (a double click) passed the check above too; the # unique index refuses this one (FO-C14). await db.rollback() raise HTTPException(status_code=400, detail="Maintenance type already assigned to this printer") from None # Re-fetch with relationship loaded for response serialization from sqlalchemy.orm import selectinload result = await db.execute( select(PrinterMaintenance) .options(selectinload(PrinterMaintenance.maintenance_type)) .where(PrinterMaintenance.id == item.id) ) item = result.scalar_one() return item @router.delete("/items/{item_id}") async def remove_maintenance_item( item_id: int, db: AsyncSession = Depends(get_db, scope="function"), _: User | None = RequirePermissionIfAuthEnabled(Permission.MAINTENANCE_DELETE), printer_scope: PrinterScope = RequestPrinterScope, ): """Remove a maintenance item (unassign a custom type from a printer).""" result = await db.execute( select(PrinterMaintenance) .where(PrinterMaintenance.id == item_id) .options(selectinload(PrinterMaintenance.maintenance_type)) ) item = result.scalar_one_or_none() if not item or not printer_scope.allows(item.printer_id): raise HTTPException(status_code=404, detail="Maintenance item not found") # Only allow removing custom (non-system) types if item.maintenance_type.is_system: raise HTTPException(status_code=400, detail="Cannot remove system maintenance types") await db.delete(item) await db.commit() return {"status": "removed"} @router.post("/items/{item_id}/perform", response_model=MaintenanceStatus) async def perform_maintenance( item_id: int, data: PerformMaintenanceRequest, db: AsyncSession = Depends(get_db, scope="function"), _: User | None = RequirePermissionIfAuthEnabled(Permission.MAINTENANCE_UPDATE), printer_scope: PrinterScope = RequestPrinterScope, ): """Mark maintenance as performed (reset the counter).""" result = await db.execute( select(PrinterMaintenance) .where(PrinterMaintenance.id == item_id) .options(selectinload(PrinterMaintenance.maintenance_type)) ) item = result.scalar_one_or_none() if not item or not printer_scope.allows(item.printer_id): raise HTTPException(status_code=404, detail="Maintenance item not found") # Get printer for name result = await db.execute(select(Printer).where(Printer.id == item.printer_id)) printer = result.scalar_one() # Get current hours current_hours = await get_printer_total_hours(db, item.printer_id) # Create history entry history = MaintenanceHistory( printer_maintenance_id=item.id, hours_at_maintenance=current_hours, notes=data.notes, ) db.add(history) # Update item item.last_performed_at = datetime.now(timezone.utc) item.last_performed_hours = current_hours await db.commit() # MQTT relay - publish maintenance reset try: from backend.app.services.mqtt_relay import mqtt_relay await mqtt_relay.on_maintenance_reset( printer_id=item.printer_id, printer_name=printer.name, maintenance_type=item.maintenance_type.name, ) except Exception: pass # Don't fail if MQTT fails # Calculate status interval = item.custom_interval_hours or item.maintenance_type.default_interval_hours # The item's own interval type first, as the status list reads it (dbcol-09). interval_type = item.custom_interval_type or getattr(item.maintenance_type, "interval_type", "hours") or "hours" hours_since = current_hours - item.last_performed_hours hours_until = interval - hours_since return MaintenanceStatus( id=item.id, printer_id=item.printer_id, printer_name=printer.name, printer_model=printer.model, maintenance_type_id=item.maintenance_type_id, maintenance_type_name=item.maintenance_type.name, maintenance_type_icon=item.maintenance_type.icon, maintenance_type_wiki_url=getattr(item.maintenance_type, "wiki_url", None), enabled=item.enabled, interval_hours=interval, interval_type=interval_type, current_hours=current_hours, hours_since_maintenance=hours_since, hours_until_due=hours_until if interval_type == "hours" else 0, days_since_maintenance=0 if interval_type == "days" else None, days_until_due=interval if interval_type == "days" else None, is_due=False, is_warning=False, last_performed_at=item.last_performed_at, ) @router.get("/items/{item_id}/history", response_model=list[MaintenanceHistoryResponse]) async def get_maintenance_history( item_id: int, db: AsyncSession = Depends(get_db, scope="function"), _: User | None = RequirePermissionIfAuthEnabled(Permission.MAINTENANCE_READ), printer_scope: PrinterScope = RequestPrinterScope, ): """Get maintenance history for a specific item.""" item_printer_id = ( await db.execute(select(PrinterMaintenance.printer_id).where(PrinterMaintenance.id == item_id)) ).scalar_one_or_none() if not printer_scope.allows(item_printer_id): raise HTTPException(status_code=404, detail="Maintenance item not found") result = await db.execute( select(MaintenanceHistory) .where(MaintenanceHistory.printer_maintenance_id == item_id) .order_by(MaintenanceHistory.performed_at.desc()) ) return result.scalars().all() @router.get("/summary") async def get_maintenance_summary( db: AsyncSession = Depends(get_db, scope="function"), _: User | None = RequirePermissionIfAuthEnabled(Permission.MAINTENANCE_READ), printer_scope: PrinterScope = RequestPrinterScope, ): """Get a summary of maintenance status across all printers.""" await ensure_default_types(db) result = await db.execute(select(Printer).where(Printer.is_active.is_(True))) printers = [p for p in result.scalars().all() if printer_scope.allows(p.id)] total_due = 0 total_warning = 0 printers_with_issues = [] for printer in printers: overview = await get_printer_maintenance(printer.id, db) total_due += overview.due_count total_warning += overview.warning_count if overview.due_count > 0 or overview.warning_count > 0: printers_with_issues.append( { "printer_id": printer.id, "printer_name": printer.name, "due_count": overview.due_count, "warning_count": overview.warning_count, } ) return { "total_due": total_due, "total_warning": total_warning, "printers_with_issues": printers_with_issues, } @router.patch("/printers/{printer_id}/hours") async def set_printer_hours( printer_id: int, total_hours: float, db: AsyncSession = Depends(get_db, scope="function"), _: User | None = RequirePrinterPermissionIfAuthEnabled(Permission.MAINTENANCE_UPDATE), ): """Set the total print hours for a printer (adjusts offset to match). The offset is calculated as: offset = total_hours - runtime_hours Where runtime_hours comes from the runtime_seconds counter that tracks actual machine active time (RUNNING state only — paused time excluded, #1521). """ # Get printer result = await db.execute(select(Printer).where(Printer.id == printer_id)) printer = result.scalar_one_or_none() if not printer: raise HTTPException(status_code=404, detail="Printer not found") # Get current runtime hours runtime_hours = (printer.runtime_seconds or 0) / 3600.0 # Calculate needed offset printer.print_hours_offset = max(0, total_hours - runtime_hours) await db.commit() # Check for maintenance items that need attention and send notification try: await ensure_default_types(db) overview = await _get_printer_maintenance_internal(printer_id, db, commit=True) items_needing_attention = [ { "name": item.maintenance_type_name, "is_due": item.is_due, "is_warning": item.is_warning, } for item in overview.maintenance_items if item.enabled and (item.is_due or item.is_warning) ] if items_needing_attention: await notification_service.on_maintenance_due(printer_id, printer.name, items_needing_attention, db) logger.info( f"Sent maintenance notification for printer {printer_id}: " f"{len(items_needing_attention)} items need attention" ) except Exception as e: logger.warning("Failed to send maintenance notification: %s", e) return { "printer_id": printer_id, "total_hours": total_hours, "runtime_hours": runtime_hours, "offset_hours": printer.print_hours_offset, }