#!/usr/bin/env python3 import ctypes import ctypes.util import os import re import select import pty import threading import queue import struct import subprocess import time import wave import json import socket import signal import base64 import hashlib import urllib.parse import ssl from pathlib import Path AF_BLUETOOTH = 31 SOCK_SEQPACKET = 5 SOCK_RAW = 3 BTPROTO_L2CAP = 0 BTPROTO_HCI = 1 ATT_CID = 4 SMP_CID = 6 BDADDR_LE_PUBLIC = 1 HCI_CHANNEL_RAW = 0 # Physical Bluetooth identities are runtime configuration. # REMOTE is selected by KodiFlix VoiceHelper Setup. # LOCAL is discovered from the current box's hci0 adapter. REMOTE = "" LOCAL = "" # VoiceHelper STT server. STT_HOST = "stt.kodiflix.net" STT_PORT = 443 STT_PATH = "/ws" STT_TLS = True STT_HEARTBEAT_SECONDS = 30.0 STT_READY_TIMEOUT = 2.0 STT_RETRY_SECONDS = 5.0 BT_READY_TIMEOUT_SECONDS = 30.0 BT_READY_STABLE_POLLS = 3 BT_READY_POLL_SECONDS = 0.20 # The LE ACL can appear in hcitool before the kernel fixed-CID ATT socket is # ready. Retrying ATT locally avoids tearing the whole Bluetooth stack down # just because L2CAP needs a few hundred milliseconds to settle. ATT_ATTACH_RETRY_SECONDS = 3.0 ATT_ATTACH_RETRY_INTERVAL = 0.15 ATT_ATTACH_RETRY_ERRNOS = {16, 38, 110} # EBUSY / ENOSYS-ish / ETIMEDOUT seen here SERVICE_DATA = Path("/storage/.kodi/userdata/addon_data/service.voicehelper") REMOTE_MAP_FILE = SERVICE_DATA / "remote-map.json" REMOTE_CONFIG_FILE = SERVICE_DATA / "remote.json" BUTTON_ACTIONS_FILE = SERVICE_DATA / "button-actions.json" CONFIG_FILE = SERVICE_DATA / "config.json" LAST_COMMAND_FILE = SERVICE_DATA / "last-command.json" UI_STATE_FILE = SERVICE_DATA / "ui-state.json" REMOTECACHE_STATE_FILE = Path("/storage/.kodi/userdata/addon_data/service.widgethelper.remotecache/state.json") HID_CCCD_HANDLES = [0x0030,0x0035,0x0039,0x0040,0x0047,0x004B,0x004F,0x0053,0x005A] DEFAULT_BUTTON_ACTIONS = { "up":"Up","down":"Down","left":"Left","right":"Right","ok":"Select", "back":"Back","home":"ActivateWindow(Home)", "volume_up":"VolumeUp","volume_down":"VolumeDown", "voice":None,"netflix":"RunScript(script.voicehelper.bridge,command=watchlist_toggle)","prime_video":None,"youtube":None, "apps":"ContextMenu", "xiaomi_tv":"RunScript(script.kodiflix.info)", "power":None } # Kodi's normal keymap changes D-pad behavior during fullscreen playback. # Because VoiceHelper injects actions directly, we reproduce that context here. FULLSCREEN_VIDEO_ACTIONS = { "left": "StepBack", "right": "StepForward", "up": "ChapterOrBigStepForward", "down": "ChapterOrBigStepBack", "ok": "ActivateWindow(videoosd)", "back": "Back", "volume_up": "VolumeUp", "volume_down": "VolumeDown", } DEFAULT_SERVICE_CONFIG = {} # Kodi JSON-RPC TCP service. This avoids HTTP credentials and keeps the # VoiceHelper process independent of Kodi's embedded Python runtime. KODI_RPC_HOST = "127.0.0.1" KODI_RPC_PORT = 9090 LOG = Path("/storage/voicehelper-service.log") BTMON = Path("/storage/voicehelper-service-btmon.log") # Exact Pairing Request from the Xiaomi/CoreELEC exchange that completed successfully. # KeyboardDisplay, Bonding/No-MITM/Legacy, key size 16, key dist 0x05/0x07. PREQ = bytes.fromhex("01 04 00 01 10 05 07") libc = ctypes.CDLL(ctypes.util.find_library("c"), use_errno=True) # --------------------------------------------------------------------------- # Minimal AES-128 implementation # --------------------------------------------------------------------------- SBOX = ( 0x63,0x7c,0x77,0x7b,0xf2,0x6b,0x6f,0xc5,0x30,0x01,0x67,0x2b,0xfe,0xd7,0xab,0x76, 0xca,0x82,0xc9,0x7d,0xfa,0x59,0x47,0xf0,0xad,0xd4,0xa2,0xaf,0x9c,0xa4,0x72,0xc0, 0xb7,0xfd,0x93,0x26,0x36,0x3f,0xf7,0xcc,0x34,0xa5,0xe5,0xf1,0x71,0xd8,0x31,0x15, 0x04,0xc7,0x23,0xc3,0x18,0x96,0x05,0x9a,0x07,0x12,0x80,0xe2,0xeb,0x27,0xb2,0x75, 0x09,0x83,0x2c,0x1a,0x1b,0x6e,0x5a,0xa0,0x52,0x3b,0xd6,0xb3,0x29,0xe3,0x2f,0x84, 0x53,0xd1,0x00,0xed,0x20,0xfc,0xb1,0x5b,0x6a,0xcb,0xbe,0x39,0x4a,0x4c,0x58,0xcf, 0xd0,0xef,0xaa,0xfb,0x43,0x4d,0x33,0x85,0x45,0xf9,0x02,0x7f,0x50,0x3c,0x9f,0xa8, 0x51,0xa3,0x40,0x8f,0x92,0x9d,0x38,0xf5,0xbc,0xb6,0xda,0x21,0x10,0xff,0xf3,0xd2, 0xcd,0x0c,0x13,0xec,0x5f,0x97,0x44,0x17,0xc4,0xa7,0x7e,0x3d,0x64,0x5d,0x19,0x73, 0x60,0x81,0x4f,0xdc,0x22,0x2a,0x90,0x88,0x46,0xee,0xb8,0x14,0xde,0x5e,0x0b,0xdb, 0xe0,0x32,0x3a,0x0a,0x49,0x06,0x24,0x5c,0xc2,0xd3,0xac,0x62,0x91,0x95,0xe4,0x79, 0xe7,0xc8,0x37,0x6d,0x8d,0xd5,0x4e,0xa9,0x6c,0x56,0xf4,0xea,0x65,0x7a,0xae,0x08, 0xba,0x78,0x25,0x2e,0x1c,0xa6,0xb4,0xc6,0xe8,0xdd,0x74,0x1f,0x4b,0xbd,0x8b,0x8a, 0x70,0x3e,0xb5,0x66,0x48,0x03,0xf6,0x0e,0x61,0x35,0x57,0xb9,0x86,0xc1,0x1d,0x9e, 0xe1,0xf8,0x98,0x11,0x69,0xd9,0x8e,0x94,0x9b,0x1e,0x87,0xe9,0xce,0x55,0x28,0xdf, 0x8c,0xa1,0x89,0x0d,0xbf,0xe6,0x42,0x68,0x41,0x99,0x2d,0x0f,0xb0,0x54,0xbb,0x16) RCON = (0x00,0x01,0x02,0x04,0x08,0x10,0x20,0x40,0x80,0x1b,0x36) def xtime(a): return ((a << 1) ^ 0x1b) & 0xff if a & 0x80 else (a << 1) & 0xff def aes_expand(key): w = list(key) i = 16 r = 1 while len(w) < 176: t = w[-4:] if i % 16 == 0: t = t[1:] + t[:1] t = [SBOX[x] for x in t] t[0] ^= RCON[r] r += 1 for x in t: w.append(w[-16] ^ x) i += 1 return w def add_round_key(s, rk): for i in range(16): s[i] ^= rk[i] def sub_bytes(s): for i in range(16): s[i] = SBOX[s[i]] def shift_rows(s): # AES state is column-major. s[1],s[5],s[9],s[13] = s[5],s[9],s[13],s[1] s[2],s[6],s[10],s[14] = s[10],s[14],s[2],s[6] s[3],s[7],s[11],s[15] = s[15],s[3],s[7],s[11] def mix_columns(s): for c in range(4): i = c * 4 a = s[i:i+4] t = a[0] ^ a[1] ^ a[2] ^ a[3] u = a[0] s[i] ^= t ^ xtime(a[0] ^ a[1]) s[i+1] ^= t ^ xtime(a[1] ^ a[2]) s[i+2] ^= t ^ xtime(a[2] ^ a[3]) s[i+3] ^= t ^ xtime(a[3] ^ u) def aes128_be(key, block): rk = aes_expand(key) s = list(block) add_round_key(s, rk[0:16]) for rnd in range(1,10): sub_bytes(s) shift_rows(s) mix_columns(s) add_round_key(s, rk[rnd*16:(rnd+1)*16]) sub_bytes(s) shift_rows(s) add_round_key(s, rk[160:176]) return bytes(s) def e_le(key_le, plaintext_le): # Bluetooth Core legacy SMP uses little-endian octet strings around AES. return aes128_be(key_le[::-1], plaintext_le[::-1])[::-1] def xor16(a,b): return bytes(x ^ y for x,y in zip(a,b)) def c1(k, r, preq, pres, iat, ia, rat, ra): """ Bluetooth LE Legacy Pairing c1(), using the same little-endian in-memory representation as the Linux kernel SMP implementation. The Core spec writes: p1 = pres || preq || rat || iat p2 = padding || ia || ra But Linux stores these Bluetooth octet strings least-significant octet first. Because e_le() below already performs the byte-order conversion around AES, the arrays passed to e_le() must match Linux's in-memory form: p1 memory = iat || rat || preq || pres p2 memory = ra || ia || padding """ p1 = bytes([iat]) + bytes([rat]) + preq + pres p2 = ra + ia + b"\x00\x00\x00\x00" return e_le( k, xor16( e_le(k, xor16(r, p1)), p2 ) ) def s1(k, initiator_random, responder_random): # Xiaomi/Linux validated Legacy SMP STK construction. # # In our captured/wire representation, the AES input is: # LSB64(initiator_random) || LSB64(responder_random) # # This reproduces the exact STK used by the successful kernel pairing: # 6e a4 a5 6d 13 14 fc 49 c7 66 9d 3b ce 2d 16 52 return e_le(k, initiator_random[:8] + responder_random[:8]) # --------------------------------------------------------------------------- # Bluetooth plumbing # --------------------------------------------------------------------------- class BdAddr(ctypes.Structure): _fields_ = [("b", ctypes.c_ubyte * 6)] class SockAddrL2(ctypes.Structure): _pack_ = 1 _fields_ = [ ("l2_family", ctypes.c_ushort), ("l2_psm", ctypes.c_ushort), ("l2_bdaddr", BdAddr), ("l2_cid", ctypes.c_ushort), ("l2_bdaddr_type", ctypes.c_ubyte), ] class SockAddrHCI(ctypes.Structure): _pack_ = 1 _fields_ = [ ("hci_family", ctypes.c_ushort), ("hci_dev", ctypes.c_ushort), ("hci_channel", ctypes.c_ushort), ] def log(s=""): print(s, flush=True) with LOG.open("a") as f: f.write(s + "\n") def run(cmd, timeout=10): try: p = subprocess.run(cmd, text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=timeout) return p.returncode, p.stdout.strip() except subprocess.TimeoutExpired as e: out=e.stdout or "" if isinstance(out,bytes): out=out.decode(errors="replace") return 124,out.strip() def normalize_bt_mac(value): raw = re.sub(r"[^0-9A-Fa-f]", "", str(value or "")).upper() if len(raw) != 12: return "" return ":".join(raw[i:i+2] for i in range(0, 12, 2)) def load_configured_remote(): """Load physical remote MAC supplied by the Kodi service supervisor.""" mac = normalize_bt_mac(os.getenv("VOICEHELPER_REMOTE_MAC", "")) if not mac: raise RuntimeError( "VOICEHELPER_REMOTE_MAC is not configured by service supervisor" ) return mac, "Xiaomi RC" def detect_local_adapter_mac(): """Discover this box's Bluetooth public address from kernel hci0.""" rc, out = run(["hciconfig", "hci0"], timeout=3) if rc != 0: raise RuntimeError(f"hciconfig hci0 failed rc={rc}: {out.strip()}") patterns = ( r"BD Address:\s*((?:[0-9A-Fa-f]{2}:){5}[0-9A-Fa-f]{2})", r"Address:\s*((?:[0-9A-Fa-f]{2}:){5}[0-9A-Fa-f]{2})", ) for pattern in patterns: m = re.search(pattern, out) if m: mac = normalize_bt_mac(m.group(1)) if mac: return mac raise RuntimeError( "could not discover hci0 Bluetooth address from hciconfig output" ) def addr_le(mac): return bytes.fromhex(mac.replace(":",""))[::-1] def bdaddr(mac): return BdAddr((ctypes.c_ubyte*6)(*addr_le(mac))) def mk_l2(mac): a=SockAddrL2() a.l2_family=AF_BLUETOOTH a.l2_psm=0 a.l2_bdaddr=bdaddr(mac) a.l2_cid=ATT_CID a.l2_bdaddr_type=BDADDR_LE_PUBLIC return a def writefd(fd,data): b=ctypes.create_string_buffer(data) if libc.write(fd,b,len(data))<0: raise OSError(ctypes.get_errno(),"write failed") def readfd(fd,timeout,size=512): r,_,_=select.select([fd],[],[],timeout) if not r: return None b=ctypes.create_string_buffer(size) n=libc.read(fd,b,size) if n<0: raise OSError(ctypes.get_errno(),"read failed") return bytes(b.raw[:n]) def recover_stale_bluez_ownership(): """ Recover from a previous VoiceHelper instance that was terminated by a Kodi restart before its finally-block could release the runtime systemd mask. /run survives a Kodi restart, so bluetooth.service can remain masked-runtime and inactive even though the previous raw HCI owner is gone. A real OS reboot clears this automatically. """ enabled_rc, enabled_out = run( ["systemctl", "is-enabled", "bluetooth.service"], timeout=3 ) enabled_state = enabled_out.strip() active_rc, active_out = run( ["systemctl", "is-active", "bluetooth.service"], timeout=3 ) active_state = active_out.strip() stale_mask = (enabled_state == "masked-runtime") if not stale_mask: log( f"Bluetooth preflight: service enabled_state={enabled_state!r} " f"active_state={active_state!r}; no stale VoiceHelper mask." ) return False log( "Bluetooth preflight: stale runtime mask detected from previous " "VoiceHelper instance; restoring BlueZ before takeover." ) rc, out = run( ["systemctl", "unmask", "--runtime", "bluetooth.service"], timeout=5 ) log(f"bluetooth.service runtime unmask rc={rc}: {out.strip()}") rc, out = run( ["systemctl", "start", "bluetooth.service"], timeout=8 ) log(f"bluetooth.service restart rc={rc}: {out.strip()}") # Best effort: CoreELEC/Amlogic can leave the controller unpowered after # ownership transitions. BlueZ will normally power it, but request power # explicitly if needed. try: run(["bluetoothctl", "power", "on"], timeout=5) except Exception: pass return True def wait_controller_ready(timeout=BT_READY_TIMEOUT_SECONDS): """ State-driven Bluetooth startup gate. No fixed boot sleep. VoiceHelper waits until: 1) bluetooth.service is active 2) hci0 exists as a Primary controller 3) btmgmt info succeeds for several consecutive polls Requiring consecutive successful reads avoids grabbing hci0 during the instant the Amlogic firmware/controller first appears but is not stable yet. """ deadline = time.monotonic() + timeout stable = 0 last_service = None last_info = "" while time.monotonic() < deadline: svc_rc, svc_out = run( ["systemctl", "is-active", "bluetooth.service"], timeout=2 ) service_active = (svc_rc == 0 and svc_out.strip() == "active") info_rc, info_out = run(["btmgmt", "info"], timeout=3) last_service = svc_out.strip() last_info = info_out controller_ready = ( info_rc == 0 and "hci0:" in info_out and "Primary controller" in info_out and "supported settings:" in info_out ) if service_active and controller_ready: stable += 1 if stable == 1: log( "Bluetooth startup signal detected; " "verifying controller stability..." ) if stable >= BT_READY_STABLE_POLLS: log( f"Bluetooth controller stable after " f"{BT_READY_STABLE_POLLS} consecutive polls." ) return True else: stable = 0 time.sleep(BT_READY_POLL_SECONDS) log( "Bluetooth startup gate timed out; " "VoiceHelper will not take over the controller." ) log(f"Last bluetooth.service state: {last_service!r}") if last_info: log("Last btmgmt info:") log(last_info) return False def restore_bluez(reason=""): """ Emergency fail-open path. If raw VoiceHelper takeover fails, return ownership to CoreELEC/BlueZ so the Xiaomi remote remains usable as a normal remote instead of being left with bluetoothd stopped. """ if reason: log(f"Restoring BlueZ: {reason}") try: run(["btmgmt", "power", "off"], timeout=3) except Exception: pass try: # Release VoiceHelper's exclusive ownership before restoring BlueZ. run(["systemctl", "unmask", "--runtime", "bluetooth.service"]) run(["systemctl", "start", "bluetooth.service"]) except Exception: pass time.sleep(1.5) # This CoreELEC build does not reliably power hci0 merely by starting # bluetooth.service. try: run(["bluetoothctl", "power", "on"], timeout=5) except Exception: pass time.sleep(1.0) # Best effort only. The remote may be asleep; a button press can wake it. try: run(["bluetoothctl", "trust", REMOTE], timeout=5) run(["bluetoothctl", "connect", REMOTE], timeout=6) except Exception: pass log("BlueZ recovery path completed.") def prepare(): """ Prepare hci0 for direct raw ownership only. BlueZ is not started, stopped, restored, or used as a handoff mechanism. bluetooth.service is expected to remain masked on this appliance. """ log("Preparing kernel controller for raw VoiceHelper ownership...") # Ensure no stale bluetoothd can compete for HCI ownership. try: run(["pkill", "-x", "bluetoothd"], timeout=2) except Exception: pass # hci0 must be up at the kernel layer. rc, out = run(["hciconfig", "hci0"], timeout=2) if rc != 0 or "hci0:" not in out: raise RuntimeError("kernel hci0 is unavailable") if "UP RUNNING" not in out: rc, out = run(["hciconfig", "hci0", "up"], timeout=3) if rc != 0: raise RuntimeError("could not bring kernel hci0 up") # Start from a known raw-controller state without invoking BlueZ. try: run(["btmgmt", "power", "off"], timeout=3) time.sleep(0.15) run(["btmgmt", "power", "on"], timeout=3) time.sleep(0.20) except Exception as exc: log(f"btmgmt power-cycle warning: {type(exc).__name__}: {exc}") # LE central role; no kernel/BlueZ pairing database is used. try: run(["hciconfig", "hci0", "leadv", "0"], timeout=2) except Exception: pass rc, out = run(["hciconfig", "hci0"], timeout=2) if rc != 0 or "UP RUNNING" not in out: raise RuntimeError("kernel hci0 failed raw preparation") def create_connection(): payload=bytearray() payload += b"\x00" # no accept list payload += b"\x00" # own public identity payload += b"\x00" # peer public payload += addr_le(REMOTE) payload += b"\x01" # LE 1M payload += struct.pack(" 1: log( f"ATT attach succeeded on retry {attempt} " f"after controller/L2CAP settle." ) return fd except OSError as exc: last_exc = exc err = getattr(exc, "errno", None) if err not in ATT_ATTACH_RETRY_ERRNOS: raise remaining = max(0.0, end - time.monotonic()) log( f"ATT attach transient failure errno={err} " f"attempt={attempt}; LE link still present; " f"retrying in {ATT_ATTACH_RETRY_INTERVAL:.2f}s " f"({remaining:.2f}s window left)." ) time.sleep(ATT_ATTACH_RETRY_INTERVAL) if last_exc is not None: raise last_exc raise TimeoutError("ATT attach retry window expired") def attach_att(): fd=libc.socket(AF_BLUETOOTH,SOCK_SEQPACKET,BTPROTO_L2CAP) if fd<0: raise OSError(ctypes.get_errno(),"ATT socket") la=mk_l2(LOCAL) if libc.bind(fd,ctypes.byref(la),ctypes.sizeof(la))<0: raise OSError(ctypes.get_errno(),"ATT bind") ra=mk_l2(REMOTE) if libc.connect(fd,ctypes.byref(ra),ctypes.sizeof(ra))<0: raise OSError(ctypes.get_errno(),"ATT connect") return fd def open_hci(): fd=libc.socket(AF_BLUETOOTH,SOCK_RAW,BTPROTO_HCI) if fd<0: raise OSError(ctypes.get_errno(),"HCI socket") a=SockAddrHCI() a.hci_family=AF_BLUETOOTH a.hci_dev=0 a.hci_channel=HCI_CHANNEL_RAW if libc.bind(fd,ctypes.byref(a),ctypes.sizeof(a))<0: raise OSError(ctypes.get_errno(),"HCI bind") return fd def acl_smp(handle,payload): hf=(handle & 0x0fff)|0x2000 acl_len=4+len(payload) return b"\x02"+struct.pack(">> SMP {label}: {payload.hex(' ')}") writefd(hci,frame) def hci_cmd(ogf,ocf,payload,timeout=2): cmd=["hcitool","cmd",f"0x{ogf:02x}",f"0x{ocf:04x}"]+[f"{b:02x}" for b in payload] try: p=subprocess.run(cmd,text=True,stdout=subprocess.PIPE, stderr=subprocess.STDOUT,timeout=timeout) log(p.stdout.strip()) except subprocess.TimeoutExpired as e: out=e.stdout or "" if isinstance(out,bytes): out=out.decode(errors="replace") log(out.strip()) def remote_version(handle): log("Read Remote Version...") hci_cmd(0x01,0x001d,struct.pack(" LE-ACL:.*?SMP: Pairing Response.*?' r'IO capability: NoInputNoOutput.*?' r'Authentication requirement:.*?\(0x01\).*?' r'Max encryption key size: 16.*?' r'Initiator key distribution:.*?\(0x01\).*?' r'Responder key distribution:.*?\(0x03\)') m,txt,_=wait_btmon(pat,timeout,pos) if not m: raise RuntimeError("Did not observe expected Xiaomi Pairing Response") # Actual bytes matching the observed response. return bytes.fromhex("02 03 00 01 10 01 03") def wait_remote_key_distribution(pos, timeout=4.0): """ After STK encryption, Legacy SMP bonding is not finished until the negotiated key-distribution phase completes. Xiaomi negotiated: initiator key distribution = EncKey (0x01) responder key distribution = EncKey + IdKey (0x03) The remote therefore sends: 0x06 Encryption Information 0x07 Central Identification 0x08 Identity Information 0x09 Identity Address Information Wait until the final responder key PDU is visible in btmon before sending our initiator EncKey material. """ end = time.monotonic() + timeout while time.monotonic() < end: try: txt = BTMON.read_text(errors="replace")[pos:] except Exception: txt = "" if "SMP: Identity Address Information" in txt: log("*** REMOTE SMP KEY DISTRIBUTION COMPLETE ***") log("Responder 0x06/0x07/0x08/0x09 observed; replying immediately.") return True time.sleep(0.05) log("*** WARNING: remote key distribution not observed in live window ***") return False def finish_legacy_key_distribution(hci, handle, pos): """ Finish the negotiated initiator key distribution instead of abandoning pairing immediately after STK encryption. This is deliberately ephemeral for the probe: the random LTK/EDIV/Rand are sent to the remote but are not installed into BlueZ's persistent key store. """ wait_remote_key_distribution(pos, 1.5) ltk = os.urandom(16) ediv = int.from_bytes(os.urandom(2), "little") rand = os.urandom(8) log("Completing initiator Legacy SMP key distribution...") log("v13 sanity: first outbound SMP key opcodes MUST print as 06 and 07.") log("Initiator LTK : " + ltk.hex(" ")) log(f"Initiator EDIV: 0x{ediv:04x}") log("Initiator Rand: " + rand.hex(" ")) send_smp(hci, handle, b"\x06" + ltk, "Encryption Information") time.sleep(0.05) send_smp( hci, handle, b"\x07" + struct.pack("' direction formatting anymore. PTY/color output and btmon formatting made that unnecessarily fragile. Instead: - strip ANSI escape sequences and CRs - find every requested SMP packet after `pos` - extract its 16-byte Confirm/Random value - ignore `skip_value` (the exact value we just transmitted) - return the first different 16-byte value, i.e. Xiaomi's response """ end_time = time.monotonic() + timeout ansi = re.compile(r"\x1b\[[0-9;?]*[ -/]*[@-~]") while time.monotonic() < end_time: try: txt = BTMON.read_text(errors="replace") except Exception: txt = "" tail = txt[pos:] tail = ansi.sub("", tail).replace("\r", "") # Bail immediately on an actual remote SMP failure. if "SMP: Pairing Failed" in tail: # Don't accidentally treat an old failure outside our marked tail. raise RuntimeError("Remote returned SMP Pairing Failed") lines = tail.splitlines() for i, line in enumerate(lines): if f"SMP: {kind}" not in line: continue # Find the value marker shortly after the SMP header. for j in range(i + 1, min(i + 5, len(lines))): if "value[16]:" not in lines[j]: continue # Find the next line containing at least sixteen standalone # hexadecimal octets. for k in range(j + 1, min(j + 4, len(lines))): octets = re.findall( r"(?>> ATT {req.hex(' ')}") writefd(fd, req) end = time.monotonic() + timeout while time.monotonic() < end: pkt = readfd(fd, max(0.01, end-time.monotonic())) if pkt is None: break if not pkt: continue if pkt[0] in (0x1b, 0x1d): h = pkt[1] | (pkt[2] << 8) if len(pkt) >= 3 else 0 log(f"ASYNC {'IND' if pkt[0]==0x1d else 'NOTIFY'} 0x{h:04x}: {pkt[3:].hex(' ')}") if pkt[0] == 0x1d: writefd(fd, b"\x1e") continue log(f"<<< ATT {pkt.hex(' ')}") if pkt[0] in expected: return pkt return pkt log("<<< ATT TIMEOUT") return None def find_information(fd, start, end): attrs = [] cur = start while cur <= end: pkt = att_xact(fd, b"\x04" + struct.pack("= end: break cur = last + 1 return attrs def discover_characteristics(fd, start, end): chars = [] cur = start while cur <= end: req = b"\x08" + struct.pack("= end: break cur = last + 1 return chars def read_handle(fd, handle): pkt = att_xact(fd, b"\x0a" + struct.pack(" 0x{handle:04x}: {payload.hex(' ')}") writefd(fd, pkt) def dump_btmon_tail(lines=80): try: time.sleep(0.25) txt = BTMON.read_text(errors="replace").splitlines() log("") log("--- btmon tail after voice event ---") for line in txt[-lines:]: log(line) log("--- end btmon tail ---") except Exception as e: log(f"Could not dump btmon tail: {e}") # --------------------------------------------------------------------------- # Kodi / TMDB Helper bridge # --------------------------------------------------------------------------- def kodi_rpc(method, params=None, timeout=3.0): req = { "jsonrpc": "2.0", "id": 1, "method": method, } if params is not None: req["params"] = params wire = (json.dumps(req, separators=(",", ":")) + "\n").encode("utf-8") with socket.create_connection((KODI_RPC_HOST, KODI_RPC_PORT), timeout=timeout) as sock: sock.settimeout(timeout) sock.sendall(wire) buf = bytearray() while True: chunk = sock.recv(4096) if not chunk: break buf.extend(chunk) try: obj = json.loads(buf.decode("utf-8")) if obj.get("id") == 1: return obj except (UnicodeDecodeError, json.JSONDecodeError): continue if len(buf) > 1024 * 1024: raise RuntimeError("Kodi JSON-RPC response too large") raise RuntimeError("No complete Kodi JSON-RPC response") def ensure_service_files(): SERVICE_DATA.mkdir(parents=True, exist_ok=True) if not CONFIG_FILE.exists(): CONFIG_FILE.write_text(json.dumps(DEFAULT_SERVICE_CONFIG, indent=2) + "\n") if not BUTTON_ACTIONS_FILE.exists(): BUTTON_ACTIONS_FILE.write_text(json.dumps(DEFAULT_BUTTON_ACTIONS, indent=2) + "\n") def load_service_config(): # STT endpoint is product-owned as of 0.8.8. ensure_service_files() def load_remote_map(): try: doc = json.loads(REMOTE_MAP_FILE.read_text()) except Exception as e: log(f"Remote map load failed: {e}") return {}, DEFAULT_BUTTON_ACTIONS.copy() by_event = {} for name, spec in doc.get("buttons", {}).items(): for ev in spec.get("events", []): try: by_event[(int(ev["handle"],16), bytes.fromhex(ev["data"]))] = name except Exception: pass try: actions = json.loads(BUTTON_ACTIONS_FILE.read_text()) except Exception: actions = DEFAULT_BUTTON_ACTIONS.copy() # Migrate defaults added after first install. Existing non-empty custom # assignments always win. changed = False for key, value in DEFAULT_BUTTON_ACTIONS.items(): if key not in actions: actions[key] = value changed = True # These two mappings were explicitly promoted to product defaults. if not actions.get("apps"): actions["apps"] = "ContextMenu" changed = True # Promote only the previously unassigned Netflix button; preserve user overrides. if actions.get("netflix") is None: actions["netflix"] = DEFAULT_BUTTON_ACTIONS["netflix"] changed = True if not actions.get("xiaomi_tv"): actions["xiaomi_tv"] = "Info" changed = True if changed: try: BUTTON_ACTIONS_FILE.write_text(json.dumps(actions, indent=2) + "\n") except Exception: pass return by_event, actions def kodi_action(action): if not action: return rc,out = run(["kodi-send","--action="+str(action)], timeout=3) if rc != 0: log(f"Kodi action failed ({action}): {out.strip()}") def active_video_player(): """Return True while Kodi has an active video player.""" try: response = kodi_rpc("Player.GetActivePlayers") result = response.get("result") or [] return any(p.get("type") == "video" for p in result) except Exception as e: log(f"Player state query failed: {e}") return False def kodi_current_window(): """Return Kodi's top/current window id and label. Window.IsActive(fullscreenvideo) remains true underneath Python dialogs (such as AIOHelper NextUp), so it is not sufficient for input routing. GUI.GetProperties(currentwindow) tells us which window actually owns focus. """ try: response = kodi_rpc( "GUI.GetProperties", {"properties": ["currentwindow", "currentcontrol"]} ) result = response.get("result") or {} win = result.get("currentwindow") or {} ctl = result.get("currentcontrol") or {} return { "id": int(win.get("id", -1)), "label": str(win.get("label", "")), "control": str(ctl.get("label", "")), } except Exception as e: log(f"Current window query failed: {e}") return {"id": -1, "label": "", "control": ""} def action_for_button(name, actions): # Kodi WINDOW_FULLSCREEN_VIDEO is 12005. Playback-oriented mappings are # valid only when that window is the actual top/current window. # # Python dialogs (13000-13099), VideoOSD (12901), AIOHelper NextUp, source # selectors, subtitle dialogs, etc. therefore receive ordinary navigation. win = kodi_current_window() if win["id"] == 12005 and name in FULLSCREEN_VIDEO_ACTIONS: action = FULLSCREEN_VIDEO_ACTIONS[name] log( f"REMOTE ROUTE fullscreen id={win['id']} " f"button={name} action={action}" ) return action action = actions.get(name) if name in ("up", "down", "left", "right", "ok", "back"): log( f"REMOTE ROUTE ui id={win['id']} label={win['label']!r} " f"control={win['control']!r} button={name} action={action}" ) return action def voice_command_info(command, query, transcript="", category="movie"): """Dispatch the interpreted voice command through Bridge 0.4.x contract.""" command = str(command or "").strip().lower() query = " ".join(str(query or "").strip().split()) transcript = str(transcript or "").strip() category = str(category or "movie").strip().lower() if category in ("show", "shows", "series", "television", "tvshow", "tv show"): category = "tv" elif category in ("film", "films", "movies"): category = "movie" if category not in ("movie", "tv"): category = "movie" if command not in ("play", "info", "search") or not query: return False log( "VOICE BRIDGE DISPATCH " f"command={command!r} category={category!r} " f"title={query!r} transcript={transcript!r}" ) call = ( "RunScript(script.voicehelper.bridge," "command=" + command + "," "category=" + category + "," "title=" + urllib.parse.quote_plus(query) + ")" ) try: rc, out = run(["kodi-send", "--action=" + call], timeout=5) log(f"VOICE BRIDGE rc={rc} output={(out or '').strip()!r}") return rc == 0 except Exception as exc: log("VOICE BRIDGE FAILED: " + f"{type(exc).__name__}: {exc}") return False _netflix_last_press = 0.0 def dispatch_remote_button(name, actions): global _netflix_last_press if name == "netflix": now = time.monotonic() if now - _netflix_last_press < 1.2: return _netflix_last_press = now action = action_for_button(name, actions) if action: kodi_action(action) elif name != "voice": log(f"Button '{name}' is unassigned") def publish_voice_command(cmdobj, transcript): SERVICE_DATA.mkdir(parents=True, exist_ok=True) doc = { "time": time.time(), "transcript": transcript, "command": cmdobj.get("command"), "query": cmdobj.get("query",""), } LAST_COMMAND_FILE.write_text(json.dumps(doc, indent=2) + "\n") log(f"Voice command: {doc['command']} {doc['query']}".rstrip()) command = (doc.get("command") or "").lower() query = (doc.get("query") or "").strip() if command == "play" and query: # VoiceHelper Bridge 0.2 owns title -> first TMDB result -> tmdb_id. action = ( "RunScript(script.voicehelper.bridge," "action=play," "type=movie," "query=" + urllib.parse.quote_plus(query) + ")" ) rc, out = run(["kodi-send", "--action=" + action], timeout=5) if rc != 0: log(f"PLAY dispatch failed: {out.strip()}") else: log(f"PLAY dispatched: {query}") set_ui_state("resolving", query, command="play") elif command == "search": # Parsed and published now; SEARCH execution is intentionally next. log(f"SEARCH pending implementation: {query}") def set_ui_state(state, text="", **extra): SERVICE_DATA.mkdir(parents=True, exist_ok=True) doc = { "time": time.time(), "state": state, "text": text, } doc.update(extra) tmp = UI_STATE_FILE.with_suffix(".tmp") tmp.write_text(json.dumps(doc, separators=(",", ":"))) tmp.replace(UI_STATE_FILE) # --------------------------------------------------------------------------- # Tiny RFC6455 WebSocket client (stdlib only) # --------------------------------------------------------------------------- # --------------------------------------------------------------------------- # Process-wide STT transport registry. # # Kodi's service supervisor terminates daemon.py during a Kodi restart. The # STT socket normally lives inside map_ab5e_and_capture(), so without a # process-level reference SIGTERM can bypass the normal function-exit cleanup # and leave the authenticated warm WebSocket alive on the server. # --------------------------------------------------------------------------- _ACTIVE_STT = None _ACTIVE_STT_LOCK = threading.Lock() _SHUTTING_DOWN = threading.Event() def register_active_stt(ws): global _ACTIVE_STT with _ACTIVE_STT_LOCK: _ACTIVE_STT = ws def unregister_active_stt(ws=None): global _ACTIVE_STT with _ACTIVE_STT_LOCK: if ws is None or _ACTIVE_STT is ws: _ACTIVE_STT = None def close_active_stt(reason="", immediate=False): """Retire the current STT WebSocket without touching Bluetooth. immediate=True is reserved for process shutdown: no WS close frame, no send lock, and no RX-thread join. """ global _ACTIVE_STT with _ACTIVE_STT_LOCK: ws = _ACTIVE_STT _ACTIVE_STT = None if ws is not None: try: if reason: log(reason) except Exception: pass try: if immediate: ws.abort_close() else: ws.close() except Exception as exc: try: log(f"STT shutdown close failed: {type(exc).__name__}: {exc}") except Exception: pass def _termination_handler(signum, frame): """Kodi restart/shutdown: retire STT before daemon process disappears.""" if _SHUTTING_DOWN.is_set(): raise SystemExit(0) _SHUTTING_DOWN.set() try: close_active_stt( f"VoiceHelper received signal {signum}; abort-closing STT WebSocket.", immediate=True, ) finally: raise SystemExit(0) class SimpleWebSocket: def __init__(self, host, port, path="/ws", timeout=3.0, tls=False): self.host = host self.port = port self.path = path self.timeout = timeout self.sock = None self.tls = tls # v0.9.3: a WebSocket connection must be read continuously even while # VoiceHelper is idle. Uvicorn/websockets can send server-side PING # control frames; if nobody reads them, this tiny stdlib client never # emits the required PONG and the server eventually closes the socket. # # The RX thread owns all socket reads and queues only application # text/binary frames for recv_message()/recv_json(). self._rx_queue = queue.Queue() self._rx_thread = None self._rx_stop = threading.Event() self._send_lock = threading.Lock() def connect(self): sock = socket.create_connection((self.host, self.port), timeout=self.timeout) if self.tls: ctx = ssl.create_default_context() sock = ctx.wrap_socket(sock, server_hostname=self.host) sock.settimeout(self.timeout) key = base64.b64encode(os.urandom(16)).decode("ascii") req = ( f"GET {self.path} HTTP/1.1\r\n" f"Host: {self.host}:{self.port}\r\n" "Upgrade: websocket\r\n" "Connection: Upgrade\r\n" f"Sec-WebSocket-Key: {key}\r\n" "Sec-WebSocket-Version: 13\r\n" "\r\n" ) sock.sendall(req.encode("ascii")) buf = bytearray() while b"\r\n\r\n" not in buf: chunk = sock.recv(4096) if not chunk: raise ConnectionError("STT WebSocket closed during handshake") buf.extend(chunk) if len(buf) > 16384: raise ConnectionError("STT WebSocket handshake response too large") header = bytes(buf).split(b"\r\n\r\n", 1)[0].decode("latin1", "replace") status = header.split("\r\n", 1)[0] if " 101 " not in status: raise ConnectionError("STT WebSocket handshake failed: " + status) accept = None for line in header.split("\r\n")[1:]: if ":" in line: k, v = line.split(":", 1) if k.strip().lower() == "sec-websocket-accept": accept = v.strip() break expected = base64.b64encode( hashlib.sha1( (key + "258EAFA5-E914-47DA-95CA-C5AB0DC85B11").encode("ascii") ).digest() ).decode("ascii") if accept != expected: raise ConnectionError("STT WebSocket Sec-WebSocket-Accept mismatch") self.sock = sock self._rx_stop.clear() self._rx_thread = threading.Thread( target=self._rx_loop, name="VoiceHelper-STT-RX", daemon=True, ) self._rx_thread.start() def _send_frame(self, opcode, payload=b""): if self.sock is None: raise ConnectionError("STT WebSocket is not connected") payload = bytes(payload) first = 0x80 | (opcode & 0x0f) mask = os.urandom(4) n = len(payload) if n < 126: header = bytes([first, 0x80 | n]) elif n <= 0xffff: header = bytes([first, 0x80 | 126]) + struct.pack("!H", n) else: header = bytes([first, 0x80 | 127]) + struct.pack("!Q", n) masked = bytes(b ^ mask[i & 3] for i, b in enumerate(payload)) frame = header + mask + masked # Audio, heartbeat PINGs, CLOSE and RX-thread PONGs can originate from # different execution contexts. Keep complete WebSocket frames atomic. with self._send_lock: if self.sock is None: raise ConnectionError("STT WebSocket is not connected") self.sock.sendall(frame) def send_text(self, text): self._send_frame(0x1, text.encode("utf-8")) def send_json(self, obj): self.send_text(json.dumps(obj, separators=(",", ":"))) def send_binary(self, data): self._send_frame(0x2, data) def send_ping(self, payload=b"vh"): """Send an RFC6455 ping. The peer/ASGI stack answers with pong.""" self._send_frame(0x9, payload) def _recv_exact(self, n): out = bytearray() while len(out) < n: if self.sock is None: raise ConnectionError("STT WebSocket is not connected") chunk = self.sock.recv(n - len(out)) if not chunk: raise ConnectionError("STT WebSocket closed") out.extend(chunk) return bytes(out) def _recv_frame(self): """Read exactly one RFC6455 frame from the socket.""" h = self._recv_exact(2) fin = bool(h[0] & 0x80) opcode = h[0] & 0x0f masked = bool(h[1] & 0x80) n = h[1] & 0x7f if n == 126: n = struct.unpack("!H", self._recv_exact(2))[0] elif n == 127: n = struct.unpack("!Q", self._recv_exact(8))[0] mask = self._recv_exact(4) if masked else None payload = self._recv_exact(n) if masked: payload = bytes(b ^ mask[i & 3] for i, b in enumerate(payload)) if not fin and opcode in (0x1, 0x2): raise ConnectionError("Fragmented STT WebSocket messages are not supported") return opcode, payload def _rx_loop(self): """Continuously service the WebSocket, including control frames.""" try: # The RX thread blocks here by design. It must always be present to # answer server PINGs, even when the microphone has been idle for # minutes or hours. if self.sock is not None: self.sock.settimeout(None) while not self._rx_stop.is_set(): opcode, payload = self._recv_frame() if opcode == 0x8: # CLOSE raise ConnectionError("STT WebSocket sent CLOSE") if opcode == 0x9: # PING from server/proxy self._send_frame(0xA, payload) continue if opcode == 0xA: # PONG for our heartbeat continue if opcode in (0x1, 0x2): self._rx_queue.put(("message", opcode, payload)) except Exception as e: if not self._rx_stop.is_set(): self._rx_queue.put(("error", e)) def recv_message(self, timeout=None): """Return the next application text/binary frame queued by RX thread.""" if self.sock is None: raise ConnectionError("STT WebSocket is not connected") try: if timeout is None: item = self._rx_queue.get() else: item = self._rx_queue.get(timeout=timeout) except queue.Empty: raise socket.timeout("timed out waiting for STT WebSocket message") if not item: raise ConnectionError("STT WebSocket receive queue failed") if item[0] == "error": raise item[1] _, opcode, payload = item return opcode, payload def recv_json(self, timeout=None): opcode, payload = self.recv_message(timeout) if opcode != 0x1: raise ValueError("Expected text JSON from STT server") return json.loads(payload.decode("utf-8")) def close(self): self._rx_stop.set() sock = self.sock if sock is not None: try: self._send_frame(0x8, b"") except Exception: pass try: sock.shutdown(socket.SHUT_RDWR) except Exception: pass try: sock.close() except Exception: pass self.sock = None # Do not join ourselves if close() is ever invoked from the RX thread. t = self._rx_thread if t is not None and t.is_alive() and threading.current_thread() is not t: try: t.join(timeout=0.5) except Exception: pass self._rx_thread = None def abort_close(self): """Immediate process-shutdown close. Never sends a WebSocket CLOSE frame, never acquires _send_lock, and never joins the RX thread. This path is specifically for SIGTERM / Kodi shutdown where blocking cleanup can prevent Kodi from exiting. """ self._rx_stop.set() # Detach first so no later heartbeat/audio path can reuse the socket. sock = self.sock self.sock = None self._rx_thread = None if sock is not None: try: sock.shutdown(socket.SHUT_RDWR) except Exception: pass try: sock.close() except Exception: pass def kodiflix_device_auth(): """Read the existing KodiFlix device identity created by RemoteCache pairing.""" try: state = json.loads(REMOTECACHE_STATE_FILE.read_text()) return { "device_id": str(state.get("device_id") or "").strip(), "device_token": str(state.get("device_token") or "").strip(), } except Exception as e: log(f"KodiFlix auth unavailable: {e}") return {"device_id": "", "device_token": ""} def connect_stt(): auth = kodiflix_device_auth() device_id = auth["device_id"] token = auth["device_token"] if not device_id or not token: set_ui_state("error", "KodiFlix device is not paired") raise PermissionError("KodiFlix device identity is missing") ws = SimpleWebSocket(STT_HOST, STT_PORT, STT_PATH, timeout=4.0, tls=STT_TLS) ws.connect() # Only functional change from 0.8.9: # present the existing paired KodiFlix device_id with its token. ws.send_json({ "event": "auth", "device_id": device_id, "token": token, }) reply = ws.recv_json(timeout=30.0) if reply.get("event") != "authorized": ws.close() raise PermissionError("STT rejected KodiFlix device authorization") return ws def pcm_samples_to_s16le(samples): out = bytearray(len(samples) * 2) off = 0 for sample in samples: struct.pack_into(" 32767: return 32767 if v < -32768: return -32768 return v def ima_decode_nibble(code, predictor, index): step = IMA_STEP_TABLE[index] diff = step >> 3 if code & 1: diff += step >> 2 if code & 2: diff += step >> 1 if code & 4: diff += step predictor = predictor - diff if (code & 8) else predictor + diff predictor = clamp16(predictor) index += IMA_INDEX_TABLE[code & 0x0f] index = max(0, min(88, index)) return predictor, index def ima_decode_bytes(data, predictor, index): out = [] for b in data: predictor, index = ima_decode_nibble((b >> 4) & 0x0f, predictor, index) out.append(predictor) predictor, index = ima_decode_nibble(b & 0x0f, predictor, index) out.append(predictor) return out, predictor, index def write_pcm16_wav(path, samples, rate=16000): with wave.open(str(path), "wb") as wf: wf.setnchannels(1) wf.setsampwidth(2) wf.setframerate(rate) pcm = bytearray() for v in samples: pcm += int(v).to_bytes(2, "little", signed=True) wf.writeframes(bytes(pcm)) def initialize_hid_session(fd): """ Bring up the ordinary remote FIRST. Voice/ATVV is deliberately not touched here. Once this returns, the HID notification path is considered the primary service and must survive any later voice/STT failure. """ enabled = 0 for h in HID_CCCD_HANDLES: try: write_cccd(fd, h) enabled += 1 except Exception as exc: log(f"HID CCCD 0x{h:04x} enable failed: {type(exc).__name__}: {exc}") if enabled == 0: raise RuntimeError("Could not enable any Xiaomi HID notification CCCDs") button_events, button_actions = load_remote_map() log("") log("******************************************************") log(f"REMOTE READY: HID notifications enabled ({enabled}/{len(HID_CCCD_HANDLES)}).") log("Voice initialization will occur only after this point.") log("******************************************************") set_ui_state("idle") return button_events, button_actions def run_hid_only_loop(fd, button_events, button_actions, reason=""): """ Fail-open ordinary-remote loop. If optional ATVV initialization fails after HID is already ready, stay on the existing encrypted ATT connection and continue routing normal buttons. No BlueZ restore, controller reset, or fresh pairing is triggered merely because voice is unavailable. """ if reason: log(f"Voice disabled; retaining HID-only remote: {reason}") set_ui_state("error", "Voice temporarily unavailable") held_button = None held_handle = None keepalive_next = time.monotonic() + 0.25 # Primary Service declaration handle 0x0001 exists on the already proven # GATT database and provides an innocuous ATT read to keep the raw link # active without depending on the optional ATVV service. keepalive_handle = 0x0001 while True: now = time.monotonic() if now >= keepalive_next: # ATT Read Request. Ignore the response below; its purpose here is # simply to prevent the direct raw link from going idle. writefd(fd, b"\x0a" + struct.pack("= 3: h = pkt[1] | (pkt[2] << 8) payload = pkt[3:] if pkt[0] == 0x1d: writefd(fd, b"\x1e") if payload and all(b == 0 for b in payload): if held_button is not None and h == held_handle: log(f"HID release: {held_button} handle=0x{h:04x}") held_button = None held_handle = None continue if payload: name = button_events.get((h, bytes(payload))) if not name: continue # Voice is optional in this mode. Never let a voice-button # press disturb the working HID session. if name == "voice": set_ui_state("error", "Voice temporarily unavailable") continue # One action per physical press/release cycle. Repeated # non-zero reports while the button is latched are ignored. if held_button == name and held_handle == h: log(f"HID duplicate held report suppressed: {name} handle=0x{h:04x}") continue held_button = name held_handle = h log(f"HID press latched: {name} handle=0x{h:04x}") dispatch_remote_button(name, button_actions) else: # Expected response to keepalive read. if pkt[0] in (0x0b, 0x01): continue def map_ab5e_and_capture(fd, button_events, button_actions): start,end=0x0093,0x009b tx_handle=0x0095 audio_handle=0x0097 ctl_handle=0x009a # Confirm exact primary service record. svc = att_xact(fd, b"\x10" + struct.pack("= STT_HEARTBEAT_SECONDS ): try: ws.send_ping() last_ping = time.monotonic() except Exception as exc: retire( "STT idle heartbeat failed; socket retired until next mic press: " f"{type(exc).__name__}: {exc}" ) if ws is not None: unregister_active_stt(ws) try: ws.abort_close() except Exception: pass stt_events.put(("availability", False)) def start_stt_worker(): nonlocal stt_worker_thread if stt_worker_thread is not None and stt_worker_thread.is_alive(): return stt_worker_stop.clear() stt_worker_thread = threading.Thread( target=stt_worker_main, name="VoiceHelper-STT-Worker", daemon=True, ) stt_worker_thread.start() def abort_voice_session(reason="", notify_remote=True): """ Abort only the current ATVV voice transaction. Never tears down Bluetooth/HID and never exits the daemon. This is used when START_SEARCH has already arrived but STT cannot service it. """ nonlocal stt_started, stt_pcm_bytes, stt_final nonlocal mic_open_sent, audio_started, close_sent nonlocal audio_deadline, stream_id, current_codec, current_rate nonlocal adpcm_predictor, adpcm_index, keepalive_next, deadline nonlocal voice_release_seen, voice_release_pending_stt_end, native_htt_active if reason: log(reason) if notify_remote and mic_open_sent: try: sid = stream_id if audio_started else 0x00 att_write_cmd( fd, tx_handle, bytes([0x0d, sid]), "MIC_CLOSE_ABORT" ) log( f"Voice session abort sent to remote " f"(stream_id=0x{sid:02x})." ) except Exception as exc: # Recovery must never sacrifice ordinary remote control. log( "Voice abort command failed; HID loop retained: " f"{type(exc).__name__}: {exc}" ) # Reset every voice transaction latch. stt_started = False stt_pcm_bytes = 0 stt_final = None mic_open_sent = False audio_started = False close_sent = False audio_deadline = None stream_id = 0x00 voice_release_seen = False voice_release_pending_stt_end = False native_htt_active = False current_codec = 0x02 current_rate = 16000 adpcm_predictor = 0 adpcm_index = 0 # Resume the normal persistent ATT/HID lifecycle immediately. keepalive_next = time.monotonic() + 0.05 deadline = float("inf") # STT is the LAST startup stage. At this point raw LE, SMP/encryption, # ordinary HID notifications, and ATVV mapping are already complete. log("Starting persistent warm STT worker now.") start_stt_worker() log( "Persistent STT worker thread started=%s alive=%s" % ( stt_worker_thread is not None, bool(stt_worker_thread and stt_worker_thread.is_alive()), ) ) log("Voice layer initialized after HID; isolated STT worker starting.") log("VoiceHelper ready.") log("Press VOICE and speak a PLAY or SEARCH command.") deadline=float("inf") keepalive_next=time.monotonic()+0.25 audio_deadline=None stream_id=0x00 audio_packets=0 audio_bytes=0 ctl_packets=0 other_packets=0 mic_open_sent=False audio_started=False close_sent=False first_audio=bytearray() # HID hold/repeat state. Xiaomi sends a non-zero report on press and an # all-zero report on release; synthesize Kodi repeats between them. held_button = None held_handle = None # 1.0.27: native ATVV Hold-to-Talk owns VOICE press/release. In HTT mode # KEYCODE_ASSIST HID events are suppressed and AUDIO_STOP reason=0x02 is # the physical button-up/utterance boundary. voice_release_seen = False voice_release_pending_stt_end = False native_htt_active = False raw_audio = bytearray() audio_events = [] decoded_pcm_16k = [] adpcm_predictor = 0 adpcm_index = 0 current_codec = 0x02 current_rate = 16000 audio_frame_no = 0 try: while time.monotonic()= keepalive_next: try: writefd(fd, b"\x0a" + struct.pack("=3: h=pkt[1] | (pkt[2]<<8) payload=pkt[3:] if pkt[0]==0x1d: writefd(fd,b"\x1e") if h==ctl_handle: ctl_packets += 1 log(f"CTL 0x{h:04x}: {payload.hex(' ')}") if payload: cmd=payload[0] if cmd==0x0b: # CAPS_RESP caps_seen = True if len(payload) >= 3: caps_version = (payload[1] << 8) | payload[2] else: caps_version = None # v1.0 CAPS_RESP: # 0B version(2) codecs(1) interaction_model(1) ... if caps_version == 0x0100 and len(payload) >= 5: caps_interaction_model = payload[4] log( f"*** CAPS_RESP v1.0 codecs=0x{payload[3]:02x} " f"interaction_model=0x{caps_interaction_model:02x} " f"raw={payload.hex(' ')} ***" ) if caps_interaction_model == 0x03: log("*** NATIVE HOLD-TO-TALK NEGOTIATED ***") elif caps_interaction_model == 0x01: log("*** PRESS-TO-TALK NEGOTIATED; legacy release fallback retained ***") else: log("*** REMOTE SELECTED ON-REQUEST; legacy MIC_OPEN path retained ***") else: # 0.4e and other legacy layouts do not contain the # v1.0 interaction-model byte in this position. log( f"*** CAPS_RESP legacy/unknown version=" f"{('0x%04x' % caps_version) if caps_version is not None else 'unknown'} " f"raw={payload.hex(' ')}; On-request fallback active ***" ) elif cmd==0x08: # START_SEARCH (legacy On-request) if native_htt_active or audio_started: # Native HTT owns the current transaction. A late # START_SEARCH must never overwrite LISTENING or # inject a legacy MIC_OPEN into the native stream. log("*** LATE START_SEARCH IGNORED DURING NATIVE HTT ***") else: voice_request_t0 = time.monotonic() voice_release_seen = False voice_release_pending_stt_end = False first_audio_logged = False first_pcm_logged = False pre_stt_pcm.clear() log("Voice request received. +0.0ms") # Legacy fallback only. t_open = time.monotonic() att_write_cmd(fd, tx_handle, bytes.fromhex("0c 00"), "MIC_OPEN") mic_open_sent = True log(f"MIC_OPEN write completed +{(time.monotonic()-voice_request_t0)*1000.0:.1f}ms") set_ui_state("processing","Starting voice…") stt_commands.put(("start",None)) log(f"STT start queued +{(time.monotonic()-voice_request_t0)*1000.0:.1f}ms") elif cmd==0x0a and len(payload)>=7: # AUDIO_SYNC sync_codec = payload[1] sync_frame = (payload[2] << 8) | payload[3] pred_u = (payload[4] << 8) | payload[5] sync_pred = pred_u - 0x10000 if pred_u & 0x8000 else pred_u sync_index = payload[6] current_codec = sync_codec current_rate = 16000 if sync_codec == 0x02 else 8000 adpcm_predictor = sync_pred adpcm_index = max(0, min(88, sync_index)) audio_frame_no = sync_frame audio_events.append({ "type":"AUDIO_SYNC", "codec":sync_codec, "frame_no":sync_frame, "predictor":sync_pred, "step_index":sync_index, "sample_rate":current_rate, }) log(f"*** AUDIO_SYNC codec=0x{sync_codec:02x} frame={sync_frame} pred={sync_pred} index={sync_index} rate={current_rate} ***") elif cmd==0x04 and len(payload)>=4: # AUDIO_START reason,codec,stream_id=payload[1],payload[2],payload[3] # Native HTT/PTT begins at AUDIO_START and may arrive # without START_SEARCH/HID. Initialize the voice/STT # transaction here when the remote owns mic activation. if reason in (0x01, 0x03) and not audio_started: if reason == 0x03: native_htt_active = True voice_request_t0 = time.monotonic() voice_release_seen = False voice_release_pending_stt_end = False first_audio_logged = False first_pcm_logged = False pre_stt_pcm.clear() log( f"Voice request received from native " f"{'HTT' if reason == 0x03 else 'PTT'} AUDIO_START. +0.0ms" ) if not stt_started: stt_commands.put(("start", None)) log("STT start queued from native AUDIO_START +0.0ms") audio_started=True audio_deadline=None current_codec = codec current_rate = 16000 if codec == 0x02 else 8000 adpcm_predictor = 0 adpcm_index = 0 audio_frame_no = 0 audio_events.append({ "type":"AUDIO_START", "reason":reason, "codec":codec, "stream_id":stream_id, "sample_rate":current_rate, }) dt = ((time.monotonic() - voice_request_t0) * 1000.0) if voice_request_t0 else 0.0 log(f"*** AUDIO_START reason=0x{reason:02x} codec=0x{codec:02x} stream_id=0x{stream_id:02x} +{dt:.1f}ms ***") if reason == 0x03: log("*** HTT AUDIO ACTIVE: release must arrive as AUDIO_STOP reason=0x02 ***") set_ui_state("listening", "Listening…") elif cmd==0x00 and len(payload)>=2: # AUDIO_STOP stop_reason = payload[1] log(f"*** AUDIO_STOP reason=0x{stop_reason:02x} ***") # Native HTT button release. The remote has already # stopped the microphone; this is the authoritative # utterance boundary and does not require MIC_CLOSE. if stop_reason == 0x02 and not voice_release_seen: native_htt_active = False voice_release_seen = True audio_deadline = None audio_started = False dt = ((time.monotonic() - voice_request_t0) * 1000.0) if voice_request_t0 else 0.0 log(f"*** VOICE BUTTON UP NATIVE HTT AUDIO_STOP 0x02 +{dt:.1f}ms ***") log("Processing speech...") set_ui_state("processing", "Processing…") if stt_started: stt_commands.put(("end", None)) log( f"*** STT END QUEUED ON HTT BUTTON UP " f"({stt_pcm_bytes} PCM bytes queued) ***" ) stt_started = False else: voice_release_pending_stt_end = True log("*** STT END DEFERRED UNTIL READY (HTT BUTTON UP) ***") elif stop_reason == 0x08: native_htt_active = False # Remote safety timeout. Finalize any still-active # session so the next voice press starts cleanly. audio_started = False if not voice_release_seen: voice_release_seen = True log("*** AUDIO TRANSFER TIMEOUT; FINALIZING STT ***") set_ui_state("processing", "Processing…") if stt_started: stt_commands.put(("end", None)) stt_started = False else: voice_release_pending_stt_end = True elif stop_reason in (0x00, 0x04, 0x10, 0x80): audio_started = False # 0x00 commonly acknowledges a host MIC_CLOSE. # Other stop reasons terminate transport but are # not interpreted as a physical HTT button-up. log( f"ATVV audio transport stopped reason=0x{stop_reason:02x}; " "physical HTT release not inferred." ) elif cmd==0x0c and len(payload)>=3: err=(payload[1]<<8)|payload[2] log(f"*** MIC_OPEN_ERROR 0x{err:04x} ***") break elif h==audio_handle: # Ignore any transport tail arriving after physical button-up. # MIC_CLOSE has already been sent and the utterance boundary is fixed. if voice_release_seen: continue audio_packets += 1 audio_bytes += len(payload) raw_audio.extend(payload) pcm, adpcm_predictor, adpcm_index = ima_decode_bytes( payload, adpcm_predictor, adpcm_index ) if current_rate == 8000: pcm16 = [] for sample in pcm: decoded_pcm_16k.append(sample) decoded_pcm_16k.append(sample) pcm16.append(sample) pcm16.append(sample) else: decoded_pcm_16k.extend(pcm) pcm16 = pcm # Queue decoded PCM only. The HID thread never writes # to the WebSocket or waits on network backpressure. chunk = pcm_samples_to_s16le(pcm16) if not first_audio_logged: dt = ((time.monotonic() - voice_request_t0) * 1000.0) if voice_request_t0 else 0.0 log(f"FIRST AUDIO PACKET +{dt:.1f}ms") first_audio_logged = True if stt_started: stt_commands.put(("pcm", chunk)) stt_pcm_bytes += len(chunk) if not first_pcm_logged: dt = ((time.monotonic() - voice_request_t0) * 1000.0) if voice_request_t0 else 0.0 log(f"FIRST PCM QUEUED TO STT +{dt:.1f}ms") first_pcm_logged = True else: # Preserve the leading edge if ATVV beats STT READY. Cap at # 8 seconds of mono 16-bit/16 kHz PCM to avoid unbounded growth. if len(pre_stt_pcm) < 256000: room = 256000 - len(pre_stt_pcm) pre_stt_pcm.extend(chunk[:room]) audio_events.append({ "type":"AUDIO", "packet":audio_packets, "frame_no":audio_frame_no, "bytes":len(payload), "codec":current_codec, "sample_rate":current_rate, "predictor_after":adpcm_predictor, "step_index_after":adpcm_index, }) audio_frame_no = (audio_frame_no + 1) & 0xffff if len(first_audio)<256: take=min(256-len(first_audio),len(payload)) first_audio.extend(payload[:take]) if audio_packets <= 12 or audio_packets % 25 == 0: pass else: if payload and all(b == 0 for b in payload): # Release report for whichever HID button is currently held. if held_button is not None and h == held_handle: if held_button == "voice": # Native HTT suppresses KEYCODE_ASSIST HID events. # If a legacy HID release is seen, do not use it # as the voice boundary; AUDIO_STOP owns HTT release. log("*** LEGACY VOICE HID RELEASE OBSERVED; awaiting ATVV AUDIO_STOP ***") else: log(f"HID release: {held_button} handle=0x{h:04x}") held_button = None held_handle = None elif payload: name = button_events.get((h, bytes(payload))) if name: # Native HTT does not use KEYCODE_ASSIST HID. If a # legacy VOICE HID report appears, ATVV still owns the # voice session and release boundary. if name != "voice": # One action per physical press/release cycle. # Ignore duplicate reports until release re-arms. if held_button == name and held_handle == h: log(f"HID duplicate held report suppressed: {name} handle=0x{h:04x}") continue held_button = name held_handle = h log(f"HID press latched: {name} handle=0x{h:04x}") dispatch_remote_button(name, button_actions) else: other_packets += 1 else: # Expected response to our keepalive read. if pkt[0] == 0x0b: pass elif pkt[0] == 0x01: log("ATT ERROR: "+pkt.hex(" ")) else: other_packets += 1 pass except (ConnectionResetError, BrokenPipeError, OSError) as e: log("") log(f"*** ATT LINK CLOSED DURING VOICE TEST: {type(e).__name__}: {e} ***") try: rc,out=run(["hcitool","con"]) log("Connection state after closure:") log(out) except Exception as ce: log(f"Could not query hcitool con: {ce}") dump_btmon_tail() stt_worker_stop.set() try: stt_commands.put_nowait(("shutdown", None)) except Exception: pass if not stt_final: log("No final STT result returned.") set_ui_state("error", "I didn't catch that") def kernel_hci_ready(timeout=20.0): """Wait for kernel hci0 only; BlueZ is deliberately irrelevant.""" end=time.monotonic()+timeout while time.monotonic()