#!/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 PIL import Image, ImageDraw, ImageFont from LXST.Primitives.Telephony import Telephone from whisplay_interface import WhisplayInterface 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.selected_index = 0 self._wm8960_device = None self._ptt_enabled = True self._ptt_active = False self.whisplay_interface = None self._gesture_thread = None if use_whisplay: self._init_whisplay() 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): self.whisplay_interface = WhisplayInterface() self.whisplay_interface.create_board( app_id="lxst-client", display_name="LXST Client", icon="LC", use_daemon_default_log=True, ) self.whisplay_interface.board.set_backlight(70) # self.whisplay_interface.board.on_exit_request(self._on_exit_request) # self.whisplay_interface.board.on_focus_revoked(self._on_focus_revoked) 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.whisplay_interface.board.LCD_WIDTH, self.whisplay_interface.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.whisplay_interface.board is None: return fb = self._render("LXST Client", ["Channels loading..."], footer="Please wait") self.whisplay_interface.board.draw_image(0, 0, self.whisplay_interface.board.LCD_WIDTH, self.whisplay_interface.board.LCD_HEIGHT, fb) def _show_channels(self): if self.whisplay_interface.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.whisplay_interface.board.draw_image(0, 0, self.whisplay_interface.board.LCD_WIDTH, self.whisplay_interface.board.LCD_HEIGHT, fb) def _show_in_call(self): if self.whisplay_interface.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.whisplay_interface.board.draw_image(0, 0, self.whisplay_interface.board.LCD_WIDTH, self.whisplay_interface.board.LCD_HEIGHT, fb) def _show_status(self, title, detail="", accent=(255, 180, 60)): if self.whisplay_interface.board is None: return lines = [detail] if detail else [] fb = self._render(title, lines, accent=accent) self.whisplay_interface.board.draw_image(0, 0, self.whisplay_interface.board.LCD_WIDTH, self.whisplay_interface.board.LCD_HEIGHT, fb) def _gesture_loop(self): while self.should_run: self.whisplay_interface.update_gestures() # Double-press (highest priority) if self.whisplay_interface.was_double_pressed and self.state == self.STATE_IN_CALL: self.telephone.hangup() time.sleep(0.02) continue # STATE_READY: release actions if self.state == self.STATE_READY and self.whisplay_interface.was_pressed: if self.whisplay_interface.was_long_pressed and self.channels: self.whisplay_interface.board.set_rgb(255, 180, 0) self.call_channel(self.selected_index) elif self.channels: self.selected_index = (self.selected_index + 1) % len(self.channels) self._show_channels() # STATE_IN_CALL: hold-to-talk + release actions if self.state == self.STATE_IN_CALL: if self.whisplay_interface.is_pressed and self._ptt_enabled and not self._ptt_active: self._ptt_active = True self.telephone.unmute_transmit() if self.whisplay_interface.board: self.whisplay_interface.board.set_rgb(255, 255, 0) if self.whisplay_interface.was_pressed: if self._ptt_enabled: if not self.whisplay_interface.was_long_pressed: self._ptt_active = False self._ptt_enabled = False if self.whisplay_interface.board: self._show_in_call() else: self._ptt_active = False self.telephone.mute_transmit() if self.whisplay_interface.board: self.whisplay_interface.board.set_rgb(0, 255, 80) elif not self.whisplay_interface.was_long_pressed: self._ptt_enabled = True self.telephone.mute_transmit() if self.whisplay_interface.board: self._show_in_call() time.sleep(0.02) def _on_exit_request(self, payload=None): self.cleanup() if self.whisplay_interface.board: self.whisplay_interface.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.whisplay_interface.board: self.whisplay_interface.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 was_in_call: print(f"\nCall ended") elif was_calling: print(f"\nCall could not be connected") if self.whisplay_interface.board: self.whisplay_interface.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.whisplay_interface.board: self._run_gui() else: self._run_cli() def _run_gui(self): self._flash_screen() self.discover(timeout=15) self.selected_index = 0 self.whisplay_interface.board.set_rgb(0, 80, 255) self._show_channels() self._gesture_thread = threading.Thread(target=self._gesture_loop, daemon=True) self._gesture_thread.start() 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._gesture_thread: self._gesture_thread.join(timeout=1) if self.whisplay_interface.board: self.whisplay_interface.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()