direct_s1_ble_print.py

#!/usr/bin/env python3
from __future__ import annotations

import argparse
import asyncio
import json
import os
from pathlib import Path
import sys

from bleak import BleakClient, BleakScanner
from bleak.exc import BleakError
from PIL import Image


ROOT = Path(__file__).resolve().parent
TIMINI_DIR = ROOT / "third_party" / "TiMini-Print"
S01_CONFIG = ROOT / "s01_printer_config.json"
DEFAULT_TARGET = os.environ.get("S1_BLUETOOTH_TARGET", "YOUR_PRINTER_NAME_OR_ADDRESS")
DEFAULT_ADDRESS = os.environ.get("S1_BLUETOOTH_ADDRESS")

sys.path.insert(0, str(TIMINI_DIR))

from timiniprint import reporting  # noqa: E402
from timiniprint.app.cli import build_print_job  # noqa: E402
from timiniprint.devices import PrinterCatalog  # noqa: E402


WRITE_CANDIDATES = [
    "0000ff02-0000-1000-8000-00805f9b34fb",
    "0000ff11-0000-1000-8000-00805f9b34fb",
    "00002af1-0000-1000-8000-00805f9b34fb",
    "49535343-8841-43f4-a8d4-ecbe34729bb3",
    "0000eee1-0000-1000-8000-00805f9b34fb",
    "0000eee3-0000-1000-8000-00805f9b34fb",
    "bef8d6c9-9c21-4c9e-b632-bd58c1009f9f",
    "0000ff82-0000-1000-8000-00805f9b34fb",
    "0000fff2-0000-1000-8000-00805f9b34fb",
]


WRITE_CANDIDATE_ALIASES = {
    "ff02": "0000ff02-0000-1000-8000-00805f9b34fb",
    "ff11": "0000ff11-0000-1000-8000-00805f9b34fb",
    "2af1": "00002af1-0000-1000-8000-00805f9b34fb",
    "microchip": "49535343-8841-43f4-a8d4-ecbe34729bb3",
    "eee1": "0000eee1-0000-1000-8000-00805f9b34fb",
    "eee3": "0000eee3-0000-1000-8000-00805f9b34fb",
    "bef8": "bef8d6c9-9c21-4c9e-b632-bd58c1009f9f",
    "ff82": "0000ff82-0000-1000-8000-00805f9b34fb",
    "fff2": "0000fff2-0000-1000-8000-00805f9b34fb",
}


def resolve_char(value: str) -> str:
    return WRITE_CANDIDATE_ALIASES.get(value.lower(), value)


def build_s01_payload(text: str, darkness: int) -> tuple[bytes, int, int]:
    return build_timini_profile_payload(
        "s01",
        text=text,
        image_path=None,
        darkness=darkness,
        text_columns=None,
        text_font=None,
    )


def build_timini_profile_payload(
    profile_key: str,
    *,
    text: str,
    image_path: Path | None,
    darkness: int,
    text_columns: int | None,
    text_font: str | None,
) -> tuple[bytes, int, int]:
    catalog = PrinterCatalog.load()
    if profile_key == "s01" and S01_CONFIG.exists():
        config = json.loads(S01_CONFIG.read_text(encoding="utf-8"))
        device = catalog.device_from_printer_config(config, transport_target=None)
    else:
        device = catalog.device_from_key(profile_key, transport_target=None)
    job = build_print_job(
        device,
        path=None if image_path is None else str(image_path),
        text_input=text if image_path is None else None,
        blackening=darkness,
        text_columns=text_columns,
        text_font=text_font,
        trim_top_bottom_margins=False if image_path is not None else True,
        reporter=reporting.DUMMY_REPORTER,
    )
    chunk_size = device.profile.stream.chunk_size
    delay_ms = device.profile.stream.delay_ms
    return job.payload, chunk_size, delay_ms


def build_escpos_payload(text: str) -> tuple[bytes, int, int]:
    normalized = text.replace("\r\n", "\n").replace("\r", "\n")
    lines = [
        b"\x1b@",  # initialize
        b"\x1ba\x01",  # center
        b"\x1bE\x01",  # bold on
        normalized.encode("cp437", errors="replace"),
        b"\x1bE\x00",  # bold off
        b"\x1ba\x00",  # left
        b"\n\n\n",
    ]
    payload = b"".join(lines)
    if not payload.endswith(b"\n"):
        payload += b"\n"
    return payload, 20, 10


def overstrike_line(line: str, passes: int) -> str:
    if passes <= 1 or not line:
        return line
    return "\r".join(line for _ in range(passes))


