This commit is contained in:
AndBondStyle
2023-10-30 02:03:02 +03:00
parent f4d66533b5
commit 3d48cb4a7b
14 changed files with 572 additions and 285 deletions
+1
View File
@@ -0,0 +1 @@
from .printer import BluetoothTransport, PrinterClient, SerialTransport
+69 -37
View File
@@ -1,44 +1,76 @@
import argparse
import printerclient
import printencoder
import re
import click
from PIL import Image
import time
# import math
# mm_to_px = lambda x: math.ceil(x / 25.4 * 203)
# px_to_mm = lambda x: math.ceil(x / 25.4 * 203)
from niimprint import BluetoothTransport, PrinterClient, SerialTransport
if __name__ == '__main__':
parser = argparse.ArgumentParser(
description="Niimbot printer client")
parser.add_argument('-a', '--address', required=True, help="MAC address of target device")
parser.add_argument('--no-check', action='store_true', help="Skips image check")
parser.add_argument('-d', '--density', type=int, default=2, help="Printer density (1~3)")
parser.add_argument('-t', '--type', type=int, default=1, help="Label type (1~3)")
parser.add_argument('-n', '--quantity', type=int, default=1, help="Number of copies")
parser.add_argument('image', help="PIL supported image file")
args = parser.parse_args()
img = Image.open(args.image)
if img.width / img.height > 1:
# rotate clockwise 90deg, upper line (left line) prints first.
img = img.transpose(Image.ROTATE_270)
assert args.no_check or (img.width == 96 and img.height < 600)
@click.command("print")
@click.option(
"-m",
"--model",
type=click.Choice(["b21", "d11"], False),
default="b21",
show_default=True,
help="Niimbot printer model",
)
@click.option(
"-c",
"--conn",
type=click.Choice(["usb", "bluetooth"]),
default="usb",
show_default=True,
help="Connection type",
)
@click.option(
"-a",
"--addr",
help="Bluetooth MAC address OR serial device path",
)
@click.option(
"-d",
"--density",
type=click.IntRange(1, 5),
default=5,
show_default=True,
help="Print density",
)
@click.option(
"-i",
"--image",
type=click.Path(exists=True),
required=True,
help="Image path",
)
def print_cmd(model, conn, addr, density, image):
assert model != "d11", "D11 support may be broken (test yourself)"
assert conn != "bluetooth", "Bluetooth support may be broken (test yourself)"
printer = printerclient.PrinterClient(args.address)
printer.set_label_type(args.type)
printer.set_label_density(args.density)
if conn == "bluetooth":
addr = addr.upper()
assert re.fullmatch(r"([0-9A-F]{2}:){5}([0-9A-F]{2})", addr), "Bad MAC address"
transport = BluetoothTransport(addr)
if conn == "usb":
port = addr if addr is not None else "auto"
transport = SerialTransport(port=port)
printer.start_print()
printer.allow_print_clear()
printer.start_page_print()
printer.set_dimension(img.height, img.width)
printer.set_quantity(args.quantity)
for pkt in printencoder.naive_encoder(img):
printer._send(pkt)
printer.end_page_print()
while (a := printer.get_print_status())['page'] != args.quantity:
# print(a)
time.sleep(0.1)
printer.end_print()
if model == "b21":
# This may be wrong, but B21 doesn't accept anything larger. It's just shy of
# 50mm * 8 px/mm = 400px (for vertical space it's always 8 px/mm), so lgtm.
max_height_px = 384
if model == "d11":
max_height_px = 100 # I don't have D11 to test
# Image is printed left-to-right. Generally, we expect image width to be larger
# than image height, because that's the usual sticker aspect ratio.
image = Image.open(image)
assert image.width > image.height, "Are you sure image rotation is right?"
assert image.height <= max_height_px, f"Image height too big for {model}"
printer = PrinterClient(transport)
printer.print_image(image, density=density)
if __name__ == "__main__":
print_cmd()
@@ -5,11 +5,11 @@ class NiimbotPacket:
@classmethod
def from_bytes(cls, pkt):
assert pkt[:2] == b'\x55\x55'
assert pkt[-2:] == b'\xaa\xaa'
assert pkt[:2] == b"\x55\x55"
assert pkt[-2:] == b"\xaa\xaa"
type_ = pkt[2]
len_ = pkt[3]
data = pkt[4:4+len_]
data = pkt[4 : 4 + len_]
checksum = type_ ^ len_
for i in data:
@@ -22,7 +22,9 @@ class NiimbotPacket:
checksum = self.type ^ len(self.data)
for i in self.data:
checksum ^= i
return bytes((0x55, 0x55, self.type, len(self.data), *self.data, checksum, 0xaa, 0xaa))
return bytes(
(0x55, 0x55, self.type, len(self.data), *self.data, checksum, 0xAA, 0xAA)
)
def __repr__(self):
return f"<NiimbotPacket type={self.type} data={self.data}>"
-29
View File
@@ -1,29 +0,0 @@
import PIL.Image as Image
import PIL.ImageOps as ImageOps
import struct
import niimbotpacket
import sys
if sys.version_info.minor >= 10:
def countbitsofbytes(data):
return int.from_bytes(data, 'big').bit_count()
else:
def countbitsofbytes(data):
n = int.from_bytes(data, 'big')
# https://stackoverflow.com/a/9830282
n = (n & 0x55555555) + ((n & 0xAAAAAAAA) >> 1)
n = (n & 0x33333333) + ((n & 0xCCCCCCCC) >> 2)
n = (n & 0x0F0F0F0F) + ((n & 0xF0F0F0F0) >> 4)
n = (n & 0x00FF00FF) + ((n & 0xFF00FF00) >> 8)
n = (n & 0x0000FFFF) + ((n & 0xFFFF0000) >> 16)
return n
def naive_encoder(img):
img_data = ImageOps.invert(img.convert("L")).convert("1").tobytes()
for x in range(img.height):
line_data = img_data[x*12:(x+1)*12]
counts = ( countbitsofbytes(line_data[i*4:(i+1)*4]) for i in range(3) )
header = struct.pack('>H3BB', x, *counts, 1)
pkt = niimbotpacket.NiimbotPacket(0x85, header+line_data)
yield pkt
+289
View File
@@ -0,0 +1,289 @@
import abc
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.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")
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 one via SERIAL_PORT env var:"
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):
# Internally, we're slicing the image vertically, so it's convenient to rotate
image = image.transpose(Image.ROTATE_270)
self.set_label_density(density)
self.set_label_type(1)
self.start_print()
# self.allow_print_clear() # Something unsupported in protocol decoding (B21)
self.start_page_print()
self.set_dimension(image.height, image.width)
# self.set_quantity(1) # Same thing (B21)
for pkt in self._encode_image(image):
self._send(pkt)
self.end_page_print()
while not self.end_print():
time.sleep(0.1)
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(" ", 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(msg)
def _transceive(self, reqcode, data, respoffset=1):
respcode = respoffset + reqcode
packet = NiimbotPacket(reqcode, data)
self._log_buffer("-->", packet.to_bytes())
self._send(packet)
resp = None
for _ in range(6):
for packet in self._recv():
if packet.type == 219:
raise ValueError
elif packet.type == 0:
raise NotImplementedError
elif packet.type == respcode:
resp = packet
if resp:
return resp
time.sleep(0.1)
return resp
def get_info(self, key):
if 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)
else:
return None
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, w, h):
packet = self._transceive(
RequestCodeEnum.SET_DIMENSION, struct.pack(">HH", w, h)
)
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)
page, progress1, progress2 = struct.unpack(">HBB", packet.data)
return {"page": page, "progress1": progress1, "progress2": progress2}
-195
View File
@@ -1,195 +0,0 @@
import niimbotpacket
import socket
import struct
import time
import enum
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
GET_RFID = 26
HEARTBEAT = 220
SET_LABEL_TYPE = 35
SET_LABEL_DENSITY = 33
START_PRINT = 1
END_PRINT = 243
START_PAGE_PRINT = 3
END_PAGE_PRINT = 227
ALLOW_PRINT_CLEAR = 32
SET_DIMENSION = 19
SET_QUANTITY = 21
GET_PRINT_STATUS = 163
_packet_to_int = lambda x: int.from_bytes(x.data, 'big')
# TODO REMOVE MAGIC NUMBER
class PrinterClient:
def __init__(self, address):
self._sock = socket.socket(socket.AF_BLUETOOTH, socket.SOCK_STREAM, socket.BTPROTO_RFCOMM)
self._sock.connect((address, 1))
self._packetbuf = bytearray()
def _recv(self):
packets = []
self._packetbuf.extend(self._sock.recv(1024))
while len(self._packetbuf) > 4:
pkt_len = self._packetbuf[3] + 7
if len(self._packetbuf) >= pkt_len:
packet = niimbotpacket.NiimbotPacket.from_bytes(self._packetbuf[:pkt_len])
# print('recv:',packet)
packets.append(packet)
del self._packetbuf[:pkt_len]
return packets
def _send(self, packet):
# print('send:',packet)
self._sock.send(packet.to_bytes())
def _transceive(self, reqcode, data, respoffset=1):
respcode = respoffset + reqcode
self._send(niimbotpacket.NiimbotPacket(reqcode, data))
resp = None
for _ in range(6):
for packet in self._recv():
if packet.type == 219:
raise ValueError
elif packet.type == 0:
raise NotImplementedError
elif packet.type == respcode:
resp = packet
if resp:
return resp
time.sleep(0.1)
return resp
def get_info(self, key):
if 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)
else:
return None
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 <= 3
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, w, h):
packet = self._transceive(RequestCodeEnum.SET_DIMENSION, struct.pack('>HH', w, h))
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)
page, progress1, progress2 = struct.unpack('>HBB', packet.data)
return {'page': page, 'progress1': progress1, 'progress2': progress2}