ble_probe.py

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

import argparse
import asyncio
from datetime import datetime, timezone
import os

from typing import Iterable

from bleak import BleakClient, BleakScanner

from direct_s1_ble_print import build_timini_profile_payload


DEFAULT_TARGET = os.environ.get("S1_BLUETOOTH_TARGET", "YOUR_PRINTER_NAME_OR_ADDRESS")
WRITE_PROBE_CHARS = [
    ("ff02", "0000ff02-0000-1000-8000-00805f9b34fb"),
    ("ff11", "0000ff11-0000-1000-8000-00805f9b34fb"),
    ("ff12", "0000ff12-0000-1000-8000-00805f9b34fb"),
    ("microchip", "49535343-8841-43f4-a8d4-ecbe34729bb3"),
    ("microchip-extra", "49535343-6daa-4d02-abf6-19569aca69fe"),
    ("eee1", "0000eee1-0000-1000-8000-00805f9b34fb"),
    ("eee3", "0000eee3-0000-1000-8000-00805f9b34fb"),
    ("2af1", "00002af1-0000-1000-8000-00805f9b34fb"),
    ("bef8", "bef8d6c9-9c21-4c9e-b632-bd58c1009f9f"),
    ("fec7", "0000fec7-0000-1000-8000-00805f9b34fb"),
    ("ff82", "0000ff82-0000-1000-8000-00805f9b34fb"),
    ("fff2", "0000fff2-0000-1000-8000-00805f9b34fb"),
]

WRITE_CHAR_ALIASES = {
    "ff02": "0000ff02-0000-1000-8000-00805f9b34fb",
    "ff11": "0000ff11-0000-1000-8000-00805f9b34fb",
    "ff12": "0000ff12-0000-1000-8000-00805f9b34fb",
    "microchip": "49535343-8841-43f4-a8d4-ecbe34729bb3",
    "microchip-extra": "49535343-6daa-4d02-abf6-19569aca69fe",
    "eee1": "0000eee1-0000-1000-8000-00805f9b34fb",
    "eee3": "0000eee3-0000-1000-8000-00805f9b34fb",
    "2af1": "00002af1-0000-1000-8000-00805f9b34fb",
    "bef8": "bef8d6c9-9c21-4c9e-b632-bd58c1009f9f",
    "fec7": "0000fec7-0000-1000-8000-00805f9b34fb",
    "ff82": "0000ff82-0000-1000-8000-00805f9b34fb",
    "fff2": "0000fff2-0000-1000-8000-00805f9b34fb",
}


def now_stamp() -> str:
    return datetime.now(timezone.utc).astimezone().strftime("%H:%M:%S")


def hex_bytes(data: bytes, limit: int = 64) -> str:
    preview = data[:limit].hex()
    if len(data) > limit:
        return f"{preview}..."
    return preview


def ascii_bytes(data: bytes, limit: int = 64) -> str:
    snippet = data[:limit]
    return "".join(chr(b) if 32 <= b < 127 else "." for b in snippet)


def char_props(char) -> set[str]:
    return {prop.lower() for prop in char.properties}


def resolve_uuid(value: str) -> str:
    return WRITE_CHAR_ALIASES.get(value.lower(), value)


def parse_payload_text(command: str) -> bytes:
    if command.startswith("hex:"):
        return bytes.fromhex(command[4:])
    return command.encode("cp437", errors="replace")


async def scan(timeout: float) -> None:
    devices = await BleakScanner.discover(timeout=timeout, return_adv=True)
    for device, adv in devices.values():
        name = device.name or adv.local_name or "<unknown>"
        print(f"{name} | {device.address} | rssi={adv.rssi}")
        for service_uuid in adv.service_uuids or []:
            print(f"  service {service_uuid}")


async def inspect(target: str, timeout: float) -> None:
    found = await BleakScanner.find_device_by_filter(
        lambda device, adv: target.lower() in ((device.name or adv.local_name or device.address or "").lower()),
        timeout=timeout,
    )
    if found is None:
        found = await BleakScanner.find_device_by_address(target, timeout=timeout)
    if found is None:
        raise SystemExit(f"Could not find BLE target {target!r}")

    print(f"Connecting to {found.name or '<unknown>'} | {found.address}")
    async with BleakClient(found, pair=False, timeout=15.0) as client:
        print(f"connected={client.is_connected} mtu={getattr(client, 'mtu_size', '<unknown>')}")
        services = client.services
        for service in services:
            print(f"service {service.uuid} {service.description}")
            for char in service.characteristics:
                props = ",".join(char.properties)
                print(f"  char {char.uuid} props={props} handle={char.handle}")
                for descriptor in char.descriptors:
                    print(f"    desc {descriptor.uuid} handle={descriptor.handle}")