def build_plain_payload(text: str, overstrike: int = 1) -> tuple[bytes, int, int]:
    normalized = text.replace("\r\n", "\n").replace("\r", "\n")
    lines = normalized.split("\n")
    trailing_newline = normalized.endswith("\n")
    rendered = "\n".join(overstrike_line(line, overstrike) for line in lines)
    if trailing_newline and not rendered.endswith("\n"):
        rendered += "\n"
    if not rendered.endswith("\n"):
        rendered += "\n"
    rendered += "\n\n"
    return rendered.encode("cp437", errors="replace"), 20, 10


def image_to_ascii(path: Path, columns: int = 32, threshold: int = 150, invert: bool = False) -> str:
    chars = (" ", "#") if not invert else ("#", " ")
    image = Image.open(path).convert("L")
    aspect = image.height / max(1, image.width)
    rows = max(1, int(columns * aspect * 0.55))
    image = image.resize((columns, rows))
    lines = []
    for y in range(rows):
        line = []
        for x in range(columns):
            pixel = image.getpixel((x, y))
            line.append(chars[1] if pixel < threshold else chars[0])
        lines.append("".join(line).rstrip())
    return "\n".join(lines)


def build_ascii_image_payload(path: Path, columns: int, threshold: int, invert: bool) -> tuple[bytes, int, int]:
    return build_plain_payload(image_to_ascii(path, columns=columns, threshold=threshold, invert=invert))


async def find_target(
    target: str,
    timeout: float,
    retries: int = 3,
    retry_delay: float = 2.0,
):
    fallback_targets = [target]
    if target == DEFAULT_TARGET and DEFAULT_ADDRESS:
        fallback_targets.append(DEFAULT_ADDRESS)
    for attempt in range(1, retries + 1):
        print(f"Scanning for {target!r} ({attempt}/{retries})...", flush=True)
        devices = await BleakScanner.discover(timeout=timeout, return_adv=True)
        for found, adv in devices.values():
            haystack = " ".join(
                [
                    found.name or "",
                    adv.local_name or "",
                    found.address or "",
                ]
            ).lower()
            if any(candidate.lower() in haystack for candidate in fallback_targets):
                return found
        if attempt < retries:
            print(
                f"BLE target {target!r} not visible "
                f"(attempt {attempt}/{retries}); waiting {retry_delay:.1f}s",
                flush=True,
            )
            await asyncio.sleep(retry_delay)
    return None


def get_characteristic(client: BleakClient, uuid: str):
    for service in client.services:
        for char in service.characteristics:
            if char.uuid.lower() == uuid.lower():
                return char
    return None


async def send_payload(
    target: str,
    char_uuid: str,
    payload: bytes,
    chunk_size: int,
    delay_ms: int,
    scan_timeout: float,
    scan_retries: int,
    retry_delay: float,
    connect_retries: int,
    post_write_delay: float,
    response: bool | None,
) -> None:
    last_error: Exception | None = None
    for connect_attempt in range(1, connect_retries + 1):
        device = await find_target(target, scan_timeout, retries=scan_retries, retry_delay=retry_delay)
        if device is None:
            raise SystemExit(
                f"Could not find BLE target {target!r}. "
                "Make sure the printer is not connected to the phone app and wait a few seconds after printing."
            )

        print(f"Connecting to {device.name or '<unknown>'} | {device.address}", flush=True)
        try:
            async with BleakClient(device, pair=False, timeout=15.0) as client:
                char = get_characteristic(client, char_uuid)
                if char is None:
                    available = [
                        c.uuid
                        for service in client.services
                        for c in service.characteristics
                        if "write" in c.properties or "write-without-response" in c.properties
                    ]
                    raise SystemExit(f"Characteristic {char_uuid} not found. Writable chars: {available}")

                props = {prop.lower() for prop in char.properties}
                if response is None:
                    response = "write-without-response" not in props
                effective_chunk = min(
                    chunk_size,
                    getattr(char, "max_write_without_response_size", chunk_size) or chunk_size,
                )
                effective_chunk = max(1, effective_chunk)
                delay_seconds = max(0, delay_ms) / 1000.0
                print(
                    f"Writing {len(payload)} bytes to {char.uuid} "
                    f"chunk={effective_chunk} delay_ms={delay_ms} response={response}",
                    flush=True,
                )
                chunks = 0
                for offset in range(0, len(payload), effective_chunk):
                    chunk = payload[offset : offset + effective_chunk]
                    await client.write_gatt_char(char, chunk, response=response)
                    chunks += 1
                    if delay_seconds:
                        await asyncio.sleep(delay_seconds)
                print(f"Wrote {chunks} chunks.", flush=True)
                if post_write_delay:
                    await asyncio.sleep(post_write_delay)
                return
        except (BleakError, TimeoutError, OSError) as exc:
            last_error = exc
            if connect_attempt >= connect_retries:
                raise
            print(
                f"BLE connect/write failed on attempt {connect_attempt}/{connect_retries}: {exc}. "
                f"Retrying in {retry_delay:.1f}s.",
                flush=True,
            )
            await asyncio.sleep(retry_delay)
    if last_error:
        raise last_error


