#!/usr/bin/env python3 """ lxst-relay - Group voice channel bridge for Reticulum Telephony with NomadNet LXMF bot for channel management. Allows multiple users to call into a shared voice channel. Users can create new channels by messaging the bot via LXMF with ``/create_channel ``. Each channel is its own Reticulum node with a unique identity. """ import json import RNS import LXMF import os import time import signal as signal_module import threading import numpy as np import argparse from LXST import Pipeline from LXST.Network import SignallingReceiver, Packetizer, LinkSource from LXST.Sinks import LocalSink from LXST.Sources import LocalSource from LXST.Codecs import Codec2 from LXST.Codecs.Codec import Codec, CodecError from LXST.Primitives.Telephony import Signalling APP_NAME = "lxst" PRIMITIVE_NAME = "telephony" FIELD_SIGNALLING = 0x00 FIELD_FRAMES = 0x01 # ====================================================================== # Audio pipeline components (unchanged) # ====================================================================== class FrameCaptureSink(LocalSink): """Stores the latest decoded float32 audio frame from a participant.""" def __init__(self, participant): self.participant = participant self.samplerate = 48000 self.channels = 1 def handle_frame(self, frame, source, decoded=True): with self.participant.frame_lock: self.participant.latest_frame = frame class ConferenceSource(LocalSource): """A source that periodically outputs the mixed audio of all channel participants except the one identified by *exclude_link_id*.""" def __init__(self, channel, exclude_link_id, frame_time_ms=60): self.channel = channel self.exclude_link_id = exclude_link_id self.frame_time_ms = frame_time_ms self.frame_time = frame_time_ms / 1000.0 self.should_run = False self._sink = None self._codec = None self.samplerate = 48000 self.channels = 1 self.mixer_thread = None @property def sink(self): return self._sink @sink.setter def sink(self, sink): self._sink = sink @property def codec(self): return self._codec @codec.setter def codec(self, codec): if codec is None: self._codec = None elif not issubclass(type(codec), Codec): raise CodecError(f"Invalid codec specified for {self}") else: self._codec = codec def start(self): if not self.should_run: self.should_run = True self.mixer_thread = threading.Thread(target=self._run, daemon=True) self.mixer_thread.start() def stop(self): self.should_run = False def _run(self): while self.should_run: loop_start = time.time() if self._sink: mixed = self.channel.get_mixed_frame( exclude=self.exclude_link_id ) if mixed is not None and self._sink.can_receive(): if self._codec: encoded = self._codec.encode(mixed) self._sink.handle_frame(encoded, self) elapsed = time.time() - loop_start sleep_time = max(0, self.frame_time - elapsed) time.sleep(sleep_time) class Participant: """A remote caller connected to a channel.""" def __init__(self, link, identity): self.link = link self.identity = identity self.identity_hash = identity.hash self.link_source = None self.packetizer = None self.conf_source = None self.conf_pipeline = None self.latest_frame = None self.frame_lock = threading.Lock() self.joined_at = time.time() self.ended = False class VoiceChannel: """A group voice channel that mixes audio from all participants. Each participant receives a mix of every *other* participant (their own audio is excluded so they don't hear themselves). """ def __init__(self, name, frame_time_ms=60): self.name = name self.participants = {} self.frame_time_ms = frame_time_ms self.frame_time = frame_time_ms / 1000.0 self.lock = threading.Lock() def add_participant(self, participant, relay=None): with self.lock: link_id = participant.link.link_id self.participants[link_id] = participant capture_sink = FrameCaptureSink(participant) link_source = LinkSource( link=participant.link, signalling_receiver=relay, sink=capture_sink, ) participant.link_source = link_source outgoing_codec = Codec2(mode=Codec2.CODEC2_1600) packetizer = Packetizer(destination=participant.link) participant.packetizer = packetizer conf_source = ConferenceSource( channel=self, exclude_link_id=link_id, frame_time_ms=self.frame_time_ms, ) participant.conf_source = conf_source conf_pipeline = Pipeline( source=conf_source, codec=outgoing_codec, sink=packetizer, ) participant.conf_pipeline = conf_pipeline link_source.start() conf_pipeline.start() RNS.log( f"Participant {RNS.prettyhexrep(participant.identity_hash)} " f"joined channel '{self.name}' " f"({len(self.participants)} participant(s))", RNS.LOG_NOTICE, ) def remove_participant(self, link_id): with self.lock: p = self.participants.pop(link_id, None) if p is None: return p.ended = True if p.conf_source: p.conf_source.stop() if p.link_source: p.link_source.stop() if p.link and p.link.status == RNS.Link.ACTIVE: p.link.teardown() count = len(self.participants) RNS.log( f"Participant left channel '{self.name}' " f"({count} participant(s) remaining)", RNS.LOG_NOTICE, ) def get_participant_count(self): with self.lock: return len(self.participants) def get_mixed_frame(self, exclude=None): """Sum the latest frame from every participant except *exclude*.""" with self.lock: frames = [] for lid, p in self.participants.items(): if exclude is not None and lid == exclude: continue with p.frame_lock: if p.latest_frame is not None: frames.append(p.latest_frame) if not frames: return None if len(frames) == 1: return frames[0] min_len = min(f.shape[0] for f in frames) trimmed = [f[:min_len] for f in frames] mixed = np.sum(trimmed, axis=0) mixed = np.clip(mixed, -1.0, 1.0) return mixed # ====================================================================== # Dynamic channel node – each is its own Reticulum identity # ====================================================================== class ChannelNode(SignallingReceiver): """A voice channel that operates as its own Reticulum node. Has an independent ``RNS.Identity``, announces its own ``lxst.telephony`` destination, and manages its own ``VoiceChannel`` instance. """ def __init__(self, identity, name, admins, frame_time_ms=60): super().__init__() self.identity = identity self.name = name self.admins = admins # set[bytes] – identity hashes self.frame_time_ms = frame_time_ms self.voice_channel = VoiceChannel(name=name, frame_time_ms=frame_time_ms) self.destination = RNS.Destination( identity, RNS.Destination.IN, RNS.Destination.SINGLE, APP_NAME, PRIMITIVE_NAME, ) self.destination.set_proof_strategy(RNS.Destination.PROVE_NONE) self.destination.set_link_established_callback( self._incoming_link_established ) # -- Link lifecycle ------------------------------------------------ def _incoming_link_established(self, link): link.is_incoming = True link.is_outgoing = False link.answered = False link.is_terminating = False link.set_remote_identified_callback(self._caller_identified) link.set_link_closed_callback(self._link_closed) self.signal(Signalling.STATUS_AVAILABLE, link) def _caller_identified(self, link, identity): participant = Participant(link, identity) self.signal(Signalling.STATUS_RINGING, link) time.sleep(0.5) self.signal(Signalling.STATUS_CONNECTING, link) self.voice_channel.add_participant(participant, relay=self) self.signal(Signalling.STATUS_ESTABLISHED, link) RNS.log( f"Call established on channel '{self.name}' with " f"{RNS.prettyhexrep(identity.hash)}, " f"{self.voice_channel.get_participant_count()} participant(s)", RNS.LOG_NOTICE, ) def _link_closed(self, link): self.voice_channel.remove_participant(link.link_id) # -- SignallingReceiver ------------------------------------------- def signalling_received(self, signals, source): for signal in signals: if signal >= Signalling.PREFERRED_PROFILE: profile = signal - Signalling.PREFERRED_PROFILE RNS.log( f"Channel '{self.name}' got preferred profile " f"0x{profile:02x}", RNS.LOG_DEBUG, ) # -- Public API --------------------------------------------------- def announce(self): self.destination.announce() def cleanup(self): for p in list(self.voice_channel.participants.values()): p.ended = True if p.conf_source: p.conf_source.stop() if p.link_source: p.link_source.stop() if p.link and p.link.status == RNS.Link.ACTIVE: p.link.teardown() self.voice_channel.participants.clear() # ====================================================================== # Channel manager – persists channels to disk # ====================================================================== class ChannelManager: """Creates, loads, and tracks ``ChannelNode`` instances.""" def __init__(self, storage_dir): self.channels_dir = os.path.join(storage_dir, "channels") os.makedirs(self.channels_dir, exist_ok=True) self.channels = {} # name -> ChannelNode # -- Creation ------------------------------------------------------ def create_channel(self, name, creator_identity_hash): """Create a new channel with its own identity. Returns ``(ChannelNode, None)`` on success or ``(None, error_message)`` on failure. """ if name in self.channels: return None, f"Channel '{name}' already exists" if not name or "/" in name or ".." in name: return None, "Invalid channel name" chan_dir = os.path.join(self.channels_dir, name) try: os.makedirs(chan_dir, exist_ok=False) except FileExistsError: return None, f"Channel '{name}' already exists" identity = RNS.Identity() identity.to_file(os.path.join(chan_dir, "identity")) admins = {creator_identity_hash} with open(os.path.join(chan_dir, "admins"), "w") as f: f.write(RNS.hexrep(creator_identity_hash, delimit=False) + "\n") node = ChannelNode(identity, name, admins) self.channels[name] = node node.announce() RNS.log( f"Created channel '{name}' with identity " f"{RNS.hexrep(identity.hash, delimit=False)}", RNS.LOG_NOTICE, ) return node, None # -- Persistence --------------------------------------------------- def load_all(self): """Restore channels from disk (called at startup).""" if not os.path.isdir(self.channels_dir): return for entry in sorted(os.listdir(self.channels_dir)): chan_dir = os.path.join(self.channels_dir, entry) if not os.path.isdir(chan_dir): continue identity_path = os.path.join(chan_dir, "identity") admin_path = os.path.join(chan_dir, "admins") if not os.path.isfile(identity_path): continue identity = RNS.Identity.from_file(identity_path) if identity is None: RNS.log( f"Could not load identity for channel '{entry}', " f"skipping", RNS.LOG_ERROR, ) continue admins = set() if os.path.isfile(admin_path): with open(admin_path) as f: for line in f: line = line.strip() if line: try: admins.add(bytes.fromhex(line)) except ValueError: pass node = ChannelNode(identity, entry, admins) self.channels[entry] = node node.announce() RNS.log( f"Restored channel '{entry}' " f"({RNS.hexrep(identity.hash, delimit=False)})", RNS.LOG_NOTICE, ) # -- Queries ------------------------------------------------------- def is_admin(self, channel_name, identity_hash): node = self.channels.get(channel_name) if node is None: return False return identity_hash in node.admins def is_admin_of_any(self, identity_hash): for node in self.channels.values(): if identity_hash in node.admins: return True return False # -- Cleanup ------------------------------------------------------- def cleanup_all(self): for node in self.channels.values(): node.cleanup() self.channels.clear() def announce_all(self): for node in self.channels.values(): node.announce() # ====================================================================== # Main relay – telephony + LXMF bot # ====================================================================== class LXSTRelay(SignallingReceiver): """Conference bridge with LXMF bot for channel management. Registers two destinations on the Reticulum network: * ``lxst.telephony`` – the default / lobby channel that anyone can call directly. * ``lxmf.delivery`` – an LXMF bot that responds to ``/create_channel ``, creating a fresh Reticulum node for each channel and assigning the creator as admin. """ def __init__( self, configdir=None, rnsconfigdir=None, channel_name="default", verbosity=0, announce_interval=0, ): super().__init__() self.configdir = configdir self.channel_name = channel_name self.should_run = False self.announce_interval = announce_interval if self.configdir is None: if os.path.isdir("/etc/lxst-relay"): self.configdir = "/etc/lxst-relay" else: userdir = RNS.Reticulum.userdir self.configdir = os.path.join(userdir, ".lxst-relay") os.makedirs(self.configdir, exist_ok=True) # ---- Identity ---- 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: RNS.log( "Could not load identity from file, creating new", RNS.LOG_ERROR, ) 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: " f"{RNS.prettyhexrep(self.identity.hash)}", RNS.LOG_NOTICE, ) self.reticulum = RNS.Reticulum( configdir=rnsconfigdir, loglevel=3 + verbosity ) # ---- Default telephony destination (lobby) ---- 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) self.destination.set_link_established_callback( self._incoming_link_established ) self.channel = VoiceChannel(name=channel_name) # ---- LXMF bot ---- 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.bot_destination = self.lxmf_router.register_delivery_identity( self.identity, display_name="LXST Channel Bot", ) self.lxmf_router.register_delivery_callback( self._lxmf_delivery_callback ) # ---- Channel manager (dynamic nodes) ---- self.channel_manager = ChannelManager(self.configdir) self.channel_manager.load_all() # ---- Startup announce ---- self.announce() RNS.log( f"Relay lobby: {RNS.prettyhexrep(self.destination.hash)}", RNS.LOG_NOTICE, ) RNS.log( f"Bot (LXMF): {RNS.prettyhexrep(self.bot_destination.hash)}", RNS.LOG_NOTICE, ) if self.channel_manager.channels: for name, node in self.channel_manager.channels.items(): RNS.log( f"Channel '{name}': " f"{RNS.prettyhexrep(node.destination.hash)}", RNS.LOG_NOTICE, ) # ================================================================== # Telephony – lobby channel # ================================================================== def _incoming_link_established(self, link): link.is_incoming = True link.is_outgoing = False link.answered = False link.is_terminating = False link.set_remote_identified_callback(self._caller_identified) link.set_link_closed_callback(self._link_closed) self.signal(Signalling.STATUS_AVAILABLE, link) RNS.log("Incoming link (lobby), awaiting identification", RNS.LOG_DEBUG) def _caller_identified(self, link, identity): RNS.log( f"Caller identified in lobby: " f"{RNS.prettyhexrep(identity.hash)}", RNS.LOG_NOTICE, ) participant = Participant(link, identity) self.signal(Signalling.STATUS_RINGING, link) time.sleep(0.5) self.signal(Signalling.STATUS_CONNECTING, link) self.channel.add_participant(participant, relay=self) self.signal(Signalling.STATUS_ESTABLISHED, link) RNS.log( f"Call established in lobby with " f"{RNS.prettyhexrep(identity.hash)}, " f"{self.channel.get_participant_count()} participant(s)", RNS.LOG_NOTICE, ) def _link_closed(self, link): RNS.log(f"Link closed (lobby): {link}", RNS.LOG_DEBUG) self.channel.remove_participant(link.link_id) # ================================================================== # LXMF bot # ================================================================== def _lxmf_delivery_callback(self, message): """Called on every incoming LXMF message addressed to the bot.""" content = message.content_as_string().strip() sender_hash = message.source_hash RNS.log( f"Bot received from " f"{RNS.hexrep(sender_hash, delimit=False)}: {content}", RNS.LOG_NOTICE, ) if not content: return if content.startswith("/create_channel"): self._cmd_create_channel(message, content, sender_hash) elif content.startswith("/channels_json"): self._cmd_list_channels_as_json(message) elif content.startswith("/channels"): self._cmd_list_channels(message) else: self._bot_reply( message, "Unknown command.\n" "Available: /create_channel , /channels", ) # -- Commands ------------------------------------------------------ def _cmd_create_channel(self, message, content, sender_hash): parts = content.split(maxsplit=1) if len(parts) < 2 or not parts[1].strip(): self._bot_reply( message, "Usage: /create_channel ", ) return channel_name = parts[1].strip() node, error = self.channel_manager.create_channel( channel_name, sender_hash ) if error: self._bot_reply(message, error) return hash_hex = RNS.hexrep(node.destination.hash, delimit=False) self._bot_reply( message, f"Channel '{channel_name}' created!\n\n" f"Call identity hash:\n{hash_hex}\n\n" f"Share this hash for others to join.\n" f"You are the admin of this channel.", ) def _cmd_list_channels(self, message): if not self.channel_manager.channels: self._bot_reply(message, "No channels created yet.") return lines = ["Active channels:"] for name, node in self.channel_manager.channels.items(): hash_hex = RNS.hexrep(node.destination.hash, delimit=False) lines.append(f" {name} → {hash_hex}") self._bot_reply(message, "\n".join(lines)) def _cmd_list_channels_as_json(self, message): if not self.channel_manager.channels: self._bot_reply(message, "No channels created yet.") return channels = [[name, RNS.hexrep(node.destination.hash, delimit=False)] for name, node in self.channel_manager.channels.items()] self._bot_reply(message, json.dumps(channels)) # -- Send reply ---------------------------------------------------- def _bot_reply(self, incoming_msg, text): """Send a reply to the author of *incoming_msg*.""" try: sender_identity = RNS.Identity.recall(incoming_msg.source_hash) except Exception: sender_identity = None if sender_identity is None: RNS.log( "Cannot reply – sender identity not cached", RNS.LOG_ERROR, ) return dest = RNS.Destination( sender_identity, RNS.Destination.OUT, RNS.Destination.SINGLE, "lxmf", "delivery", ) reply = LXMF.LXMessage( dest, self.bot_destination, text, desired_method=LXMF.LXMessage.DIRECT, ) self.lxmf_router.handle_outbound(reply) # ================================================================== # SignallingReceiver (lobby) # ================================================================== def signalling_received(self, signals, source): for signal in signals: if signal >= Signalling.PREFERRED_PROFILE: profile = signal - Signalling.PREFERRED_PROFILE RNS.log( f"Lobby got preferred profile: 0x{profile:02x}", RNS.LOG_DEBUG, ) else: RNS.log( f"Lobby got signal: 0x{signal:02x}", RNS.LOG_DEBUG, ) # ================================================================== # Public API # ================================================================== def announce(self): self.destination.announce() self.bot_destination.announce() self.channel_manager.announce_all() RNS.log("Relay announced on network", RNS.LOG_NOTICE) def _announce_loop(self): while self.should_run: time.sleep(self.announce_interval) self.announce() def start(self): if self.should_run: return signal_module.signal(signal_module.SIGINT, self._sigint_handler) signal_module.signal(signal_module.SIGTERM, self._sigterm_handler) self.should_run = True RNS.log( f"LXST Relay ready on channel '{self.channel_name}'", RNS.LOG_NOTICE, ) RNS.log( f"Identity: {RNS.prettyhexrep(self.identity.hash)}", RNS.LOG_NOTICE, ) RNS.log( f"Bot at: {RNS.prettyhexrep(self.bot_destination.hash)}", RNS.LOG_NOTICE, ) RNS.log( "Message the bot (via NomadNet / LXMF) with " "/create_channel to create a new channel", RNS.LOG_NOTICE, ) RNS.log("Waiting for calls and messages...", RNS.LOG_NOTICE) if self.announce_interval > 0: threading.Thread( target=self._announce_loop, daemon=True, ).start() while self.should_run: time.sleep(1) def stop(self): self.should_run = False self.channel_manager.cleanup_all() RNS.log("Relay stopped.", RNS.LOG_NOTICE) # ================================================================== # Signal handlers # ================================================================== def _sigint_handler(self, sig, frame): self.stop() def _sigterm_handler(self, sig, frame): self.stop() # ====================================================================== # Entry point # ====================================================================== def main(): parser = argparse.ArgumentParser( description=( "LXST Relay – Group voice channel bridge for Reticulum " "Telephony with NomadNet LXMF bot" ) ) parser.add_argument( "--config", "-c", default=None, help="path to configuration directory", type=str, ) parser.add_argument( "--rnsconfig", default=None, help="path to alternative Reticulum config directory", type=str, ) parser.add_argument( "--channel", "-n", default="default", help="lobby channel name (default: 'default')", type=str, ) parser.add_argument( "--version", action="version", version="lxst-relay 0.2.0", ) parser.add_argument( "-v", "--verbose", action="count", default=0 ) parser.add_argument( "--announce_interval", default=0, help="seconds between announces (0 = announce once at startup, default)", type=int, ) args = parser.parse_args() relay = LXSTRelay( configdir=args.config, rnsconfigdir=args.rnsconfig, channel_name=args.channel, verbosity=args.verbose, announce_interval=args.announce_interval, ) try: relay.start() except KeyboardInterrupt: relay.stop() print("") if __name__ == "__main__": main()