async def write_probe(target: str, timeout: float, delay: float) -> None:
    found = await BleakScanner.find_device_by_filter(
        lambda device, adv: target.lower() in ((device.name or adv.local_name or device.address or "").lower()),
        timeout=timeout,
    )
    if found is None:
        found = await BleakScanner.find_device_by_address(target, timeout=timeout)
    if found is None:
        raise SystemExit(f"Could not find BLE target {target!r}")

    print(f"Connecting to {found.name or '<unknown>'} | {found.address}")
    async with BleakClient(found, pair=False, timeout=15.0) as client:
        chars = {
            char.uuid.lower(): char
            for service in client.services
            for char in service.characteristics
            if "write" in char.properties or "write-without-response" in char.properties
        }
        for label, uuid in WRITE_PROBE_CHARS:
            char = chars.get(uuid.lower())
            if char is None:
                print(f"skip {label}: missing")
                continue
            props = char_props(char)
            response = "write-without-response" not in props
            payload = f"probe {label}\n\n".encode("ascii")
            print(f"write {label}: {uuid} response={response}")
            await client.write_gatt_char(char, payload, response=response)
            await asyncio.sleep(delay)


async def read_and_listen(
    target: str,
    timeout: float,
    listen_seconds: float,
    read_once: bool,
    poke_text: str | None,
    poke_char: str,
    poke_repeat: int,
    native_text: str | None,
    native_darkness: int,
) -> None:
    found = await BleakScanner.find_device_by_filter(
        lambda device, adv: target.lower() in ((device.name or adv.local_name or device.address or "").lower()),
        timeout=timeout,
    )
    if found is None:
        found = await BleakScanner.find_device_by_address(target, timeout=timeout)
    if found is None:
        raise SystemExit(f"Could not find BLE target {target!r}")

    print(f"Connecting to {found.name or '<unknown>'} | {found.address}")
    async with BleakClient(found, pair=False, timeout=15.0) as client:
        readable = []
        notify_chars = []
        for service in client.services:
            for char in service.characteristics:
                props = char_props(char)
                if "read" in props:
                    readable.append(char)
                if "notify" in props or "indicate" in props:
                    notify_chars.append(char)

        print(f"readable={len(readable)} notify_or_indicate={len(notify_chars)}")

        if read_once:
            for char in readable:
                try:
                    value = await client.read_gatt_char(char)
                    print(
                        f"[{now_stamp()}] read {char.uuid} props={','.join(char.properties)} "
                        f"len={len(value)} hex={hex_bytes(value)} ascii={ascii_bytes(value)}"
                    )
                except Exception as exc:
                    print(f"[{now_stamp()}] read {char.uuid} failed: {exc}")

        seen: set[str] = set()

        def make_handler(char_uuid: str):
            def handler(_sender: int, data: bytearray) -> None:
                payload = bytes(data)
                print(
                    f"[{now_stamp()}] notify {char_uuid} len={len(payload)} "
                    f"hex={hex_bytes(payload)} ascii={ascii_bytes(payload)}"
                )
                seen.add(char_uuid)

            return handler

        for char in notify_chars:
            try:
                await client.start_notify(char, make_handler(char.uuid))
                print(f"subscribed {char.uuid} props={','.join(char.properties)}")
            except Exception as exc:
                print(f"skip notify {char.uuid}: {exc}")

        if poke_text:
            target_uuid = resolve_uuid(poke_char)
            poke = next(
                (
                    char
                    for service in client.services
                    for char in service.characteristics
                    if char.uuid.lower() == target_uuid.lower()
                ),
                None,
            )
            if poke is None:
                print(f"poke characteristic missing: {poke_char}")
            else:
                payload = (poke_text * max(1, poke_repeat)).encode("cp437", errors="replace")
                response = "write-without-response" not in char_props(poke)
                print(
                    f"poke write {poke.uuid} len={len(payload)} response={response} "
                    f"text={poke_text!r} repeat={poke_repeat}"
                )
                await client.write_gatt_char(poke, payload, response=response)
                await asyncio.sleep(1.0)

        if native_text:
            payload, chunk_size, delay_ms = build_timini_profile_payload(
                "phomemo_m02",
                text=native_text,
                image_path=None,
                darkness=native_darkness,
                text_columns=16,
                text_font=None,
            )
            poke = next(
                (
                    char
                    for service in client.services
                    for char in service.characteristics
                    if char.uuid.lower() == resolve_uuid("ff02").lower()
                ),
                None,
            )
            if poke is None:
                print("native print characteristic missing: ff02")
            else:
                response = "write-without-response" not in char_props(poke)
                effective_chunk = min(
                    chunk_size,
                    getattr(poke, "max_write_without_response_size", chunk_size) or chunk_size,
                )
                effective_chunk = max(1, effective_chunk)
                print(
                    f"native print write {poke.uuid} len={len(payload)} chunk={effective_chunk} "
                    f"delay_ms={delay_ms} response={response} text={native_text!r} darkness={native_darkness}"
                )
                for offset in range(0, len(payload), effective_chunk):
                    chunk = payload[offset : offset + effective_chunk]
                    await client.write_gatt_char(poke, chunk, response=response)
                    if delay_ms:
                        await asyncio.sleep(delay_ms / 1000.0)
                await asyncio.sleep(1.0)

        try:
            if listen_seconds > 0:
                await asyncio.sleep(listen_seconds)
        finally:
            for char in notify_chars:
                try:
                    await client.stop_notify(char)
                except Exception:
                    pass

        if not seen:
            print("no notifications observed")


