feat(printer-models): PrinterModel Protocol + ModelRegistry for plugin discovery

backend/app/printer_models/base.py

@@ -0,0 +1,50 @@
+"""Protocol contract for printer-model plugins.
+
+Each printer-model family (PT-Series, QL-Series, future series) lives in its
+own module under app.printer_models.<series>, implements this Protocol, and
+registers itself in app.printer_models.registry.ModelRegistry.
+
+The Protocol is `@runtime_checkable` so plugin authors and guard utilities
+can validate candidates with isinstance() if desired; the registry itself
+does not enforce this at registration.
+"""
+
+from __future__ import annotations
+
+from typing import Protocol, runtime_checkable
+
+from PIL import Image
+
+from app.models.tape import TapeSpec
+from app.services.status_block import StatusBlock
+
+
+@runtime_checkable
+class PrinterModel(Protocol):
+    """Per-model-family driver contract."""
+
+    model_id: str  # canonical, e.g. "PT-P750W"
+    pjl_signatures: list[str]  # PJL MDL substrings this plugin handles
+    snmp_model_oid_value_substr: str  # substring of the SNMP OID 2435.2.3.9.1.1.7.0 value
+    dpi: tuple[int, int]  # (width_dpi, height_dpi)
+    print_head_pins: int  # physical pin count
+
+    async def query_status(
+        self,
+        host: str,
+        port: int = 9100,
+        timeout_s: float = 5.0,
+    ) -> StatusBlock:
+        """Send ESC i S to the printer, read the 32-byte reply, return parsed status."""
+
+    def width_to_pixels(self, tape_spec: TapeSpec) -> int:
+        """Return the number of pixels along the print-head axis for the given tape."""
+
+    def build_print_job(
+        self,
+        image: Image.Image,
+        tape_spec: TapeSpec,
+        auto_cut: bool = True,
+        high_resolution: bool = False,
+    ) -> bytes:
+        """Encode an image into the Brother raster byte-stream for this model."""

backend/app/printer_models/registry.py

@@ -0,0 +1,69 @@
+"""Discover printer-model plugins by PJL or SNMP fingerprint."""
+
+from __future__ import annotations
+
+from typing import ClassVar
+
+from app.printer_models.base import PrinterModel
+
+
+class ModelNotFoundError(Exception):
+    """No registered plugin claims the given printer fingerprint."""
+
+
+class ModelRegistry:
+    """Class-level registry of PrinterModel plugins.
+
+    Plugins register themselves at import time. Production code calls
+    `find_by_pjl()` or `find_by_snmp_oid_value()` to resolve a discovered
+    printer to its driver.
+    """
+
+    _models: ClassVar[list[PrinterModel]] = []
+
+    @classmethod
+    def register(cls, model: PrinterModel) -> None:
+        """Append *model* to the registry. Duplicate registrations are not prevented."""
+        if any(not sig for sig in model.pjl_signatures):
+            raise ValueError(
+                f"PrinterModel {model.model_id!r} has an empty PJL signature; "
+                "empty substrings match every input and would shadow other plugins"
+            )
+        if not model.snmp_model_oid_value_substr:
+            raise ValueError(
+                f"PrinterModel {model.model_id!r} has an empty SNMP OID substring; "
+                "empty substrings match every input and would shadow other plugins"
+            )
+        cls._models.append(model)
+
+    @classmethod
+    def all(cls) -> list[PrinterModel]:
+        """Return a copy of all registered plugins."""
+        return list(cls._models)
+
+    @classmethod
+    def find_by_pjl(cls, pjl_string: str) -> PrinterModel:
+        """Match a plugin by PJL MDL substring.
+
+        Returns the first registered plugin whose signature matches. Registration
+        order determines priority if multiple plugins match.
+
+        Example pjl_string: 'MFG:Brother;CMD:PJL;MDL:PT-P750W;CLS:PRINTER;'
+        """
+        for model in cls._models:
+            for sig in model.pjl_signatures:
+                if sig in pjl_string:
+                    return model
+        raise ModelNotFoundError(f"No plugin matched PJL string: {pjl_string!r}")
+
+    @classmethod
+    def find_by_snmp_oid_value(cls, oid_value: str) -> PrinterModel:
+        """Match a plugin by SNMP model-OID value substring.
+
+        Returns the first registered plugin whose signature matches. Registration
+        order determines priority if multiple plugins match.
+        """
+        for model in cls._models:
+            if model.snmp_model_oid_value_substr in oid_value:
+                return model
+        raise ModelNotFoundError(f"No plugin matched SNMP OID value: {oid_value!r}")

