diff --git a/lib/whisplay.py b/lib/whisplay.py new file mode 100644 index 0000000..5a041a0 --- /dev/null +++ b/lib/whisplay.py @@ -0,0 +1,661 @@ +import spidev +import time +import os +import threading +import gpiod + +# Detect gpiod API version: v1 has LINE_REQ_DIR_OUT; v2 has LineSettings +_GPIOD_V2 = hasattr(gpiod, 'LineSettings') + +if _GPIOD_V2: + from gpiod.line import Direction, Value, Bias + + +# ==================== Platform Detection ==================== +def _detect_platform(): + """Detect hardware platform type""" + try: + with open("/proc/device-tree/model", "r") as f: + model = f.read().strip('\0').strip() + if "Raspberry" in model: + return "rpi", model + elif "Radxa" in model: + return "radxa", model + except Exception: + pass + # Also check compatible string (some SoCs use non-descriptive model names) + try: + with open("/proc/device-tree/compatible", "r") as f: + compat = f.read() + if "radxa" in compat.lower(): + # Extract first compatible string as readable model name + parts = compat.split('\0') + model = parts[0] if parts else "Unknown Radxa" + return "radxa", model + except Exception: + pass + return "unknown", "Unknown" + + +PLATFORM, PLATFORM_MODEL = _detect_platform() + + +# ==================== Pin Mappings ==================== +# Physical 40-pin header pin number -> (gpiochip number, line offset) + +# Raspberry Pi: BOARD pin -> BCM GPIO number +_RPI_BOARD_TO_BCM = { + 3: 2, 5: 3, 7: 4, 8: 14, + 10: 15, 11: 17, 12: 18, 13: 27, + 15: 22, 16: 23, 18: 24, 19: 10, + 21: 9, 22: 25, 23: 11, 24: 8, + 26: 7, 27: 0, 28: 1, 29: 5, + 31: 6, 32: 12, 33: 13, 35: 19, + 36: 16, 37: 26, 38: 20, 40: 21, +} + + +def _build_rpi_pin_map(): + """Build RPi BOARD-pin -> (gpiochip, line_offset) mapping. + RPi 5 uses gpiochip4 (RP1), earlier models use gpiochip0.""" + chip_num = 4 if "Raspberry Pi 5" in PLATFORM_MODEL else 0 + return {pin: (chip_num, bcm) for pin, bcm in _RPI_BOARD_TO_BCM.items()} + +# Based on RK3566 Radxa ZERO 3W +RADXA_ZERO3_PIN_MAP = { + 3: (1, 0), 5: (1, 1), 7: (3, 20), 8: (0, 25), + 10: (0, 24), 11: (3, 1), 12: (3, 3), 13: (3, 2), + 15: (3, 8), 16: (3, 9), 18: (3, 10), 19: (4, 19), + 21: (4, 21), 22: (3, 17), 23: (4, 18), 24: (4, 22), + 26: (4, 25), 27: (4, 10), 28: (4, 11), 29: (3, 11), + 31: (3, 12), 32: (3, 18), 33: (3, 19), 35: (3, 4), + 36: (3, 7), 37: (1, 4), 38: (3, 6), 40: (3, 5), +} + +# Based on Allwinner A733 Radxa Cubie A7Z +# gpiochip0 (2000000.pinctrl, 352 lines): PA=0-31, PB=32-63, ..., PJ=288-319, PK=320-351 +# gpiochip1 (7025000.pinctrl, 64 lines): PL=0-31, PM=32-63 +RADXA_CUBIE_A7Z_PIN_MAP = { + 3: (0, 311), 5: (0, 310), 7: (0, 32), 8: (0, 41), + 10: (0, 42), 11: (0, 33), 12: (0, 37), 13: (1, 6), + 15: (1, 7), 16: (0, 312), 18: (0, 313), 19: (0, 108), + 21: (0, 109), 22: (1, 5), 23: (0, 107), 24: (0, 106), + 26: (0, 110), 27: (0, 113), 28: (0, 112), 29: (0, 34), + 31: (0, 35), 32: (1, 37), 33: (1, 35), 35: (0, 38), + 36: (0, 36), 37: (1, 36), 38: (0, 40), 40: (0, 39), +} + + +def _detect_radxa_board(): + """Detect specific Radxa board variant from device tree compatible string""" + try: + with open("/proc/device-tree/compatible", "r") as f: + compat = f.read().lower() + if "cubie-a7z" in compat: + return "cubie-a7z" + elif "cubie-a7a" in compat: + return "cubie-a7a" + elif "cubie-a7s" in compat: + return "cubie-a7s" + except Exception: + pass + # Default to zero3w for backward compatibility + return "zero3w" + + +# ==================== gpiod v1/v2 Compatibility ==================== +class _LineHandle: + """Unified thin wrapper that exposes set_value(int) / get_value()->int + regardless of whether the underlying library is gpiod v1 or v2.""" + + def __init__(self, v2_request=None, v2_offset=None, v1_line=None): + self._v2_req = v2_request + self._v2_off = v2_offset + self._v1 = v1_line + + def set_value(self, val): + if self._v2_req is not None: + self._v2_req.set_value( + self._v2_off, Value.ACTIVE if val else Value.INACTIVE + ) + else: + self._v1.set_value(val) + + def get_value(self): + if self._v2_req is not None: + return 1 if self._v2_req.get_value(self._v2_off) == Value.ACTIVE else 0 + return self._v1.get_value() + + def release(self): + try: + if self._v2_req is not None: + self._v2_req.release() + else: + self._v1.release() + except Exception: + pass + + +def _request_output(chip, line_offset, consumer='whisplay'): + """Request a single GPIO line as output (LOW initial).""" + if _GPIOD_V2: + settings = gpiod.LineSettings( + direction=Direction.OUTPUT, output_value=Value.INACTIVE + ) + req = chip.request_lines(consumer=consumer, config={line_offset: settings}) + return _LineHandle(v2_request=req, v2_offset=line_offset) + else: + line = chip.get_line(line_offset) + line.request(consumer=consumer, type=gpiod.LINE_REQ_DIR_OUT, default_val=0) + return _LineHandle(v1_line=line) + + +def _request_input(chip, line_offset, consumer='whisplay-btn'): + """Request a single GPIO line as input with bias disabled.""" + if _GPIOD_V2: + try: + settings = gpiod.LineSettings( + direction=Direction.INPUT, bias=Bias.DISABLED + ) + except Exception: + settings = gpiod.LineSettings(direction=Direction.INPUT) + req = chip.request_lines(consumer=consumer, config={line_offset: settings}) + return _LineHandle(v2_request=req, v2_offset=line_offset) + else: + line = chip.get_line(line_offset) + try: + line.request( + consumer=consumer, + type=gpiod.LINE_REQ_DIR_IN, + flags=gpiod.LINE_REQ_FLAG_BIAS_DISABLE + ) + except Exception: + line.request(consumer=consumer, type=gpiod.LINE_REQ_DIR_IN) + return _LineHandle(v1_line=line) + + +# ==================== Software PWM ==================== +class SoftPWM: + """Software PWM implementation for GPIO platforms without hardware PWM support""" + + def __init__(self, set_value_func, frequency=100, stop_value=0): + self._set_value = set_value_func + self.frequency = frequency + self.stop_value = stop_value + self.duty_cycle = 0.0 + self._running = False + self._thread = None + + def start(self, duty_cycle=0): + self.duty_cycle = float(duty_cycle) + self._running = True + self._thread = threading.Thread(target=self._pwm_loop, daemon=True) + self._thread.start() + + def ChangeDutyCycle(self, duty_cycle): + self.duty_cycle = max(0.0, min(100.0, float(duty_cycle))) + + def stop(self): + self._running = False + if self._thread: + self._thread.join(timeout=1) + try: + self._set_value(self.stop_value) + except Exception: + pass + + def _pwm_loop(self): + while self._running: + period = 1.0 / self.frequency + dc = self.duty_cycle + if dc <= 0: + self._set_value(0) + time.sleep(period) + elif dc >= 100: + self._set_value(1) + time.sleep(period) + else: + on_time = period * dc / 100.0 + off_time = period - on_time + self._set_value(1) + time.sleep(on_time) + self._set_value(0) + time.sleep(off_time) + + +class WhisplayBoard: + # LCD parameters + LCD_WIDTH = 240 + LCD_HEIGHT = 280 + BUTTON_POLL_INTERVAL_SEC = 0.005 + CornerHeight = 20 # Rounded corner height in pixels + + # Physical pin definitions (BOARD mode - shared by both platforms) + DC_PIN = 13 + RST_PIN = 7 + LED_PIN = 15 + + # RGB LED pins + RED_PIN = 22 + GREEN_PIN = 18 + BLUE_PIN = 16 + + # Button pin + BUTTON_PIN = 11 + + def __init__(self): + self.platform = PLATFORM + self.backlight_pwm = None + self._current_r = 0 + self._current_g = 0 + self._current_b = 0 + self.button_press_callback = None + self.button_release_callback = None + + # Select pin map and SPI config based on platform + if self.platform == "rpi": + self._pin_map = _build_rpi_pin_map() + self._spi_bus = 0 + self._spi_cs = 0 + self._spi_speed = 100_000_000 + elif self.platform == "radxa": + self._radxa_board = _detect_radxa_board() + if self._radxa_board == "cubie-a7z": + self._pin_map = RADXA_CUBIE_A7Z_PIN_MAP + self._spi_bus = 1 # SPI1, CS0 (Allwinner A733) + self._spi_cs = 0 + self._spi_speed = 48_000_000 + else: + self._pin_map = RADXA_ZERO3_PIN_MAP + self._spi_bus = 3 # SPI3, CS0 (RK3566 Radxa Zero 3W) + self._spi_cs = 0 + self._spi_speed = 48_000_000 + else: + raise RuntimeError( + f"Unsupported platform: {self.platform}\n" + "Install python3-libgpiod: sudo apt install python3-libgpiod" + ) + + self._init_gpio() + self._init_spi() + + self.previous_frame = None + # Detect hardware version and set backlight mode + self._detect_hardware_version() + self._detect_wm8960() + self.set_backlight(0) + self._reset_lcd() + self._init_display() + self.fill_screen(0) + + # ==================== GPIO Initialization (gpiod, unified) ==================== + def _init_gpio(self): + """Initialize all GPIO lines via gpiod (works on both RPi and Radxa).""" + self._gpio_chips = {} + self._gpio_lines = {} + + pins_used = [self.DC_PIN, self.RST_PIN, self.LED_PIN, + self.RED_PIN, self.GREEN_PIN, self.BLUE_PIN, + self.BUTTON_PIN] + + for pin in pins_used: + if pin not in self._pin_map: + raise RuntimeError(f"Physical pin {pin} is not defined in pin map") + chip_num, _ = self._pin_map[pin] + if chip_num not in self._gpio_chips: + self._gpio_chips[chip_num] = gpiod.Chip(f'/dev/gpiochip{chip_num}') + + # Request output pins + output_pins = [self.DC_PIN, self.RST_PIN, self.LED_PIN, + self.RED_PIN, self.GREEN_PIN, self.BLUE_PIN] + for pin in output_pins: + chip_num, line_offset = self._pin_map[pin] + chip = self._gpio_chips[chip_num] + self._gpio_lines[pin] = _request_output(chip, line_offset) + + # Enable backlight (LOW = on) + self._gpio_lines[self.LED_PIN].set_value(0) + + # Initialize RGB LED (using software PWM) + red_line = self._gpio_lines[self.RED_PIN] + green_line = self._gpio_lines[self.GREEN_PIN] + blue_line = self._gpio_lines[self.BLUE_PIN] + self.red_pwm = SoftPWM(red_line.set_value, 100, stop_value=1) + self.green_pwm = SoftPWM(green_line.set_value, 100, stop_value=1) + self.blue_pwm = SoftPWM(blue_line.set_value, 100, stop_value=1) + self.red_pwm.start(0) + self.green_pwm.start(0) + self.blue_pwm.start(0) + + # Initialize button (input, polled for state changes) + # The WhisPlay HAT has an external pull-down resistor on the button line. + # Button pressed = HIGH, released = LOW. No internal pull needed. + chip_num, line_offset = self._pin_map[self.BUTTON_PIN] + chip = self._gpio_chips[chip_num] + self._gpio_lines[self.BUTTON_PIN] = _request_input(chip, line_offset) + + # Start button event listener thread (polling) + self._btn_thread_running = True + self._btn_thread = threading.Thread(target=self._button_monitor, daemon=True) + self._btn_thread.start() + + def _init_spi(self): + """Initialize SPI bus.""" + self.spi = spidev.SpiDev() + self.spi.open(self._spi_bus, self._spi_cs) + self.spi.max_speed_hz = self._spi_speed + self.spi.mode = 0b00 + + def _button_monitor(self): + """Button state polling thread. + HIGH (1) = pressed, LOW (0) = released. + Poll interval is configurable; shorter values improve responsiveness. + """ + btn_line = self._gpio_lines[self.BUTTON_PIN] + last_state = btn_line.get_value() + poll_interval = self.BUTTON_POLL_INTERVAL_SEC + while self._btn_thread_running: + try: + state = btn_line.get_value() + if state != last_state: + last_state = state + if state == 1: + if self.button_press_callback: + self.button_press_callback() + else: + if self.button_release_callback: + self.button_release_callback() + except Exception: + if self._btn_thread_running: + pass + time.sleep(poll_interval) + + # ==================== GPIO Helpers ==================== + def _gpio_output(self, pin, value): + """Set GPIO pin output value""" + self._gpio_lines[pin].set_value(1 if value else 0) + + def _gpio_input(self, pin): + """Read GPIO pin input value""" + return self._gpio_lines[pin].get_value() + + # ==================== Hardware Detection ==================== + def _detect_hardware_version(self): + """Detect hardware version and set backlight mode accordingly""" + try: + model = PLATFORM_MODEL + if self.platform == "rpi": + if "Zero" in model and "2" not in model: + self.backlight_mode = False # Use simple on/off mode + else: + self.backlight_mode = True # Use PWM mode + elif self.platform == "radxa": + # Radxa uses software PWM mode + self.backlight_mode = True + else: + self.backlight_mode = True + print( + f"Detected hardware: {model}, Backlight mode: {'PWM' if self.backlight_mode else 'Simple Switch'}") + except Exception as e: + print(f"Error detecting hardware version: {e}") + self.backlight_mode = True + + def _detect_wm8960(self): + """Detect if a sound card containing wm8960 exists""" + try: + with open("/proc/asound/cards", "r") as f: + lines = f.readlines() + for line in lines: + if "wm8960" in line.lower(): + print("wm8960 sound card detected.") + return True + except Exception as e: + print(f"Error detecting wm8960 sound card: {e}") + return False + + print("wm8960 sound card not detected. Please refer to the following page for installation instructions.") + print("https://docs.pisugar.com/") + return False + + # ========== Backlight Control ========== + def set_backlight(self, brightness): + if self.backlight_mode: # PWM mode + if self.backlight_pwm is None: + led_line = self._gpio_lines[self.LED_PIN] + self.backlight_pwm = SoftPWM(led_line.set_value, 1000, stop_value=1) + self.backlight_pwm.start(100) + if 0 <= brightness <= 100: + duty_cycle = 100 - brightness + self.backlight_pwm.ChangeDutyCycle(duty_cycle) + else: # Simple on/off mode + if brightness == 0: + self._gpio_output(self.LED_PIN, 1) # Turn off backlight + else: + self._gpio_output(self.LED_PIN, 0) # Turn on backlight + + def set_backlight_mode(self, mode): + """ + Set backlight mode + :param mode: True for PWM brightness control, False for simple on/off + """ + if mode == self.backlight_mode: + return # Mode unchanged, no action needed + + if mode: # Switch to PWM mode + led_line = self._gpio_lines[self.LED_PIN] + self.backlight_pwm = SoftPWM(led_line.set_value, 1000, stop_value=1) + self.backlight_pwm.start(100) + else: # Switch to simple on/off mode + if self.backlight_pwm is not None: + self.backlight_pwm.stop() + self.backlight_pwm = None + self._gpio_output(self.LED_PIN, 1) # Ensure backlight is on + self.backlight_mode = mode + + def _reset_lcd(self): + self._gpio_output(self.RST_PIN, 1) + time.sleep(0.1) + self._gpio_output(self.RST_PIN, 0) + time.sleep(0.1) + self._gpio_output(self.RST_PIN, 1) + time.sleep(0.12) + + def _init_display(self): + self._send_command(0x11) + time.sleep(0.12) + USE_HORIZONTAL = 1 + direction = {0: 0x00, 1: 0xC0, 2: 0x70, + 3: 0xA0}.get(USE_HORIZONTAL, 0x00) + self._send_command(0x36, direction) + self._send_command(0x3A, 0x05) + self._send_command(0xB2, 0x0C, 0x0C, 0x00, 0x33, 0x33) + self._send_command(0xB7, 0x35) + self._send_command(0xBB, 0x32) + self._send_command(0xC2, 0x01) + self._send_command(0xC3, 0x15) + self._send_command(0xC4, 0x20) + self._send_command(0xC6, 0x0F) + self._send_command(0xD0, 0xA4, 0xA1) + self._send_command( + 0xE0, + 0xD0, + 0x08, + 0x0E, + 0x09, + 0x09, + 0x05, + 0x31, + 0x33, + 0x48, + 0x17, + 0x14, + 0x15, + 0x31, + 0x34, + ) + self._send_command( + 0xE1, + 0xD0, + 0x08, + 0x0E, + 0x09, + 0x09, + 0x15, + 0x31, + 0x33, + 0x48, + 0x17, + 0x14, + 0x15, + 0x31, + 0x34, + ) + self._send_command(0x21) + self._send_command(0x29) + + def _send_command(self, cmd, *args): + self._gpio_output(self.DC_PIN, 0) + self.spi.xfer2([cmd]) + if args: + self._gpio_output(self.DC_PIN, 1) + self._send_data(list(args)) + + def _send_data(self, data): + self._gpio_output(self.DC_PIN, 1) + try: + self.spi.writebytes2(data) + except AttributeError: + max_chunk = 4096 + for i in range(0, len(data), max_chunk): + self.spi.writebytes(data[i : i + max_chunk]) + + def _send_data_bytes(self, data: bytes | bytearray): + """Fast path for bytes/bytearray — avoids Python list overhead.""" + self._gpio_output(self.DC_PIN, 1) + try: + self.spi.writebytes2(data) + except AttributeError: + max_chunk = 4096 + for i in range(0, len(data), max_chunk): + self.spi.writebytes(list(data[i : i + max_chunk])) + + def set_window(self, x0, y0, x1, y1, use_horizontal=0): + if use_horizontal in (0, 1): + self._send_command(0x2A, x0 >> 8, x0 & 0xFF, x1 >> 8, x1 & 0xFF) + self._send_command( + 0x2B, (y0 + 20) >> 8, (y0 + 20) & 0xFF, (y1 + + 20) >> 8, (y1 + 20) & 0xFF + ) + elif use_horizontal in (2, 3): + self._send_command( + 0x2A, (x0 + 20) >> 8, (x0 + 20) & 0xFF, (x1 + + 20) >> 8, (x1 + 20) & 0xFF + ) + self._send_command(0x2B, y0 >> 8, y0 & 0xFF, y1 >> 8, y1 & 0xFF) + self._send_command(0x2C) + + def draw_pixel(self, x, y, color): + if x >= self.LCD_WIDTH or y >= self.LCD_HEIGHT: + return + self.set_window(x, y, x, y) + self._send_data([(color >> 8) & 0xFF, color & 0xFF]) + + def draw_line(self, x0, y0, x1, y1, color): + dx = abs(x1 - x0) + dy = abs(y1 - y0) + sx = 1 if x0 < x1 else -1 + sy = 1 if y0 < y1 else -1 + err = dx - dy + + while True: + self.draw_pixel(x0, y0, color) + if x0 == x1 and y0 == y1: + break + e2 = 2 * err + if e2 > -dy: + err -= dy + x0 += sx + if e2 < dx: + err += dx + y0 += sy + + def fill_screen(self, color): + self.set_window(0, 0, self.LCD_WIDTH - 1, self.LCD_HEIGHT - 1) + high = (color >> 8) & 0xFF + low = color & 0xFF + buffer = bytes([high, low]) * (self.LCD_WIDTH * self.LCD_HEIGHT) + self._send_data_bytes(buffer) + + def draw_image(self, x, y, width, height, pixel_data): + if (x + width > self.LCD_WIDTH) or (y + height > self.LCD_HEIGHT): + raise ValueError("Image dimensions exceed screen bounds") + self.set_window(x, y, x + width - 1, y + height - 1) + if isinstance(pixel_data, (bytes, bytearray)): + self._send_data_bytes(pixel_data) + else: + self._send_data(pixel_data) + + # ========== RGB LED & Button ========== + def set_rgb(self, r, g, b): + self.red_pwm.ChangeDutyCycle(100 - (r / 255 * 100)) + self.green_pwm.ChangeDutyCycle(100 - (g / 255 * 100)) + self.blue_pwm.ChangeDutyCycle(100 - (b / 255 * 100)) + self._current_r = r + self._current_g = g + self._current_b = b + + def set_rgb_fade(self, r_target, g_target, b_target, duration_ms=100): + steps = 20 # Adjust steps to control fade smoothness + delay_ms = duration_ms / steps + + r_step = (r_target - self._current_r) / steps + g_step = (g_target - self._current_g) / steps + b_step = (b_target - self._current_b) / steps + + for _ in range(steps + 1): + r_interim = int(self._current_r + _ * r_step) + g_interim = int(self._current_g + _ * g_step) + b_interim = int(self._current_b + _ * b_step) + self.set_rgb( + max(0, min(255, r_interim)), + max(0, min(255, g_interim)), + max(0, min(255, b_interim)), + ) + time.sleep(delay_ms / 1000.0) + + def button_pressed(self): + return self._gpio_input(self.BUTTON_PIN) == 1 + + def on_button_press(self, callback): + self.button_press_callback = callback + + def on_button_release(self, callback): + self.button_release_callback = callback + + # ========== Cleanup ========== + def cleanup(self): + # Stop backlight PWM + if self.backlight_pwm is not None: + self.backlight_pwm.stop() + # Close SPI + self.spi.close() + # Stop RGB LED PWM + self.red_pwm.stop() + self.green_pwm.stop() + self.blue_pwm.stop() + + # Stop button listener thread + self._btn_thread_running = False + if hasattr(self, '_btn_thread') and self._btn_thread: + self._btn_thread.join(timeout=2) + # Release GPIO resources + for line in self._gpio_lines.values(): + try: + line.release() + except Exception: + pass + for chip in self._gpio_chips.values(): + try: + chip.close() + except Exception: + pass diff --git a/lib/whisplay_client.py b/lib/whisplay_client.py new file mode 100644 index 0000000..4909bbf --- /dev/null +++ b/lib/whisplay_client.py @@ -0,0 +1,281 @@ +import json +import mmap +import socket +import threading +import time + +from .whisplay import WhisplayBoard + + +DEFAULT_DAEMON_SOCKET_PATH = "/tmp/whisplay-daemon.sock" +DEFAULT_APP_ID = "whisplay-ai-chatbot" +DEFAULT_APP_DISPLAY_NAME = "AI Chatbot" +DEFAULT_APP_ICON = "AI" +DEFAULT_EXIT_GESTURE = "quad_click" +DEFAULT_PRIORITY = 0 +DEFAULT_USE_DAEMON_DEFAULT_LOG = False + + +class WhisplayDaemonProxy: + LCD_WIDTH = 240 + LCD_HEIGHT = 280 + CornerHeight = 20 + + def __init__( + self, + socket_path: str = DEFAULT_DAEMON_SOCKET_PATH, + app_id: str = DEFAULT_APP_ID, + display_name: str = DEFAULT_APP_DISPLAY_NAME, + icon: str = DEFAULT_APP_ICON, + launch_command: str | None = None, + launch_cwd: str | None = None, + persist: bool = True, + exit_gesture: str = DEFAULT_EXIT_GESTURE, + priority: int = DEFAULT_PRIORITY, + use_daemon_default_log: bool = DEFAULT_USE_DAEMON_DEFAULT_LOG, + ): + self.socket_path = socket_path + self.button_press_callback = None + self.button_release_callback = None + self.exit_request_callback = None + self.focus_revoked_callback = None + self._button_down = False + self._subscriber = None + self._running = False + self._mmap = None + self._fb_file = None + self._fb_stride = self.LCD_WIDTH * 2 + self._fb_path = None + self._session_token = None + self._app_id = app_id + self._display_name = display_name + self._icon = icon + self._launch_command = launch_command + self._launch_cwd = launch_cwd + self._persist = persist + self._exit_gesture = str(exit_gesture or DEFAULT_EXIT_GESTURE) + self._priority = int(priority) + self._use_daemon_default_log = bool(use_daemon_default_log) + + def _send_request(self, cmd: str, payload: dict | None = None) -> dict: + body = {"version": 1, "cmd": cmd, "payload": payload or {}} + with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as client: + client.connect(self.socket_path) + client.sendall((json.dumps(body) + "\n").encode("utf-8")) + line = client.makefile("r").readline().strip() + if not line: + raise RuntimeError("empty response from whisplay-daemon") + response = json.loads(line) + if not response.get("ok"): + raise RuntimeError(response.get("error", "whisplay-daemon request failed")) + return response + + def ping(self) -> bool: + try: + self._send_request("health.ping") + return True + except Exception: + return False + + def register(self): + payload = { + "app_id": self._app_id, + "display_name": self._display_name, + "icon": self._icon, + "persist": self._persist, + } + if self._launch_command is not None: + payload["launch_command"] = self._launch_command + if self._launch_cwd is not None: + payload["cwd"] = self._launch_cwd + payload["exit_gesture"] = self._exit_gesture + payload["priority"] = self._priority + payload["use_daemon_default_log"] = self._use_daemon_default_log + self._send_request("app.register", payload) + + def acquire_foreground(self, timeout_sec: float = 5.0): + deadline = time.time() + timeout_sec + last_error = None + while time.time() < deadline: + try: + response = self._send_request("app.focus.acquire", {"app_id": self._app_id}) + self._session_token = response["payload"]["session_token"] + fb = self._send_request( + "framebuffer.acquire", + {"app_id": self._app_id, "session_token": self._session_token}, + )["payload"] + self._attach_framebuffer(fb["buffer_handle"], int(fb["stride"])) + return + except Exception as exc: + last_error = exc + time.sleep(0.2) + raise RuntimeError(f"failed to acquire foreground: {last_error}") + + def _attach_framebuffer(self, buffer_handle: str, stride: int): + self._detach_framebuffer() + self._fb_path = buffer_handle + self._fb_stride = stride + self._fb_file = open(buffer_handle, "r+b") + self._mmap = mmap.mmap(self._fb_file.fileno(), 0) + + def _detach_framebuffer(self): + if self._mmap is not None: + try: + self._mmap.close() + except Exception: + pass + self._mmap = None + if self._fb_file is not None: + try: + self._fb_file.close() + except Exception: + pass + self._fb_file = None + self._fb_path = None + + def release_focus(self): + if self._session_token: + try: + self._send_request( + "app.focus.release", + {"app_id": self._app_id, "session_token": self._session_token}, + ) + except Exception: + pass + self._session_token = None + self._detach_framebuffer() + + def prepare_exit(self): + self.release_focus() + + def set_backlight(self, brightness): + self._send_request("backlight.set", {"brightness": int(brightness)}) + + def set_rgb(self, r, g, b): + self._send_request("led.set", {"r": int(r), "g": int(g), "b": int(b)}) + + def set_rgb_fade(self, r_target, g_target, b_target, duration_ms=100): + self._send_request( + "led.fade", + { + "r": int(r_target), + "g": int(g_target), + "b": int(b_target), + "duration_ms": int(duration_ms), + }, + ) + + def draw_image(self, x, y, width, height, pixel_data): + if self._mmap is None: + return + frame_bytes = bytes(pixel_data if not isinstance(pixel_data, bytes) else pixel_data) + row_bytes = width * 2 + for row in range(height): + src = row * row_bytes + dst = ((y + row) * self._fb_stride) + (x * 2) + self._mmap[dst:dst + row_bytes] = frame_bytes[src:src + row_bytes] + + def fill_screen(self, color): + if self._mmap is None: + return + high = (int(color) >> 8) & 0xFF + low = int(color) & 0xFF + self._mmap.seek(0) + self._mmap.write(bytes([high, low]) * (self.LCD_WIDTH * self.LCD_HEIGHT)) + self._mmap.seek(0) + + def button_pressed(self): + return self._button_down + + def on_button_press(self, callback): + self.button_press_callback = callback + + def on_button_release(self, callback): + self.button_release_callback = callback + + def on_exit_request(self, callback): + self.exit_request_callback = callback + + def on_focus_revoked(self, callback): + self.focus_revoked_callback = callback + + def start_event_listener(self): + if self._subscriber is not None: + return + self._running = True + self._subscriber = threading.Thread(target=self._event_loop, daemon=True) + self._subscriber.start() + + def _event_loop(self): + while self._running: + try: + with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as client: + client.connect(self.socket_path) + body = {"version": 1, "cmd": "events.subscribe", "payload": {"app_id": self._app_id}} + client.sendall((json.dumps(body) + "\n").encode("utf-8")) + reader = client.makefile("r") + ack = reader.readline().strip() + if not ack: + raise RuntimeError("subscription ack missing") + for line in reader: + if not self._running: + return + line = line.strip() + if not line: + continue + event = json.loads(line) + name = event.get("event") + payload = event.get("payload", {}) or {} + if name == "button_pressed": + self._button_down = True + if self.button_press_callback: + self.button_press_callback() + elif name == "button_released": + self._button_down = False + if self.button_release_callback: + self.button_release_callback() + elif name == "app_exit_requested": + if self.exit_request_callback: + self.exit_request_callback() + elif name == "app_focus_revoked": + self._session_token = None + self._detach_framebuffer() + if self.focus_revoked_callback: + self.focus_revoked_callback(payload) + except Exception: + time.sleep(0.5) + + def cleanup(self): + self._running = False + self.release_focus() + + +def create_whisplay_hardware( + app_id: str = DEFAULT_APP_ID, + display_name: str = DEFAULT_APP_DISPLAY_NAME, + icon: str = DEFAULT_APP_ICON, + launch_command: str | None = None, + launch_cwd: str | None = None, + persist: bool = True, + exit_gesture: str = DEFAULT_EXIT_GESTURE, + priority: int = DEFAULT_PRIORITY, + use_daemon_default_log: bool = DEFAULT_USE_DAEMON_DEFAULT_LOG, +): + daemon = WhisplayDaemonProxy( + socket_path=DEFAULT_DAEMON_SOCKET_PATH, + app_id=app_id, + display_name=display_name, + icon=icon, + launch_command=launch_command, + launch_cwd=launch_cwd, + persist=persist, + exit_gesture=exit_gesture, + priority=priority, + use_daemon_default_log=use_daemon_default_log, + ) + if daemon.ping(): + daemon.register() + daemon.start_event_listener() + daemon.acquire_foreground() + return daemon + return WhisplayBoard() diff --git a/lxst-whisplay-client b/lxst-whisplay-client new file mode 100755 index 0000000..b6a4804 --- /dev/null +++ b/lxst-whisplay-client @@ -0,0 +1,2 @@ +#!/bin/sh +exec python3 "$(dirname "$0")/lxst_client.py" "$@" diff --git a/lxst_client.py b/lxst_client.py new file mode 100644 index 0000000..dd16d5f --- /dev/null +++ b/lxst_client.py @@ -0,0 +1,605 @@ +#!/usr/bin/env python3 +""" +LXST Whisplay Client - Group voice chat for Raspberry Pi Zero 2W +with PiSugar Whisplay HAT. + +Connects to lxst-relay voice channels via Reticulum/LXST. +Uses the Whisplay HAT for LCD display, button (PTT), RGB LED, +and WM8960 audio I/O. +""" + +import RNS +import os +import sys +import time +import threading +import argparse +import numpy as np +import signal as signal_module + +from LXST import Pipeline +from LXST.Network import SignallingReceiver, Packetizer, LinkSource +from LXST.Sinks import LineSink +from LXST.Sources import LineSource +from LXST.Codecs import Opus +from LXST.Primitives.Telephony import Signalling + +script_dir = os.path.dirname(os.path.abspath(__file__)) +whisplay_runtime = os.path.abspath(os.path.join(script_dir, "..", "Whisplay", "runtime")) +if whisplay_runtime not in sys.path: + sys.path.insert(0, whisplay_runtime) + +from lib.whisplay_client import create_whisplay_hardware + +HAS_PIL = False +try: + from PIL import Image, ImageDraw, ImageFont + HAS_PIL = True +except ImportError: + pass + +APP_NAME = "lxst" +PRIMITIVE_NAME = "telephony" + +STATE_IDLE = 0 +STATE_CONNECTING = 1 +STATE_CONNECTED = 2 +STATE_TRANSMITTING = 3 +STATE_FAILED = 4 + +STATUS_STR = { + STATE_IDLE: "Disconnected", + STATE_CONNECTING: "Connecting...", + STATE_CONNECTED: "Connected", + STATE_TRANSMITTING: "TX", + STATE_FAILED: "Failed", +} + +ACCENT_IDLE = (60, 150, 255) +ACCENT_CONNECTING = (255, 180, 0) +ACCENT_CONNECTED = (0, 200, 80) +ACCENT_TX = (255, 50, 50) +ACCENT_FAILED = (200, 50, 50) + +RGB_IDLE = (0, 0, 100) +RGB_CONNECTING = (100, 80, 0) +RGB_CONNECTED = (0, 100, 0) +RGB_TX = (255, 0, 0) +RGB_FAILED = (100, 0, 0) + +WHISPLAY_APP_ID = "lxst-whisplay-client" +DISPLAY_NAME = "LXST Voice" + +class LXSTWhisplayClient(SignallingReceiver): + def __init__(self, channel_hash, identity_path=None, rns_config=None, verbosity=0): + super().__init__() + self.channel_hash = channel_hash + self.state = STATE_IDLE + self.link = None + self.link_source = None + self.packetizer = None + self.line_source = None + self.line_sink = None + self.tx_pipeline = None + self.participant_count = 0 + self._exit = False + self._cancel_connect = False + self._lock = threading.Lock() + self._press_time = 0 + self._press_state = STATE_IDLE + self._error_msg = "" + + self._init_rns(identity_path, rns_config, verbosity) + self._init_whisplay() + self._init_fonts() + + def _init_rns(self, identity_path, rns_config, verbosity): + if identity_path and os.path.isfile(identity_path): + self.identity = RNS.Identity.from_file(identity_path) + if self.identity is None: + RNS.log("Could not load identity, generating new", RNS.LOG_ERROR) + self.identity = RNS.Identity() + if identity_path: + self.identity.to_file(identity_path) + else: + RNS.log("No identity found, generating new", RNS.LOG_NOTICE) + self.identity = RNS.Identity() + if identity_path: + self.identity.to_file(identity_path) + + self.reticulum = RNS.Reticulum(configdir=rns_config, loglevel=3 + verbosity) + + self.destination = RNS.Destination( + self.identity, + RNS.Destination.IN, + RNS.Destination.SINGLE, + APP_NAME, + PRIMITIVE_NAME, + ) + self.destination.set_proof_strategy(RNS.Destination.PROVE_NONE) + + RNS.log( + f"LXST Client identity: {RNS.prettyhexrep(self.identity.hash)}", + RNS.LOG_NOTICE, + ) + + def _init_whisplay(self): + self.board = create_whisplay_hardware( + app_id=WHISPLAY_APP_ID, + display_name=DISPLAY_NAME, + icon="LX", + use_daemon_default_log=True, + ) + self.board.set_backlight(70) + self.board.on_button_press(self._on_button_press) + self.board.on_button_release(self._on_button_release) + if hasattr(self.board, "on_exit_request"): + self.board.on_exit_request(self._on_exit_request) + if hasattr(self.board, "on_focus_revoked"): + self.board.on_focus_revoked(self._on_focus_revoked) + + if not HAS_PIL: + self.board.fill_screen(0) + + def _init_fonts(self): + self._fonts = {} + if not HAS_PIL: + return + for size, bold in [(22, True), (18, False), (14, False), (12, False)]: + self._fonts[(size, bold)] = self._load_font(size, bold) + + def _load_font(self, size, bold=False): + candidates = [] + if bold: + candidates.extend([ + "/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf", + ]) + candidates.extend([ + "/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf", + "/usr/share/fonts/truetype/freefont/FreeSans.ttf", + ]) + for path in candidates: + if os.path.exists(path): + try: + return ImageFont.truetype(path, size) + except Exception: + continue + return ImageFont.load_default() + + def _wrap_text(self, draw, text, font, max_width): + if not text: + return [""] + words = text.split() + if not words: + return [text] + lines = [] + current = words[0] + for word in words[1:]: + cand = f"{current} {word}" + bbox = draw.textbbox((0, 0), cand, font=font) + if (bbox[2] - bbox[0]) <= max_width: + current = cand + else: + lines.append(current) + current = word + lines.append(current) + return lines + + def _rgb565_bytes(self, image): + rgb = image.convert("RGB") + w, h = rgb.width, rgb.height + buf = bytearray(w * h * 2) + i = 0 + for y in range(h): + for x in range(w): + r, g, b = rgb.getpixel((x, y)) + v = ((r & 0xF8) << 8) | ((g & 0xFC) << 3) | (b >> 3) + buf[i] = (v >> 8) & 0xFF + buf[i + 1] = v & 0xFF + i += 2 + return bytes(buf) + + def _render_frame(self, accent=None): + if not HAS_PIL: + return None + w = self.board.LCD_WIDTH + h = self.board.LCD_HEIGHT + accent = accent or ACCENT_IDLE + bg = (11, 16, 24) + + img = Image.new("RGB", (w, h), bg) + draw = ImageDraw.Draw(img) + + font_title = self._fonts.get((22, True), ImageFont.load_default()) + font_body = self._fonts.get((18, False), ImageFont.load_default()) + font_small = self._fonts.get((14, False), ImageFont.load_default()) + font_tiny = self._fonts.get((12, False), ImageFont.load_default()) + + draw.rounded_rectangle( + (12, 16, w - 12, h - 16), radius=18, + fill=(20, 28, 40), outline=(40, 55, 72), width=2, + ) + + draw.rounded_rectangle( + (20, 22, w - 20, 58), radius=12, fill=accent, + ) + + state_str = STATUS_STR.get(self.state, "Unknown") + r_accent, g_accent, b_accent = accent + luminance = 0.299 * r_accent + 0.587 * g_accent + 0.114 * b_accent + text_color = (18, 24, 32) if luminance >= 170 else (255, 255, 255) + + draw.text((30, 31), state_str, fill=text_color, font=font_small) + + cy = 74 + draw.text((24, cy), "LXST Voice", fill=(255, 255, 255), font=font_title) + cy += 32 + + ch = self.channel_hash + display_ch = ch[:16] + "..." if len(ch) > 19 else ch + draw.text((24, cy), display_ch, fill=(180, 190, 200), font=font_body) + cy += 26 + + if self.state in (STATE_CONNECTED, STATE_TRANSMITTING): + pc = self.participant_count + draw.text( + (24, cy), f"Participants: {pc}", + fill=(150, 200, 150), font=font_body, + ) + cy += 28 + + cy += 8 + + if self.state == STATE_IDLE: + for line in self._wrap_text(draw, "Press button to connect", font_body, w - 56): + draw.text((24, cy), line, fill=(160, 180, 210), font=font_body) + cy += 24 + elif self.state == STATE_CONNECTING: + for line in self._wrap_text(draw, "Connecting to channel...", font_body, w - 56): + draw.text((24, cy), line, fill=(255, 200, 100), font=font_body) + cy += 24 + for line in self._wrap_text(draw, "Press to cancel", font_small, w - 56): + draw.text((24, cy), line, fill=(200, 170, 100), font=font_small) + cy += 20 + elif self.state == STATE_CONNECTED: + for line in self._wrap_text(draw, "Hold button to talk", font_body, w - 56): + draw.text((24, cy), line, fill=(160, 220, 180), font=font_body) + cy += 24 + for line in self._wrap_text(draw, "Long press 3s to leave", font_small, w - 56): + draw.text((24, cy), line, fill=(150, 180, 160), font=font_small) + cy += 20 + elif self.state == STATE_TRANSMITTING: + draw.rounded_rectangle( + (28, cy, w - 28, cy + 36), radius=8, + fill=(220, 40, 40), outline=(255, 80, 80), width=2, + ) + tx_label = "TRANSMITTING" + bbox = draw.textbbox((0, 0), tx_label, font=font_title) + tw = bbox[2] - bbox[0] + draw.text( + ((w - tw) // 2, cy + 6), tx_label, + fill=(255, 255, 255), font=font_title, + ) + cy += 48 + for line in self._wrap_text(draw, "Release to stop", font_small, w - 56): + draw.text((24, cy), line, fill=(255, 150, 150), font=font_small) + cy += 20 + elif self.state == STATE_FAILED: + for line in self._wrap_text(draw, self._error_msg, font_body, w - 56): + draw.text((24, cy), line, fill=(255, 150, 150), font=font_body) + cy += 24 + for line in self._wrap_text(draw, "Press to retry", font_small, w - 56): + draw.text((24, cy), line, fill=(200, 150, 150), font=font_small) + cy += 20 + + y_pos = h - 40 + draw.rounded_rectangle( + (20, y_pos, w - 20, h - 22), radius=12, + fill=(28, 37, 50), + ) + if self.state == STATE_CONNECTED: + draw.text( + (28, y_pos + 8), "RX", fill=(100, 200, 100), font=font_tiny, + ) + bar_w = w - 80 + bar_x = 60 + draw.rounded_rectangle( + (bar_x, y_pos + 10, bar_x + bar_w, y_pos + 22), + radius=4, fill=(40, 55, 70), + ) + draw.rounded_rectangle( + (bar_x, y_pos + 10, bar_x + int(bar_w * 0.6), y_pos + 22), + radius=4, fill=(0, 200, 80), + ) + elif self.state == STATE_TRANSMITTING: + draw.text( + (28, y_pos + 8), "TX", fill=(255, 100, 100), font=font_tiny, + ) + bar_w = w - 80 + bar_x = 60 + draw.rounded_rectangle( + (bar_x, y_pos + 10, bar_x + bar_w, y_pos + 22), + radius=4, fill=(40, 55, 70), + ) + draw.rounded_rectangle( + (bar_x, y_pos + 10, bar_x + bar_w, y_pos + 22), + radius=4, fill=(255, 50, 50), + ) + elif self.state == STATE_IDLE: + draw.text( + (28, y_pos + 8), "Standby", fill=(150, 170, 190), font=font_tiny, + ) + + return self._rgb565_bytes(img) + + def _update_display(self): + if self.state == STATE_IDLE: + accent = ACCENT_IDLE + elif self.state == STATE_CONNECTING: + accent = ACCENT_CONNECTING + elif self.state == STATE_CONNECTED: + accent = ACCENT_CONNECTED + elif self.state == STATE_TRANSMITTING: + accent = ACCENT_TX + elif self.state == STATE_FAILED: + accent = ACCENT_FAILED + else: + accent = ACCENT_IDLE + + frame = self._render_frame(accent=accent) + if frame: + self.board.draw_image( + 0, 0, self.board.LCD_WIDTH, self.board.LCD_HEIGHT, frame, + ) + + def _set_state(self, state, error_msg=""): + with self._lock: + self.state = state + self._error_msg = error_msg + self._update_display() + + def _fail(self, msg): + self._cleanup_all() + self.board.set_rgb(*RGB_FAILED) + self._set_state(STATE_FAILED, msg) + RNS.log(msg, RNS.LOG_ERROR) + time.sleep(4) + with self._lock: + if self.state == STATE_FAILED: + self.state = STATE_IDLE + self._update_display() + self.board.set_rgb(*RGB_IDLE) + + def connect(self): + with self._lock: + if self.state not in (STATE_IDLE, STATE_FAILED): + return + self.state = STATE_CONNECTING + self._update_display() + + self.board.set_rgb(*RGB_CONNECTING) + + try: + channel_id = RNS.Identity.from_hex(self.channel_hash) + except Exception as e: + self._fail(f"Invalid channel hash: {e}") + return + + dest = RNS.Destination( + channel_id, + RNS.Destination.OUT, + RNS.Destination.SINGLE, + APP_NAME, + PRIMITIVE_NAME, + ) + + self.link = RNS.Link(dest) + self.link.set_link_closed_callback(self._on_link_closed) + + deadline = time.time() + 20 + while self.link.status != RNS.Link.ACTIVE and time.time() < deadline: + time.sleep(0.05) + if self._cancel_connect: + self._cancel_connect = False + self._cleanup_link() + self._set_state(STATE_IDLE) + self.board.set_rgb(*RGB_IDLE) + return + + if self.link.status != RNS.Link.ACTIVE: + self._fail("Connection timeout - channel unreachable") + return + + try: + self.line_sink = LineSink() + self.link_source = LinkSource(self.link, self, sink=self.line_sink) + self.link_source.start() + + self.tx_codec = Opus(profile=Opus.PROFILE_VOICE_MEDIUM) + self.packetizer = Packetizer(self.link) + self.line_source = LineSource(codec=self.tx_codec, sink=self.packetizer) + self.tx_pipeline = Pipeline( + source=self.line_source, + codec=self.tx_codec, + sink=self.packetizer, + ) + except Exception as e: + self._fail(f"Audio setup failed: {e}") + return + + with self._lock: + self.state = STATE_CONNECTED + self._update_display() + self.board.set_rgb(*RGB_CONNECTED) + + RNS.log( + f"Connected to channel {RNS.prettyhexrep(channel_id.hash)}", + RNS.LOG_NOTICE, + ) + + def _cleanup_link(self): + if self.link: + try: + if self.link.status == RNS.Link.ACTIVE: + self.link.teardown() + except Exception: + pass + self.link = None + + def _cleanup_pipelines(self): + if self.tx_pipeline and self.tx_pipeline.running: + self.tx_pipeline.stop() + self.tx_pipeline = None + if self.link_source: + self.link_source.stop() + self.link_source = None + if self.line_sink: + self.line_sink.stop() + self.line_sink = None + if self.line_source: + self.line_source = None + if self.packetizer: + self.packetizer.stop() + self.packetizer = None + + def _cleanup_all(self): + self._cleanup_pipelines() + self._cleanup_link() + + def disconnect(self): + with self._lock: + was_connected = self.state not in (STATE_IDLE,) + self._cleanup_all() + self.state = STATE_IDLE + self.participant_count = 0 + if was_connected: + self._update_display() + self.board.set_rgb(*RGB_IDLE) + + RNS.log("Disconnected from channel", RNS.LOG_NOTICE) + + def _on_link_closed(self, link): + self.disconnect() + + def _ptt_start(self): + with self._lock: + if self.state != STATE_CONNECTED: + return + if self.tx_pipeline and not self.tx_pipeline.running: + self.tx_pipeline.start() + self.state = STATE_TRANSMITTING + self.board.set_rgb(*RGB_TX) + self._update_display() + + def _ptt_stop(self): + with self._lock: + if self.state != STATE_TRANSMITTING: + return + if self.tx_pipeline and self.tx_pipeline.running: + self.tx_pipeline.stop() + self.state = STATE_CONNECTED + self.board.set_rgb(*RGB_CONNECTED) + self._update_display() + + def _on_button_press(self): + self._press_time = time.time() + with self._lock: + self._press_state = self.state + state = self.state + + if state == STATE_CONNECTED: + self._ptt_start() + elif state == STATE_IDLE or state == STATE_FAILED: + threading.Thread(target=self.connect, daemon=True).start() + elif state == STATE_CONNECTING: + self._cancel_connect = True + + def _on_button_release(self): + held = time.time() - self._press_time + if self._press_state == STATE_CONNECTED and held >= 3.0: + threading.Thread(target=self.disconnect, daemon=True).start() + return + self._ptt_stop() + + def _on_exit_request(self, payload=None): + self._exit = True + self.disconnect() + + def _on_focus_revoked(self, payload=None): + self.disconnect() + + def signalling_received(self, signals, source): + pass + + def run(self): + self.board.set_rgb(*RGB_IDLE) + self._update_display() + + signal_module.signal(signal_module.SIGINT, self._sigint_handler) + signal_module.signal(signal_module.SIGTERM, self._sigterm_handler) + + RNS.log( + f"LXST Whisplay Client ready on channel hash: {self.channel_hash}", + RNS.LOG_NOTICE, + ) + RNS.log("Press button to connect", RNS.LOG_NOTICE) + + while not self._exit: + time.sleep(0.1) + + self.cleanup() + + def _sigint_handler(self, sig, frame): + self._exit = True + + def _sigterm_handler(self, sig, frame): + self._exit = True + + def cleanup(self): + self._exit = True + self._cancel_connect = True + self.disconnect() + try: + self.board.set_rgb(0, 0, 0) + self.board.cleanup() + except Exception: + pass + + +def main(): + parser = argparse.ArgumentParser( + description="LXST Whisplay Client - Group voice chat for Pi Zero 2W", + ) + parser.add_argument( + "channel", + help="Channel destination hash (hex string from lxst-relay bot)", + ) + parser.add_argument( + "--identity", "-i", default=None, + help="Path to persistent identity file", + ) + parser.add_argument( + "--rns-config", default=None, + help="Path to alternative Reticulum config directory", + ) + parser.add_argument( + "-v", "--verbose", action="count", default=0, + ) + args = parser.parse_args() + + client = LXSTWhisplayClient( + channel_hash=args.channel, + identity_path=args.identity, + rns_config=args.rns_config, + verbosity=args.verbose, + ) + + try: + client.run() + except KeyboardInterrupt: + client.cleanup() + print("") + + +if __name__ == "__main__": + main()