bambuddy/backend/app/schemas/printer.py

from datetime import datetime

from pydantic import BaseModel, Field, field_validator, model_validator

from backend.app.schemas.not_null import IgnoresNullForRequiredColumns
from backend.app.schemas.printer_location import normalize_location_name


def _mask_url_login(url: str | None) -> str | None:
    """A stored camera URL with its login secret masked.

    One URL, not a log line: the login runs to the last ``@`` before the
    first ``/`` after the scheme, whitespace, ``?`` and ``#`` included (when
    in doubt it is masked). The user part up to the first ``:`` survives;
    without a ``:`` the login may be a token and goes whole.
    """
    if not url or "://" not in url:
        return url
    start = url.index("://") + 3
    slash = url.find("/", start)
    end = len(url) if slash == -1 else slash
    if "@" not in url[start:end]:
        return url
    login, host = url[start:end].rsplit("@", 1)
    shown = f"{login.split(':', 1)[0]}:[REDACTED]" if ":" in login else "[REDACTED]"
    return f"{url[:start]}{shown}@{host}{url[end:]}"


class PrinterBase(BaseModel):
    name: str = Field(..., min_length=1, max_length=100)
    serial_number: str = Field(..., min_length=1, max_length=50)

    @field_validator("serial_number")
    @classmethod
    def _normalize_serial_number(cls, v: str) -> str:
        """Uppercase and trim the serial number.

        Bambu serial numbers are uppercase alphanumeric, and the MQTT report
        topic ``device/<serial>/report`` is case-sensitive. A serial entered
        in the wrong case (or with stray whitespace) connects and subscribes
        without error but never receives a message — the printer publishes to
        the correctly-cased topic, so every status field stays unknown (#1465).
        Normalising on input makes the subscribed topic always match.
        """
        normalized = v.strip().upper()
        if not normalized:
            raise ValueError("serial_number must not be blank")
        return normalized

    ip_address: str = Field(
        ...,
        max_length=253,
        pattern=r"^(\d{1,3}(\.\d{1,3}){3}|[a-zA-Z0-9]([a-zA-Z0-9\-]{0,61}[a-zA-Z0-9])?(\.[a-zA-Z0-9]([a-zA-Z0-9\-]{0,61}[a-zA-Z0-9])?)*)$",
    )
    model: str | None = None
    location: str | None = None  # Group/location name
    auto_archive: bool = True
    external_camera_url: str | None = None
    external_camera_type: str | None = None  # "mjpeg", "rtsp", "snapshot", "usb"
    external_camera_enabled: bool = False
    external_camera_snapshot_url: str | None = None  # Optional single-frame override; #1177
    camera_rotation: int = 0  # 0, 90, 180, 270 degrees
    camera_light_auto: bool = False  # #1655, used when camera_light_mode is "selected"


class PrinterCreate(PrinterBase):
    # access_code lives on the input shapes only — never on the default
    # PrinterResponse. Direct exposure on PRINTERS_READ would let a Viewer
    # connect to the printer's MQTT and bypass Bambuddy's RBAC.
    access_code: str = Field(..., min_length=1, max_length=20)

    # Input only, not on PrinterBase: SQLite never enforced the column width,
    # so a stored location may be longer, and the response must still read it.
    _location = field_validator("location")(normalize_location_name)


class PlateDetectionROI(BaseModel):
    """Region of interest for plate detection (percentages 0.0-1.0)."""

    x: float = Field(..., ge=0.0, le=1.0)  # X start %
    y: float = Field(..., ge=0.0, le=1.0)  # Y start %
    w: float = Field(..., ge=0.0, le=1.0)  # Width %
    h: float = Field(..., ge=0.0, le=1.0)  # Height %