backend/tests/unit/printer_models/conftest.py

@@ -0,0 +1,10 @@
+"""Test isolation for printer-model registry tests."""
+
+import pytest
+from app.printer_models.registry import ModelRegistry
+
+
+@pytest.fixture(autouse=True)
+def _clear_model_registry(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Reset the class-level _models list before each test."""
+    monkeypatch.setattr(ModelRegistry, "_models", [])

backend/tests/unit/printer_models/test_registry.py

@@ -0,0 +1,87 @@
+from __future__ import annotations
+
+from typing import ClassVar
+
+import pytest
+from app.models.tape import TapeSpec
+from app.printer_models.registry import ModelNotFoundError, ModelRegistry
+from app.services.status_block import StatusBlock
+from PIL import Image
+
+
+class FakePtModel:
+    model_id = "PT-P750W"
+    pjl_signatures: ClassVar[list[str]] = ["MDL:PT-P750W"]
+    snmp_model_oid_value_substr = "PT-P750W"
+    dpi: ClassVar[tuple[int, int]] = (180, 180)
+    print_head_pins = 128
+
+    async def query_status(
+        self,
+        host: str,
+        port: int = 9100,
+        timeout_s: float = 5.0,
+    ) -> StatusBlock:
+        raise NotImplementedError("test double — not exercised")
+
+    def width_to_pixels(self, tape_spec: TapeSpec) -> int:
+        raise NotImplementedError("test double — not exercised")
+
+    def build_print_job(
+        self,
+        image: Image.Image,
+        tape_spec: TapeSpec,
+        auto_cut: bool = True,
+        high_resolution: bool = False,
+    ) -> bytes:
+        raise NotImplementedError("test double — not exercised")
+
+
+def test_registry_register_and_find_by_pjl() -> None:
+    fake = FakePtModel()
+    ModelRegistry.register(fake)
+    pjl = "MFG:Brother;CMD:PJL;MDL:PT-P750W;CLS:PRINTER;"
+    assert ModelRegistry.find_by_pjl(pjl) is fake
+
+
+def test_registry_find_by_snmp_oid_value() -> None:
+    fake = FakePtModel()
+    ModelRegistry.register(fake)
+    oid_value = "Brother PT-P750W"
+    assert ModelRegistry.find_by_snmp_oid_value(oid_value) is fake
+
+
+def test_registry_unknown_pjl_raises() -> None:
+    pjl = "MDL:UnknownModel;"
+    with pytest.raises(ModelNotFoundError, match="UnknownModel"):
+        ModelRegistry.find_by_pjl(pjl)
+
+
+def test_registry_unknown_snmp_raises() -> None:
+    oid = "Unknown printer"
+    with pytest.raises(ModelNotFoundError, match="Unknown printer"):
+        ModelRegistry.find_by_snmp_oid_value(oid)
+
+
+def test_registry_all_returns_copy() -> None:
+    fake = FakePtModel()
+    ModelRegistry.register(fake)
+    snapshot = ModelRegistry.all()
+    snapshot.clear()  # mutating the copy must not affect the registry
+    assert len(ModelRegistry.all()) == 1
+
+
+def test_register_rejects_empty_pjl_signature() -> None:
+    class BadModel(FakePtModel):
+        pjl_signatures: ClassVar[list[str]] = [""]
+
+    with pytest.raises(ValueError, match="empty PJL signature"):
+        ModelRegistry.register(BadModel())
+
+
+def test_register_rejects_empty_snmp_substring() -> None:
+    class BadModel(FakePtModel):
+        snmp_model_oid_value_substr = ""
+
+    with pytest.raises(ValueError, match="empty SNMP OID substring"):
+        ModelRegistry.register(BadModel())