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 <noreply@anthropic.com>
2026-10-21 16:56:45 +03:00
commit 9e06f3ebd0
12 changed files with 765 additions and 0 deletions
@@ -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)