class PrinterUpdate(IgnoresNullForRequiredColumns):
    __orm_table__ = "printers"
    name: str | None = None
    ip_address: str | None = Field(
        default=None,
        max_length=253,
        pattern=r"^(\d{1,3}(\.\d{1,3}){3}|[a-zA-Z0-9]([a-zA-Z0-9\-]{0,61}[a-zA-Z0-9])?(\.[a-zA-Z0-9]([a-zA-Z0-9\-]{0,61}[a-zA-Z0-9])?)*)$",
    )
    access_code: str | None = None
    model: str | None = None
    location: str | None = None
    _location = field_validator("location")(normalize_location_name)
    is_active: bool | None = None
    auto_archive: bool | None = None
    print_hours_offset: float | None = None
    # Wear cost per printing hour (#694); null or 0 turns it off.
    wear_cost_per_hour: float | None = Field(default=None, ge=0, le=100000)
    external_camera_url: str | None = None
    external_camera_type: str | None = None
    external_camera_enabled: bool | None = None
    external_camera_snapshot_url: str | None = None  # #1177
    camera_rotation: int | None = None  # 0, 90, 180, 270 degrees
    camera_light_auto: bool | None = None  # #1655
    plate_detection_enabled: bool | None = None
    plate_detection_roi: PlateDetectionROI | None = None


class PrinterResponse(PrinterBase):
    id: int
    is_active: bool
    nozzle_count: int = 1  # 1 or 2, auto-detected from MQTT
    # Whether the model is sold with both Standard and High Flow nozzles, so a
    # K-profile's flow type is a real choice rather than a meaningless field.
    # Derived from the model, not from nozzle_count — see
    # printer_models.supports_nozzle_flow_type.
    supports_nozzle_flow_type: bool = True
    print_hours_offset: float = 0.0
    wear_cost_per_hour: float | None = None  # #694
    external_camera_url: str | None = None
    external_camera_type: str | None = None
    external_camera_enabled: bool = False
    external_camera_snapshot_url: str | None = None  # #1177
    camera_rotation: int = 0  # 0, 90, 180, 270 degrees
    camera_light_auto: bool = False  # #1655
    plate_detection_enabled: bool = False
    plate_detection_roi: PlateDetectionROI | None = None
    created_at: datetime
    updated_at: datetime

    class Config:
        from_attributes = True

    @field_validator("external_camera_url", "external_camera_snapshot_url", mode="after")
    @classmethod
    def _camera_credentials(cls, value: str | None) -> str | None:
        """Readers see the camera URLs without the password in them."""
        return _mask_url_login(value)


class PrinterResponseWithSecret(PrinterResponse):
    """PrinterResponse + access_code and the full camera URLs. Returned ONLY to callers with
    PRINTERS_UPDATE (Admin / Operator JWTs, or single-trust auth-disabled mode).

    Viewers and API keys never receive this shape — they get the bare
    PrinterResponse without access_code, since holding the access_code lets
    the caller talk to the printer's MQTT directly and bypass Bambuddy's RBAC.
    """

    access_code: str

    @field_validator("external_camera_url", "external_camera_snapshot_url", mode="after")
    @classmethod
    def _camera_credentials(cls, value: str | None) -> str | None:
        """Editors keep the full camera URLs, to edit them."""
        return value


class HMSErrorResponse(BaseModel):
    code: str
    attr: int = 0  # Attribute value for constructing wiki URL
    module: int
    severity: int  # Bambu alert level: 1 error (stopped), 2 warning (paused), 3 notification, 0 invalid
    actions: list[str] = []  # List of user-facing action keys (e.g. "CHECK_FILAMENT")
    job_id: str | None = None  # Optional job ID for actions that require it (e.g. "CHECK_ASSISTANT")
    # Canonical hex identifier the firmware uses to match HMS-related commands.
    # 16 chars for `hms[]`-array faults (full 64-bit attr+code), 8 chars for
    # `print_error` faults. The frontend echoes this back as
    # HmsActionBody.print_error so we send the firmware-recognised key, not the
    # truncated short_code that historically caused silent command rejection
    # (#1830, H2D wrong-plate verification).
    full_code: str = ""
    # The bundled catalogue's sentence for this fault, so a client does not have
    # to carry its own copy of the same table to tell a user why a print halted
    # (#2926). English only and not localized — the catalogue ships one language.
    # Generated from Bambu Studio's HMS files, keyed by `full_code` and the
    # printer model (#2728). None when Bambu publishes no text for the code,
    # which it does for some codes it lists. A client should treat null as "no
    # text available", never as "no fault" — `full_code` is what identifies the
    # fault, and it is always present.
    description: str | None = None