async def matrix_probe(target: str, timeout: float, delay: float, payload_text: str) -> None:
    found = await BleakScanner.find_device_by_filter(
        lambda device, adv: target.lower() in ((device.name or adv.local_name or device.address or "").lower()),
        timeout=timeout,
    )
    if found is None:
        found = await BleakScanner.find_device_by_address(target, timeout=timeout)
    if found is None:
        raise SystemExit(f"Could not find BLE target {target!r}")

    print(f"Connecting to {found.name or '<unknown>'} | {found.address}")
    async with BleakClient(found, pair=False, timeout=15.0) as client:
        chars = {
            char.uuid.lower(): char
            for service in client.services
            for char in service.characteristics
        }
        notify_chars = [
            char
            for service in client.services
            for char in service.characteristics
            if "notify" in char_props(char) or "indicate" in char_props(char)
        ]
        print(f"subscribing to {len(notify_chars)} notify/indicate characteristics")

        def make_handler(char_uuid: str):
            def handler(_sender: int, data: bytearray) -> None:
                payload = bytes(data)
                print(
                    f"[{now_stamp()}] notify {char_uuid} len={len(payload)} "
                    f"hex={hex_bytes(payload)} ascii={ascii_bytes(payload)}"
                )

            return handler

        for char in notify_chars:
            try:
                await client.start_notify(char, make_handler(char.uuid))
            except Exception as exc:
                print(f"skip notify {char.uuid}: {exc}")

        payload = payload_text.encode("cp437", errors="replace")
        for label, uuid in WRITE_PROBE_CHARS:
            char = chars.get(uuid.lower())
            if char is None:
                print(f"skip {label}: missing")
                continue
            props = char_props(char)
            response = "write-without-response" not in props
            print(f"write {label}: {uuid} response={response} payload={payload_text!r}")
            await client.write_gatt_char(char, payload, response=response)
            await asyncio.sleep(delay)

        for char in notify_chars:
            try:
                await client.stop_notify(char)
            except Exception:
                pass


