commit 9e06f3ebd037129e99d61525907cc2021a3eb866 Author: Alex Cube Date: Wed Oct 21 16:56:45 2026 +0300 LX-D02 BLE thermal printer: PC, Windows printer, Home Assistant - printer.py: protocol with flow control (pause / resend / done) - lxd02.py: print text or pictures from the command line - windows/: POS-58 driver + local ESC/POS server = a regular Windows printer, Explorer menu for pictures - custom_components/lxd02: Home Assistant integration with a persistent queue Co-Authored-By: Claude Opus 5.5 diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..f1239a1 --- /dev/null +++ b/.gitignore @@ -0,0 +1,3 @@ +__pycache__/ +*.log +*.bin diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..e3874b4 --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2026 Alex Cube + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/README.md b/README.md new file mode 100644 index 0000000..b30e773 --- /dev/null +++ b/README.md @@ -0,0 +1,186 @@ +# lx-d02 + +Печать на Bluetooth-термопринтере **LX-D02** (он же Funny Print, Xiqi, DOLEWA и другие +бренды) без приложения на смартфоне: + +- **как на обычном принтере Windows**: из Блокнота, браузера, Word, из меню «Печать»; +- **картинки правым кликом** в проводнике, растянутые на всю ширину ленты; +- **из командной строки**: `python lxd02.py "Привет"`; +- **из Home Assistant**: действие `lxd02.print_text` с очередью, которая переживает + выключенный принтер и перезапуск HA. + +Принтер работает только по BLE (Bluetooth Low Energy): USB-порт у него лишь для зарядки, +и Windows без этого проекта его как принтер не видит. + +Протокол разобран в [ValdikSS/printer-driver-funnyprint](https://github.com/ValdikSS/printer-driver-funnyprint), +управление потоком сделано по образцу [klimromanyuk/funnyprint](https://github.com/klimromanyuk/funnyprint). +Подробный рассказ, как это сделано: [блог «Доктор Лексиум»](https://www.lexium.ru/). + +## Что нужно + +- Python 3.9+ и Bluetooth на компьютере. +- `pip install -r requirements.txt` +- Принтер включён и **не подключён к телефону**: LX-D02 держит только одно соединение, + и пока открыто Funny Print, компьютер его не найдёт. + +Проверка: `python lxd02.py "Привет"` должен напечатать строку. + +На Python 3.9 bleak 1.x падает при импорте, поэтому в `requirements.txt` стоит `bleak<1.0`. +На более новом Python ограничение можно снять. + +## Печать из командной строки + +``` +python lxd02.py "Первая строка" "Вторая строка" +python lxd02.py фото.jpg скриншот.png +``` + +Текст печатается шрифтом DejaVu Sans (кириллица есть). Картинки растягиваются на ширину +ленты (48 мм, 384 точки), альбомные поворачиваются вдоль ленты, чтобы выйти крупнее, +полутона передаются растром. + +Принтер ищется по имени `LX-D02`. Чтобы не тратить на поиск несколько секунд, задайте MAC: +`set LXD02_MAC=AA:BB:CC:DD:EE:FF` (MAC покажет любой BLE-сканер, например nRF Connect +на телефоне). + +## Принтер в Windows + +Как это устроено: Windows печатает на принтер с драйвером **POS-58** (стандартный драйвер +58-мм чековых принтеров), драйвер раскладывает страницу на ленту 58 мм и отдаёт её +командами ESC/POS в порт `127.0.0.1:9100`. Там их принимает `windows/print_server.py` +и печатает по Bluetooth. + +Почему не встроенные драйверы Windows: «Microsoft Print to PDF» знает только офисные +форматы, и лист A4, ужатый до 48 мм, нечитаем. IPP Class Driver принимает только +стандартные названия бумаги, самое узкое из которых A6. + +### Драйвер + +Драйвер POS-58 обычно идёт с любым дешёвым 58-мм чековым принтером, ищите +«POS-58 driver» на сайте Xprinter или продавца. Проверено с `POS-58 11.2.0.0` +(Zjiang, `pos58.inf`). Сам принтер POS-58 не нужен, нужен только драйвер. + +### Установка + +В PowerShell **от имени администратора**: + +```powershell +Set-ExecutionPolicy -Scope Process Bypass +.\windows\install.ps1 +``` + +Скрипт: + +- добавляет принтер **LX-D02** на драйвере POS-58 и порту `127.0.0.1:9100`; +- ставит в Планировщик задание «LX-D02 print server» (сервер стартует при входе в систему); +- добавляет в меню картинок пункт «Print on LX-D02» и ярлык в «Отправить». + +Удаление: `.\windows\install.ps1 -Uninstall`. Журнал: `windows\print_server.log`. + +### Советы + +- **Бумага.** У драйвера по умолчанию лента 3276 мм. Пусть вас не пугает превью: сервер + срезает белое сверху и снизу и сжимает пустые промежутки до 6 мм. Для удобного превью + выберите в настройках печати размер 58×210. +- **Поля.** Ставьте поля «Минимальные» (в браузере) или 0 (в редакторах). Левое поле + сервер убирает сам, но правое срежет драйвер: печатная полоса 48 мм начинается + от левого края листа, а браузер с обычными полями сдвигает текст вправо. +- **Картинки** лучше печатать правым кликом, а не из браузера или «Фотографий»: + те раскладывают картинку по листу, и она выходит мелкой или обрезанной. +- Странный результат? Последнее задание сохраняется в `windows\last_job.bin`, + `python windows\print_server.py --test windows\last_job.bin` покажет его картинкой. + +## Home Assistant + +Нужен Bluetooth на хосте HA (встроенный адаптер или прокси ESPHome). + +Скопируйте `custom_components/lxd02` в `/config/custom_components/`, добавьте +в `configuration.yaml` и перезапустите HA: + +```yaml +lxd02: + mac: AA:BB:CC:DD:EE:FF +``` + +Действие: + +```yaml +action: lxd02.print_text +data: + text: "Привет из Home Assistant" + size: 28 # размер шрифта, 10..120 + density: 3 # плотность, 0..7 + feed: 40 # прогон ленты после текста + separator: true # линия под текстом +``` + +Задания попадают в очередь на диске (`/config/.storage/lxd02`). Если принтер выключен, +вне зоны или занят телефоном, задание ждёт, и очередь печатается, как только HA снова +увидит принтер. Пока очередь не пуста, HA пробует раз в минуту. Действие не падает, +если принтера нет: ищите `LX-D02` в журнале HA. + +### Пример: чат MeshCore на ленте + +Печатает входящие сообщения канала `my_channel` из интеграции MeshCore: отправитель +и время, ниже текст, под ним черта. Свои исходящие пропускаются: интеграция присылает +их заново на каждое обновление статуса доставки. + +```yaml +triggers: +- trigger: event + event_type: meshcore_message + event_data: + message_type: channel +conditions: +- condition: template + value_template: "{{ trigger.event.data.channel | lower == 'my_channel' and not trigger.event.data.outgoing | default(false) }}" +actions: +- action: lxd02.print_text + continue_on_error: true + data: + size: 24 + separator: true + text: "{{ trigger.event.data.sender_name }} · {{ now().strftime('%d.%m %H:%M') }}\n{{ trigger.event.data.message }}" +mode: queued +max: 20 +``` + +### Если Bluetooth на хосте зависает + +На ТВ-приставках встроенный чип раз в несколько недель может зависнуть +(`hci0: Opcode 0x0c03 failed: -110` в `dmesg`, HA пишет `NotReady`). Помогает +перепривязать чип к драйверу. Пример сторожа на systemd для чипа на UART (`hci_uart_bcm`): + +```sh +#!/bin/sh +# /usr/local/sbin/bt-watchdog.sh: run every minute by bt-watchdog.timer +D=/sys/bus/serial/drivers/hci_uart_bcm +up() { hciconfig hci0 2>/dev/null | grep -q "UP RUNNING"; } +up && exit 0 +sleep 20 # HA/BlueZ may be resetting the adapter itself +up && exit 0 +logger -t bt-watchdog "hci0 down, rebinding serial0-0" +echo serial0-0 > $D/unbind 2>/dev/null +sleep 2 +echo serial0-0 > $D/bind +sleep 8 +timeout 5 bluetoothctl power on >/dev/null 2>&1 +``` + +Имена драйвера и устройства у вас могут отличаться: смотрите `/sys/bus/serial/drivers/`. + +## Протокол коротко + +| | | +|---|---| +| Имя в BLE-рекламе | `LX-D02` | +| Сервис / запись / уведомления | `FFE6` / `FFE1` / `FFE2` | +| Ширина печати | 384 точки, 48 байт на строку, по две строки в пакете из 96 байт | +| Рукопожатие | принтер шлёт вызов `5a0a`, ответ `5a0b` + байт CRC16 от MAC × 10 | +| Управление потоком | `5a08` пауза (буфер полон), `5a05` + номер строки (дослать с неё), `5a06` готово | + +Весь протокол: `custom_components/lxd02/printer.py`, самопроверка `python printer.py`. + +## Лицензия + +MIT, см. `LICENSE`. diff --git a/custom_components/lxd02/DejaVuSans.ttf b/custom_components/lxd02/DejaVuSans.ttf new file mode 100644 index 0000000..5267218 Binary files /dev/null and b/custom_components/lxd02/DejaVuSans.ttf differ diff --git a/custom_components/lxd02/__init__.py b/custom_components/lxd02/__init__.py new file mode 100644 index 0000000..214951b --- /dev/null +++ b/custom_components/lxd02/__init__.py @@ -0,0 +1,107 @@ +"""LX-D02 thermal printer: service lxd02.print_text. + +Jobs go to a persistent queue (.storage/lxd02). If the printer is off, they wait +and print when it shows up in Bluetooth again. + +configuration.yaml: + lxd02: + mac: AA:BB:CC:DD:EE:FF +""" +import asyncio +from datetime import timedelta +import logging + +from bleak.exc import BleakError +import voluptuous as vol +from bleak_retry_connector import BleakClientWithServiceCache, establish_connection + +from homeassistant.components import bluetooth +from homeassistant.core import HomeAssistant, ServiceCall, callback +from homeassistant.exceptions import HomeAssistantError +import homeassistant.helpers.config_validation as cv +from homeassistant.helpers.event import async_track_time_interval +from homeassistant.helpers.storage import Store + +from .printer import print_payloads, render, to_payloads + +DOMAIN = "lxd02" +_LOGGER = logging.getLogger(__name__) +RETRY = timedelta(seconds=60) + +CONFIG_SCHEMA = vol.Schema( + {DOMAIN: vol.Schema({vol.Required("mac"): cv.string})}, extra=vol.ALLOW_EXTRA +) + +PRINT_SCHEMA = vol.Schema( + { + vol.Required("text"): cv.string, + vol.Optional("size", default=28): vol.All(vol.Coerce(int), vol.Range(min=10, max=120)), + vol.Optional("density", default=3): vol.All(vol.Coerce(int), vol.Range(min=0, max=7)), + vol.Optional("feed", default=40): vol.All(vol.Coerce(int), vol.Range(min=0, max=500)), + vol.Optional("separator", default=False): cv.boolean, + } +) + + +async def async_setup(hass: HomeAssistant, config) -> bool: + mac = config[DOMAIN]["mac"].upper() + lock = asyncio.Lock() + store = Store(hass, 1, DOMAIN) + queue = await store.async_load() or [] + + async def print_job(device, job): + img = await hass.async_add_executor_job( + render, job["text"], job["size"], job["separator"] + ) + payloads = to_payloads(img) + try: + client = await establish_connection(BleakClientWithServiceCache, device, "LX-D02") + except BleakError as err: + raise HomeAssistantError(f"LX-D02 connect failed: {err!r}") from err + try: + await print_payloads(client, mac, payloads, job["density"], job["feed"]) + except (asyncio.TimeoutError, RuntimeError, BleakError) as err: + raise HomeAssistantError(f"LX-D02 print failed: {err!r}") from err + finally: + await client.disconnect() + + async def flush(): + async with lock: + while queue: + device = bluetooth.async_ble_device_from_address(hass, mac, connectable=True) + if device is None: + _LOGGER.info("LX-D02 not in range, %d job(s) queued", len(queue)) + return + try: + await print_job(device, queue[0]) + except HomeAssistantError as err: + _LOGGER.warning("%s; %d job(s) stay queued", err, len(queue)) + return + except Exception: # bad job (render etc.): drop it, don't block the queue + _LOGGER.exception("LX-D02 job dropped: %r", queue[0]) + queue.pop(0) + await store.async_save(queue) + + async def print_text(call: ServiceCall) -> None: + queue.append(dict(call.data)) + await store.async_save(queue) + await flush() + + @callback + def seen(service_info, change) -> None: + if queue and not lock.locked(): + hass.async_create_task(flush()) + + # HA calls `seen` only when the advert changes or the printer comes back after + # being marked unavailable, not on every packet. So a failed attempt (or a replay + # of stale history right after HA restart) is never retried by adverts alone: + # the timer retries while the queue is not empty. + async_track_time_interval(hass, callback(lambda _: seen(None, None)), RETRY) + bluetooth.async_register_callback( + hass, + seen, + bluetooth.BluetoothCallbackMatcher(address=mac, connectable=True), + bluetooth.BluetoothScanningMode.ACTIVE, + ) + hass.services.async_register(DOMAIN, "print_text", print_text, schema=PRINT_SCHEMA) + return True diff --git a/custom_components/lxd02/manifest.json b/custom_components/lxd02/manifest.json new file mode 100644 index 0000000..4bc4f32 --- /dev/null +++ b/custom_components/lxd02/manifest.json @@ -0,0 +1,10 @@ +{ + "domain": "lxd02", + "name": "LX-D02 Thermal Printer", + "version": "0.2.0", + "documentation": "https://git.08h.ru/alex/lx-d02", + "dependencies": ["bluetooth"], + "requirements": [], + "codeowners": [], + "iot_class": "local_push" +} diff --git a/custom_components/lxd02/printer.py b/custom_components/lxd02/printer.py new file mode 100644 index 0000000..8ae3192 --- /dev/null +++ b/custom_components/lxd02/printer.py @@ -0,0 +1,143 @@ +"""LX-D02 (FunnyPrint) protocol. No Home Assistant imports: also used by ../../lxd02.py. + +Protocol from github.com/ValdikSS/printer-driver-funnyprint. +""" +import asyncio +from pathlib import Path + +from PIL import Image, ImageDraw, ImageFont + +WRITE = "0000ffe1-0000-1000-8000-00805f9b34fb" +NOTIFY = "0000ffe2-0000-1000-8000-00805f9b34fb" +WIDTH = 384 # dots, 48 bytes per row +FONT = Path(__file__).with_name("DejaVuSans.ttf") + + +def crc16(data): + crc = 0 + for b in data: + for i in range(8): + if ((crc >> 15) & 1) ^ ((b >> (7 - i)) & 1): + crc = ((crc << 1) & 0xFFFF) ^ 0x1021 + else: + crc = (crc << 1) & 0xFFFF + return crc + + +def handshake_reply(mac): + r = crc16(b"\x00" + bytes.fromhex(mac.replace(":", ""))) >> 8 + return b"\x5a\x0b" + bytes([r]) * 10 + + +def wrap(text, font, width): + out = [] + for para in text.split("\n"): + line = "" + for word in para.split(" "): + cand = f"{line} {word}" if line else word + if line and font.getlength(cand) > width: + out.append(line) + line = word + else: + line = cand + out.append(line) + return out + + +def render(text, size=28, separator=False): + font = ImageFont.truetype(str(FONT), size) + lines = wrap(text, font, WIDTH - 8) + text_h = (size + 6) * len(lines) + 4 + h = text_h + (16 if separator else 0) + img = Image.new("1", (WIDTH, h + h % 2), 1) + d = ImageDraw.Draw(img) + for i, line in enumerate(lines): + d.text((4, 2 + i * (size + 6)), line, font=font, fill=0) + if separator: + d.rectangle((0, text_h + 6, WIDTH - 1, text_h + 9), fill=0) + return img + + +def to_payloads(img): + """1-bit image, WIDTH wide -> 96-byte payloads, two rows each, 1 = black.""" + if img.height % 2: + img = img.crop((0, 0, WIDTH, img.height + 1)) # pads with 0 = black + img.paste(1, (0, img.height - 1, WIDTH, img.height)) + raw = bytes(b ^ 0xFF for b in img.tobytes()) + rows = [raw[i:i + 48] for i in range(0, len(raw), 48)] + return [rows[i] + rows[i + 1] for i in range(0, len(rows), 2)] + + +async def print_payloads(client, mac, payloads, density=3, feed=40): + """Handshake and print on an already connected BleakClient.""" + payloads = payloads + [bytes(96)] * feed + events = asyncio.Queue() + + def w(data): + return client.write_gatt_char(WRITE, data, response=False) + + async def wait_for(prefix, timeout=5): + while True: + data = await asyncio.wait_for(events.get(), timeout) + if data[:2] == prefix: + return data + + await client.start_notify(NOTIFY, lambda _, data: events.put_nowait(bytes(data))) + await w(b"\x5a\x01" + bytes(10)) # device info + await asyncio.sleep(0.5) + await w(b"\x5a\x0a" + bytes(10)) # challenge + await wait_for(b"\x5a\x0a") + await w(handshake_reply(mac)) + if (await wait_for(b"\x5a\x0b"))[2] != 1: + raise RuntimeError("LX-D02 handshake failed") + + await w(b"\x5a\x0c" + bytes([max(0, min(7, density))])) + await asyncio.sleep(0.3) + n = len(payloads) + await w(b"\x5a\x04" + n.to_bytes(2, "big") + b"\x00\x00") + await asyncio.sleep(0.1) + + # Flow control as in klimromanyuk/funnyprint: the printer's buffer is small. + # 5a08 = "pause, buffer full"; 5a05 + row number = "lost rows, resend from here"; + # 5a06 = "done". Without this, long prints (pictures) come out cut short. + cur, paused, rests = 0, False, 0 + try: + while True: + if paused or cur >= n: + data = await asyncio.wait_for(events.get(), 90 if paused else max(60, n * 0.5)) + elif not events.empty(): + data = events.get_nowait() + else: + await w(b"\x55" + cur.to_bytes(2, "big") + payloads[cur] + b"\x00") + cur += 1 + await asyncio.sleep(0.025) + continue + kind = data[:2] + if kind == b"\x5a\x06": + return + if kind == b"\x5a\x08": + paused = True + elif kind == b"\x5a\x05": + lost = int.from_bytes(data[2:4], "big") + if paused: # let the printer print out its buffer, then resend + rests += 1 + await asyncio.sleep(min(8.0, 2.0 + rests * 0.5)) + while not events.empty(): + more = events.get_nowait() + if more[:2] == b"\x5a\x06": + return + if more[:2] == b"\x5a\x05": + lost = int.from_bytes(more[2:4], "big") + paused = False + cur = lost + finally: + await w(b"\x5a\x04" + n.to_bytes(2, "big") + b"\x01\x00") + + +if __name__ == "__main__": + assert handshake_reply("00:00:00:00:00:00") == b"\x5a\x0b" + bytes(10) + assert to_payloads(Image.new("1", (WIDTH, 2), 0)) == [b"\xff" * 96] + assert to_payloads(Image.new("1", (WIDTH, 1), 0)) == [b"\xff" * 48 + bytes(48)] + assert render("x", separator=True).height == render("x").height + 16 + assert len(wrap("a " * 200, ImageFont.truetype(str(FONT), 28), WIDTH)) > 1 + print("ok") diff --git a/custom_components/lxd02/services.yaml b/custom_components/lxd02/services.yaml new file mode 100644 index 0000000..7a6d611 --- /dev/null +++ b/custom_components/lxd02/services.yaml @@ -0,0 +1,38 @@ +print_text: + name: Print text + description: Print text on the LX-D02 thermal printer. + fields: + text: + name: Text + required: true + example: "Temperature: 21.5 °C" + selector: + text: + multiline: true + size: + name: Font size + default: 28 + selector: + number: + min: 10 + max: 120 + density: + name: Density + default: 3 + selector: + number: + min: 0 + max: 7 + separator: + name: Separator line + description: Draw a horizontal line under the text. + default: false + selector: + boolean: + feed: + name: Feed after (rows of 2 dots) + default: 40 + selector: + number: + min: 0 + max: 500 diff --git a/lxd02.py b/lxd02.py new file mode 100644 index 0000000..c2f8772 --- /dev/null +++ b/lxd02.py @@ -0,0 +1,72 @@ +"""Print on the LX-D02 thermal printer from the PC. + +Usage: python lxd02.py "text" [more lines...] + python lxd02.py picture.jpg [more pictures...] (stretched to the tape width) +Set LXD02_MAC=AA:BB:CC:DD:EE:FF to skip the Bluetooth search. +""" +import asyncio +import os +import sys +from pathlib import Path + +from bleak import BleakClient, BleakScanner +from PIL import Image, ImageOps + +sys.path.insert(0, str(Path(__file__).resolve().parent / "custom_components" / "lxd02")) +from printer import WIDTH, print_payloads, render, to_payloads # noqa: E402 + +PICTURES = {".jpg", ".jpeg", ".png", ".bmp", ".gif", ".webp", ".tif", ".tiff"} + + +def picture(path): + """Any picture -> 1-bit image WIDTH wide. Landscape ones are turned to run + along the tape, so they come out bigger.""" + img = ImageOps.exif_transpose(Image.open(path)).convert("RGBA") + white = Image.new("RGBA", img.size, "white") + img = Image.alpha_composite(white, img).convert("L") # transparent -> white, not black + if img.width > img.height: + img = img.rotate(90, expand=True) + img = img.resize((WIDTH, max(2, round(img.height * WIDTH / img.width))), Image.LANCZOS) + return ImageOps.autocontrast(img, cutoff=1).convert("1") # dithering for halftones + + +async def send(img, mac=None): + mac = mac or os.environ.get("LXD02_MAC") + if not mac: + device = await BleakScanner.find_device_by_name("LX-D02", timeout=20) + if device is None: + raise RuntimeError("LX-D02 not found: is it on and not connected to a phone?") + mac = device.address + for attempt in range(3): # Windows often fails the first connect to a sleeping printer + try: + client = BleakClient(mac, timeout=20) + await client.connect() + break + except Exception: + if attempt == 2: + raise + await asyncio.sleep(2) + try: + await print_payloads(client, mac, to_payloads(img)) + finally: + await client.disconnect() + + +def main(args): + if args and all(Path(a).suffix.lower() in PICTURES and Path(a).is_file() for a in args): + for a in args: + print("printing", a) + asyncio.run(send(picture(a))) + else: + asyncio.run(send(render("\n".join(args) or "Hello, LX-D02!\nПривет с ПК"))) + print("ok") + + +if __name__ == "__main__": + try: + main(sys.argv[1:]) + except Exception as err: # started from "Send to": keep the window open to show why + print("ERROR:", err) + if sys.stdin and sys.stdin.isatty(): + input("Press Enter to close") + sys.exit(1) diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..96fdc8b --- /dev/null +++ b/requirements.txt @@ -0,0 +1,2 @@ +bleak<1.0 +pillow diff --git a/windows/install.ps1 b/windows/install.ps1 new file mode 100644 index 0000000..82e5049 --- /dev/null +++ b/windows/install.ps1 @@ -0,0 +1,60 @@ +# Adds the "LX-D02" printer to Windows, starts print_server.py at every logon and adds +# "Print on LX-D02" for pictures to the Explorer menu and to "Send to". +# Needs the POS-58 driver (stock driver for 58 mm receipt printers), see README. +# +# Run in PowerShell as administrator: .\install.ps1 (printer found by name) +# .\install.ps1 AA:BB:CC:DD:EE:FF +# Remove: .\install.ps1 -Uninstall +param([string]$Mac = "", [switch]$Uninstall) +$ErrorActionPreference = "Stop" +$Name = "LX-D02" +$Port = "LX-D02-9100" +$Task = "LX-D02 print server" +$Root = Split-Path $PSScriptRoot +$Menu = "HKCU:\Software\Classes\SystemFileAssociations\image\shell\LXD02" +$SendTo = Join-Path ([Environment]::GetFolderPath("SendTo")) "LX-D02.lnk" + +# Clean up first, so running it again simply reinstalls +Stop-ScheduledTask -TaskName $Task -ErrorAction SilentlyContinue +Unregister-ScheduledTask -TaskName $Task -Confirm:$false -ErrorAction SilentlyContinue +Get-CimInstance Win32_Process -Filter "Name like 'python%.exe'" | + Where-Object { $_.CommandLine -like "*print_server.py*" } | + ForEach-Object { Stop-Process -Id $_.ProcessId -Force } +Remove-Printer -Name $Name -ErrorAction SilentlyContinue +Remove-PrinterPort -Name $Port -ErrorAction SilentlyContinue +Remove-Item $Menu -Recurse -ErrorAction SilentlyContinue +Remove-Item $SendTo -ErrorAction SilentlyContinue +if ($Uninstall) { "Removed."; return } + +$driver = Get-PrinterDriver | Where-Object Name -like "POS-58*" | Select-Object -First 1 +if (-not $driver) { throw "POS-58 driver is not installed. See README: 'Driver'." } +$py = (Get-Command python.exe).Source +$pyw = (Get-Command pythonw.exe).Source + +# The printer: the POS-58 driver sends ESC/POS to our server instead of a COM port +Add-PrinterPort -Name $Port -PrinterHostAddress 127.0.0.1 -PortNumber 9100 +Add-Printer -Name $Name -DriverName $driver.Name -PortName $Port + +# The server, started at logon +$action = New-ScheduledTaskAction -Execute $pyw -Argument "`"$PSScriptRoot\print_server.py`" $Mac" -WorkingDirectory $PSScriptRoot +$trigger = New-ScheduledTaskTrigger -AtLogOn -User $env:USERNAME +$settings = New-ScheduledTaskSettingsSet -ExecutionTimeLimit 0 -AllowStartIfOnBatteries -DontStopIfGoingOnBatteries +Register-ScheduledTask -TaskName $Task -Action $action -Trigger $trigger -Settings $settings -Force | Out-Null +Start-ScheduledTask -TaskName $Task + +# Pictures: right click -> "Print on LX-D02", or several at once via "Send to" +$cmd = "`"$py`" `"$Root\lxd02.py`"" +New-Item -Path "$Menu\command" -Force | Out-Null +Set-ItemProperty -Path $Menu -Name "(default)" -Value "Print on LX-D02" +Set-ItemProperty -Path $Menu -Name "Icon" -Value "$env:SystemRoot\System32\imageres.dll,46" +Set-ItemProperty -Path $Menu -Name "MultiSelectModel" -Value "Single" # one Bluetooth connection at a time +Set-ItemProperty -Path "$Menu\command" -Name "(default)" -Value "$cmd `"%1`"" +$link = (New-Object -ComObject WScript.Shell).CreateShortcut($SendTo) +$link.TargetPath = $py +$link.Arguments = "`"$Root\lxd02.py`"" +$link.WorkingDirectory = $Root +$link.IconLocation = "$env:SystemRoot\System32\imageres.dll,46" +$link.Save() + +"Done: printer '$Name' (driver '$($driver.Name)'), print server, Explorer menu." +"Log: $PSScriptRoot\print_server.log" diff --git a/windows/print_server.py b/windows/print_server.py new file mode 100644 index 0000000..958a56c --- /dev/null +++ b/windows/print_server.py @@ -0,0 +1,123 @@ +"""LX-D02 as a regular Windows printer. + +Windows prints to the "LX-D02" printer: the stock POS-58 driver for 58 mm receipt +printers, on a TCP port 127.0.0.1:9100 (see install.ps1). The driver turns the page into +ESC/POS raster; this server takes it from the port and prints it over Bluetooth. + +Usage: python print_server.py [MAC] (without MAC the printer is found by name) + python print_server.py --test FILE (saved job -> FILE.png, no printing) + python print_server.py --selftest +""" +import asyncio +import logging +import queue +import socketserver +import sys +import threading +from pathlib import Path + +from PIL import Image, ImageOps + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) +from lxd02 import WIDTH, send # noqa: E402 + +PORT = 9100 +MAX_GAP = 48 # dots: longer white gaps are cut down to this (6 mm) +LAST_JOB = Path(__file__).with_name("last_job.bin") +log = logging.getLogger("lx-d02") + + +def render(data): + """ESC/POS from the POS-58 driver -> 1-bit image WIDTH wide, ready to print. + + The driver sends the page as GS v 0 raster bands. Its ESC J feeds between the bands + only move the paper by the band height, so the bands are just stacked.""" + rows, pos = [], 0 + while True: + pos = data.find(b"\x1dv0", pos) + if pos < 0 or pos + 8 > len(data): + break + w = data[pos + 4] | data[pos + 5] << 8 + h = data[pos + 6] | data[pos + 7] << 8 + band = data[pos + 8:pos + 8 + w * h] + rows += [band[r * w:(r + 1) * w].ljust(WIDTH // 8, b"\0")[:WIDTH // 8] for r in range(h)] + pos += 8 + w * h + if not rows: + raise ValueError(f"no ESC/POS raster in the job: {data[:16]!r} (is the driver POS-58?)") + img = ImageOps.invert(Image.frombytes("1", (WIDTH, len(rows)), b"".join(rows)).convert("L")) + box = ImageOps.invert(img).getbbox() + if not box: + raise ValueError("blank page") + # Apps keep their page margins: drop the left one and the white above and below + shift = max(0, box[0] - 4) + img = img.crop((shift, max(0, box[1] - 4), WIDTH + shift, min(img.height, box[3] + 4))) + img.paste(255, (WIDTH - shift, 0, WIDTH, img.height)) # the crop past the right edge is black + return squeeze(img).convert("1") + + +def squeeze(img, gap=MAX_GAP): + """Shorten white gaps longer than `gap` rows: page layouts put the footer a whole + page below the text, and the tape is not a page.""" + raw = img.tobytes() + keep, blank = [], 0 + for y in range(img.height): + row = raw[y * img.width:(y + 1) * img.width] + blank = blank + 1 if min(row) == 255 else 0 + if blank <= gap: + keep.append(row) + return Image.frombytes("L", (img.width, len(keep)), b"".join(keep)) + + +class Job(socketserver.BaseRequestHandler): + jobs = queue.Queue() + + def handle(self): + chunks = [] + while True: + chunk = self.request.recv(65536) + if not chunk: + break + chunks.append(chunk) + if chunks: # an empty connection is Windows checking the port + self.jobs.put(b"".join(chunks)) + + +def printer_loop(mac): + """One job at a time: the printer takes a single Bluetooth connection.""" + while True: + data = Job.jobs.get() + LAST_JOB.write_bytes(data) # for debugging a strange print: --test last_job.bin + try: + img = render(data) + log.info("job: %d bytes, %d dots long", len(data), img.height) + asyncio.run(send(img, mac)) + log.info("printed") + except Exception: + log.exception("job failed") + + +def main(mac): + threading.Thread(target=printer_loop, args=(mac,), daemon=True).start() + server = socketserver.ThreadingTCPServer(("127.0.0.1", PORT), Job) + log.info("waiting for jobs on 127.0.0.1:%d", PORT) + server.serve_forever() + + +if __name__ == "__main__": + handlers = [logging.FileHandler(Path(__file__).with_name("print_server.log"), encoding="utf-8")] + if sys.stderr: # None under pythonw + handlers.append(logging.StreamHandler()) + logging.basicConfig(level=logging.INFO, format="%(asctime)s %(message)s", handlers=handlers) + if sys.argv[1:2] == ["--selftest"]: + # one blank row, then one row with a single dot at x = 7: comes out shifted left + job = b"\x1b@\x1dv0\x00" + bytes([48, 0, 2, 0]) + bytes(48) + b"\x01" + bytes(47) + b"\x1bJ\x15" + img = render(job) + assert img.width == WIDTH and img.getpixel((4, 1)) == 0 and img.getpixel((7, 1)) == 255 + assert squeeze(Image.new("L", (8, 100), 255), 10).height == 10 + print("ok") + elif sys.argv[1:2] == ["--test"]: + src = Path(sys.argv[2]) + render(src.read_bytes()).save(src.with_suffix(".png")) + print("ok", src.with_suffix(".png")) + else: + main(sys.argv[1] if len(sys.argv) > 1 else None)