def hms_error_responses(errors) -> list[HMSErrorResponse]:
    """A printer's live HMS faults (``PrinterState.hms_errors``) as API rows.

    Shared by the printer status route and the webhook status route, so a
    fault reads the same to the UI and to an API-key client (#2919).
    """
    return [
        HMSErrorResponse(
            code=e.code,
            attr=e.attr,
            module=e.module,
            severity=e.severity,
            actions=e.actions,
            job_id=e.job_id,
            full_code=e.full_code,
            description=e.description,
        )
        for e in (errors or [])
    ]


class AMSTray(BaseModel):
    id: int
    tray_color: str | None = None
    tray_type: str | None = None
    tray_sub_brands: str | None = None  # Full name like "PLA Basic", "PETG HF"
    tray_id_name: str | None = None  # Bambu filament ID like "A00-Y2" (can decode to color)
    tray_info_idx: str | None = None  # Filament preset ID like "GFA00"
    remain: int = 0
    k: float | None = None  # Pressure advance value (from tray or K-profile lookup)
    cali_idx: int | None = None  # Calibration index for K-profile lookup
    tag_uid: str | None = None  # RFID tag UID (any tag)
    tray_uuid: str | None = None  # Bambu Lab spool UUID (32-char hex)
    nozzle_temp_min: int | None = None  # Min nozzle temperature
    nozzle_temp_max: int | None = None  # Max nozzle temperature
    drying_temp: int | None = None  # RFID-recommended drying temp
    drying_time: int | None = None  # RFID-recommended drying time (hours)
    state: int | None = None  # AMS tray state: 9=empty, 10=spool present not loaded, 11=loaded
    # Firmware's authoritative "spool physically present" bit (from tray_exist_bits).
    # True for a non-RFID spool the firmware can't identify — the UI shows "?" rather
    # than "Empty" (#2527). None when the bitmask was unavailable (→ state-based fallback).
    exists: bool | None = None


class AMSUnit(BaseModel):
    id: int
    humidity: int | None = None
    temp: float | None = None
    is_ams_ht: bool = False  # True for AMS-HT (single spool), False for regular AMS (4 spools)
    tray: list[AMSTray] = []
    serial_number: str = ""  # AMS unit serial number (sn from MQTT)
    sw_ver: str = ""  # AMS firmware version (from get_version info.module)
    dry_time: int = 0  # Minutes remaining (0 = not drying, >0 = drying active)
    # True when dry_time > 0 but the countdown has not ticked for over
    # DRY_COUNTDOWN_STALL_SECONDS with no active dry_status phase: the timer is
    # set but no cycle is running (never started, or paused partway).
    dry_countdown_stalled: bool = False
    dry_status: int = 0  # 0=Off, 1=Checking, 2=Drying, 3=Cooling, 4=Stopping, 5=Error
    dry_sub_status: int = 0  # 0=Off, 1=Heating, 2=Dehumidify
    dry_sf_reason: list[int] = []  # Cannot-dry reasons from firmware (see CannotDryReason)
    dry_target_temp: int | None = None  # Active-cycle target °C (Bambu doesn't echo this)
    dry_filament: str | None = None  # Active-cycle filament name we sent
    module_type: str = ""  # "ams", "n3f", "n3s"


class NozzleInfoResponse(BaseModel):
    nozzle_type: str = ""  # "stainless_steel" or "hardened_steel"
    nozzle_diameter: str = ""  # e.g., "0.4"


class NozzleRackSlot(BaseModel):
    """H2C nozzle rack slot (6-position tool-changer dock)."""

    id: int = 0
    nozzle_type: str = ""
    nozzle_diameter: str = ""
    wear: int | None = None
    stat: int | None = None  # Nozzle status (e.g. mounted/docked)
    max_temp: int = 0  # Max temperature rating °C (0 = not set)
    serial_number: str = ""  # Nozzle serial number
    filament_color: str = ""  # RGBA hex ("00000000" = no filament)
    filament_id: str = ""  # Bambu filament ID
    filament_type: str = ""  # Material type (e.g. "PLA", "PETG")


class AmsLabelBody(BaseModel):
    label: str = Field(..., min_length=1, max_length=100)
    ams_serial: str = Field(default="", max_length=50)


