diff --git a/backend/app/printer_models/base.py b/backend/app/printer_models/base.py new file mode 100644 index 0000000..b6d1575 --- /dev/null +++ b/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., 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.""" diff --git a/backend/app/printer_models/registry.py b/backend/app/printer_models/registry.py new file mode 100644 index 0000000..62c3544 --- /dev/null +++ b/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}") diff --git a/backend/tests/unit/printer_models/conftest.py b/backend/tests/unit/printer_models/conftest.py new file mode 100644 index 0000000..6a63a39 --- /dev/null +++ b/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", []) diff --git a/backend/tests/unit/printer_models/test_registry.py b/backend/tests/unit/printer_models/test_registry.py new file mode 100644 index 0000000..a7fb20a --- /dev/null +++ b/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())