Add model table, PrinterError, and port upstream PRs #28/#12

Library changes on top of upstream 35c41a7:

- models.py: MODELS table (max width / density / needs_print_clear per
  model) and prepare_print(), replacing the duplicated width/density
  logic in the CLI. Adds d101 (192px head, tested on 12mm tape).
- _transceive raises PrinterError instead of returning None (callers
  used to crash with AttributeError on a silent printer).
- PR #28: send ALLOW_PRINT_CLEAR / SET_QUANTITY for D-series models
  (labels taller than ~210px failed on D11 without them) and pause 1ms
  between line packets.
- PR #12: replace the fixed sleep after END_PAGE_PRINT with
  get_print_status() polling (30s deadline, fallback for firmwares that
  do not answer). get_print_status returns PrintStatus(finished,
  progress, feed_progress, error).
- set_dimension parameter names match their use.

Verified on a D101 over Bluetooth.

.gitignore

@@ -136,3 +136,6 @@ dmypy.json
 
 # Cython debug symbols
 cython_debug/
+
+# generated label output
+label.png

niimprint/__init__.py

@@ -1 +1,2 @@
+from .models import MODELS, PrinterError, PrinterModel, get_model, prepare_print
 from .printer import BluetoothTransport, PrinterClient, SerialTransport

niimprint/__main__.py

@@ -4,14 +4,20 @@ import re
 import click
 from PIL import Image
 