async def command_sweep(
    target: str,
    timeout: float,
    delay: float,
    char_uuid: str,
    commands: list[str],
) -> None:
    found = await BleakScanner.find_device_by_filter(
        lambda device, adv: target.lower() in ((device.name or adv.local_name or device.address or "").lower()),
        timeout=timeout,
    )
    if found is None:
        found = await BleakScanner.find_device_by_address(target, timeout=timeout)
    if found is None:
        raise SystemExit(f"Could not find BLE target {target!r}")

    target_uuid = resolve_uuid(char_uuid)
    print(f"Connecting to {found.name or '<unknown>'} | {found.address}")
    async with BleakClient(found, pair=False, timeout=15.0) as client:
        chars = {
            char.uuid.lower(): char
            for service in client.services
            for char in service.characteristics
        }
        notify_chars = [
            char
            for service in client.services
            for char in service.characteristics
            if "notify" in char_props(char) or "indicate" in char_props(char)
        ]

        def make_handler(char_uuid: str):
            def handler(_sender: int, data: bytearray) -> None:
                payload = bytes(data)
                print(
                    f"[{now_stamp()}] notify {char_uuid} len={len(payload)} "
                    f"hex={hex_bytes(payload)} ascii={ascii_bytes(payload)}"
                )

            return handler

        for char in notify_chars:
            try:
                await client.start_notify(char, make_handler(char.uuid))
            except Exception as exc:
                print(f"skip notify {char.uuid}: {exc}")

        char = chars.get(target_uuid.lower())
        if char is None:
            raise SystemExit(f"Missing target characteristic {char_uuid!r}")
        response = "write-without-response" not in char_props(char)
        for command in commands:
            payload = parse_payload_text(command)
            print(f"write {char.uuid} response={response} payload={command!r}")
            await client.write_gatt_char(char, payload, response=response)
            await asyncio.sleep(delay)

        for char in notify_chars:
            try:
                await client.stop_notify(char)
            except Exception:
                pass


def main() -> int:
    parser = argparse.ArgumentParser()
    sub = parser.add_subparsers(dest="command", required=True)
    scan_parser = sub.add_parser("scan")
    scan_parser.add_argument("--timeout", type=float, default=8.0)
    inspect_parser = sub.add_parser("inspect")
    inspect_parser.add_argument("--target", default=DEFAULT_TARGET)
    inspect_parser.add_argument("--timeout", type=float, default=12.0)
    probe_parser = sub.add_parser("write-probe")
    probe_parser.add_argument("--target", default=DEFAULT_TARGET)
    probe_parser.add_argument("--timeout", type=float, default=12.0)
    probe_parser.add_argument("--delay", type=float, default=0.8)
    status_parser = sub.add_parser("status-probe")
    status_parser.add_argument("--target", default=DEFAULT_TARGET)
    status_parser.add_argument("--timeout", type=float, default=12.0)
    status_parser.add_argument("--listen-seconds", type=float, default=12.0)
    status_parser.add_argument("--read-once", action="store_true", help="Read all readable characteristics once before listening.")
    status_parser.add_argument("--poke-text", help="Optional tiny raw text to write after subscribing, to trigger status notifications.")
    status_parser.add_argument("--poke-char", default="ff02", help="Write characteristic alias or UUID for --poke-text.")
    status_parser.add_argument("--poke-repeat", type=int, default=1, help="Repeat --poke-text this many times to make a longer job.")
    status_parser.add_argument("--native-text", help="Optional native raster text job to send while listening.")
    status_parser.add_argument("--native-darkness", type=int, default=3, choices=range(1, 6), help="Darkness level for --native-text.")
    matrix_parser = sub.add_parser("matrix-probe")
    matrix_parser.add_argument("--target", default=DEFAULT_TARGET)
    matrix_parser.add_argument("--timeout", type=float, default=12.0)
    matrix_parser.add_argument("--delay", type=float, default=1.0)
    matrix_parser.add_argument("--payload", default=".")
    command_parser = sub.add_parser("command-sweep")
    command_parser.add_argument("--target", default=DEFAULT_TARGET)
    command_parser.add_argument("--timeout", type=float, default=12.0)
    command_parser.add_argument("--delay", type=float, default=1.0)
    command_parser.add_argument("--char", default="ff82")
    command_parser.add_argument("commands", nargs="+")
    args = parser.parse_args()
    if args.command == "scan":
        asyncio.run(scan(args.timeout))
    elif args.command == "inspect":
        asyncio.run(inspect(args.target, args.timeout))
    elif args.command == "write-probe":
        asyncio.run(write_probe(args.target, args.timeout, args.delay))
    elif args.command == "status-probe":
        asyncio.run(
            read_and_listen(
                args.target,
                args.timeout,
                args.listen_seconds,
                args.read_once,
                args.poke_text,
                args.poke_char,
                args.poke_repeat,
                args.native_text,
                args.native_darkness,
            )
        )
    elif args.command == "matrix-probe":
        asyncio.run(matrix_probe(args.target, args.timeout, args.delay, args.payload))
    elif args.command == "command-sweep":
        asyncio.run(command_sweep(args.target, args.timeout, args.delay, args.char, args.commands))
    return 0


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