label_printer/capture_print_job.py

"""
Capture the raw print job bytes sent to the PT-E560BT.

This is ground truth: it shows exactly what reached the printer, independent of
SumatraPDF and of what our PDF claims. Comparing our intended geometry against
the actual raster tells us whether a discrepancy originates in our PDF, in
SumatraPDF, or in the driver/device.

How it works: the printer is redirected to a FILE port, the job is printed, and
the resulting file is inspected. The original port is then restored.

Usage:
    python capture_print_job.py            # capture one job to ./captures
    python capture_print_job.py --inspect FILE
"""

from __future__ import annotations

import argparse
import os
import subprocess
import sys
import time

import win32print

PRINTER = "Brother PT-E560BT"
CAPTURE_DIR = "captures"


def get_port(printer: str) -> str:
    handle = win32print.OpenPrinter(printer)
    try:
        return win32print.GetPrinter(handle, 2)["pPortName"]
    finally:
        win32print.ClosePrinter(handle)


def set_port(printer: str, port: str) -> None:
    """Repoint a printer at a different port. Needs admin rights."""
    handle = win32print.OpenPrinter(printer, {"DesiredAccess": win32print.PRINTER_ALL_ACCESS})
    try:
        info = win32print.GetPrinter(handle, 2)
        info["pPortName"] = port
        win32print.SetPrinter(handle, 2, info, 0)
    finally:
        win32print.ClosePrinter(handle)


def dump_structure(path: str) -> None:
    """Summarise a captured job: size, and any recognisable P-touch commands."""
    size = os.path.getsize(path)
    with open(path, "rb") as handle:
        data = handle.read()

    print(f"\nFile       : {path}")
    print(f"Size       : {size} bytes")
    print(f"First bytes: {data[:32].hex(' ')}")

    # Brother raster printers use an ESC/P-style command set. Recognise the
    # common ones so the page setup the device received can be read back.
    known = {
        b"\x1b@": "ESC @ (initialise)",
        b"\x1bia": "ESC i a (set media/tape type)",
        b"\x1biM": "ESC i M (set print mode)",
        b"\x1biA": "ESC i A (set auto-cut)",
        b"\x1biK": "ESC i K (set halftone)",
        b"\x1biD": "ESC i D (set margin)",
        b"\x1biz": "ESC i z (set page size)",
        b"\x1biE": "ESC i E (set compression)",
        b"g": "g (raster graphics transfer)",
        b"\x1a": "SUB (cut/end)",
    }
    print("\nRecognised commands present:")
    found_any = False
    for token, label in known.items():
        if token in data:
            count = data.count(token)
            print(f"  {label:<32} x{count}")
            found_any = True
    if not found_any:
        print("  (none recognised - job may be compressed or in a different format)")

    # IF the device received an explicit page size (ESC i z), decode it:
    # ESC i z <len> <valid> <width> <length> <raster lines> ...
    idx = data.find(b"\x1biz")
    if idx >= 0:
        print("\nESC i z page-size command found:")
        chunk = data[idx: idx + 24]
        print(f"  raw: {chunk.hex(' ')}")
        if len(chunk) >= 10:
            # Byte layout: 1b 69 7a <len=10> <valid> <width(2)> <length(2)> <lines(4)>
            valid = chunk[4]
            width = int.from_bytes(chunk[5:7], "little")
            length = int.from_bytes(chunk[7:9], "little")
            lines = int.from_bytes(chunk[9:13], "little")
            print(f"  valid flags : 0x{valid:02x}")
            print(f"  width       : {width} dots = {width / 180 * 25.4:.2f} mm")
            print(f"  length      : {length} dots = {length / 180 * 25.4:.2f} mm")
            print(f"  raster lines: {lines} = {lines / 180 * 25.4:.2f} mm")


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--inspect", help="inspect an existing capture instead of printing")
    args = parser.parse_args()

    if args.inspect:
        dump_structure(args.inspect)
        return 0

    os.makedirs(CAPTURE_DIR, exist_ok=True)
    target = os.path.abspath(os.path.join(CAPTURE_DIR, "job.prn"))
    if os.path.exists(target):
        os.remove(target)

    # Find our most recent label PDF.
    import glob

    pdfs = sorted(glob.glob("pdf_backup/ptouch_*.pdf"), key=os.path.getmtime)
    if not pdfs:
        print("No label PDF found in pdf_backup/")
        return 1
    pdf = pdfs[-1]

    original_port = get_port(PRINTER)
    print(f"Printer    : {PRINTER}")
    print(f"Original   : {original_port}")
    print(f"PDF        : {pdf}")
    print(f"Capture to : {target}")

    if original_port.upper().startswith("FILE:"):
        print("\nPort is already a FILE: port; capturing directly.")
        print(f"Check the file it writes to. Set port to FILE:{target} manually if needed.")
        return 1

    # Create a local port that writes to our file.
    port_name = rf"FILE:{target}"
    try:
        win32print.AddPort(None, port_name)
        print(f"Created port: {port_name}")
    except Exception as exc:  # noqa: BLE001
        print(f"AddPort failed (may already exist): {exc}")

    try:
        set_port(PRINTER, port_name)
        print("Redirected printer to capture port.")
    except Exception as exc:  # noqa: BLE001
        print(f"\nCould not redirect the printer: {exc}")
        print("This step needs administrator rights and a local printer port.")
        return 1

    try:
        from print_label import _sumatra_print

        sumatra = os.path.join("SumatraPDF", "SumatraPDF.exe")
        ok, msg = _sumatra_print(sumatra, PRINTER, pdf, "noscale,paperkind=261")
        print(f"Print      : {'sent' if ok else 'failed'} - {msg}")

        # Give the spooler time to flush to the file.
        for _ in range(20):
            time.sleep(0.5)
            if os.path.exists(target) and os.path.getsize(target) > 0:
                break
    finally:
        set_port(PRINTER, original_port)
        print(f"Restored port: {original_port}")

    if os.path.exists(target):
        dump_structure(target)
    else:
        print("\nNo capture file was produced.")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())