#!/usr/bin/env python3
# SPDX-License-Identifier: MIT
# Copyright (c) 2026 jacq.cc
"""GameSir T1d (Tello edition) -> virtual Xbox 360 gamepad, Linux only.

Reads the controller's BLE input reports with bleak and feeds a uinput device
through python-evdev, so Steam, SDL games and anything else that understands
an Xbox 360 pad can use it. The T1d has no Bluetooth HID profile, so this
bridge is the only way it can drive games.

Calibration (the sticks drift, see README.md):
  * Centre is sampled from the first --settle seconds after a connect in
    which the sticks are still (hands off), or fixed with --center LX,LY,RX,RY.
    Until then the nominal 512 is used, so the pad works at once.
  * Hold C1 + C2 for 1.5 s at any time to re-sample the centre.
  * Radial dead zone (--deadzone, fraction of half travel) and the remaining
    travel is rescaled, so full deflection still gives full output.
  * Range: run once with --calibrate and move both sticks around their full
    circle for 10 s. The real raw extremes per axis are saved to
    ~/.config/t1d-bridge/<MAC>.json ($T1D_CALIB_DIR if set; the .deb uses
    /var/lib/gamesir-t1d) and used for scaling from then on (a worn
    pot that only reaches raw 720 instead of 1023 still gives full output).
    Exact rail values (0, 1023) are never learned: on a worn pot they are
    wiper lift-offs, not travel. Without a file the nominal 0..1023 is used.
  * Each side of an axis is scaled by its own travel, so an off-centre rest
    position doesn't make one direction faster.
  * --rail-timeout N: an axis that sits at exactly 0 or 1023 (the ADC rail)
    for N seconds is treated as centred until it moves again. This is the
    work-around for a potentiometer wiper that loses contact (the left stick
    Y axis on the author's unit). Off by default because it also cuts a deliberate
    full-deflection hold after N seconds.

Setup, once:
  python3 -m venv ~/.venvs/t1d && ~/.venvs/t1d/bin/pip install bleak evdev
  sudo install -m 644 gamesir-ctrl/60-t1d-bridge-uinput.rules /etc/udev/rules.d/
  sudo udevadm control --reload && sudo udevadm trigger --name-match=uinput
Run:
  ~/.venvs/t1d/bin/python gamesir-ctrl/t1d-bridge.py --selftest   # no controller needed
  ~/.venvs/t1d/bin/python gamesir-ctrl/t1d-bridge.py --calibrate  # once: sticks in full circles for 10 s
  ~/.venvs/t1d/bin/python gamesir-ctrl/t1d-bridge.py -v           # controller on, LEDs blinking
"""
from __future__ import annotations

import argparse
import asyncio
import json
import logging
import os
import pathlib
import math
import signal
import statistics
import sys
import time

from bleak import BleakClient, BleakScanner
from bleak.exc import BleakError
from evdev import AbsInfo, UInput, ecodes as e

NAME_PREFIX = "Gamesir-T1d"
INPUT_CHAR = "00008651-0000-1000-8000-00805f9b34fb"
INPUT_TYPE = b"\xa1\xc5"      # anything else (c9 c6 = status blob every ~5 s) is dropped
REPORT_LEN = 20
RAW_MAX = 1023                # 10-bit stick axes
STICK_MAX = 32767
NOMINAL = RAW_MAX // 2 + 1  # 512
# T1D_CALIB_DIR overrides: the .deb points every user at the shared /var/lib/gamesir-t1d
# (the range belongs to the controller, not to whoever is logged in).
CALIB_DIR = pathlib.Path(os.environ.get("T1D_CALIB_DIR")
                         or pathlib.Path(os.environ.get("XDG_CONFIG_HOME", os.path.expanduser("~/.config"))) / "t1d-bridge")
AXES = ("lx", "ly", "rx", "ry")

# Byte 9 bits: seen identically in dg-hub, ElishaAz and Isaac-dot-py decoders.
BUTTONS_B9 = {
    0x01: e.BTN_A, 0x02: e.BTN_B, 0x04: e.BTN_MODE, 0x08: e.BTN_X,
    0x10: e.BTN_Y, 0x40: e.BTN_TL, 0x80: e.BTN_TR,
}
# Byte 10 bits: L2/R2 digital, C1, C2 verified live 2026-10-01; L3/R3 (stick
# clicks) verified 2026-10-12. The pair button sends nothing: it drops the connection.
BUTTONS_B10 = {0x01: e.BTN_TL2, 0x02: e.BTN_TR2, 0x04: e.BTN_SELECT, 0x08: e.BTN_START,
               0x20: e.BTN_THUMBL, 0x40: e.BTN_THUMBR}
