From e6ef7f8a1f51c690a7d47a8347311868444b3bc8 Mon Sep 17 00:00:00 2001 From: hjoungjoo <149982795+hjoungjoo@users.noreply.github.com> Date: Sat, 27 Jun 2026 14:18:41 +0900 Subject: [PATCH] Add Bluetooth keyboard support --- pifinder_setup.sh | 9 +- python/PiFinder/keyboard_interface.py | 13 + python/PiFinder/keyboard_pi.py | 293 ++++++++++++++- python/PiFinder/main.py | 34 +- python/PiFinder/sys_utils.py | 271 ++++++++++++++ python/PiFinder/sys_utils_fake.py | 47 +++ python/PiFinder/ui/base.py | 3 + python/PiFinder/ui/bluetooth_keyboard.py | 451 +++++++++++++++++++++++ python/PiFinder/ui/menu_manager.py | 9 + python/PiFinder/ui/menu_structure.py | 6 + python/PiFinder/ui/textentry.py | 8 + python/tests/test_keyboard_interface.py | 13 + python/tests/test_sys_utils.py | 55 +++ 13 files changed, 1194 insertions(+), 18 deletions(-) create mode 100644 python/PiFinder/ui/bluetooth_keyboard.py create mode 100644 python/tests/test_keyboard_interface.py diff --git a/pifinder_setup.sh b/pifinder_setup.sh index e8ac1534..c8245cf5 100755 --- a/pifinder_setup.sh +++ b/pifinder_setup.sh @@ -45,6 +45,14 @@ sudo systemctl unmask hostapd # open permissisons on wpa_supplicant file so we can adjust network config sudo chmod 666 /etc/wpa_supplicant/wpa_supplicant.conf +# Bluetooth HID keyboards +if [[ -f /etc/bluetooth/input.conf ]]; then + sudo sed -i \ + -e 's/^#\?UserspaceHID=.*/UserspaceHID=true/' \ + -e 's/^#\?LEAutoSecurity=.*/LEAutoSecurity=true/' \ + /etc/bluetooth/input.conf +fi + # Samba config sudo cp ~/PiFinder/pi_config_files/smb.conf /etc/samba/smb.conf @@ -92,4 +100,3 @@ sudo systemctl enable pifinder sudo systemctl enable pifinder_splash echo "PiFinder setup complete, please restart the Pi" - diff --git a/python/PiFinder/keyboard_interface.py b/python/PiFinder/keyboard_interface.py index bf0fa9d2..456fdde8 100644 --- a/python/PiFinder/keyboard_interface.py +++ b/python/PiFinder/keyboard_interface.py @@ -7,6 +7,7 @@ class KeyboardInterface: + TEXT_BASE = 1000 NA = 10 PLUS = 11 MINUS = 12 @@ -30,6 +31,18 @@ class KeyboardInterface: LNG_SQUARE = 204 POWER_BTN = 300 + @classmethod + def text_key(cls, char: str) -> int: + return cls.TEXT_BASE + ord(char) + + @classmethod + def is_text_key(cls, keycode: int) -> bool: + return cls.TEXT_BASE <= keycode < cls.TEXT_BASE + 256 + + @classmethod + def text_from_keycode(cls, keycode: int) -> str: + return chr(keycode - cls.TEXT_BASE) + def __init__(self, q=None): self.q = q diff --git a/python/PiFinder/keyboard_pi.py b/python/PiFinder/keyboard_pi.py index 44702fdc..7ba22faf 100644 --- a/python/PiFinder/keyboard_pi.py +++ b/python/PiFinder/keyboard_pi.py @@ -6,7 +6,7 @@ """ -from time import sleep, time +from time import monotonic, sleep, time import libinput from PiFinder.keyboard_interface import KeyboardInterface import RPi.GPIO as GPIO @@ -16,6 +16,80 @@ logger = logging.getLogger("Keyboard.Pi") +KEY_ESC = 1 +KEY_1 = 2 +KEY_2 = 3 +KEY_3 = 4 +KEY_4 = 5 +KEY_5 = 6 +KEY_6 = 7 +KEY_7 = 8 +KEY_8 = 9 +KEY_9 = 10 +KEY_0 = 11 +KEY_MINUS = 12 +KEY_EQUAL = 13 +KEY_BACKSPACE = 14 +KEY_Q = 16 +KEY_W = 17 +KEY_E = 18 +KEY_R = 19 +KEY_T = 20 +KEY_Y = 21 +KEY_U = 22 +KEY_I = 23 +KEY_O = 24 +KEY_P = 25 +KEY_ENTER = 28 +KEY_LEFTCTRL = 29 +KEY_A = 30 +KEY_S = 31 +KEY_D = 32 +KEY_F = 33 +KEY_G = 34 +KEY_H = 35 +KEY_J = 36 +KEY_K = 37 +KEY_L = 38 +KEY_LEFTSHIFT = 42 +KEY_Z = 44 +KEY_X = 45 +KEY_C = 46 +KEY_V = 47 +KEY_B = 48 +KEY_N = 49 +KEY_M = 50 +KEY_RIGHTSHIFT = 54 +KEY_LEFTALT = 56 +KEY_SPACE = 57 +KEY_KP7 = 71 +KEY_KP8 = 72 +KEY_KP9 = 73 +KEY_KPMINUS = 74 +KEY_KP4 = 75 +KEY_KP5 = 76 +KEY_KP6 = 77 +KEY_KPPLUS = 78 +KEY_KP1 = 79 +KEY_KP2 = 80 +KEY_KP3 = 81 +KEY_KP0 = 82 +KEY_KPENTER = 96 +KEY_RIGHTCTRL = 97 +KEY_RIGHTALT = 100 +KEY_UP = 103 +KEY_LEFT = 105 +KEY_RIGHT = 106 +KEY_DOWN = 108 + +PHYSICAL_HOLD_SECONDS = 1.0 +PHYSICAL_REPEAT_SECONDS = 0.08 +PHYSICAL_CTRL_KEYS = {KEY_LEFTCTRL, KEY_RIGHTCTRL} +PHYSICAL_SHIFT_KEYS = {KEY_LEFTSHIFT, KEY_RIGHTSHIFT} +PHYSICAL_ALT_KEYS = {KEY_LEFTALT, KEY_RIGHTALT} +PHYSICAL_MODIFIER_KEYS = PHYSICAL_CTRL_KEYS | PHYSICAL_SHIFT_KEYS | PHYSICAL_ALT_KEYS + + class KeyboardPi(KeyboardInterface): def __init__(self, q): self.q = q @@ -68,36 +142,223 @@ def __init__(self, q): # physical keyboard support init self.li_kb = libinput.LibInput(context_type=libinput.ContextType.UDEV) self.li_kb.assign_seat("seat0") + self.physical_pressed = set() + self.physical_press_times: dict[int, float] = {} + self.physical_last_repeat_times: dict[int, float] = {} + self.physical_hold_sent: set[int] = set() + self.physical_press_modifiers: dict[int, set[int]] = {} + + self.text_physical_key_mapping: dict[int, int] = { + KEY_SPACE: self.text_key(" "), + KEY_A: self.text_key("a"), + KEY_B: self.text_key("b"), + KEY_C: self.text_key("c"), + KEY_D: self.text_key("d"), + KEY_E: self.text_key("e"), + KEY_F: self.text_key("f"), + KEY_G: self.text_key("g"), + KEY_H: self.text_key("h"), + KEY_I: self.text_key("i"), + KEY_J: self.text_key("j"), + KEY_K: self.text_key("k"), + KEY_L: self.text_key("l"), + KEY_M: self.text_key("m"), + KEY_N: self.text_key("n"), + KEY_O: self.text_key("o"), + KEY_P: self.text_key("p"), + KEY_Q: self.text_key("q"), + KEY_R: self.text_key("r"), + KEY_S: self.text_key("s"), + KEY_T: self.text_key("t"), + KEY_U: self.text_key("u"), + KEY_V: self.text_key("v"), + KEY_W: self.text_key("w"), + KEY_X: self.text_key("x"), + KEY_Y: self.text_key("y"), + KEY_Z: self.text_key("z"), + } + self.shift_text_physical_key_mapping: dict[int, int] = { + key: self.text_key(chr(value - self.TEXT_BASE).upper()) + for key, value in self.text_physical_key_mapping.items() + } + self.physical_key_mapping: dict[int, int] = { + KEY_UP: self.UP, + KEY_DOWN: self.DOWN, + KEY_LEFT: self.LEFT, + KEY_RIGHT: self.RIGHT, + KEY_ENTER: self.SQUARE, + KEY_KPENTER: self.SQUARE, + KEY_ESC: self.LEFT, + KEY_BACKSPACE: self.MINUS, + KEY_EQUAL: self.PLUS, + KEY_KPPLUS: self.PLUS, + KEY_MINUS: self.MINUS, + KEY_KPMINUS: self.MINUS, + KEY_1: 1, + KEY_2: 2, + KEY_3: 3, + KEY_4: 4, + KEY_5: 5, + KEY_6: 6, + KEY_7: 7, + KEY_8: 8, + KEY_9: 9, + KEY_0: 0, + KEY_KP1: 1, + KEY_KP2: 2, + KEY_KP3: 3, + KEY_KP4: 4, + KEY_KP5: 5, + KEY_KP6: 6, + KEY_KP7: 7, + KEY_KP8: 8, + KEY_KP9: 9, + KEY_KP0: 0, + } + self.alt_physical_key_mapping: dict[int, int] = { + KEY_UP: self.ALT_UP, + KEY_DOWN: self.ALT_DOWN, + KEY_LEFT: self.ALT_LEFT, + KEY_RIGHT: self.ALT_RIGHT, + KEY_EQUAL: self.ALT_PLUS, + KEY_KPPLUS: self.ALT_PLUS, + KEY_MINUS: self.ALT_MINUS, + KEY_KPMINUS: self.ALT_MINUS, + KEY_0: self.ALT_0, + KEY_KP0: self.ALT_0, + KEY_ENTER: self.ALT_SQUARE, + KEY_KPENTER: self.ALT_SQUARE, + } + self.long_physical_key_mapping: dict[int, int] = { + KEY_UP: self.LNG_UP, + KEY_DOWN: self.LNG_DOWN, + KEY_LEFT: self.LNG_LEFT, + KEY_RIGHT: self.LNG_RIGHT, + KEY_ENTER: self.LNG_SQUARE, + KEY_KPENTER: self.LNG_SQUARE, + } + + def _remember_physical_press(self, key: int) -> None: + if key in self.physical_pressed: + return + + self.physical_press_times[key] = monotonic() + self.physical_last_repeat_times.pop(key, None) + self.physical_hold_sent.discard(key) + self.physical_press_modifiers[key] = set( + self.physical_pressed & PHYSICAL_MODIFIER_KEYS + ) + self.physical_pressed.add(key) + + def _forget_physical_press(self, key: int) -> None: + self.physical_pressed.discard(key) + self.physical_press_times.pop(key, None) + self.physical_last_repeat_times.pop(key, None) + self.physical_hold_sent.discard(key) + self.physical_press_modifiers.pop(key, None) + + def _physical_key_modifiers(self, key: int) -> set[int]: + return ( + set(self.physical_press_modifiers.get(key, set())) + | (self.physical_pressed & PHYSICAL_MODIFIER_KEYS) + ) + + def _get_physical_hold_key(self) -> int: + now = monotonic() + held_keys = sorted( + self.physical_pressed, + key=lambda key: self.physical_press_times.get(key, now), + ) + + for key in held_keys: + if key in PHYSICAL_MODIFIER_KEYS: + continue + + press_time = self.physical_press_times.get(key) + if press_time is None or now - press_time < PHYSICAL_HOLD_SECONDS: + continue + + modifiers = self._physical_key_modifiers(key) + if modifiers & PHYSICAL_ALT_KEYS: + continue + + if key in [KEY_UP, KEY_DOWN]: + if modifiers: + continue + + last_repeat = self.physical_last_repeat_times.get(key) + if last_repeat is None or now - last_repeat >= PHYSICAL_REPEAT_SECONDS: + self.physical_last_repeat_times[key] = now + return self.physical_key_mapping[key] + continue + + if key in self.physical_hold_sent: + continue + + mapped_key = self.long_physical_key_mapping.get(key) + if mapped_key is not None: + self.physical_hold_sent.add(key) + return mapped_key + + return 0 def get_keyboard_key(self) -> int: """ - Checks libinput keyboard, if keyrelesed - map to our keycode and return + Checks libinput keyboard and maps key events to PiFinder keycodes. Returns 0 for no key registered """ - key_mapping: dict[int, int] = { - 103: self.UP, - 108: self.DOWN, - 105: self.LEFT, - 106: self.RIGHT, - 28: self.SQUARE, - 78: self.MINUS, - 74: self.PLUS, - } - while True: while True: + if hold_key := self._get_physical_hold_key(): + return hold_key + self.li_kb._libinput.libinput_dispatch(self.li_kb._li) hevent = self.li_kb._libinput.libinput_get_event(self.li_kb._li) if not hevent: - return 0 + return self._get_physical_hold_key() type_ = self.li_kb._libinput.libinput_event_get_type(hevent) if type_.is_keyboard(): kbev = libinput.KeyboardEvent(hevent, self.li_kb._libinput) - if kbev.key_state == libinput.constant.KeyState.RELEASED: - return key_mapping.get(kbev.key, 0) + if kbev.key_state == libinput.constant.KeyState.PRESSED: + self._remember_physical_press(kbev.key) + continue + if kbev.key_state != libinput.constant.KeyState.RELEASED: + continue + + press_modifiers = self.physical_press_modifiers.get( + kbev.key, set() + ) + release_modifiers = ( + set(self.physical_pressed) | set(press_modifiers) + ) + was_hold_sent = kbev.key in self.physical_hold_sent + self._forget_physical_press(kbev.key) + + if kbev.key in PHYSICAL_MODIFIER_KEYS: + continue + if was_hold_sent: + continue + + alt_pressed = bool(PHYSICAL_ALT_KEYS & release_modifiers) + if alt_pressed: + mapped_key = self.alt_physical_key_mapping.get(kbev.key) + return mapped_key if mapped_key is not None else 0 + + ctrl_pressed = bool(PHYSICAL_CTRL_KEYS & release_modifiers) + shift_pressed = bool(PHYSICAL_SHIFT_KEYS & release_modifiers) + if ctrl_pressed or shift_pressed: + mapped_key = self.long_physical_key_mapping.get(kbev.key) + if mapped_key is not None: + return mapped_key + if ctrl_pressed: + return 0 + return self.shift_text_physical_key_mapping.get(kbev.key, 0) + + return self.physical_key_mapping.get( + kbev.key, self.text_physical_key_mapping.get(kbev.key, 0) + ) def run_keyboard(self, log_queue): """ diff --git a/python/PiFinder/main.py b/python/PiFinder/main.py index 8754ff10..dd3bf4c5 100644 --- a/python/PiFinder/main.py +++ b/python/PiFinder/main.py @@ -25,6 +25,7 @@ import logging import argparse import pickle +import threading from pathlib import Path from PIL import Image, ImageOps from multiprocessing import Process, Queue @@ -341,6 +342,32 @@ def _build_pygame_keymaps(): return key_map, ctrl_key_map +def start_bluetooth_keyboard_autoreconnect() -> None: + """ + Start non-blocking Bluetooth keyboard reconnection for paired devices. + """ + if hardware_platform != "Pi": + return + + try: + from PiFinder import sys_utils + + reconnect_thread = threading.Thread( + target=sys_utils.auto_reconnect_bluetooth_keyboards, + kwargs={ + "attempts": 12, + "delay_seconds": 5, + "connect_timeout": 10, + }, + name="BluetoothKeyboardReconnect", + daemon=True, + ) + reconnect_thread.start() + logger.info("Bluetooth keyboard auto-reconnect started") + except Exception as e: + logger.warning("Could not start Bluetooth keyboard auto-reconnect: %s", e) + + def main( log_helper: MultiprocLogging, script_name=None, @@ -617,6 +644,8 @@ def main( ) posserver_process.start() + start_bluetooth_keyboard_autoreconnect() + # Initialize Catalogs console.write(" Catalogs") logger.info(" Catalogs") @@ -856,7 +885,10 @@ def main( if keycode != keyboard_base.POWER_BTN: sound.request(sound_queue, Earcon.KEYPRESS) # ignore keystroke if we have been asleep - if keycode > 99: + if keyboard_base.is_text_key(keycode): + menu_manager.key_text(keyboard_base.text_from_keycode(keycode)) + + elif keycode > 99: # Long left is return to top if keycode == keyboard_base.LNG_LEFT: menu_manager.key_long_left() diff --git a/python/PiFinder/sys_utils.py b/python/PiFinder/sys_utils.py index a3094dc0..02426fd9 100644 --- a/python/PiFinder/sys_utils.py +++ b/python/PiFinder/sys_utils.py @@ -1,6 +1,8 @@ import glob import json import re +import subprocess +import time from typing import Dict, Any import pam @@ -16,6 +18,16 @@ logger = logging.getLogger("SysUtils") +BLUETOOTHCTL_COMMAND = "bluetoothctl" +BLUETOOTH_DEVICE_RE = re.compile( + r"(?:\[[^\]]+\]\s*)?Device\s+([0-9A-Fa-f:]{17})\s+(.+)" +) +BLUETOOTH_DEVICE_FIELD_RE = re.compile( + r"(?:\[[^\]]+\]\s*)?Device\s+([0-9A-Fa-f:]{17})\s+([^:]+):\s+(.+)" +) +BLUETOOTH_MAC_RE = re.compile(r"^[0-9A-Fa-f:]{17}$") +ANSI_ESCAPE_RE = re.compile(r"\x1b\[[0-9;?]*[A-Za-z]") + class Network: """ @@ -246,6 +258,265 @@ def go_wifi_cli(): return True +def _clean_bluetoothctl_output(output: str) -> str: + output = ANSI_ESCAPE_RE.sub("", output) + return output.replace("\r", "\n") + + +def _bluetoothctl(commands: list[str], timeout: int = 20) -> str: + script = "\n".join(commands + ["quit"]) + "\n" + result = subprocess.run( + [BLUETOOTHCTL_COMMAND], + input=script, + capture_output=True, + text=True, + timeout=timeout, + check=False, + ) + return _clean_bluetoothctl_output(result.stdout + result.stderr) + + +def _parse_bluetooth_bool(value: str) -> bool: + return value.strip().lower() == "yes" + + +def _new_bluetooth_device(address: str, name: str = "") -> dict[str, Any]: + return { + "address": address, + "name": name, + "paired": False, + "trusted": False, + "connected": False, + "blocked": False, + "icon": "", + } + + +def _has_bluetooth_name(device: dict[str, Any]) -> bool: + name = str(device.get("name", "")).strip() + return bool(name) and not BLUETOOTH_MAC_RE.match(name) + + +def _merge_bluetooth_device_name(device: dict[str, Any], name: str) -> None: + name = name.strip() + if not name or BLUETOOTH_MAC_RE.match(name): + return + if not _has_bluetooth_name(device): + device["name"] = name + + +def _parse_bluetooth_devices(output: str) -> dict[str, dict[str, Any]]: + devices: dict[str, dict[str, Any]] = {} + for line in _clean_bluetoothctl_output(output).splitlines(): + field_match = BLUETOOTH_DEVICE_FIELD_RE.search(line) + if field_match: + address = field_match.group(1).upper() + field = field_match.group(2).strip().lower() + value = field_match.group(3).strip() + device = devices.setdefault(address, _new_bluetooth_device(address)) + if field in ["name", "alias"]: + _merge_bluetooth_device_name(device, value) + continue + elif field in ["paired", "trusted", "connected", "blocked"]: + device[field] = _parse_bluetooth_bool(value) + continue + elif field == "icon": + device["icon"] = value + continue + elif field in ["rssi", "txpower", "uuids", "servicesresolved"]: + continue + + match = BLUETOOTH_DEVICE_RE.search(line) + if not match: + continue + address = match.group(1).upper() + name = match.group(2).strip() + device = devices.setdefault(address, _new_bluetooth_device(address)) + _merge_bluetooth_device_name(device, name) + return devices + + +def _parse_bluetooth_info(output: str) -> dict[str, Any]: + info: dict[str, Any] = {} + for raw_line in _clean_bluetoothctl_output(output).splitlines(): + line = raw_line.strip() + if ":" not in line: + continue + key, value = line.split(":", 1) + value = value.strip() + if key in ["Name", "Alias", "Icon"]: + info[key.lower()] = value + elif key in ["Paired", "Trusted", "Connected", "Blocked"]: + info[key.lower()] = _parse_bluetooth_bool(value) + return info + + +def is_bluetooth_keyboard(device: dict[str, Any]) -> bool: + """ + Best-effort keyboard detection for reconnect filtering. + """ + name = str(device.get("name", "")).lower() + icon = str(device.get("icon", "")).lower() + return "keyboard" in name or "keys" in name or icon == "input-keyboard" + + +def list_bluetooth_devices(scan_output: str = "") -> list[dict[str, Any]]: + """ + Return cached Bluetooth devices with paired/trusted/connected status. + """ + logger.info("SYS: Listing Bluetooth devices") + devices = _parse_bluetooth_devices(scan_output) + output = _bluetoothctl(["power on", "devices", "devices Paired"], timeout=12) + for address, scanned_device in _parse_bluetooth_devices(output).items(): + device = devices.setdefault(address, _new_bluetooth_device(address)) + _merge_bluetooth_device_name(device, str(scanned_device.get("name", ""))) + + for address, device in devices.items(): + info = _parse_bluetooth_info(_bluetoothctl([f"info {address}"], timeout=8)) + for key, value in info.items(): + if key in ["name", "alias"]: + _merge_bluetooth_device_name(device, str(value)) + else: + device[key] = value + device["address"] = address + device["name"] = device.get("name") or device.get("alias") or address + + return sorted( + devices.values(), + key=lambda item: ( + not bool(item.get("connected")), + not bool(item.get("paired")), + str(item.get("name", "")).lower(), + ), + ) + + +def scan_bluetooth_devices(scan_seconds: int = 12) -> list[dict[str, Any]]: + """ + Scan for nearby Bluetooth devices and return the refreshed cached device list. + """ + logger.info("SYS: Scanning for Bluetooth devices for %s seconds", scan_seconds) + process = subprocess.Popen( + [BLUETOOTHCTL_COMMAND], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + ) + assert process.stdin is not None + scan_output = "" + try: + for command in [ + "power on", + "agent KeyboardDisplay", + "default-agent", + "pairable on", + "scan on", + ]: + process.stdin.write(command + "\n") + process.stdin.flush() + + time.sleep(scan_seconds) + + for command in ["scan off", "devices", "devices Paired", "quit"]: + process.stdin.write(command + "\n") + process.stdin.flush() + scan_output, _ = process.communicate(timeout=8) + except subprocess.TimeoutExpired: + process.kill() + scan_output, _ = process.communicate() + finally: + if process.poll() is None: + process.terminate() + + return list_bluetooth_devices(scan_output) + + +def connect_bluetooth_device(address: str, timeout: int = 25) -> str: + logger.info("SYS: Connecting Bluetooth device %s", address) + return _bluetoothctl( + [ + "power on", + "agent KeyboardDisplay", + "default-agent", + f"trust {address}", + f"connect {address}", + ], + timeout=timeout, + ) + + +def disconnect_bluetooth_device(address: str) -> str: + logger.info("SYS: Disconnecting Bluetooth device %s", address) + return _bluetoothctl([f"disconnect {address}"], timeout=15) + + +def remove_bluetooth_device(address: str) -> str: + logger.info("SYS: Removing Bluetooth device %s", address) + return _bluetoothctl([f"remove {address}"], timeout=15) + + +def reconnect_bluetooth_keyboards(connect_timeout: int = 25) -> int: + """ + Connect paired keyboard-like devices. If none are identifiable as keyboards, + try all paired devices so generic HID names still work. + """ + devices = [ + d for d in list_bluetooth_devices() if d.get("paired") or d.get("trusted") + ] + targets = [d for d in devices if is_bluetooth_keyboard(d)] or devices + count = 0 + for device in targets: + if device.get("connected"): + continue + connect_bluetooth_device(str(device["address"]), timeout=connect_timeout) + count += 1 + return count + + +def auto_reconnect_bluetooth_keyboards( + attempts: int = 12, + delay_seconds: int = 5, + connect_timeout: int = 10, +) -> int: + """ + Retry Bluetooth keyboard reconnection in the background during startup. + + Bluetooth controllers and HID devices can appear a few seconds after the + PiFinder service starts, especially after a reboot. This helper is designed + for a daemon thread: it logs failures and keeps retrying without raising. + """ + total_attempted = 0 + for attempt in range(1, attempts + 1): + try: + logger.info( + "SYS: Bluetooth keyboard reconnect attempt %s/%s", + attempt, + attempts, + ) + total_attempted += reconnect_bluetooth_keyboards(connect_timeout) + devices = [ + d + for d in list_bluetooth_devices() + if d.get("paired") or d.get("trusted") + ] + targets = [d for d in devices if is_bluetooth_keyboard(d)] or devices + if targets and any(d.get("connected") for d in targets): + logger.info("SYS: Bluetooth keyboard reconnect complete") + return total_attempted + except Exception as e: + logger.warning("SYS: Bluetooth keyboard reconnect failed: %s", e) + + if attempt < attempts: + time.sleep(delay_seconds) + + logger.info( + "SYS: Bluetooth keyboard reconnect finished after %s attempts", + attempts, + ) + return total_attempted + + def remove_backup(): """ Removes backup file diff --git a/python/PiFinder/sys_utils_fake.py b/python/PiFinder/sys_utils_fake.py index 77ca7150..703e53bc 100644 --- a/python/PiFinder/sys_utils_fake.py +++ b/python/PiFinder/sys_utils_fake.py @@ -252,6 +252,53 @@ def go_wifi_cli(): return True +BLUETOOTHCTL_COMMAND = "bluetoothctl" + + +def is_bluetooth_keyboard(device): + name = str(device.get("name", "")).lower() + icon = str(device.get("icon", "")).lower() + return "keyboard" in name or "keys" in name or icon == "input-keyboard" + + +def list_bluetooth_devices(scan_output=""): + return [] + + +def scan_bluetooth_devices(scan_seconds=12): + logger.info("SYS: Fake Bluetooth scan for %s seconds", scan_seconds) + return [] + + +def connect_bluetooth_device(address, timeout=25): + logger.info("SYS: Fake Bluetooth connect %s", address) + return "Done" + + +def disconnect_bluetooth_device(address): + logger.info("SYS: Fake Bluetooth disconnect %s", address) + return "Done" + + +def remove_bluetooth_device(address): + logger.info("SYS: Fake Bluetooth remove %s", address) + return "Done" + + +def reconnect_bluetooth_keyboards(connect_timeout=25): + logger.info("SYS: Fake Bluetooth keyboard reconnect") + return 0 + + +def auto_reconnect_bluetooth_keyboards( + attempts=12, + delay_seconds=5, + connect_timeout=10, +): + logger.info("SYS: Fake Bluetooth keyboard auto-reconnect") + return 0 + + def verify_password(username, password): """ Checks the provided password against the provided user diff --git a/python/PiFinder/ui/base.py b/python/PiFinder/ui/base.py index 812c2bfa..5ee72a77 100644 --- a/python/PiFinder/ui/base.py +++ b/python/PiFinder/ui/base.py @@ -513,6 +513,9 @@ def cycle_display_mode(self): def key_number(self, number): pass + def key_text(self, char: str): + pass + def key_plus(self): pass diff --git a/python/PiFinder/ui/bluetooth_keyboard.py b/python/PiFinder/ui/bluetooth_keyboard.py new file mode 100644 index 00000000..843792a8 --- /dev/null +++ b/python/PiFinder/ui/bluetooth_keyboard.py @@ -0,0 +1,451 @@ +#!/usr/bin/python +# -*- coding:utf-8 -*- +""" +Bluetooth keyboard pairing and connection UI. +""" + +import fcntl +import os +import re +import select +import subprocess +import time +from typing import Any, TYPE_CHECKING + +from PiFinder import utils +from PiFinder.ui.text_menu import UITextMenu + +if TYPE_CHECKING: + + def _(a) -> Any: + return a + + +sys_utils = utils.get_sys_utils() + +ANSI_ESCAPE_RE = re.compile(r"\x1b\[[0-9;?]*[A-Za-z]") +PASSKEY_RE = re.compile(r"Passkey:\s*([0-9]{4,8})", re.IGNORECASE) +MAC_ADDRESS_RE = re.compile(r"^[0-9A-Fa-f:]{17}$") +DEVICE_NAME_MAX = 20 +SCAN_SECONDS = 12 +PAIR_TIMEOUT = 90 + + +class UIBluetoothKeyboard(UITextMenu): + """ + Small Bluetooth HID manager for pairing keyboards from the PiFinder UI. + USB keyboards need no pairing; they are handled by the normal libinput path. + """ + + __title__ = "Keyboard" + + def __init__(self, *args, **kwargs): + self.devices: list[dict[str, Any]] = [] + self.status = "" + self.action_menu_active = False + self.action_index = 0 + self.selected_device: dict[str, Any] | None = None + + self.pair_process: subprocess.Popen | None = None + self.pair_address = "" + self.pair_name = "" + self.pair_started = 0.0 + self.pair_status = "" + self.pair_output = "" + self.pair_done_at: float | None = None + self.pair_success = False + self.pair_connect_sent = False + self.pair_yes_sent = False + + self._refresh_devices() + kwargs["item_definition"] = self._create_menu_definition() + super().__init__(*args, **kwargs) + + def _refresh_devices(self): + try: + self.devices = sys_utils.list_bluetooth_devices() + self.status = "" + except Exception as e: + self.devices = [] + self.status = f"BT error: {e}" + + def _create_menu_definition(self): + items = [ + {"name": _("Scan / Pair"), "value": "__scan__"}, + {"name": _("Reconnect"), "value": "__reconnect__"}, + {"name": _("Refresh"), "value": "__refresh__"}, + ] + for device in self.devices: + items.append( + { + "name": self._device_label(device), + "value": device["address"], + "device": device, + } + ) + if not self.devices: + items.append({"name": _("No BT devices"), "value": None}) + return {"name": _("Keyboard"), "select": "single", "items": items} + + def _rebuild_menu(self): + self.item_definition = self._create_menu_definition() + self._menu_items = [x["name"] for x in self.item_definition["items"]] + if self._current_item_index >= len(self._menu_items): + self._current_item_index = max(0, len(self._menu_items) - 1) + + def _device_label(self, device: dict[str, Any]) -> str: + if device.get("connected"): + marker = "*" + elif device.get("paired"): + marker = "+" + else: + marker = "-" + name = self._device_name(device) + available = max(5, DEVICE_NAME_MAX - 2) + if len(name) > available: + name = name[: available - 3] + "..." + return f"{marker} {name}" + + def _device_name(self, device: dict[str, Any]) -> str: + name = str(device.get("name") or device.get("alias") or "").strip() + address = str(device.get("address") or "").strip() + if not name or MAC_ADDRESS_RE.match(name): + return f"Unknown {address[-5:]}" if address else "Unknown" + return name + + def _short_address(self, device: dict[str, Any]) -> str: + address = str(device.get("address") or "").strip() + if not address: + return "" + return f"MAC ...{address[-8:]}" + + def _selected_item(self) -> dict[str, Any]: + return self.item_definition["items"][self._current_item_index] + + def _run_scan(self): + self.message(_("BT scanning"), SCAN_SECONDS) + try: + self.devices = sys_utils.scan_bluetooth_devices(SCAN_SECONDS) + self.status = f"Found {len(self.devices)}" + except Exception as e: + self.status = f"Scan error: {e}" + self._rebuild_menu() + + def _run_reconnect(self): + self.message(_("Connecting"), 2) + try: + count = sys_utils.reconnect_bluetooth_keyboards() + self.status = f"Reconnect {count}" + except Exception as e: + self.status = f"Conn error: {e}" + self._refresh_devices() + self._rebuild_menu() + + def _open_action_menu(self, device: dict[str, Any]): + self.selected_device = device + self.action_menu_active = True + self.action_index = 0 + + def _action_items(self) -> list[tuple[str, str]]: + device = self.selected_device or {} + actions: list[tuple[str, str]] = [] + if not device.get("paired"): + actions.append((_("Pair+Connect"), "pair")) + else: + actions.append((_("Pair Again"), "pair")) + actions.append((_("Connect"), "connect")) + if device.get("connected"): + actions.append((_("Disconnect"), "disconnect")) + actions.append((_("Remove"), "remove")) + actions.append((_("Cancel"), "cancel")) + return actions + + def _perform_action(self): + if self.selected_device is None: + self.action_menu_active = False + return + + label, action = self._action_items()[self.action_index] + address = str(self.selected_device["address"]) + if action == "cancel": + self.action_menu_active = False + return + if action == "pair": + self._start_pairing(address, str(self.selected_device.get("name", ""))) + return + + self.message(label, 1) + try: + if action == "connect": + output = sys_utils.connect_bluetooth_device(address) + elif action == "disconnect": + output = sys_utils.disconnect_bluetooth_device(address) + elif action == "remove": + output = sys_utils.remove_bluetooth_device(address) + else: + output = "" + self.status = self._short_result(output) + except Exception as e: + self.status = f"BT error: {e}" + self.action_menu_active = False + self._refresh_devices() + self._rebuild_menu() + + def _short_result(self, output: str) -> str: + clean = self._clean_text(output) + for line in reversed([x.strip() for x in clean.splitlines() if x.strip()]): + if "successful" in line.lower() or "failed" in line.lower(): + return line[:24] + return "Done" + + def _start_pairing(self, address: str, name: str): + self._close_pair_process() + self.pair_address = address + self.pair_name = name or address + self.pair_started = time.time() + self.pair_status = "Starting" + self.pair_output = "" + self.pair_done_at = None + self.pair_success = False + self.pair_connect_sent = False + self.pair_yes_sent = False + + try: + self.pair_process = subprocess.Popen( + [sys_utils.BLUETOOTHCTL_COMMAND], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=False, + ) + assert self.pair_process.stdout is not None + flags = fcntl.fcntl(self.pair_process.stdout.fileno(), fcntl.F_GETFL) + fcntl.fcntl( + self.pair_process.stdout.fileno(), fcntl.F_SETFL, flags | os.O_NONBLOCK + ) + for command in [ + "power on", + "agent KeyboardDisplay", + "default-agent", + "pairable on", + f"pair {address}", + ]: + self._send_pair_command(command) + self.pair_status = "Pairing" + except Exception as e: + self.pair_status = f"Start failed: {e}" + self._finish_pairing(False) + + def _send_pair_command(self, command: str): + if self.pair_process is None or self.pair_process.stdin is None: + return + try: + self.pair_process.stdin.write((command + "\n").encode()) + self.pair_process.stdin.flush() + except BrokenPipeError: + pass + + def _clean_text(self, text: str) -> str: + return ANSI_ESCAPE_RE.sub("", text).replace("\r", "\n") + + def _read_pair_output(self): + if self.pair_process is None or self.pair_process.stdout is None: + return + fd = self.pair_process.stdout.fileno() + while True: + ready, _, _ = select.select([fd], [], [], 0) + if not ready: + break + try: + chunk = os.read(fd, 4096) + except BlockingIOError: + break + if not chunk: + break + text = chunk.decode(errors="replace") + self.pair_output = (self.pair_output + text)[-4000:] + self._handle_pair_text(self._clean_text(self.pair_output)) + + def _handle_pair_text(self, text: str): + passkey = PASSKEY_RE.search(text) + if passkey: + self.pair_status = f"Type {passkey.group(1)}" + + if ( + not self.pair_yes_sent + and ( + "Confirm passkey" in text + or "Authorize service" in text + or "Accept pairing" in text + ) + ): + self._send_pair_command("yes") + self.pair_yes_sent = True + + pair_ok = "Pairing successful" in text or "AlreadyExists" in text + if pair_ok and not self.pair_connect_sent: + self.pair_status = "Connecting" + self._send_pair_command(f"trust {self.pair_address}") + self._send_pair_command(f"connect {self.pair_address}") + self.pair_connect_sent = True + + if "Connection successful" in text: + self.pair_status = "Connected" + self._finish_pairing(True) + return + + failure_tokens = [ + "AuthenticationCanceled", + "AuthenticationFailed", + "Failed to pair", + "Failed to connect", + "not available", + "No default controller", + ] + if any(token in text for token in failure_tokens): + if pair_ok: + self.pair_status = "Paired" + else: + self.pair_status = "Pair failed" + self._finish_pairing(pair_ok) + + def _finish_pairing(self, success: bool): + if self.pair_done_at is not None: + return + self.pair_success = success + self.pair_done_at = time.time() + self._send_pair_command("quit") + + def _close_pair_process(self): + if self.pair_process is None: + return + try: + if self.pair_process.poll() is None: + self.pair_process.terminate() + self.pair_process.wait(timeout=1) + except Exception: + try: + self.pair_process.kill() + except Exception: + pass + self.pair_process = None + + def _update_pairing(self): + if self.pair_process is None: + return + self._read_pair_output() + if time.time() - self.pair_started > PAIR_TIMEOUT: + self.pair_status = "Pair timeout" + self._finish_pairing(False) + if self.pair_process.poll() is not None and self.pair_done_at is None: + self._finish_pairing(self.pair_success) + if self.pair_done_at and time.time() - self.pair_done_at > 2.5: + self._close_pair_process() + self.action_menu_active = False + self._refresh_devices() + self._rebuild_menu() + + def _draw_lines(self, lines: list[str], selected: int | None = None): + self.clear_screen() + draw_y = self.display_class.titlebar_height + 2 + max_chars = max(4, (self.display_class.resX - 4) // self.fonts.base.width) + for idx, line in enumerate(lines): + if draw_y > self.display_class.resY - self.fonts.base.height: + break + if len(line) > max_chars: + line = line[: max_chars - 3] + "..." + fill = self.colors.get(255 if selected == idx else 128) + font = self.fonts.bold.font if selected == idx else self.fonts.base.font + self.draw.text((2, draw_y), line, font=font, fill=fill) + draw_y += self.fonts.base.height + 2 + + def _draw_action_menu(self): + device = self.selected_device or {} + name = self._device_name(device) + state = [] + if device.get("connected"): + state.append("Conn") + if device.get("paired"): + state.append("Pair") + if device.get("trusted"): + state.append("Trust") + lines = [name, self._short_address(device), " ".join(state) or "New device"] + lines.extend(label for label, _ in self._action_items()) + self._draw_lines(lines, self.action_index + 3) + return self.screen_update() + + def _draw_pairing(self): + self._update_pairing() + lines = [ + "Pair Keyboard", + self.pair_name, + self.pair_status, + ] + if self.pair_done_at: + lines.append("OK" if self.pair_success else "Failed") + else: + lines.append("Left cancels") + self._draw_lines(lines) + return self.screen_update() + + def update(self, force=False): + if self.pair_process is not None: + return self._draw_pairing() + if self.action_menu_active: + return self._draw_action_menu() + result = super().update(force) + if self.status: + y = self.display_class.resY - self.fonts.base.height - 1 + max_chars = max(4, (self.display_class.resX - 4) // self.fonts.base.width) + status = self.status[:max_chars] + self.draw.rectangle( + [0, y - 1, self.display_class.resX, self.display_class.resY], + fill=self.colors.get(0), + ) + self.draw.text((2, y), status, font=self.fonts.base.font, fill=self.colors.get(192)) + return self.screen_update() + return result + + def key_up(self): + if self.action_menu_active: + self.action_index = (self.action_index - 1) % len(self._action_items()) + return + super().key_up() + + def key_down(self): + if self.action_menu_active: + self.action_index = (self.action_index + 1) % len(self._action_items()) + return + super().key_down() + + def key_right(self): + if self.pair_process is not None: + return + if self.action_menu_active: + return self._perform_action() + + selected_item = self._selected_item() + value = selected_item.get("value") + if value == "__scan__": + return self._run_scan() + if value == "__reconnect__": + return self._run_reconnect() + if value == "__refresh__": + self._refresh_devices() + self._rebuild_menu() + return + if selected_item.get("device"): + return self._open_action_menu(selected_item["device"]) + + def key_left(self) -> bool: + if self.pair_process is not None: + self._close_pair_process() + self.pair_status = "Canceled" + self.action_menu_active = False + self._refresh_devices() + self._rebuild_menu() + return False + if self.action_menu_active: + self.action_menu_active = False + return False + return True diff --git a/python/PiFinder/ui/menu_manager.py b/python/PiFinder/ui/menu_manager.py index f6b6c993..dc27ff62 100644 --- a/python/PiFinder/ui/menu_manager.py +++ b/python/PiFinder/ui/menu_manager.py @@ -352,6 +352,15 @@ def key_number(self, number): self.stack[-1].key_number(number) + def key_text(self, char: str): + if self.help_images is not None: + # Exit help + self.help_images = None + self.update() + return + + self.stack[-1].key_text(char) + def key_plus(self): self.stack[-1].key_plus() diff --git a/python/PiFinder/ui/menu_structure.py b/python/PiFinder/ui/menu_structure.py index 5a8b97f6..da9f2abf 100644 --- a/python/PiFinder/ui/menu_structure.py +++ b/python/PiFinder/ui/menu_structure.py @@ -19,6 +19,7 @@ from PiFinder.ui.locationentry import UILocationEntry from PiFinder.ui.radec_entry import UIRADecEntry from PiFinder.ui.telemetry_list import UITelemetryList +from PiFinder.ui.bluetooth_keyboard import UIBluetoothKeyboard import PiFinder.ui.callbacks as callbacks @@ -1149,6 +1150,11 @@ def _(key: str) -> Any: }, ], }, + { + "name": _("Keyboard"), + "class": UIBluetoothKeyboard, + "label": "keyboard_settings", + }, ], }, { diff --git a/python/PiFinder/ui/textentry.py b/python/PiFinder/ui/textentry.py index 284ec20d..bacfd940 100644 --- a/python/PiFinder/ui/textentry.py +++ b/python/PiFinder/ui/textentry.py @@ -363,6 +363,14 @@ def key_long_minus(self): self.current_text = "" self.update_search_results() + def key_text(self, char: str): + if len(char) != 1: + return + self.last_key = None + self.char_index = 0 + self.last_key_press_time = time.time() + self.add_char(char) + def key_number(self, number): current_time = time.time() number_key = str(number) diff --git a/python/tests/test_keyboard_interface.py b/python/tests/test_keyboard_interface.py new file mode 100644 index 00000000..c750916a --- /dev/null +++ b/python/tests/test_keyboard_interface.py @@ -0,0 +1,13 @@ +from PiFinder.keyboard_interface import KeyboardInterface + + +def test_text_key_round_trip(): + keycode = KeyboardInterface.text_key("A") + + assert KeyboardInterface.is_text_key(keycode) + assert KeyboardInterface.text_from_keycode(keycode) == "A" + + +def test_regular_keys_are_not_text_keys(): + assert not KeyboardInterface.is_text_key(KeyboardInterface.LEFT) + assert not KeyboardInterface.is_text_key(KeyboardInterface.POWER_BTN) diff --git a/python/tests/test_sys_utils.py b/python/tests/test_sys_utils.py index ce9b1f1b..03b1e06f 100644 --- a/python/tests/test_sys_utils.py +++ b/python/tests/test_sys_utils.py @@ -106,6 +106,61 @@ def test_rewrite_hosts_ignores_commented_line(): assert "# 127.0.1.1 oldname\n" in result assert result.endswith("127.0.1.1\tpf-rich\n") + @pytest.mark.unit + def test_parse_bluetooth_devices_merges_fields(): + output = """ + \x1b[0;94m[bluetooth]# Device AA:BB:CC:DD:EE:FF Keychron K2\r + Device AA:BB:CC:DD:EE:FF Paired: yes + Device AA:BB:CC:DD:EE:FF Connected: no + Device AA:BB:CC:DD:EE:FF Icon: input-keyboard + Device 11:22:33:44:55:66 11:22:33:44:55:66 + Device 11:22:33:44:55:66 Name: Travel Mouse + """ + + devices = sys_utils._parse_bluetooth_devices(output) + + keyboard = devices["AA:BB:CC:DD:EE:FF"] + assert keyboard["name"] == "Keychron K2" + assert keyboard["paired"] is True + assert keyboard["connected"] is False + assert keyboard["icon"] == "input-keyboard" + assert sys_utils.is_bluetooth_keyboard(keyboard) + + mouse = devices["11:22:33:44:55:66"] + assert mouse["name"] == "Travel Mouse" + assert not sys_utils.is_bluetooth_keyboard(mouse) + + @pytest.mark.unit + def test_list_bluetooth_devices_uses_info_status(monkeypatch): + def fake_bluetoothctl(commands, timeout=20): + if commands == ["power on", "devices", "devices Paired"]: + return "Device AA:BB:CC:DD:EE:FF AA:BB:CC:DD:EE:FF\n" + if commands == ["info AA:BB:CC:DD:EE:FF"]: + return """ + Name: Compact Keyboard + Paired: yes + Trusted: yes + Connected: yes + Icon: input-keyboard + """ + return "" + + monkeypatch.setattr(sys_utils, "_bluetoothctl", fake_bluetoothctl) + + devices = sys_utils.list_bluetooth_devices() + + assert devices == [ + { + "address": "AA:BB:CC:DD:EE:FF", + "name": "Compact Keyboard", + "paired": True, + "trusted": True, + "connected": True, + "blocked": False, + "icon": "input-keyboard", + } + ] + except ImportError: pass