-from niimprint import BluetoothTransport, PrinterClient, SerialTransport
+from niimprint import (
+    MODELS,
+    BluetoothTransport,
+    PrinterClient,
+    SerialTransport,
+    prepare_print,
+)
 
 
 @click.command("print")
 @click.option(
     "-m",
     "--model",
-    type=click.Choice(["b1", "b18", "b21", "d11", "d110"], False),
+    type=click.Choice(list(MODELS), False),
     default="b21",
     show_default=True,
     help="Niimbot printer model",
@@ -73,23 +79,10 @@ def print_cmd(model, conn, addr, density, rotate, image, verbose):
         port = addr if addr is not None else "auto"
         transport = SerialTransport(port=port)
 
-    if model in ("b1", "b18", "b21"):
+    image, density = prepare_print(model, Image.open(image), density, int(rotate))
-        max_width_px = 384
-    if model in ("d11", "d110"):
-        max_width_px = 96
-
-    if model in ("b18", "d11", "d110") and density > 3:
-        logging.warning(f"{model.upper()} only supports density up to 3")
-        density = 3
-
-    image = Image.open(image)
-    if rotate != "0":
-        # PIL library rotates counter clockwise, so we need to multiply by -1
-        image = image.rotate(-int(rotate), expand=True)
-    assert image.width <= max_width_px, f"Image width too big for {model.upper()}"
 
     printer = PrinterClient(transport)
-    printer.print_image(image, density=density)
+    printer.print_image(image, density=density, model=model)
 
 
 if __name__ == "__main__":

niimprint/models.py

@@ -0,0 +1,68 @@
+import logging
+from dataclasses import dataclass
+
+from PIL import Image
+
+
+class PrinterError(Exception):
+    """Raised when the printer returns an error or does not respond."""
+
+
+@dataclass(frozen=True)
+class PrinterModel:
+    name: str
+    max_width_px: int
+    max_density: int
+    # D-series printers need ALLOW_PRINT_CLEAR / SET_QUANTITY before page data
+    # (labels taller than ~210px fail otherwise). B21 rejects these commands.
+    needs_print_clear: bool = False
+
+
+MODELS = {
+    m.name: m
+    for m in (
+        PrinterModel("b1", max_width_px=384, max_density=5),
+        PrinterModel("b18", max_width_px=384, max_density=3),
+        PrinterModel("b21", max_width_px=384, max_density=5),
+        PrinterModel("d11", max_width_px=96, max_density=3, needs_print_clear=True),
+        # D101 takes up to 25mm tape (192px head); tested on 12mm tape
+        PrinterModel("d101", max_width_px=192, max_density=3, needs_print_clear=True),
+        PrinterModel("d110", max_width_px=96, max_density=3, needs_print_clear=True),
+    )
+}
+
+
+def get_model(name: str) -> PrinterModel:
+    try:
+        return MODELS[name.lower()]
+    except KeyError:
+        raise ValueError(f"Unknown printer model: {name!r}") from None
+
+
+def prepare_print(
+    model: str | PrinterModel, image: Image.Image, density: int, rotate: int = 0
+) -> tuple[Image.Image, int]:
+    """Rotate/validate an image and clamp density for the given model.
+
+    Returns the (possibly rotated) image and the effective density.
+    """
+    if isinstance(model, str):
+        model = get_model(model)
+
+    if density > model.max_density:
+        logging.warning(
+            f"{model.name.upper()} only supports density up to {model.max_density}"
+        )
+        density = model.max_density
+
+    rotate = int(rotate)
+    if rotate % 360 != 0:
+        # PIL library rotates counter clockwise, so we need to multiply by -1
+        image = image.rotate(-rotate, expand=True)
+
+    if image.width > model.max_width_px:
+        raise ValueError(
+            f"Image width {image.width}px too big for {model.name.upper()} "
+            f"(max {model.max_width_px}px)"
+        )
+    return image, density

niimprint/printer.py

@@ -1,4 +1,5 @@
 import abc
+import dataclasses
 import enum
 import logging
 import math
@@ -10,6 +11,7 @@ import serial
 from PIL import Image, ImageOps
 from serial.tools.list_ports import comports as list_comports
 
+from niimprint.models import PrinterError, PrinterModel, get_model
 from niimprint.packet import NiimbotPacket
 
 
@@ -46,6 +48,14 @@ def _packet_to_int(x):
     return int.from_bytes(x.data, "big")
 
 
+@dataclasses.dataclass(frozen=True)
+class PrintStatus:
+    finished: bool
+    progress: int  # page print progress, %
+    feed_progress: int  # label feed progress, % (lags behind `progress`)
+    error: bool
+
+
 class BaseTransport(metaclass=abc.ABCMeta):
     @abc.abstractmethod
     def read(self, length: int) -> bytes:
@@ -100,21 +110,56 @@ class PrinterClient:
         self._transport = transport
         self._packetbuf = bytearray()
 
-    def print_image(self, image: Image, density: int = 3):
+    def print_image(
+        self,
+        image: Image,
+        density: int = 3,
+        model: str | PrinterModel | None = None,
+        timeout: float = 30.0,
+    ):
+        """Print a single image. `model` enables model-specific protocol quirks."""
+        if isinstance(model, str):
+            model = get_model(model)
+        needs_print_clear = model.needs_print_clear if model else False
+
         self.set_label_density(density)
         self.set_label_type(1)
         self.start_print()
-        # self.allow_print_clear()  # Something unsupported in protocol decoding (B21)
+        if needs_print_clear:
+            self.allow_print_clear()
         self.start_page_print()
         self.set_dimension(image.height, image.width)
-        # self.set_quantity(1)  # Same thing (B21)
+        if needs_print_clear:
+            self.set_quantity(1)
         for pkt in self._encode_image(image):
             self._send(pkt)
+            time.sleep(0.001)  # don't outrun the printer on tall labels (D11)
         self.end_page_print()
-        time.sleep(0.3)  # FIXME: Check get_print_status()
+        self._wait_for_print(timeout)
         while not self.end_print():
             time.sleep(0.1)
 
+    def _wait_for_print(self, timeout: float):
+        """Poll print status until the printer reports the page is finished."""
+        deadline = time.monotonic() + timeout
+        while time.monotonic() < deadline:
+            try:
+                status = self.get_print_status()
+            except PrinterError as exc:
+                # Some firmwares don't answer status queries; fall back to waiting
+                logging.debug(f"print status unavailable: {exc}")
+                time.sleep(0.3)
+                return
+            if status.error:
+                raise PrinterError("Printer reported an error during printing")
+            if status.finished:
+                return
+            logging.info(
+                f"Printing.. {status.progress}% (feed {status.feed_progress}%)"
+            )
+            time.sleep(0.1)
+        logging.warning(f"Print did not finish within {timeout}s, ending anyway")
+
     def _encode_image(self, image: Image):
         img = ImageOps.invert(image.convert("L")).convert("1")
         for y in range(img.height):
@@ -150,33 +195,28 @@ class PrinterClient:
         packet = NiimbotPacket(reqcode, data)
         self._log_buffer("send", packet.to_bytes())
         self._send(packet)
-        resp = None
         for _ in range(6):
             for packet in self._recv():
                 if packet.type == 219:
-                    raise ValueError
+                    raise PrinterError(f"Printer rejected request {reqcode:#04x}")
                 elif packet.type == 0:
-                    raise NotImplementedError
+                    raise PrinterError(f"Request {reqcode:#04x} not supported")
                 elif packet.type == respcode:
-                    resp = packet
+                    return packet
-            if resp:
-                return resp
             time.sleep(0.1)
-        return resp
+        raise PrinterError(f"No response to request {reqcode:#04x}")
 
     def get_info(self, key):
-        if packet := self._transceive(RequestCodeEnum.GET_INFO, bytes((key,)), key):
+        packet = self._transceive(RequestCodeEnum.GET_INFO, bytes((key,)), key)
-            match key:
+        match key:
-                case InfoEnum.DEVICESERIAL:
+            case InfoEnum.DEVICESERIAL:
-                    return packet.data.hex()
+                return packet.data.hex()
-                case InfoEnum.SOFTVERSION:
+            case InfoEnum.SOFTVERSION:
-                    return _packet_to_int(packet) / 100
+                return _packet_to_int(packet) / 100
-                case InfoEnum.HARDVERSION:
+            case InfoEnum.HARDVERSION:
-                    return _packet_to_int(packet) / 100
+                return _packet_to_int(packet) / 100
-                case _:
+            case _:
-                    return _packet_to_int(packet)
+                return _packet_to_int(packet)
-        else:
-            return None
 
     def get_rfid(self):
         packet = self._transceive(RequestCodeEnum.GET_RFID, b"\x01")
@@ -272,9 +312,9 @@ class PrinterClient:
         packet = self._transceive(RequestCodeEnum.ALLOW_PRINT_CLEAR, b"\x01", 16)
         return bool(packet.data[0])
 
-    def set_dimension(self, w, h):
+    def set_dimension(self, h, w):
         packet = self._transceive(
-            RequestCodeEnum.SET_DIMENSION, struct.pack(">HH", w, h)
+            RequestCodeEnum.SET_DIMENSION, struct.pack(">HH", h, w)
         )
         return bool(packet.data[0])
 
@@ -284,5 +324,10 @@ class PrinterClient:
 
     def get_print_status(self):
         packet = self._transceive(RequestCodeEnum.GET_PRINT_STATUS, b"\x01", 16)
-        page, progress1, progress2 = struct.unpack(">HBB", packet.data)
+        data = packet.data
-        return {"page": page, "progress1": progress1, "progress2": progress2}
+        return PrintStatus(
+            finished=bool(data[1]),
+            progress=data[2],
+            feed_progress=data[3] if len(data) > 3 else data[2],
+            error=bool(data[6]) if len(data) > 6 else False,
+        )