RECENTER_MASK_B10 = 0x04 | 0x08          # C1 + C2 held together
RECENTER_HOLD_S = 1.5
# D-pad byte 11: 1 up, 3 right, 5 down, 7 left, even values are the diagonals.
HAT = {0: (0, 0), 1: (0, -1), 2: (1, -1), 3: (1, 0), 4: (1, 1), 5: (0, 1), 6: (-1, 1), 7: (-1, 0), 8: (-1, -1)}

log = logging.getLogger("t1d-bridge")


def decode(d: bytes) -> dict:
    return {
        "lx": (d[2] << 2) | (d[3] >> 6),
        "ly": ((d[3] & 0x3F) << 4) | (d[4] >> 4),
        "rx": ((d[4] & 0x0F) << 6) | (d[5] >> 2),
        "ry": ((d[5] & 0x03) << 8) | d[6],
        "l2": d[7],
        "r2": d[8],
        "b9": d[9],
        "b10": d[10],
        "dpad": d[11],
    }


class Calibration:
    QUIET_SPAN = 24      # raw units of max-min inside the window that still count as "still"
    GIVE_UP_S = 15.0     # after this long without a still window, use what we have
    MIN_TRAVEL = 100     # an axis whose calibrated travel is below this probably wasn't moved

    def __init__(self, deadzone: float, settle: float, center: list[int] | None):
        self.deadzone = min(max(deadzone, 0.0), 0.9)
        self.settle = max(settle, 0.3)
        self.fixed = center is not None
        self.center: dict[str, int] = dict(zip(AXES, center)) if center else {a: NOMINAL for a in AXES}
        self.lo: dict[str, int] = {a: 0 for a in AXES}          # learned raw extremes
        self.hi: dict[str, int] = {a: RAW_MAX for a in AXES}
        self.path: pathlib.Path | None = None
        self.dirty = False
        self.recording: list[dict[str, int]] | None = None
        self._buf: list[tuple[float, dict[str, int]]] = []
        self._done: set[str] = set()
        self._started: float | None = None

    # ----- persisted range -----
    def load(self, path: pathlib.Path) -> None:
        self.path = path
        try:
            data = json.loads(path.read_text())
            self.lo = {a: int(data["min"][a]) for a in AXES}
            self.hi = {a: int(data["max"][a]) for a in AXES}
        except FileNotFoundError:
            log.info("no calibration file (%s): using the nominal 0..1023 range. Run once with --calibrate.", path)
            return
        except (OSError, ValueError, KeyError, TypeError) as exc:
            log.warning("ignoring unreadable calibration file %s: %s", path, exc)
            return
        log.info("calibration %s (saved %s): L x %d..%d y %d..%d  R x %d..%d y %d..%d", path.name, data.get("saved", "?"),
                 self.lo["lx"], self.hi["lx"], self.lo["ly"], self.hi["ly"], self.lo["rx"], self.hi["rx"], self.lo["ry"], self.hi["ry"])

    def save(self) -> bool:
        if self.path is None:
            return False
        # Write and rename: in the shared directory another user's file can be replaced but not rewritten.
        tmp = self.path.with_name(f".{self.path.name}.{os.getpid()}")
        try:
            self.path.parent.mkdir(parents=True, exist_ok=True)
            tmp.write_text(json.dumps({"saved": time.strftime("%Y-%m-%d %H:%M"), "center": self.center,
                                       "min": self.lo, "max": self.hi}, indent=1) + "\n")
            os.replace(tmp, self.path)
        except OSError as exc:
            log.warning("cannot save calibration to %s: %s", self.path, exc)
            return False
        self.dirty = False
        log.info("calibration saved to %s", self.path)
        return True

    def finish_calibration(self) -> bool:
        samples, self.recording = self.recording or [], None
        if len(samples) < 50:
            log.warning("only %d samples: the sticks hardly moved, calibration not saved", len(samples))
            return False
        for a in AXES:
            values = sorted(v[a] for v in samples if v[a] not in (0, RAW_MAX))   # rails = wiper lift-off
            if len(values) < 10:
                continue
            self.lo[a], self.hi[a] = values[0], values[-1]
            if self.hi[a] - self.center[a] < self.MIN_TRAVEL or self.center[a] - self.lo[a] < self.MIN_TRAVEL:
                log.warning("%s travel looks short (%d..%d around %d): push that stick to its limits and calibrate again",
                            a.upper(), self.lo[a], self.hi[a], self.center[a])
        log.info("calibrated range: L x %d..%d y %d..%d  R x %d..%d y %d..%d",
                 self.lo["lx"], self.hi["lx"], self.lo["ly"], self.hi["ly"], self.lo["rx"], self.hi["rx"], self.lo["ry"], self.hi["ry"])
        self.dirty = True
        return True

    def _learn(self, raw: dict) -> None:
        for a in AXES:
            v = raw[a]
            if v in (0, RAW_MAX):
                continue
            if v > self.hi[a]:
                self.hi[a], self.dirty = v, True
            elif v < self.lo[a]:
                self.lo[a], self.dirty = v, True

    # ----- centre -----
    def start_sampling(self) -> None:
        if self.fixed:
            return
        self._buf = []
        self._done = set()
        self.center = {a: NOMINAL for a in AXES}
        self._started = time.monotonic()
        log.info("sampling the stick centre: leave the sticks alone for %.1f s", self.settle)

    def feed(self, raw: dict) -> None:
        if self.recording is not None:
            self.recording.append({a: raw[a] for a in AXES})
        else:
            self._learn(raw)
        if self._started is None:
            return
        now = time.monotonic()
        self._buf.append((now, {a: raw[a] for a in AXES}))
        self._buf = [(t, r) for t, r in self._buf if now - t <= self.settle]
        if len(self._buf) >= 3 and now - self._buf[0][0] >= 0.8 * self.settle:
            for a in AXES:
                if a in self._done:
                    continue
                values = [r[a] for _, r in self._buf]
                if max(values) - min(values) <= self.QUIET_SPAN:   # this axis held still for a window
                    self.center[a] = int(statistics.median(values))
                    self._done.add(a)
            if len(self._done) == len(AXES):
                self._finish("")
                return
        if now - self._started > self.GIVE_UP_S:
            missing = [a for a in AXES if a not in self._done]
            self._finish(f" ({', '.join(missing).upper()} never still, kept 512; hold C1+C2 to redo)")

    def _finish(self, note: str) -> None:
        self._started = None
        log.info("centre = L %d/%d  R %d/%d (raw, 512 is nominal)%s",
                 self.center["lx"], self.center["ly"], self.center["rx"], self.center["ry"], note)

    # ----- mapping -----
    def _norm(self, a: str, value: int) -> float:
        centre = self.center[a]
        delta = value - centre
        span = (self.hi[a] - centre) if delta > 0 else (centre - self.lo[a])
        return max(-1.0, min(1.0, delta / max(span, 1)))

    def stick(self, raw: dict, ax: str, ay: str) -> tuple[int, int]:
        x = self._norm(ax, raw[ax])
        y = self._norm(ay, raw[ay])
        mag = math.hypot(x, y)
        if mag <= self.deadzone:
            return 0, 0
        k = min(1.0, (mag - self.deadzone) / (1.0 - self.deadzone)) / mag
        return (int(round(max(-1.0, min(1.0, x * k)) * STICK_MAX)),
                int(round(max(-1.0, min(1.0, y * k)) * STICK_MAX)))


