#!/usr/bin/env python3 import json import os import sys import time import threading import signal import argparse import subprocess import RNS import LXMF from LXST.Primitives.Telephony import Telephone # ── Whisplay HAT GUI support ────────────────────────────────────────── _RUNTIME_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "runtime") if os.path.isdir(_RUNTIME_DIR): sys.path.insert(0, _RUNTIME_DIR) _WHISPLAY_AVAILABLE = False try: from whisplay_client import create_whisplay_hardware _WHISPLAY_AVAILABLE = True except ImportError: pass _HAS_PIL = False if _WHISPLAY_AVAILABLE: try: from PIL import Image, ImageDraw, ImageFont _HAS_PIL = True except ImportError: pass class Terminal: if not RNS.vendor.platformutils.is_windows(): UNDERLINE = "\033[4m" BOLD = "\033[1m" END = "\033[0m" else: UNDERLINE = "" BOLD = "" END = "" class ChannelClient: STATE_IDLE = 0 STATE_DISCOVERING = 1 STATE_READY = 2 STATE_CALLING = 3 STATE_IN_CALL = 4 def __init__(self, configdir=None, rnsconfigdir=None, whitelist_path=None, verbosity=0, announce_interval=0, use_whisplay=False): self.state = self.STATE_IDLE self.channels = [] self.telephone = None self.should_run = False self.announce_interval = announce_interval self._responses = [] self._response_event = threading.Event() # ── Whisplay HAT init ── self.board = None self.selected_index = 0 self._wm8960_device = None self._ptt_enabled = True self._ptt_active = False self.gestures = None if use_whisplay and _WHISPLAY_AVAILABLE and _HAS_PIL: try: self._init_whisplay() except Exception as e: RNS.log(f"Whisplay init failed: {e}", RNS.LOG_WARNING) self.board = None if configdir is None: if os.path.isdir("/etc/lxst-client"): self.configdir = "/etc/lxst-client" else: userdir = RNS.Reticulum.userdir self.configdir = os.path.join(userdir, ".lxst-client") else: self.configdir = configdir os.makedirs(self.configdir, exist_ok=True) self.whitelist_path = whitelist_path if self.whitelist_path is None: candidates = [ os.path.join(os.getcwd(), "bridge_whitelist.json"), os.path.join(self.configdir, "bridge_whitelist.json"), ] for p in candidates: if os.path.isfile(p): self.whitelist_path = p break if self.whitelist_path is None: self.whitelist_path = candidates[0] identity_path = os.path.join(self.configdir, "identity") if os.path.isfile(identity_path): self.identity = RNS.Identity.from_file(identity_path) if self.identity is None: self.identity = RNS.Identity() self.identity.to_file(identity_path) else: RNS.log("No identity found, generating new", RNS.LOG_NOTICE) self.identity = RNS.Identity() self.identity.to_file(identity_path) RNS.log( f"Generated identity: {RNS.prettyhexrep(self.identity.hash)}", RNS.LOG_NOTICE, ) self.reticulum = RNS.Reticulum( configdir=rnsconfigdir, loglevel=3 + verbosity ) self.lxmf_storage = os.path.join(self.configdir, "lxmf") os.makedirs(self.lxmf_storage, exist_ok=True) self.lxmf_router = LXMF.LXMRouter( identity=self.identity, storagepath=self.lxmf_storage, ) self.client_destination = self.lxmf_router.register_delivery_identity( self.identity, display_name="LXST Channel Client", ) self.lxmf_router.register_delivery_callback(self._delivery_callback) self.client_destination.announce() if self.announce_interval > 0: threading.Thread(target=self._announce_loop, daemon=True).start() # ── Whisplay HAT GUI ───────────────────────────────────────────── def _init_whisplay(self): from whisplay_client import create_whisplay_hardware self.board = create_whisplay_hardware( app_id="lxst-client", display_name="LXST Client", icon="LC", use_daemon_default_log=True, ) self.board.set_backlight(70) from whisplay_interface import WhisplayInterface self.gestures = WhisplayInterface(self.board) self.gestures.on_press(self._on_button_press) self.gestures.on_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 hasattr(self.board, "start_event_listener"): self.board.start_event_listener() self._load_fonts() self._setup_whisplay_audio() def _find_wm8960_card(self): try: with open("/proc/asound/cards") as f: for line in f: if "wm8960" in line.lower(): return line.split()[0] except Exception: return None return None def _setup_whisplay_audio(self): card = self._find_wm8960_card() if card is None: RNS.log("Whisplay WM8960 sound card not found", RNS.LOG_WARNING) return mixer_cmds = [ ["amixer", "-c", card, "sset", "Left Output Mixer PCM", "on"], ["amixer", "-c", card, "sset", "Right Output Mixer PCM", "on"], ["amixer", "-c", card, "sset", "Speaker", "121"], ["amixer", "-c", card, "sset", "Playback", "230"], ] for cmd in mixer_cmds: try: subprocess.run(cmd, check=False, capture_output=True, text=True) except Exception: pass self._wm8960_device = f"plughw:{card},0" RNS.log(f"Whisplay audio: WM8960 card {card} ready", RNS.LOG_DEBUG) def _load_fonts(self): def load(size, bold=False): cand = [] if bold: cand.extend([ "/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf", "/usr/share/fonts/truetype/freefont/FreeSansBold.ttf", ]) cand.extend([ "/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf", "/usr/share/fonts/truetype/freefont/FreeSans.ttf", ]) for c in cand: if os.path.exists(c): try: return ImageFont.truetype(c, size) except Exception: continue return ImageFont.load_default() self._title_font = load(20, True) self._body_font = load(16) self._small_font = load(12) def _rgb565(self, image): rgb = image.convert("RGB") out = bytearray() for y in range(rgb.height): for x in range(rgb.width): r, g, b = rgb.getpixel((x, y)) v = ((r & 0xF8) << 8) | ((g & 0xFC) << 3) | (b >> 3) out.append((v >> 8) & 0xFF) out.append(v & 0xFF) return bytes(out) def _wrap(self, draw, text, font, max_w): if not text: return [""] words = text.split() if not words: return [text] lines, cur = [], words[0] for w in words[1:]: test = f"{cur} {w}" bb = draw.textbbox((0, 0), test, font=font) if (bb[2] - bb[0]) <= max_w: cur = test else: lines.append(cur) cur = w lines.append(cur) return lines def _render(self, title, body_lines, accent=(60, 150, 255), footer=""): w, h = self.board.LCD_WIDTH, self.board.LCD_HEIGHT img = Image.new("RGB", (w, h), (11, 16, 24)) draw = ImageDraw.Draw(img) 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, 54), radius=12, fill=accent) draw.text((28, 30), title, fill=(255, 255, 255), font=self._title_font) y = 70 max_w = w - 56 for line in body_lines: if y >= h - 50: break if not line: y += 8 continue is_sel = False txt = line if isinstance(line, tuple): txt, is_sel = line wrapped = self._wrap(draw, txt, self._body_font, max_w) for wl in wrapped: if is_sel: bb = draw.textbbox((0, 0), wl, font=self._body_font) lw = bb[2] - bb[0] draw.rounded_rectangle((22, y - 2, 22 + lw + 12, y + 20), radius=6, fill=(accent[0] // 3, accent[1] // 3, accent[2] // 3)) draw.text((28, y), wl, fill=(255, 255, 255), font=self._body_font) else: draw.text((28, y), wl, fill=(180, 195, 210), font=self._body_font) y += 22 if footer: draw.rounded_rectangle((20, h - 38, w - 20, h - 22), radius=10, fill=(28, 37, 50)) draw.text((28, h - 34), footer, fill=(140, 195, 255), font=self._small_font) return self._rgb565(img) def _flash_screen(self): if self.board is None: return fb = self._render("LXST Client", ["Channels loading..."], footer="Please wait") self.board.draw_image(0, 0, self.board.LCD_WIDTH, self.board.LCD_HEIGHT, fb) def _show_channels(self): if self.board is None: return if not self.channels: fb = self._render("LXST Client", ["No channels", "", "Re-discover to find channels"], accent=(200, 100, 60), footer="No channels") else: lines = [] for i, ch in enumerate(self.channels): lines.append((f"{i+1}: {ch['hash'][:16]}..", i == self.selected_index)) fb = self._render(f"Channels ({len(self.channels)})", lines, footer="Short: next | Long: select") self.board.draw_image(0, 0, self.board.LCD_WIDTH, self.board.LCD_HEIGHT, fb) def _show_in_call(self): if self.board is None: return ch = self.channels[self.selected_index] if self.channels else None lines = [f"Channel {self.selected_index + 1}"] if ch: lines.append(ch["hash"][:16] + "..") if self._ptt_enabled: lines += ["", "PTT: hold to talk", "Short: always-on"] footer = "Dbl: leave" else: lines += ["", "Always-on (unmuted)"] footer = "Short: PTT Dbl: leave" fb = self._render("In Call", lines, accent=(60, 220, 120), footer=footer) self.board.draw_image(0, 0, self.board.LCD_WIDTH, self.board.LCD_HEIGHT, fb) def _show_status(self, title, detail="", accent=(255, 180, 60)): if self.board is None: return lines = [detail] if detail else [] fb = self._render(title, lines, accent=accent) self.board.draw_image(0, 0, self.board.LCD_WIDTH, self.board.LCD_HEIGHT, fb) def _on_button_press(self): if self.state == self.STATE_IN_CALL: if self.gestures.is_double_pressed: self.telephone.hangup() return if self._ptt_enabled: self._ptt_active = True self.telephone.unmute_transmit() if self.board: self.board.set_rgb(255, 255, 0) def _on_button_release(self): if self.gestures.was_double_pressed: self.gestures.acknowledge_double_press() return if self.state == self.STATE_IN_CALL: elapsed = self.gestures.press_duration if self._ptt_enabled: if elapsed < self.gestures.long_press_threshold: self._ptt_active = False self._ptt_enabled = False if self.board: self._show_in_call() else: self._ptt_active = False self.telephone.mute_transmit() if self.board: self.board.set_rgb(0, 255, 80) else: if elapsed < self.gestures.long_press_threshold: self._ptt_enabled = True self.telephone.mute_transmit() if self.board: self._show_in_call() return if self.state != self.STATE_READY: return elapsed = self.gestures.press_duration if elapsed >= self.gestures.long_press_threshold: if self.channels: self.board.set_rgb(255, 180, 0) self.call_channel(self.selected_index) else: if self.channels: self.selected_index = (self.selected_index + 1) % len(self.channels) self._show_channels() self.gestures.acknowledge_press() def _on_exit_request(self, payload=None): self.cleanup() if self.board: self.board.prepare_exit() def _on_focus_revoked(self, payload=None): self.cleanup() def _announce_loop(self): while self.should_run: time.sleep(self.announce_interval) self.client_destination.announce() def _delivery_callback(self, message): content = message.content_as_string().strip() sender_hex = RNS.hexrep(message.source_hash, delimit=False) try: data = json.loads(content) if isinstance(data, list): for h in data: if isinstance(h, str) and len(h) in (32, 64): self._responses.append({ "hash": h, "relay": sender_hex, }) except (json.JSONDecodeError, ValueError): pass self._response_event.set() def _discover_from_relay(self, address): try: addr_bytes = bytes.fromhex(address) identity = RNS.Identity.recall(addr_bytes) if identity is None: RNS.Transport.request_path(addr_bytes) for _ in range(50): identity = RNS.Identity.recall(addr_bytes) if identity is not None: break time.sleep(0.1) if identity is None: print(f" Could not discover relay {address[:16]}...") return dest = RNS.Destination( identity, RNS.Destination.OUT, RNS.Destination.SINGLE, "lxmf", "delivery", ) msg = LXMF.LXMessage( dest, self.client_destination, "/channels_json", desired_method=LXMF.LXMessage.DIRECT, ) self.lxmf_router.handle_outbound(msg) except Exception as e: RNS.log(f"Error sending to {address[:16]}...: {e}", RNS.LOG_ERROR) def discover(self, timeout=30): if not os.path.isfile(self.whitelist_path): print(f"bridge_whitelist.json not found at {self.whitelist_path}") return [] with open(self.whitelist_path) as f: whitelist = json.load(f) if not isinstance(whitelist, list): print("bridge_whitelist.json must contain a list of LXMF addresses") return [] self.state = self.STATE_DISCOVERING self._responses = [] self._response_event.clear() print(f"\nQuerying {len(whitelist)} relay(s) for channels...", flush=True) for addr in whitelist: self._discover_from_relay(addr) deadline = time.time() + timeout while time.time() < deadline: remaining = deadline - time.time() self._response_event.wait(timeout=min(remaining, 1)) if self._responses: break seen = set() deduped = [] for r in self._responses: if r["hash"] not in seen: seen.add(r["hash"]) deduped.append(r) self.channels = deduped print( f" Got {len(self.channels)} unique channel(s) " f"from {len(self._responses)} response(s)" ) self.state = self.STATE_READY return self.channels def _init_telephone(self): if self.telephone is None: self.telephone = Telephone(self.identity) if self._wm8960_device: self.telephone.set_speaker(self._wm8960_device) self.telephone.set_microphone(self._wm8960_device) self.telephone.set_ringer(self._wm8960_device) self.telephone.set_established_callback(self._on_call_established) self.telephone.set_ended_callback(self._on_call_ended) self.telephone.announce() def _on_call_established(self, remote_identity): self.state = self.STATE_IN_CALL print( f"\nCall established with " f"{RNS.prettyhexrep(remote_identity.hash)}" ) if self._ptt_enabled: self.telephone.mute_transmit() if self.board: self.board.set_rgb(0, 255, 80) self._show_in_call() def _on_call_ended(self, remote_identity): was_in_call = self.state == self.STATE_IN_CALL was_calling = self.state == self.STATE_CALLING self.state = self.STATE_READY self._ptt_active = False if self.gestures: self.gestures.acknowledge_double_press() if was_in_call: print(f"\nCall ended") elif was_calling: print(f"\nCall could not be connected") if self.board: self.board.set_rgb(0, 80, 255) if was_in_call: self._show_status("Call Ended") time.sleep(1) self._show_channels() def call_channel(self, index): if index < 0 or index >= len(self.channels): print("Invalid channel index") return channel = self.channels[index] dest_hash_hex = channel["hash"] dest_hash = bytes.fromhex(dest_hash_hex) print( f"\nConnecting to channel {index + 1}: " f"{dest_hash_hex[:16]}..." ) if not RNS.Transport.has_path(dest_hash): RNS.Transport.request_path(dest_hash) print(" Requesting path...", end="", flush=True) for _ in range(100): if RNS.Transport.has_path(dest_hash): break time.sleep(0.1) print() if not RNS.Transport.has_path(dest_hash): print(" Could not find path to channel") return hops = RNS.Transport.hops_to(dest_hash) print(f" Path found ({hops} hop{'s' if hops != 1 else ''})") identity = RNS.Identity.recall(dest_hash) if identity is None: print(" Could not discover channel identity") return self._init_telephone() self.state = self.STATE_CALLING self.telephone.call(identity) def show_channels(self): if not self.channels: print("No channels discovered. Run 'discover' first.") return print( f"\n{Terminal.BOLD}" f"Available channels ({len(self.channels)}):" f"{Terminal.END}" ) print(f" {'#':<4} {'Destination Hash':<48} {'Relay':<16}") print(f" {'-'*4} {'-'*48} {'-'*16}") for i, ch in enumerate(self.channels): relay_short = ( ch["relay"][:14] + ".." if len(ch["relay"]) > 16 else ch["relay"] ) print(f" {i + 1:<4} {ch['hash']:<48} {relay_short}") def show_help(self): print("") print(f"{Terminal.BOLD}Commands:{Terminal.END}") print(f" Call the channel at that number") print(f" d Discover channels (query relays)") print(f" l List channels") print(f" h / ? Show this help") print(f" q Quit") def run(self): signal.signal(signal.SIGINT, self._sigint_handler) signal.signal(signal.SIGTERM, self._sigterm_handler) self.should_run = True if self.board: self._run_gui() else: self._run_cli() def _run_gui(self): self._flash_screen() self.discover(timeout=15) self.selected_index = 0 self.board.set_rgb(0, 80, 255) self._show_channels() while self.should_run: if self.state == self.STATE_CALLING: self._show_status("Calling...", f"Channel {self.selected_index + 1}", accent=(255, 180, 0)) time.sleep(0.5) continue time.sleep(0.1) def _run_cli(self): print( f"\n{Terminal.BOLD}" f"LXST Channel Client" f"{Terminal.END}" ) print(f" Identity: {RNS.prettyhexrep(self.identity.hash)}") print(f" Whitelist: {self.whitelist_path}") print() self.discover(timeout=15) self.show_channels() self.show_help() while self.should_run: if self.state == self.STATE_IN_CALL: try: input(" In call, press Enter to hang up > ") self.telephone.hangup() self.state = self.STATE_READY except (EOFError, KeyboardInterrupt): break continue if self.state == self.STATE_CALLING: time.sleep(0.5) continue if self.state == self.STATE_DISCOVERING: time.sleep(0.5) continue try: inp = input("> ").strip().lower() except (EOFError, KeyboardInterrupt): print() break if not inp: continue if inp == "q": break elif inp in ("h", "?"): self.show_help() elif inp == "d": self.discover() self.show_channels() elif inp == "l": self.show_channels() else: try: idx = int(inp) - 1 if 0 <= idx < len(self.channels): self.call_channel(idx) else: print( f"Invalid index. " f"Enter a number between 1 and {len(self.channels)}" ) except ValueError: print(f"Unknown command: {inp}. Type 'h' for help.") def cleanup(self): if self.telephone and self.state == self.STATE_IN_CALL: self.telephone.hangup() self.should_run = False if self.gestures: self.gestures.stop() if self.board: self.board.cleanup() def _sigint_handler(self, sig, frame): self.cleanup() sys.exit(0) def _sigterm_handler(self, sig, frame): self.cleanup() sys.exit(0) def main(): parser = argparse.ArgumentParser(description="LXST Channel Client") parser.add_argument( "--config", "-c", default=None, help="config directory", ) parser.add_argument( "--rnsconfig", default=None, help="path to alternative Reticulum config directory", ) parser.add_argument( "--whitelist", "-w", default=None, help="path to bridge_whitelist.json", ) parser.add_argument( "--version", action="version", version="lxst-client 0.1.0", ) parser.add_argument("-v", "--verbose", action="count", default=0) parser.add_argument( "--announce_interval", default=0, help="seconds between LXMF destination announces (0 = announce once at startup, default)", type=int, ) parser.add_argument( "--whisplay", "-W", action="store_true", help="enable Whisplay HAT GUI mode", ) args = parser.parse_args() client = ChannelClient( configdir=args.config, rnsconfigdir=args.rnsconfig, whitelist_path=args.whitelist, verbosity=args.verbose, announce_interval=args.announce_interval, use_whisplay=args.whisplay, ) try: client.run() except KeyboardInterrupt: client.cleanup() print() if __name__ == "__main__": main()