class HmsActionBody(BaseModel):
    # Canonical hex identifier (HMSErrorResponse.full_code): 8 chars for
    # `print_error`-sourced faults, 16 chars for `hms[]`-array faults whose
    # full 64-bit code is the firmware's matching key. Length-bounded to
    # those two valid shapes to keep stray input from reaching the dispatcher.
    print_error: str = Field(..., min_length=8, max_length=16, pattern=r"^[0-9A-Fa-f]{8}([0-9A-Fa-f]{8})?$")
    # One of the HMSAction enum values. Length-capped to keep stray input from
    # reaching the dispatcher's `match` statement.
    action: str = Field(..., min_length=1, max_length=64)
    # The `subtask_id` snapshot from the HMSError that surfaced this dialog.
    # Bambu echoes it back in HMS-aware commands. Optional for idle errors.
    job_id: str | None = Field(default=None, max_length=64)


class FilaSwitchResponse(BaseModel):
    """Filament Track Switch (FTS) state — accessory that mediates AMS-to-extruder routing.

    When installed, the AMS info field reports bits 8-11 = 0xE (uninitialized)
    because slots are dynamically routed via the FTS rather than tied to a
    specific extruder. Frontend uses `installed` to suppress the per-extruder
    slot filter in the print modal. See #1162.
    """

    installed: bool = False
    # in[track] = currently loaded slot for that track (-1 = empty)
    in_slots: list[int] = []
    # out[track] = extruder this track terminates at (0 = right, 1 = left)
    out_extruders: list[int] = []
    stat: int = 0
    info: int = 0
    # Whether the switch is set up: every AMS bound to one of its two inlets.
    # A load cannot be routed until it is, so the UI blocks on this rather than
    # sending a command the firmware will drop.
    ready: bool = False


class ExtruderSlotResponse(BaseModel):
    """Which AMS slot one hotend is currently fed from.

    From ``device.extruder.info[i].snow``. Needed because ``tray_now`` is a
    single printer-wide value: on a dual-nozzle machine with both hotends
    loaded it names only one of them, so it cannot say which hotend holds a
    given slot.
    """

    # None when the hotend is not fed from any slot.
    ams_id: int | None = None
    slot_id: int | None = None
    has_filament: bool = False


class PrintOptionsResponse(BaseModel):
    """AI detection and print options from xcam data."""

    # Core AI detectors
    spaghetti_detector: bool = False
    print_halt: bool = False
    halt_print_sensitivity: str = "medium"  # Spaghetti sensitivity
    first_layer_inspector: bool = False
    printing_monitor: bool = False
    buildplate_marker_detector: bool = False
    allow_skip_parts: bool = False
    # Additional AI detectors (decoded from cfg bitmask)
    nozzle_clumping_detector: bool = True
    nozzle_clumping_sensitivity: str = "medium"
    pileup_detector: bool = True
    pileup_sensitivity: str = "medium"
    airprint_detector: bool = True
    airprint_sensitivity: str = "medium"
    auto_recovery_step_loss: bool = True
    filament_tangle_detect: bool = False