def make_uinput(generic_ids: bool) -> UInput:
    stick = AbsInfo(value=0, min=-STICK_MAX - 1, max=STICK_MAX, fuzz=16, flat=128, resolution=0)
    trigger = AbsInfo(value=0, min=0, max=255, fuzz=0, flat=0, resolution=0)
    hat = AbsInfo(value=0, min=-1, max=1, fuzz=0, flat=0, resolution=0)
    caps = {
        e.EV_KEY: sorted(set(BUTTONS_B9.values()) | set(BUTTONS_B10.values())),
        e.EV_ABS: [
            (e.ABS_X, stick), (e.ABS_Y, stick), (e.ABS_RX, stick), (e.ABS_RY, stick),
            (e.ABS_Z, trigger), (e.ABS_RZ, trigger),
            (e.ABS_HAT0X, hat), (e.ABS_HAT0Y, hat),
        ],
    }
    if generic_ids:
        ids = dict(name="GameSir T1d (bridge)", vendor=0x3537, product=0x01D1, version=0x0100)
    else:  # Xbox 360 wired pad IDs: SDL and Steam apply their built-in mapping
        ids = dict(name="Microsoft X-Box 360 pad (GameSir T1d bridge)", vendor=0x045E, product=0x028E, version=0x0110)
    try:
        return UInput(caps, bustype=e.BUS_USB, **ids)
    except PermissionError:
        sys.exit("cannot open /dev/uinput: install gamesir-ctrl/60-t1d-bridge-uinput.rules "
                 "(see the docstring) or run with sudo")


