t1d-bridge.py
Files / gamesir-t1d / t1d-bridge.py
Files in gamesir-t1d/
Directorygamesir-t1d/
- 60-t1d-bridge-uinput.rules 350 bytes
- LICENSE 1.0 KB
- t1d-bridge.py 24.1 KB
- t1d-probe.py 3.8 KB
From Dictation on Linux with Handy, and tidy text with OpenCode Go and Use the Tello's GameSir T1d controller as a Linux gamepad. MIT licence: LICENSE.
#!/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 devicethrough python-evdev, so Steam, SDL games and anything else that understandsan Xbox 360 pad can use it. The T1d has no Bluetooth HID profile, so thisbridge 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=uinputRun: ~/.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 argparseimport asyncioimport jsonimport loggingimport osimport pathlibimport mathimport signalimport statisticsimport sysimport time
from bleak import BleakClient, BleakScannerfrom bleak.exc import BleakErrorfrom 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 droppedREPORT_LEN = 20RAW_MAX = 1023 # 10-bit stick axesSTICK_MAX = 32767NOMINAL = 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 togetherRECENTER_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())Text, photos and drawings: CC BY 4.0. Code: MIT. Share and adapt them, crediting HomeLab Guides (blog.jacq.cc). Licence and privacy