class PrinterStatus(BaseModel):
    id: int
    name: str
    connected: bool
    state: str | None = None
    current_print: str | None = None
    subtask_name: str | None = None
    gcode_file: str | None = None
    progress: float | None = None
    remaining_time: int | None = None
    layer_num: int | None = None
    total_layers: int | None = None
    temperatures: dict | None = None
    cover_url: str | None = None
    hms_errors: list[HMSErrorResponse] = []
    ams: list[AMSUnit] = []
    ams_exists: bool = False
    vt_tray: list[AMSTray] = []  # Virtual tray / external spool(s)
    sdcard: bool = False  # SD card inserted
    store_to_sdcard: bool = False  # Store sent files on SD card
    timelapse: bool = False  # Timelapse recording active
    ipcam: bool = False  # Live view enabled
    wifi_signal: int | None = None  # WiFi signal strength in dBm
    wired_network: bool = False  # Ethernet connection detected
    door_open: bool = False  # Enclosure door open (X1/P1S/P2S/H2*)
    nozzles: list[NozzleInfoResponse] = []  # Nozzle hardware info (index 0=left/primary, 1=right)
    nozzle_rack: list[NozzleRackSlot] = []  # H2C 6-nozzle tool-changer rack
    print_options: PrintOptionsResponse | None = None  # AI detection and print options
    # Calibration stage tracking
    stg_cur: int = -1  # Current stage number (-1 = not calibrating)
    stg_cur_name: str | None = None  # Human-readable current stage name
    stg: list[int] = []  # List of stage numbers in calibration sequence
    # Air conditioning mode (0=cooling, 1=heating)
    airduct_mode: int = 0
    # Print speed level (1=silent, 2=standard, 3=sport, 4=ludicrous)
    speed_level: int = 2
    # Chamber light on/off
    chamber_light: bool = False
    # Active extruder for dual nozzle (0=right, 1=left)
    active_extruder: int = 0
    # AMS mapping for dual nozzle: which AMS is connected to which nozzle
    ams_mapping: list[int] = []
    # Per-AMS extruder map: {ams_id: extruder_id} where 0=right, 1=left
    ams_extruder_map: dict[str, int] = {}
    # Filament Track Switch (FTS) accessory — when installed, AMS reports
    # bits 8-11 = 0xE (uninitialized) and routing is dynamic via the FTS. See #1162.
    fila_switch: FilaSwitchResponse | None = None
    # Per-AMS FTS inlet binding: {ams_id: "A" | "B"}, from AMS info bits 24-27.
    # Which of the switch's two inlets each AMS is plumbed into, as set on the
    # printer's "Manual AMS Setup" screen. Empty unless an FTS is installed —
    # an FTS-bound AMS reaches BOTH nozzles, so it has no entry in
    # ams_extruder_map and must not be labelled left or right.
    ams_switch_inlet: dict[str, str] = {}
    # Which AMS slot each hotend is fed from, keyed by extruder id as a string
    # ("0" = right/main, "1" = left/deputy). Empty on printers that do not
    # report ``device.extruder.info``.
    extruder_slots: dict[str, ExtruderSlotResponse] = {}
    # Currently loaded tray (global ID): 254 = external spool, 255 = no filament
    tray_now: int = 255
    # The trays this print has drawn from, in order: [[global tray ID, layer the
    # switch happened at], ...]. A tray outside the job's AMS mapping is a backup
    # spool the printer switched to (AMS Filament Backup); the queue card shows it
    # in place of the slot that ran out. Reset at the start of each print.
    tray_change_log: list[list[int]] = []
    # Runout / filament-replacement guidance (#2587). Populated only while the
    # print is PAUSED. Both are globalised tray IDs (ams_id*4+slot, or 128-135 for
    # AMS-HT, or 254 for external) so the frontend can highlight them with the same
    # logic it uses for tray_now:
    #   expected_tray = the slot the firmware now expects filament in (from tray_tar).
    #                   None when idle, not paused, or the slot can't be resolved
    #                   (multi-AMS ambiguity) — the UI then says "check the printer".
    #   previous_tray = the slot loaded before the pause, i.e. the one that ran out
    #                   (from tray_pre). None when unknown.
    expected_tray: int | None = None
    previous_tray: int | None = None
    # AMS status for filament change tracking
    # Main status: 0=idle, 1=filament_change, 2=rfid_identifying, 3=assist, 4=calibration
    ams_status_main: int = 0
    # Sub status: specific step within filament change (when main=1)
    # Known values: 4=retraction, 6=load verification, 7=purge
    ams_status_sub: int = 0
    # mc_print_sub_stage - filament change step indicator used by OrcaSlicer/BambuStudio
    mc_print_sub_stage: int = 0
    # Timestamp of last AMS data update (for RFID refresh detection)
    last_ams_update: float = 0.0
    # Number of printable objects in current print (for skip objects feature)
    printable_objects_count: int = 0
    # Fan speeds (0-100 percentage, None if not available for this model)
    cooling_fan_speed: int | None = None  # Part cooling fan
    big_fan1_speed: int | None = None  # Auxiliary fan
    big_fan2_speed: int | None = None  # Chamber/exhaust fan
    heatbreak_fan_speed: int | None = None  # Hotend heatbreak fan
    # Left auxiliary part cooling fan (optional P2S/X2D accessory, airduct part id 10).
    # None = not installed / not reported by this model.
    left_aux_fan_speed: int | None = None
    # Chamber exhaust fan present (P2S/X2D External Exhaust Fan kit; airduct part id 3).
    exhaust_fan_present: bool = False
    # Firmware version (from info.module[name="ota"].sw_ver)
    firmware_version: str | None = None
    # Developer LAN mode: True = enabled, False = disabled (MQTT encryption), None = unknown
    developer_mode: bool | None = None
    # AMS Filament Backup ("auto-switch" to a second spool when one runs out).
    # True = ON, False = OFF, None = unknown / unsupported (A1 family — protocol field
    # not yet identified). UI treats None as "status unavailable", not as a hard disable.
    ams_filament_backup: bool | None = None
    # Queue: printer is awaiting the user to acknowledge the build plate is cleared
    # after a finished/failed print. Persisted across restarts (#961).
    awaiting_plate_clear: bool = False
    # AMS drying support
    supports_drying: bool = False
    # AMS "Print While Drying" — drying mid-print. Verified per Bambu wiki release notes;
    # see _DRY_WHILE_PRINTING_MIN_FIRMWARE in printer_manager.py for the matrix.
    supports_drying_while_printing: bool = False
    # The AMS can dry, but only from the printer's own screen (P1 series, #2533).
    # supports_drying is False on these; the UI keeps the control visible but disabled
    # and says why, rather than dropping it without explanation.
    drying_screen_only: bool = False
    # Active chamber heater (responds to M141). True only for H2C/H2D/H2DPro/H2S/X2D.
    supports_chamber_heater: bool = False
    # Linked archive for the active print (resolved via subtask_id). Frontend uses
    # this to fetch plate metadata and show the plate name when the source 3MF is
    # multi-plate (#881 follow-up).
    current_archive_id: int | None = None
    # 1-indexed plate number parsed from gcode_file (e.g. /Metadata/plate_2.gcode).
    # Set for every active print regardless of plate count; the frontend decides
    # whether to render it based on current_archive_id's is_multi_plate flag.
    current_plate_id: int | None = None