class Bridge:
    def __init__(self, ui: UInput, cal: Calibration, flip_y: bool, debug: bool, rail_timeout: float = 0.0):
        self.ui, self.cal, self.flip_y, self.debug = ui, cal, flip_y, debug
        self.rail_timeout = rail_timeout
        self._rail_since: dict[str, float | None] = {a: None for a in AXES}
        self._rail_muted: set[str] = set()
        self._hold_since: float | None = None
        self._hold_done = False
        self._last_dbg = None
        self.packets = 0
        self.dropped = 0
        self._raw_logged = 0

    def on_notify(self, _handle, data: bytearray) -> None:
        if self.debug and self._raw_logged < 12:
            self._raw_logged += 1
            log.debug("raw %d bytes: %s", len(data), bytes(data).hex())
        if len(data) != REPORT_LEN or bytes(data[:2]) != INPUT_TYPE:
            self.dropped += 1
            return
        raw = decode(bytes(data))
        self.packets += 1
        self.cal.feed(raw)
        self._recenter_hotkey(raw["b10"])
        if self.debug and (raw["b9"], raw["b10"], raw["dpad"]) != self._last_dbg:
            self._last_dbg = (raw["b9"], raw["b10"], raw["dpad"])
            log.debug("byte9=%s byte10=%s dpad=%d  L=%d/%d R=%d/%d L2=%d R2=%d",
                      format(raw["b9"], "08b"), format(raw["b10"], "08b"), raw["dpad"],
                      raw["lx"], raw["ly"], raw["rx"], raw["ry"], raw["l2"], raw["r2"])
        self._emit(raw)

    def _recenter_hotkey(self, b10: int) -> None:
        if b10 & RECENTER_MASK_B10 == RECENTER_MASK_B10:
            now = time.monotonic()
            if self._hold_since is None:
                self._hold_since = now
            elif not self._hold_done and now - self._hold_since >= RECENTER_HOLD_S:
                self._hold_done = True
                self.cal.start_sampling()
        else:
            self._hold_since, self._hold_done = None, False

    def _rail_filter(self, raw: dict) -> dict:
        """Replace an axis stuck at a rail for longer than --rail-timeout with its centre."""
        if self.rail_timeout <= 0:
            return raw
        now = time.monotonic()
        out = dict(raw)
        for a in AXES:
            if raw[a] in (0, RAW_MAX):
                if self._rail_since[a] is None:
                    self._rail_since[a] = now
                elif now - self._rail_since[a] >= self.rail_timeout:
                    if a not in self._rail_muted:
                        self._rail_muted.add(a)
                        log.warning("%s stuck at %d for %.1f s: treating it as centred until it moves (wiper dropout?)",
                                    a.upper(), raw[a], self.rail_timeout)
                    out[a] = self.cal.center[a]
            else:
                if a in self._rail_muted:
                    self._rail_muted.discard(a)
                    log.info("%s moved again (%d), back to normal", a.upper(), raw[a])
                self._rail_since[a] = None
        return out

    def _emit(self, raw: dict) -> None:
        ui = self.ui
        raw = self._rail_filter(raw)
        lx, ly = self.cal.stick(raw, "lx", "ly")
        rx, ry = self.cal.stick(raw, "rx", "ry")
        if self.flip_y:
            ly, ry = -ly, -ry
        ui.write(e.EV_ABS, e.ABS_X, lx)
        ui.write(e.EV_ABS, e.ABS_Y, ly)
        ui.write(e.EV_ABS, e.ABS_RX, rx)
        ui.write(e.EV_ABS, e.ABS_RY, ry)
        ui.write(e.EV_ABS, e.ABS_Z, raw["l2"])
        ui.write(e.EV_ABS, e.ABS_RZ, raw["r2"])
        hx, hy = HAT.get(raw["dpad"], (0, 0))
        ui.write(e.EV_ABS, e.ABS_HAT0X, hx)
        ui.write(e.EV_ABS, e.ABS_HAT0Y, hy)
        for mask, code in BUTTONS_B9.items():
            ui.write(e.EV_KEY, code, 1 if raw["b9"] & mask else 0)
        for mask, code in BUTTONS_B10.items():
            ui.write(e.EV_KEY, code, 1 if raw["b10"] & mask else 0)
        ui.syn()

    def release_all(self) -> None:
        self._emit({**{a: self.cal.center[a] for a in AXES}, "l2": 0, "r2": 0, "b9": 0, "b10": 0, "dpad": 0})