async def main_async(args: argparse.Namespace) -> int:
    if args.profile:
        payload, chunk_size, delay_ms = build_timini_profile_payload(
            args.profile,
            text=args.text,
            image_path=Path(args.image) if args.image else None,
            darkness=args.darkness,
            text_columns=args.text_columns,
            text_font=args.text_font,
        )
        build_label = f"profile:{args.profile}"
    elif args.image:
        payload, chunk_size, delay_ms = build_ascii_image_payload(
            Path(args.image),
            columns=args.columns,
            threshold=args.threshold,
            invert=args.invert,
        )
        build_label = "ascii-image"
    elif args.mode == "s01":
        payload, chunk_size, delay_ms = build_s01_payload(args.text, args.darkness)
        build_label = args.mode
    elif args.mode == "escpos":
        payload, chunk_size, delay_ms = build_escpos_payload(args.text)
        build_label = args.mode
    else:
        payload, chunk_size, delay_ms = build_plain_payload(args.text, overstrike=args.overstrike)
        build_label = args.mode
    print(
        f"Built {build_label} payload: {len(payload)} bytes, "
        f"head={payload[:16].hex()}, tail={payload[-16:].hex()}",
        flush=True,
    )
    response = None
    if args.response:
        response = True
    if args.no_response:
        response = False
    await send_payload(
        target=args.bluetooth,
        char_uuid=args.char,
        payload=payload,
        chunk_size=args.chunk_size or chunk_size,
        delay_ms=args.delay_ms if args.delay_ms is not None else delay_ms,
        scan_timeout=args.scan_timeout,
        scan_retries=args.scan_retries,
        retry_delay=args.retry_delay,
        connect_retries=args.connect_retries,
        post_write_delay=args.post_write_delay,
        response=response,
    )
    return 0


def main() -> int:
    parser = argparse.ArgumentParser(description="Direct BLE S01 print test using TiMini payloads.")
    parser.add_argument("text", nargs="?", default="ayo from direct bleak")
    parser.add_argument("--mode", choices=("plain", "escpos", "s01"), default="plain")
    parser.add_argument("--bluetooth", default=DEFAULT_TARGET)
    parser.add_argument(
        "--char",
        default="ff02",
        help="Write characteristic UUID or alias: " + ", ".join(WRITE_CANDIDATE_ALIASES),
    )
    parser.add_argument("--darkness", type=int, choices=range(1, 6), default=3)
    parser.add_argument("--text-columns", type=int, help="Native profile text width; lower means larger text.")
    parser.add_argument("--text-font", help="Path to a .ttf/.otf font for native profile text rendering.")
    parser.add_argument(
        "--profile",
        help="Build a native TiMini protocol payload from a known profile, e.g. phomemo_m02 or tspl_p1.",
    )
    parser.add_argument("--overstrike", type=int, choices=range(1, 5), default=1, help="Plain-text thickness hack.")
    parser.add_argument("--image", help="Image path. With --profile this builds native raster; without it prints ASCII art.")
    parser.add_argument("--columns", type=int, default=32, help="ASCII image width in text columns.")
    parser.add_argument("--threshold", type=int, default=150, help="0-255 darkness threshold for ASCII images.")
    parser.add_argument("--invert", action="store_true", help="Invert ASCII image black/white.")
    parser.add_argument("--chunk-size", type=int)
    parser.add_argument("--delay-ms", type=int)
    parser.add_argument("--scan-timeout", type=float, default=6.0)
    parser.add_argument("--scan-retries", type=int, default=3)
    parser.add_argument("--connect-retries", type=int, default=2)
    parser.add_argument("--retry-delay", type=float, default=2.0)
    parser.add_argument("--post-write-delay", type=float, default=2.0)
    mode = parser.add_mutually_exclusive_group()
    mode.add_argument("--response", action="store_true", help="Force write-with-response.")
    mode.add_argument("--no-response", action="store_true", help="Force write-without-response.")
    args = parser.parse_args()
    args.char = resolve_char(args.char)
    return asyncio.run(main_async(args))


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