Gracious
90137dd111
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.
334 lines
11 KiB
Python
334 lines
11 KiB
Python
import abc
|
|
import dataclasses
|
|
import enum
|
|
import logging
|
|
import math
|
|
import socket
|
|
import struct
|
|
import time
|
|
|
|
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
|
|
|
|
|
|
class InfoEnum(enum.IntEnum):
|
|
DENSITY = 1
|
|
PRINTSPEED = 2
|
|
LABELTYPE = 3
|
|
LANGUAGETYPE = 6
|
|
AUTOSHUTDOWNTIME = 7
|
|
DEVICETYPE = 8
|
|
SOFTVERSION = 9
|
|
BATTERY = 10
|
|
DEVICESERIAL = 11
|
|
HARDVERSION = 12
|
|
|
|
|
|
class RequestCodeEnum(enum.IntEnum):
|
|
GET_INFO = 64 # 0x40
|
|
GET_RFID = 26 # 0x1A
|
|
HEARTBEAT = 220 # 0xDC
|
|
SET_LABEL_TYPE = 35 # 0x23
|
|
SET_LABEL_DENSITY = 33 # 0x21
|
|
START_PRINT = 1 # 0x01
|
|
END_PRINT = 243 # 0xF3
|
|
START_PAGE_PRINT = 3 # 0x03
|
|
END_PAGE_PRINT = 227 # 0xE3
|
|
ALLOW_PRINT_CLEAR = 32 # 0x20
|
|
SET_DIMENSION = 19 # 0x13
|
|
SET_QUANTITY = 21 # 0x15
|
|
GET_PRINT_STATUS = 163 # 0xA3
|
|
|
|
|
|
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:
|
|
raise NotImplementedError
|
|
|
|
@abc.abstractmethod
|
|
def write(self, data: bytes):
|
|
raise NotImplementedError
|
|
|
|
|
|
class BluetoothTransport(BaseTransport):
|
|
def __init__(self, address: str):
|
|
self._sock = socket.socket(
|
|
socket.AF_BLUETOOTH,
|
|
socket.SOCK_STREAM,
|
|
socket.BTPROTO_RFCOMM,
|
|
)
|
|
self._sock.connect((address, 1))
|
|
|
|
def read(self, length: int) -> bytes:
|
|
return self._sock.recv(length)
|
|
|
|
def write(self, data: bytes):
|
|
return self._sock.send(data)
|
|
|
|
|
|
class SerialTransport(BaseTransport):
|
|
def __init__(self, port: str = "auto"):
|
|
port = port if port != "auto" else self._detect_port()
|
|
self._serial = serial.Serial(port=port, baudrate=115200, timeout=0.5)
|
|
|
|
def _detect_port(self):
|
|
all_ports = list(list_comports())
|
|
if len(all_ports) == 0:
|
|
raise RuntimeError("No serial ports detected")
|
|
if len(all_ports) > 1:
|
|
msg = "Too many serial ports, please select specific one:"
|
|
for port, desc, hwid in all_ports:
|
|
msg += f"\n- {port} : {desc} [{hwid}]"
|
|
raise RuntimeError(msg)
|
|
return all_ports[0][0]
|
|
|
|
def read(self, length: int) -> bytes:
|
|
return self._serial.read(length)
|
|
|
|
def write(self, data: bytes):
|
|
return self._serial.write(data)
|
|
|
|
|
|
class PrinterClient:
|
|
def __init__(self, transport):
|
|
self._transport = transport
|
|
self._packetbuf = bytearray()
|
|
|
|
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()
|
|
if needs_print_clear:
|
|
self.allow_print_clear()
|
|
self.start_page_print()
|
|
self.set_dimension(image.height, image.width)
|
|
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()
|
|
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):
|
|
line_data = [img.getpixel((x, y)) for x in range(img.width)]
|
|
line_data = "".join("0" if pix == 0 else "1" for pix in line_data)
|
|
line_data = int(line_data, 2).to_bytes(math.ceil(img.width / 8), "big")
|
|
counts = (0, 0, 0) # It seems like you can always send zeros
|
|
header = struct.pack(">H3BB", y, *counts, 1)
|
|
pkt = NiimbotPacket(0x85, header + line_data)
|
|
yield pkt
|
|
|
|
def _recv(self):
|
|
packets = []
|
|
self._packetbuf.extend(self._transport.read(1024))
|
|
while len(self._packetbuf) > 4:
|
|
pkt_len = self._packetbuf[3] + 7
|
|
if len(self._packetbuf) >= pkt_len:
|
|
packet = NiimbotPacket.from_bytes(self._packetbuf[:pkt_len])
|
|
self._log_buffer("recv", packet.to_bytes())
|
|
packets.append(packet)
|
|
del self._packetbuf[:pkt_len]
|
|
return packets
|
|
|
|
def _send(self, packet):
|
|
self._transport.write(packet.to_bytes())
|
|
|
|
def _log_buffer(self, prefix: str, buff: bytes):
|
|
msg = ":".join(f"{i:#04x}"[-2:] for i in buff)
|
|
logging.debug(f"{prefix}: {msg}")
|
|
|
|
def _transceive(self, reqcode, data, respoffset=1):
|
|
respcode = respoffset + reqcode
|
|
packet = NiimbotPacket(reqcode, data)
|
|
self._log_buffer("send", packet.to_bytes())
|
|
self._send(packet)
|
|
for _ in range(6):
|
|
for packet in self._recv():
|
|
if packet.type == 219:
|
|
raise PrinterError(f"Printer rejected request {reqcode:#04x}")
|
|
elif packet.type == 0:
|
|
raise PrinterError(f"Request {reqcode:#04x} not supported")
|
|
elif packet.type == respcode:
|
|
return packet
|
|
time.sleep(0.1)
|
|
raise PrinterError(f"No response to request {reqcode:#04x}")
|
|
|
|
def get_info(self, key):
|
|
packet = self._transceive(RequestCodeEnum.GET_INFO, bytes((key,)), key)
|
|
match key:
|
|
case InfoEnum.DEVICESERIAL:
|
|
return packet.data.hex()
|
|
case InfoEnum.SOFTVERSION:
|
|
return _packet_to_int(packet) / 100
|
|
case InfoEnum.HARDVERSION:
|
|
return _packet_to_int(packet) / 100
|
|
case _:
|
|
return _packet_to_int(packet)
|
|
|
|
def get_rfid(self):
|
|
packet = self._transceive(RequestCodeEnum.GET_RFID, b"\x01")
|
|
data = packet.data
|
|
|
|
if data[0] == 0:
|
|
return None
|
|
uuid = data[0:8].hex()
|
|
idx = 8
|
|
|
|
barcode_len = data[idx]
|
|
idx += 1
|
|
barcode = data[idx : idx + barcode_len].decode()
|
|
|
|
idx += barcode_len
|
|
serial_len = data[idx]
|
|
idx += 1
|
|
serial = data[idx : idx + serial_len].decode()
|
|
|
|
idx += serial_len
|
|
total_len, used_len, type_ = struct.unpack(">HHB", data[idx:])
|
|
return {
|
|
"uuid": uuid,
|
|
"barcode": barcode,
|
|
"serial": serial,
|
|
"used_len": used_len,
|
|
"total_len": total_len,
|
|
"type": type_,
|
|
}
|
|
|
|
def heartbeat(self):
|
|
packet = self._transceive(RequestCodeEnum.HEARTBEAT, b"\x01")
|
|
closingstate = None
|
|
powerlevel = None
|
|
paperstate = None
|
|
rfidreadstate = None
|
|
|
|
match len(packet.data):
|
|
case 20:
|
|
paperstate = packet.data[18]
|
|
rfidreadstate = packet.data[19]
|
|
case 13:
|
|
closingstate = packet.data[9]
|
|
powerlevel = packet.data[10]
|
|
paperstate = packet.data[11]
|
|
rfidreadstate = packet.data[12]
|
|
case 19:
|
|
closingstate = packet.data[15]
|
|
powerlevel = packet.data[16]
|
|
paperstate = packet.data[17]
|
|
rfidreadstate = packet.data[18]
|
|
case 10:
|
|
closingstate = packet.data[8]
|
|
powerlevel = packet.data[9]
|
|
rfidreadstate = packet.data[8]
|
|
case 9:
|
|
closingstate = packet.data[8]
|
|
|
|
return {
|
|
"closingstate": closingstate,
|
|
"powerlevel": powerlevel,
|
|
"paperstate": paperstate,
|
|
"rfidreadstate": rfidreadstate,
|
|
}
|
|
|
|
def set_label_type(self, n):
|
|
assert 1 <= n <= 3
|
|
packet = self._transceive(RequestCodeEnum.SET_LABEL_TYPE, bytes((n,)), 16)
|
|
return bool(packet.data[0])
|
|
|
|
def set_label_density(self, n):
|
|
assert 1 <= n <= 5 # B21 has 5 levels, not sure for D11
|
|
packet = self._transceive(RequestCodeEnum.SET_LABEL_DENSITY, bytes((n,)), 16)
|
|
return bool(packet.data[0])
|
|
|
|
def start_print(self):
|
|
packet = self._transceive(RequestCodeEnum.START_PRINT, b"\x01")
|
|
return bool(packet.data[0])
|
|
|
|
def end_print(self):
|
|
packet = self._transceive(RequestCodeEnum.END_PRINT, b"\x01")
|
|
return bool(packet.data[0])
|
|
|
|
def start_page_print(self):
|
|
packet = self._transceive(RequestCodeEnum.START_PAGE_PRINT, b"\x01")
|
|
return bool(packet.data[0])
|
|
|
|
def end_page_print(self):
|
|
packet = self._transceive(RequestCodeEnum.END_PAGE_PRINT, b"\x01")
|
|
return bool(packet.data[0])
|
|
|
|
def allow_print_clear(self):
|
|
packet = self._transceive(RequestCodeEnum.ALLOW_PRINT_CLEAR, b"\x01", 16)
|
|
return bool(packet.data[0])
|
|
|
|
def set_dimension(self, h, w):
|
|
packet = self._transceive(
|
|
RequestCodeEnum.SET_DIMENSION, struct.pack(">HH", h, w)
|
|
)
|
|
return bool(packet.data[0])
|
|
|
|
def set_quantity(self, n):
|
|
packet = self._transceive(RequestCodeEnum.SET_QUANTITY, struct.pack(">H", n))
|
|
return bool(packet.data[0])
|
|
|
|
def get_print_status(self):
|
|
packet = self._transceive(RequestCodeEnum.GET_PRINT_STATUS, b"\x01", 16)
|
|
data = packet.data
|
|
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,
|
|
)
|