async def find_controller(address: str | None, prefix: str, timeout: float):
    if address:
        return await BleakScanner.find_device_by_address(address, timeout=timeout)
    return await BleakScanner.find_device_by_filter(
        lambda d, _ad: (d.name or "").startswith(prefix), timeout=timeout)


def write_status(path: str | None, text: str) -> None:
    """One line for the KDE menu (gamesir-t1d): waiting, connected <name>, stopped."""
    if not path:
        return
    try:
        pathlib.Path(path).write_text(text + "\n")
    except OSError as exc:
        log.warning("cannot write status file %s: %s", path, exc)


async def run(args) -> int:
    ui = make_uinput(args.generic_ids)
    log.info("virtual gamepad created: %s (%s)", ui.device.path, ui.name)
    cal = Calibration(args.deadzone, args.settle, args.center)
    bridge = Bridge(ui, cal, args.flip_y, args.debug, args.rail_timeout)
    stop = asyncio.Event()
    loop = asyncio.get_running_loop()
    for sig in (signal.SIGINT, signal.SIGTERM):
        loop.add_signal_handler(sig, stop.set)

    announced = scan_failed = False
    write_status(args.status_file, "waiting")
    while not stop.is_set():
        scan = asyncio.ensure_future(find_controller(args.address, args.name_prefix, timeout=8))
        stopping = asyncio.ensure_future(stop.wait())
        await asyncio.wait([scan, stopping], return_when=asyncio.FIRST_COMPLETED)
        stopping.cancel()
        if stop.is_set():
            scan.cancel()
            break
        try:
            dev = scan.result()
        except (BleakError, OSError, asyncio.TimeoutError) as exc:   # adapter off or missing, BlueZ restarting
            if not scan_failed:
                log.warning("Bluetooth scan failed: %s; retrying every 2 s", exc)
                scan_failed = True
            await asyncio.sleep(2)
            continue
        scan_failed = False
        if dev is None:
            if not announced:
                log.info("waiting for the controller (hold Power 2 s; the LEDs should blink)")
                announced = True
            await asyncio.sleep(1)
            continue
        announced = False
        disconnected = asyncio.Event()
        try:
            async with BleakClient(dev, timeout=20, disconnected_callback=lambda _c: disconnected.set()) as client:
                log.info("connected to %s (%s)", dev.name, dev.address)
                write_status(args.status_file, f"connected {dev.name}")
                if not args.no_calib_file:
                    path = pathlib.Path(args.calib_file) if args.calib_file else CALIB_DIR / f"{dev.address.replace(':', '-')}.json"
                    if cal.path != path:
                        cal.load(path)
                cal.start_sampling()
                await client.start_notify(INPUT_CHAR, bridge.on_notify)
                # The T1d sends reports only when something changes, so read the
                # current state once; right after power-on this returns the status blob.
                first = await client.read_gatt_char(INPUT_CHAR)
                log.debug("initial read: %s", bytes(first).hex())
                bridge.on_notify(None, first)
                if args.calibrate:
                    log.info("CALIBRATION: for the next %d s move BOTH sticks slowly around their full circle "
                             "and into every corner, then let them go", args.calibrate_seconds)
                    cal.recording = []
                    await asyncio.sleep(args.calibrate_seconds)
                    if cal.finish_calibration():
                        cal.save()
                    stop.set()
                waiters = [asyncio.ensure_future(disconnected.wait()), asyncio.ensure_future(stop.wait())]
                await asyncio.wait(waiters, return_when=asyncio.FIRST_COMPLETED)
                for w in waiters:
                    w.cancel()
                if stop.is_set():
                    try:
                        await client.stop_notify(INPUT_CHAR)
                    except BleakError:
                        pass
                else:
                    log.info("controller disconnected (sleep or out of range); reconnecting")
        except (BleakError, asyncio.TimeoutError, OSError) as exc:
            log.warning("BLE error: %s; retrying", exc)
            await asyncio.sleep(2)
        finally:
            bridge.release_all()
            if cal.dirty:
                cal.save()
            write_status(args.status_file, "waiting")
    write_status(args.status_file, "stopped")
    ui.close()
    log.info("stopped after %d input reports (%d other packets dropped)", bridge.packets, bridge.dropped)
    return 0


