From 8dbf94d9e070068ac26550d3a50ceb95187010ae Mon Sep 17 00:00:00 2001 From: kjy00302 Date: Thu, 16 Mar 2023 19:46:09 +0900 Subject: [PATCH] Initial commit --- .gitignore | 138 ++++++++++++++++++++++++++++ README.md | 19 ++++ niimprint/__main__.py | 44 +++++++++ niimprint/niimbotpacket.py | 28 ++++++ niimprint/printencoder.py | 29 ++++++ niimprint/printerclient.py | 179 +++++++++++++++++++++++++++++++++++++ requirements.txt | 1 + 7 files changed, 438 insertions(+) create mode 100644 .gitignore create mode 100644 README.md create mode 100644 niimprint/__main__.py create mode 100644 niimprint/niimbotpacket.py create mode 100644 niimprint/printencoder.py create mode 100644 niimprint/printerclient.py create mode 100644 requirements.txt diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..a81c8ee --- /dev/null +++ b/.gitignore @@ -0,0 +1,138 @@ +# Byte-compiled / optimized / DLL files +__pycache__/ +*.py[cod] +*$py.class + +# C extensions +*.so + +# Distribution / packaging +.Python +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +share/python-wheels/ +*.egg-info/ +.installed.cfg +*.egg +MANIFEST + +# PyInstaller +# Usually these files are written by a python script from a template +# before PyInstaller builds the exe, so as to inject date/other infos into it. +*.manifest +*.spec + +# Installer logs +pip-log.txt +pip-delete-this-directory.txt + +# Unit test / coverage reports +htmlcov/ +.tox/ +.nox/ +.coverage +.coverage.* +.cache +nosetests.xml +coverage.xml +*.cover +*.py,cover +.hypothesis/ +.pytest_cache/ +cover/ + +# Translations +*.mo +*.pot + +# Django stuff: +*.log +local_settings.py +db.sqlite3 +db.sqlite3-journal + +# Flask stuff: +instance/ +.webassets-cache + +# Scrapy stuff: +.scrapy + +# Sphinx documentation +docs/_build/ + +# PyBuilder +.pybuilder/ +target/ + +# Jupyter Notebook +.ipynb_checkpoints + +# IPython +profile_default/ +ipython_config.py + +# pyenv +# For a library or package, you might want to ignore these files since the code is +# intended to run in multiple environments; otherwise, check them in: +# .python-version + +# pipenv +# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control. +# However, in case of collaboration, if having platform-specific dependencies or dependencies +# having no cross-platform support, pipenv may install dependencies that don't work, or not +# install all needed dependencies. +#Pipfile.lock + +# PEP 582; used by e.g. github.com/David-OConnor/pyflow +__pypackages__/ + +# Celery stuff +celerybeat-schedule +celerybeat.pid + +# SageMath parsed files +*.sage.py + +# Environments +.env +.venv +env/ +venv/ +ENV/ +env.bak/ +venv.bak/ + +# Spyder project settings +.spyderproject +.spyproject + +# Rope project settings +.ropeproject + +# mkdocs documentation +/site + +# mypy +.mypy_cache/ +.dmypy.json +dmypy.json + +# Pyre type checker +.pyre/ + +# pytype static type analyzer +.pytype/ + +# Cython debug symbols +cython_debug/ diff --git a/README.md b/README.md new file mode 100644 index 0000000..3105e6e --- /dev/null +++ b/README.md @@ -0,0 +1,19 @@ +# (WIP) Niimbot printer client + + usage: niimprint [-h] -a ADDRESS [--no-check] [-d DENSITY] [-t TYPE] [-n QUANTITY] image + + Niimbot printer client + + positional arguments: + image PIL supported image file + + options: + -h, --help show this help message and exit + -a ADDRESS, --address ADDRESS + MAC address of target device + --no-check Skips image check + -d DENSITY, --density DENSITY + Printer density (1~3) + -t TYPE, --type TYPE Label type (1~3) + -n QUANTITY, --quantity QUANTITY + Number of copies diff --git a/niimprint/__main__.py b/niimprint/__main__.py new file mode 100644 index 0000000..28a5280 --- /dev/null +++ b/niimprint/__main__.py @@ -0,0 +1,44 @@ +import argparse +import printerclient +import printencoder + +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) + +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) + + printer = printerclient.PrinterClient(args.address) + printer.set_label_type(args.type) + printer.set_label_density(args.density) + + 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() diff --git a/niimprint/niimbotpacket.py b/niimprint/niimbotpacket.py new file mode 100644 index 0000000..951520d --- /dev/null +++ b/niimprint/niimbotpacket.py @@ -0,0 +1,28 @@ +class NiimbotPacket: + def __init__(self, type_, data): + self.type = type_ + self.data = data + + @classmethod + def from_bytes(cls, pkt): + assert pkt[:2] == b'\x55\x55' + assert pkt[-2:] == b'\xaa\xaa' + type_ = pkt[2] + len_ = pkt[3] + data = pkt[4:4+len_] + + checksum = type_ ^ len_ + for i in data: + checksum ^= i + assert checksum == pkt[-3] + + return cls(type_, data) + + def to_bytes(self): + 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)) + + def __repr__(self): + return f"" diff --git a/niimprint/printencoder.py b/niimprint/printencoder.py new file mode 100644 index 0000000..09be883 --- /dev/null +++ b/niimprint/printencoder.py @@ -0,0 +1,29 @@ +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 diff --git a/niimprint/printerclient.py b/niimprint/printerclient.py new file mode 100644 index 0000000..9c7a918 --- /dev/null +++ b/niimprint/printerclient.py @@ -0,0 +1,179 @@ +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 + +_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, respcode): + 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(64, bytes((key,)), 64+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(26, b'\x01', 27) + 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(220, b'\x01', 221) + 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(35, bytes((n,)), 51) + return bool(packet.data[0]) + + def set_label_density(self, n): + assert 1 <= n <= 3 + packet = self._transceive(33, bytes((n,)), 49) + return bool(packet.data[0]) + + def start_print(self): + packet = self._transceive(1, b'\x01', 2) + return bool(packet.data[0]) + + def end_print(self): + packet = self._transceive(243, b'\x01', 244) + return bool(packet.data[0]) + + def start_page_print(self): + packet = self._transceive(3, b'\x01', 4) + return bool(packet.data[0]) + + def end_page_print(self): + packet = self._transceive(227, b'\x01', 228) + return bool(packet.data[0]) + + def allow_print_clear(self): + packet = self._transceive(32, b'\x01', 48) + return bool(packet.data[0]) + + def set_dimension(self, w, h): + packet = self._transceive(19, struct.pack('>HH', w, h), 20) + return bool(packet.data[0]) + + def set_quantity(self, n): + packet = self._transceive(21, struct.pack('>H', n), 22) + return bool(packet.data[0]) + + def get_print_status(self): + packet = self._transceive(163, b'\x01', 179) + page, progress1, progress2 = struct.unpack('>HBB', packet.data) + return {'page': page, 'progress1': progress1, 'progress2': progress2} diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..0799398 --- /dev/null +++ b/requirements.txt @@ -0,0 +1 @@ +Pillow >= 8.0