class DiagnosticCheck(BaseModel):
    """One connection-diagnostic check result.

    ``id`` is a stable key (port_mqtt, port_ftps, port_rtsps,
    macos_local_network, network_mode, subnet, external_storage, mqtt_auth,
    developer_mode, printer_publishing); the frontend renders the localized
    title and fix text from id + status. ``params`` carries interpolation
    values (e.g. network mode, IP addresses) for that text.

    Not every check is emitted on every run: ``macos_local_network`` appears
    only on macOS, where it is the only platform it can say anything about.
    """

    id: str
    status: str  # "pass" | "fail" | "warn" | "skip"
    params: dict = Field(default_factory=dict)


class PrinterDiagnosticResult(BaseModel):
    """Result of a printer connection diagnostic run."""

    printer_id: int | None = None
    ip_address: str
    overall: str  # "ok" | "warnings" | "problems"
    checks: list[DiagnosticCheck]


class DiagnosticRequest(BaseModel):
    """Pre-save (Add Printer) connection diagnostic request.

    serial_number + access_code are optional: when both are present the
    diagnostic also probes MQTT credentials, otherwise only the
    network-level checks run.
    """

    ip_address: str
    serial_number: str | None = None
    access_code: str | None = None


class PrinterFilesDownloadRequest(BaseModel):
    """Printer paths selected for a bulk download."""

    paths: list[str] = Field(..., max_length=1000)
    sizes: dict[str, int] = Field(default_factory=dict, max_length=1000)

    @model_validator(mode="after")
    def _validate_sizes(self):
        """Validate optional FTP-reported sizes used for early rejection."""

        if self.sizes and set(self.sizes) != set(self.paths):
            raise ValueError("A size is required for every selected printer path")
        if any(size < 0 for size in self.sizes.values()):
            raise ValueError("Printer file sizes must not be negative")
        return self


class PrinterFilesJobRequest(PrinterFilesDownloadRequest):
    """Browser preparation request, including native download presentation."""

    filename: str = Field(default="printer-files.zip", min_length=1, max_length=255)
    as_zip: bool = True