def selftest(generic_ids: bool) -> int:
    ui = make_uinput(generic_ids)
    print(f"virtual gamepad created: {ui.device.path}  name={ui.name!r}")
    print("sweeping the left stick and pressing A B X Y for 4 s; watch it in evtest, "
          "jstest or Steam > Settings > Controller")
    steps = 40
    for i in range(steps + 1):
        v = int(-STICK_MAX + 2 * STICK_MAX * i / steps)
        ui.write(e.EV_ABS, e.ABS_X, v)
        ui.write(e.EV_ABS, e.ABS_Y, -v)
        ui.syn()
        time.sleep(2.0 / steps)
    for code in (e.BTN_A, e.BTN_B, e.BTN_X, e.BTN_Y):
        ui.write(e.EV_KEY, code, 1); ui.syn(); time.sleep(0.25)
        ui.write(e.EV_KEY, code, 0); ui.syn(); time.sleep(0.25)
    ui.write(e.EV_ABS, e.ABS_X, 0); ui.write(e.EV_ABS, e.ABS_Y, 0); ui.syn()
    ui.close()
    print("selftest done")
    return 0


def parse_center(text: str) -> list[int]:
    parts = [int(p) for p in text.split(",")]
    if len(parts) != 4 or not all(0 <= p <= RAW_MAX for p in parts):
        raise argparse.ArgumentTypeError("expected LX,LY,RX,RY with values 0-1023")
    return parts


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0], formatter_class=argparse.RawDescriptionHelpFormatter,
                                 epilog=__doc__.split("\n", 1)[1])
    ap.add_argument("--deadzone", type=float, default=0.15, help="radial dead zone, fraction of half travel (default 0.15)")
    ap.add_argument("--settle", type=float, default=1.0, help="length of the still window sampled for the centre (default 1.0 s)")
    ap.add_argument("--center", type=parse_center, help="fixed raw centre LX,LY,RX,RY instead of sampling")
    ap.add_argument("--flip-y", action="store_true", help="invert both stick Y axes")
    ap.add_argument("--calibrate", action="store_true", help="record the sticks' real range, save it, exit")
    ap.add_argument("--calibrate-seconds", type=int, default=10, metavar="N", help="recording time for --calibrate (default 10)")
    ap.add_argument("--calib-file", metavar="PATH", help=f"calibration file (default {CALIB_DIR}/<MAC>.json)")
    ap.add_argument("--no-calib-file", action="store_true", help="ignore any calibration file, use the nominal 0..1023 range")
    ap.add_argument("--rail-timeout", type=float, default=0.0, metavar="SECONDS",
                    help="treat an axis stuck at 0 or 1023 for this long as centred (wiper dropout work-around; 0 = off)")
    ap.add_argument("--address", help="controller MAC, skips the name scan")
    ap.add_argument("--name-prefix", default=NAME_PREFIX, help=f"BLE name prefix to look for (default {NAME_PREFIX})")
    ap.add_argument("--status-file", metavar="PATH", help="keep a one-line state (waiting, connected <name>, stopped) here")
    ap.add_argument("--generic-ids", action="store_true", help="present a plain gamepad instead of Xbox 360 IDs")
    ap.add_argument("--selftest", action="store_true", help="create the virtual pad, move it, exit; no controller needed")
    ap.add_argument("-v", "--verbose", action="store_true")
    ap.add_argument("--debug", action="store_true", help="log raw button bytes on change (find unknown bits)")
    args = ap.parse_args()
    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname).1s %(message)s", datefmt="%H:%M:%S")
    log.setLevel(logging.DEBUG if args.debug else logging.INFO)
    for name in ("bleak", "dbus_fast"):
        logging.getLogger(name).setLevel(logging.INFO if args.verbose else logging.WARNING)
    if args.selftest:
        return selftest(args.generic_ids)
    return asyncio.run(run(args))


if __name__ == "__main__":
    sys.exit(main())