import RNS import time import nomadnet import threading from math import log10, pow from nomadnet.vendor.additional_urwid_widgets.FormWidgets import * from nomadnet.vendor.AsciiChart import AsciiChart from .ReadlineEdit import ReadlineEdit from .Helpers import ClickableIcon from nomadnet.util import write_config_atomic ### GYLPHS ### INTERFACE_GLYPHS = { # Glyph name # Plain # Unicode # Nerd Font ("NetworkInterfaceType", "(IP)", "\U0001f5a7", "\U000f0200"), ("SerialInterfaceType", "(<->)", "\u2194", "\U000f065c"), ("RNodeInterfaceType", "(R)" , "\u16b1", "\U000f043a"), ("OtherInterfaceType", "(#)" , "\U0001f67e", "\ued95"), } ### HELPER ### PLATFORM_IS_LINUX = False try: PLATFORM_IS_LINUX = (RNS.vendor.platformutils.is_android() or RNS.vendor.platformutils.is_linux()) except Exception: pass def _get_interface_icon(glyphset, iface_type): glyphset_index = 1 # Default to unicode if glyphset == "plain": glyphset_index = 0 # plain elif glyphset == "nerdfont": glyphset_index = 2 # nerdfont type_to_glyph_tuple = { "BackboneInterface": "NetworkInterfaceType", "AutoInterface": "NetworkInterfaceType", "TCPClientInterface": "NetworkInterfaceType", "TCPServerInterface": "NetworkInterfaceType", "UDPInterface": "NetworkInterfaceType", "I2PInterface": "NetworkInterfaceType", "RNodeInterface": "RNodeInterfaceType", "RNodeMultiInterface": "RNodeInterfaceType", "SerialInterface": "SerialInterfaceType", "KISSInterface": "SerialInterfaceType", "AX25KISSInterface": "SerialInterfaceType", "PipeInterface": "OtherInterfaceType" } glyph_tuple_name = type_to_glyph_tuple.get(iface_type, "OtherInterfaceType") for glyph_tuple in INTERFACE_GLYPHS: if glyph_tuple[0] == glyph_tuple_name: return glyph_tuple[glyphset_index + 1] # Fallback return "(#)" if glyphset == "plain" else "\U0001f67e" if glyphset == "unicode" else "\ued95" def format_bytes(bytes_value): units = ['bytes', 'KB', 'MB', 'GB', 'TB'] size = float(bytes_value) unit_index = 0 while size >= 1024.0 and unit_index < len(units) - 1: size /= 1024.0 unit_index += 1 if unit_index == 0: return f"{int(size)} {units[unit_index]}" else: return f"{size:.1f} {units[unit_index]}" def format_announce_rate(hz): try: hz = float(hz or 0) except (TypeError, ValueError): return "-" if hz <= 0: return "-" per_hour = hz * 3600 if per_hour < 100: return "<1/hr" if per_hour < 1 else "%d/hr" % round(per_hour) return "%d/min" % round(hz * 60) def format_speed(bps): try: bps = float(bps or 0) except (TypeError, ValueError): return "-" try: return RNS.prettyspeed(bps) except Exception: return "%d bps" % int(bps) def interface_mode_str(mode): modes = { 1: "Full", # MODE_FULL 2: "Point-to-Point", # MODE_POINT_TO_POINT 3: "Access Point", # MODE_ACCESS_POINT 4: "Roaming", # MODE_ROAMING 5: "Boundary", # MODE_BOUNDARY 6: "Gateway", # MODE_GATEWAY } return modes.get(mode, "Full") def connection_markup(glyphs, is_connected, label=None): icon = glyphs.get("connected" if is_connected else "disconnected", "") if label is None: label = "Connected" if is_connected else "Disconnected" attr = "connected_status" if is_connected else "disconnected_status" sep = " " if icon else "" return (attr, "%s%s%s" % (icon, sep, label)) def _get_cols_rows(): return nomadnet.NomadNetworkApp.get_shared_instance().ui.screen.get_cols_rows() ### PORT FUNCTIONS ### PYSERIAL_AVAILABLE = False # If NomadNet is installed on environments with rnspure instead of rns, pyserial won't be available try: import serial.tools.list_ports PYSERIAL_AVAILABLE = True except ImportError: class DummyPort: def __init__(self, device, description=None, manufacturer=None, hwid=None): self.device = device self.description = description or device self.manufacturer = manufacturer self.hwid = hwid self.vid = None self.pid = None def get_port_info(): if not PYSERIAL_AVAILABLE: return [] try: ports = serial.tools.list_ports.comports() port_info = [] # Ports are sorted into categories for dropdown, priority ports appear first priority_ports = [] # USB, ACM, bluetooth, etc standard_ports = [] # COM, tty/s ports for port in ports: desc = f"{port.device}" if port.description and port.description != port.device: desc += f" ({port.description})" if port.manufacturer: desc += f" - {port.manufacturer}" is_standard = ( port.device.startswith("COM") or # windows "/dev/ttyS" in port.device or # Linux "Serial" in port.description ) port_data = { 'device': port.device, 'description': desc, 'hwid': port.hwid, 'vid': port.vid, 'pid': port.pid, 'is_standard': is_standard } if is_standard: standard_ports.append(port_data) else: priority_ports.append(port_data) priority_ports.sort(key=lambda x: x['device']) standard_ports.sort(key=lambda x: x['device']) return priority_ports + standard_ports except Exception as e: RNS.log(f"error accessing serial ports: {str(e)}", RNS.LOG_ERROR) return [] def get_port_field(): if not PYSERIAL_AVAILABLE: return { "config_key": "port", "type": "edit", "label": "Port: ", "default": "", "placeholder": "/dev/ttyUSB0 or COM port (pyserial not installed)", "validation": ["required"], "transform": lambda x: x.strip() } port_info = get_port_info() if len(port_info) > 1: options = [p['description'] for p in port_info] device_map = {p['description']: p['device'] for p in port_info} return { "config_key": "port", "type": "dropdown", "label": "Port: ", "options": options, "default": options[0] if options else "", "validation": ["required"], "transform": lambda x: device_map[x] } else: # single or no ports - use text field default = port_info[0]['device'] if port_info else "" placeholder = "/dev/ttyXXX (or COM port on Windows)" return { "config_key": "port", "type": "edit", "label": "Port: ", "default": default, "placeholder": placeholder, "validation": ["required"], "transform": lambda x: x.strip() } ### RNODE #### def calculate_rnode_parameters(bandwidth, spreading_factor, coding_rate, noise_floor=6, antenna_gain=0, transmit_power=17): crn = { 5: 1, 6: 2, 7: 3, 8: 4, } coding_rate_n = crn.get(coding_rate, 1) sfn = { 5: -2.5, 6: -5, 7: -7.5, 8: -10, 9: -12.5, 10: -15, 11: -17.5, 12: -20 } data_rate = spreading_factor * ( (4 / (4 + coding_rate_n)) / (pow(2, spreading_factor) / (bandwidth / 1000))) * 1000 sensitivity = -174 + 10 * log10(bandwidth) + noise_floor + (sfn.get(spreading_factor, 0)) if bandwidth == 203125 or bandwidth == 406250 or bandwidth > 500000: sensitivity = -165.6 + 10 * log10(bandwidth) + noise_floor + (sfn.get(spreading_factor, 0)) link_budget = (transmit_power - sensitivity) + antenna_gain if data_rate < 1000: data_rate_str = f"{data_rate:.0f} bps" else: data_rate_str = f"{(data_rate / 1000):.2f} kbps" return { "data_rate": data_rate_str, "link_budget": f"{link_budget:.1f} dB", "sensitivity": f"{sensitivity:.1f} dBm", "raw_data_rate": data_rate, "raw_link_budget": link_budget, "raw_sensitivity": sensitivity } class RNodeCalculator(urwid.WidgetWrap): def __init__(self, parent_view): self.parent_view = parent_view self.update_alarm = None self.data_rate_widget = urwid.Text("Data Rate: Calculating...") self.link_budget_widget = urwid.Text("Link Budget: Calculating...") self.sensitivity_widget = urwid.Text("Sensitivity: Calculating...") self.noise_floor_edit = ReadlineEdit("", "0") self.antenna_gain_edit = ReadlineEdit("", "0") layout = urwid.Pile([ urwid.Divider("-"), urwid.Columns([ (28, urwid.Text(("key", "Enter Noise Floor (dB): "), align="right")), self.noise_floor_edit ]), urwid.Columns([ (28, urwid.Text(("key", "Enter Antenna Gain (dBi): "), align="right")), self.antenna_gain_edit ]), urwid.Divider(), urwid.Text(("connected_status", "On-Air Calculations:"), align="left"), self.data_rate_widget, self.link_budget_widget, self.sensitivity_widget, urwid.Divider(), urwid.Text([ "These calculations will update as you change RNode parameters" ]) ]) super().__init__(layout) self.connect_all_field_signals() self.update_calculation() def connect_all_field_signals(self): urwid.connect_signal(self.noise_floor_edit, 'change', self._queue_update) urwid.connect_signal(self.antenna_gain_edit, 'change', self._queue_update) rnode_fields = ['bandwidth', 'spreadingfactor', 'codingrate', 'txpower'] for field_name in rnode_fields: if field_name in self.parent_view.fields: field_widget = self.parent_view.fields[field_name]['widget'] if hasattr(field_widget, 'edit_text'): urwid.connect_signal(field_widget, 'change', self._queue_update) elif hasattr(field_widget, '_emit') and 'change' in getattr(field_widget, 'signals', []): urwid.connect_signal(field_widget, 'change', self._queue_update) def _queue_update(self, widget, new_text): if self.update_alarm is not None: try: self.parent_view.parent.app.ui.loop.remove_alarm(self.update_alarm) except: pass self.update_alarm = self.parent_view.parent.app.ui.loop.set_alarm_in( 0.3, self._delayed_update) def _delayed_update(self, loop, user_data): self.update_alarm = None self.update_calculation() def update_calculation(self): try: try: bandwidth_widget = self.parent_view.fields.get('bandwidth', {}).get('widget') bandwidth = int(bandwidth_widget.get_value()) if bandwidth_widget else 125000 except (ValueError, AttributeError): bandwidth = 125000 try: sf_widget = self.parent_view.fields.get('spreadingfactor', {}).get('widget') spreading_factor = int(sf_widget.get_value()) if sf_widget else 7 except (ValueError, AttributeError): spreading_factor = 7 try: cr_widget = self.parent_view.fields.get('codingrate', {}).get('widget') coding_rate = int(cr_widget.get_value()) if cr_widget else 5 if isinstance(coding_rate, str) and ":" in coding_rate: coding_rate = int(coding_rate.split(":")[1]) except (ValueError, AttributeError): coding_rate = 5 try: txpower_widget = self.parent_view.fields.get('txpower', {}).get('widget') if hasattr(txpower_widget, 'edit_text'): txpower_text = txpower_widget.edit_text.strip() txpower = int(txpower_text) if txpower_text else 17 else: txpower = int(txpower_widget.get_value()) if txpower_widget else 17 except (ValueError, AttributeError): txpower = 17 try: noise_floor_text = self.noise_floor_edit.edit_text.strip() noise_floor = int(noise_floor_text) if noise_floor_text else 0 except (ValueError, AttributeError): noise_floor = 0 try: antenna_gain_text = self.antenna_gain_edit.edit_text.strip() antenna_gain = int(antenna_gain_text) if antenna_gain_text else 0 except (ValueError, AttributeError): antenna_gain = 0 result = calculate_rnode_parameters( bandwidth=bandwidth, spreading_factor=spreading_factor, coding_rate=coding_rate, noise_floor=noise_floor, antenna_gain=antenna_gain, transmit_power=txpower ) self.data_rate_widget.set_text(f"Data Rate: {result['data_rate']}") self.link_budget_widget.set_text(f"Link Budget: {result['link_budget']}") self.sensitivity_widget.set_text(f"Sensitivity: {result['sensitivity']}") except (ValueError, KeyError, TypeError) as e: self.data_rate_widget.set_text(f"Data Rate: Waiting for parameters...") self.link_budget_widget.set_text(f"Link Budget: Waiting for valid parameters...") self.sensitivity_widget.set_text(f"Sensitivity: Waiting for parameters...") ### INTERFACE FIELDS ### COMMON_INTERFACE_OPTIONS = [ { "config_key": "network_name", "type": "edit", "label": "Virtual Network Name: ", "placeholder": "Optional virtual network name", "default": "", "validation": [], "transform": lambda x: x.strip() }, { "config_key": "passphrase", "type": "edit", "label": "IFAC Passphrase: ", "placeholder": "IFAC authentication passphrase", "default": "", "validation": [], "transform": lambda x: x.strip() }, { "config_key": "ifac_size", "type": "edit", "label": "IFAC Size: ", "placeholder": "8 - 512", "default": "", "validation": ['number'], "transform": lambda x: x.strip() }, { "config_key": "bitrate", "type": "edit", "label": "Inferred Bitrate: ", "placeholder": "Automatically determined", "default": "", "validation": ['number'], "transform": lambda x: x.strip() }, ] INTERFACE_FIELDS = { "BackboneInterface": [ { "config_key": "listen_on", "type": "edit", "label": "Listen On: ", "default": "", "placeholder": "e.g., 0.0.0.0", "transform": lambda x: x.strip() }, { "config_key": "port", "type": "edit", "label": "Port: ", "default": "", "placeholder": "e.g., 4242", "validation": ["number"], "transform": lambda x: int(x.strip()) if x.strip() else None }, { "config_key": "device", "type": "edit", "label": "Device: ", "default": "", "placeholder": "e.g., eth0", "transform": lambda x: x.strip() }, { "config_key": "remote", "type": "edit", "label": "Remote: ", "default": "", "placeholder": "e.g., a remote TCPServerInterface location", "transform": lambda x: x.strip() }, { "config_key": "target_host", "type": "edit", "label": "Target Host: ", "default": "", "placeholder": "e.g., 201:5d78:af73:5caf:a4de:a79f:3278:71e5", "transform": lambda x: x.strip() }, { "config_key": "port", "type": "edit", "label": "Target Port: ", "default": "", "placeholder": "e.g., 4242", "validation": ["number"], "transform": lambda x: int(x.strip()) if x.strip() else None }, { "config_key": "prefer_ipv6", "type": "checkbox", "label": "", "default": False, "validation": [], "transform": lambda x: bool(x) }, ], "AutoInterface": [ { }, { "additional_options": [ { "config_key": "devices", "type": "multilist", "label": "Devices: ", "validation": [], "transform": lambda x: ",".join(x) }, { "config_key": "ignored_devices", "type": "multilist", "label": "Ignored Devices: ", "validation": [], "transform": lambda x: ",".join(x) }, { "config_key": "group_id", "type": "edit", "label": "Group ID: ", "default": "", "placeholder": "e.g., my_custom_network", "validation": [], "transform": lambda x: x.strip() }, { "config_key": "discovery_scope", "type": "dropdown", "label": "Discovery Scope: ", "options": ["None", "link", "admin", "site", "organisation", "global"], "default": "None", "validation": [], "transform": lambda x: "" if x == "None" else x.strip() } ] }, ], "I2PInterface": [ { "config_key": "peers", "type": "multilist", "label": "Peers: ", "placeholder": "", "validation": ["required"], "transform": lambda x: ",".join(x) } ], "TCPServerInterface": [ { "config_key": "listen_ip", "type": "edit", "label": "Listen IP: ", "default": "", "placeholder": "e.g., 0.0.0.0", "validation": ["required"], "transform": lambda x: x.strip() }, { "config_key": "listen_port", "type": "edit", "label": "Listen Port: ", "default": "", "placeholder": "e.g., 4242", "validation": ["number"], "transform": lambda x: int(x.strip()) if x.strip() else None }, { "additional_options": [ { "config_key": "prefer_ipv6", "type": "checkbox", "label": "Prefer IPv6?", "default": False, "validation": [], "transform": lambda x: bool(x) }, { "config_key": "i2p_tunneled", "type": "checkbox", "label": "I2P Tunneled?", "default": False, "validation": [], "transform": lambda x: bool(x) }, { "config_key": "device", "type": "edit", "label": "Device: ", "placeholder": "A specific network device to listen on - e.g. eth0", "default": "", "validation": [], "transform": lambda x: x.strip() if x.strip() else None }, { "config_key": "port", "type": "edit", "label": "Port: ", "default": "", "placeholder": "e.g., 4242", "validation": ["number"], "transform": lambda x: int(x.strip()) if x.strip() else None }, ] } ], "TCPClientInterface": [ { "config_key": "target_host", "type": "edit", "label": "Target Host: ", "default": "", "placeholder": "e.g., 127.0.0.1", "validation": ["required"], "transform": lambda x: x.strip() }, { "config_key": "target_port", "type": "edit", "label": "Target Port: ", "default": "", "placeholder": "e.g., 4242", "validation": ["required", "number"], "transform": lambda x: int(x.strip()) if x.strip() else None }, { "additional_options": [ { "config_key": "i2p_tunneled", "type": "checkbox", "label": "I2P Tunneled?", "default": False, "validation": [], "transform": lambda x: bool(x) }, { "config_key": "kiss_framing", "type": "checkbox", "label": "KISS Framing?", "default": False, "validation": [], "transform": lambda x: bool(x) } ] } ], "UDPInterface": [ { "config_key": "listen_ip", "type": "edit", "label": "Listen IP: ", "default": "", "placeholder": "e.g., 0.0.0.0", "validation": ["required"], "transform": lambda x: x.strip() }, { "config_key": "listen_port", "type": "edit", "label": "Listen Port: ", "default": "", "placeholder": "e.g., 4242", "validation": ["number"], "transform": lambda x: int(x.strip()) if x.strip() else None }, { "config_key": "forward_ip", "type": "edit", "label": "Forward IP: ", "default": "", "placeholder": "e.g., 255.255.255.255", "validation": ["required"], "transform": lambda x: x.strip() }, { "config_key": "forward_port", "type": "edit", "label": "Forward Port: ", "default": "", "placeholder": "e.g., 4242", "validation": ["required", "number"], "transform": lambda x: int(x.strip()) if x.strip() else None }, { "additional_options": [ { "config_key": "device", "type": "edit", "label": "Device: ", "placeholder": "A specific network device to listen on - e.g. eth0", "default": "", "validation": [], "transform": lambda x: x.strip() }, { "config_key": "port", "type": "edit", "label": "Port: ", "default": "", "placeholder": "e.g., 4242", "validation": ["number"], "transform": lambda x: int(x.strip()) if x.strip() else None }, ] } ], "RNodeInterface": [ get_port_field(), { "config_key": "frequency", "type": "edit", "label": "Frequency (MHz): ", "default": "", "placeholder": "868.5", "validation": ["required", "float"], "transform": lambda x: int(float(x.strip()) * 1000000) if x.strip() else 868500000 }, { "config_key": "txpower", "type": "edit", "label": "Transmit Power (dBm): ", "default": "", "placeholder": "17", "validation": ["required", "number"], "transform": lambda x: int(x.strip()) if x.strip() else 17 }, { "config_key": "bandwidth", "type": "dropdown", "label": "Bandwidth (Hz): ", "options": ["7800", "10400", "15600", "20800", "31250", "41700", "62500", "125000", "250000", "500000", "1625000"], "default": "7800", "validation": ["required"], "transform": lambda x: int(x) }, { "config_key": "spreadingfactor", "type": "dropdown", "label": "Spreading Factor: ", "options": ["7", "8", "9", "10", "11", "12"], "default": "7", "validation": ["required"], "transform": lambda x: int(x) }, { "config_key": "codingrate", "type": "dropdown", "label": "Coding Rate: ", "options": ["4:5", "4:6", "4:7", "4:8"], "default": "4:5", "validation": ["required"], "transform": lambda x: int(x.split(":")[1]) }, { "additional_options": [ { "config_key": "id_callsign", "type": "edit", "label": "Callsign: ", "default": "", "placeholder": "e.g. MYCALL-0", "validation": [""], "transform": lambda x: x.strip() }, { "config_key": "id_interval", "type": "edit", "label": "ID Interval (Seconds): ", "placeholder": "e.g. 600", "default": "", "validation": ['number'], "transform": lambda x: "" if x == "" else int(x) }, { "config_key": "airtime_limit_long", "type": "edit", "label": "Airtime Limit Long (Seconds): ", "placeholder": "e.g. 1.5", "default": "", "validation": ['number'], "transform": lambda x: "" if x == "" else int(x) }, { "config_key": "airtime_limit_short", "type": "edit", "label": "Airtime Limit Short (Seconds): ", "placeholder": "e.g. 33", "default": "", "validation": ['number'], "transform": lambda x: "" if x == "" else int(x) }, ] } ], "RNodeMultiInterface": [ get_port_field(), { "config_key": "subinterfaces", "type": "multitable", "fields": { "frequency": { "label": "Freq (Hz)", "type": "edit", "validation": ["required", "float"], "transform": lambda x: int(x) if x else None }, "bandwidth": { "label": "BW (Hz)", "type": "edit", "options": ["7800", "10400", "15600", "20800", "31250", "41700", "62500", "125000", "250000", "500000", "1625000"], "transform": lambda x: int(x) if x else None }, "txpower": { "label": "TX (dBm)", "type": "edit", "validation": ["required", "number"], "transform": lambda x: int(x) if x else None }, "vport": { "label": "V.Port", "type": "edit", "validation": ["required", "number"], "transform": lambda x: int(x) if x else None }, "spreadingfactor": { "label": "SF", "type": "edit", "transform": lambda x: int(x) if x else None }, "codingrate": { "label": "CR", "type": "edit", "transform": lambda x: int(x) if x else None } }, "validation": ["required"], "transform": lambda x: x }, { "additional_options": [ { "config_key": "id_callsign", "type": "edit", "label": "Callsign: ", "default": "", "placeholder": "e.g. MYCALL-0", "validation": [""], "transform": lambda x: x.strip() }, { "config_key": "id_interval", "type": "edit", "label": "ID Interval (Seconds): ", "placeholder": "e.g. 600", "default": "", "validation": ['number'], "transform": lambda x: "" if x == "" else int(x) } ] } ], "SerialInterface": [ get_port_field(), { "config_key": "speed", "type": "edit", "label": "Speed (bps): ", "default": "", "placeholder": "e.g. 115200", "validation": ["required", "number"], "transform": lambda x: int(x.strip()) }, { "config_key": "databits", "type": "edit", "label": "Databits: ", "default": "", "placeholder": "e.g. 8", "validation": ["required", "number"], "transform": lambda x: int(x.strip()) }, { "config_key": "parity", "type": "edit", "label": "Parity: ", "default": "", "placeholder": "", "validation": ["number"], "transform": lambda x: "" if x == "" else int(x.strip()) }, { "config_key": "stopbits", "type": "edit", "label": "Stopbits: ", "default": "", "placeholder": "e.g. 1", "validation": ["number"], "transform": lambda x: "" if x == "" else int(x.strip()) }, ], "PipeInterface": [ { "config_key": "command", "type": "edit", "label": "Command: ", "default": "", "placeholder": "e.g. netcat -l 5757", "validation": ["required"], "transform": lambda x: x.strip() }, { "config_key": "respawn_delay", "type": "edit", "label": "Respawn Delay (seconds): ", "default": "", "placeholder": "e.g. 5", "validation": ["number"], "transform": lambda x: x.strip() }, ], "KISSInterface": [ get_port_field(), { "config_key": "speed", "type": "edit", "label": "Speed (bps): ", "default": "", "placeholder": "e.g. 115200", "validation": ["required", "number"], "transform": lambda x: int(x.strip()) }, { "config_key": "databits", "type": "edit", "label": "Databits: ", "default": "", "placeholder": "e.g. 8", "validation": ["required", "number"], "transform": lambda x: int(x.strip()) }, { "config_key": "parity", "type": "edit", "label": "Parity: ", "default": "", "placeholder": "", "validation": ["number"], "transform": lambda x: "" if x == "" else int(x.strip()) }, { "config_key": "stopbits", "type": "edit", "label": "Stopbits: ", "default": "", "placeholder": "e.g. 1", "validation": ["number"], "transform": lambda x: "" if x == "" else int(x.strip()) }, { "config_key": "preamble", "type": "edit", "label": "Preamble (miliseconds): ", "default": "", "placeholder": "e.g. 150", "validation": ["required", "number"], "transform": lambda x: "" if x == "" else int(x.strip()) }, { "config_key": "txtail", "type": "edit", "label": "TX Tail (miliseconds): ", "default": "", "placeholder": "e.g. 10", "validation": ["required", "number"], "transform": lambda x: "" if x == "" else int(x.strip()) }, { "config_key": "slottime", "type": "edit", "label": "slottime (miliseconds): ", "default": "", "placeholder": "e.g. 20", "validation": ["required", "number"], "transform": lambda x: "" if x == "" else int(x.strip()) }, { "config_key": "persistence", "type": "edit", "label": "Persistence (miliseconds): ", "default": "", "placeholder": "e.g. 200", "validation": ["required", "number"], "transform": lambda x: "" if x == "" else int(x.strip()) }, { "additional_options": [ { "config_key": "id_callsign", "type": "edit", "label": "ID Callsign: ", "default": "", "placeholder": "e.g. MYCALL-0", "validation": [""], "transform": lambda x: x.strip() }, { "config_key": "id_interval", "type": "edit", "label": "ID Interval (Seconds): ", "placeholder": "e.g. 600", "default": "", "validation": ['number'], "transform": lambda x: "" if x == "" else int(x) }, { "config_key": "flow_control", "type": "checkbox", "label": "Flow Control ", "validation": [], "transform": lambda x: "" if x == "" else bool(x) }, ] } ], "AX25KISSInterface": [ get_port_field(), { "config_key": "callsign", "type": "edit", "label": "Callsign: ", "default": "", "placeholder": "e.g. NO1CLL", "validation": ["required"], "transform": lambda x: x.strip() }, { "config_key": "ssid", "type": "edit", "label": "SSID: ", "default": "", "placeholder": "e.g. 0", "validation": ["required"], "transform": lambda x: x.strip() }, { "config_key": "speed", "type": "edit", "label": "Speed (bps): ", "default": "", "placeholder": "e.g. 115200", "validation": ["required", "number"], "transform": lambda x: int(x.strip()) }, { "config_key": "databits", "type": "edit", "label": "Databits: ", "default": "", "placeholder": "e.g. 8", "validation": ["required", "number"], "transform": lambda x: int(x.strip()) }, { "config_key": "parity", "type": "edit", "label": "Parity: ", "default": "", "placeholder": "", "validation": ["number"], "transform": lambda x: "" if x == "" else int(x.strip()) }, { "config_key": "stopbits", "type": "edit", "label": "Stopbits: ", "default": "", "placeholder": "e.g. 1", "validation": ["number"], "transform": lambda x: "" if x == "" else int(x.strip()) }, { "config_key": "preamble", "type": "edit", "label": "Preamble (miliseconds): ", "default": "", "placeholder": "e.g. 150", "validation": ["required", "number"], "transform": lambda x: "" if x == "" else int(x.strip()) }, { "config_key": "txtail", "type": "edit", "label": "TX Tail (miliseconds): ", "default": "", "placeholder": "e.g. 10", "validation": ["required", "number"], "transform": lambda x: "" if x == "" else int(x.strip()) }, { "config_key": "slottime", "type": "edit", "label": "Slottime (miliseconds): ", "default": "", "placeholder": "e.g. 20", "validation": ["required", "number"], "transform": lambda x: "" if x == "" else int(x.strip()) }, { "config_key": "persistence", "type": "edit", "label": "Persistence (miliseconds): ", "default": "", "placeholder": "e.g. 200", "validation": ["required", "number"], "transform": lambda x: "" if x == "" else int(x.strip()) }, { "additional_options": [ { "config_key": "flow_control", "type": "checkbox", "label": "Flow Control ", "validation": [], "transform": lambda x: "" if x == "" else bool(x) }, ] } ], "CustomInterface": [ { "config_key": "type", "type": "edit", "label": "Interface Type: ", "default": "", "placeholder": "Name of custom interface class", "validation": ["required"], "transform": lambda x: x.strip() }, { "config_key": "custom_parameters", "type": "keyvaluepairs", "label": "Parameters: ", "validation": [], "transform": lambda x: x }, ], "default": [ { }, ] } ### INTERFACE WIDGETS #### class SelectableInterfaceItem(urwid.WidgetWrap): def __init__(self, parent, name, is_connected, is_enabled, iface_type, tx, rx, icon="?", iface_options=None, profiles_label=None, announce_rate="N/A"): self.parent = parent self._selectable = True self.icon = icon self.name = name self.is_connected = is_connected self.is_enabled = is_enabled self.iface_options = iface_options if is_enabled: enabled_txt = ("connected_status", "Enabled") else: enabled_txt = ("disabled_status", "Disabled") connected_txt = connection_markup(parent.g, is_connected) self.selection_txt = urwid.Text(" ") self.title_widget = urwid.Text(("interface_title", f"{icon} {name}")) title_content = urwid.Columns([ (4, self.selection_txt), self.title_widget, ]) self.tx_widget = urwid.Text(("value", format_bytes(tx))) self.rx_widget = urwid.Text(("value", format_bytes(rx))) self.announce_widget = urwid.Text(("value", announce_rate)) self.status_widget = urwid.Text(enabled_txt) self.connection_widget = urwid.Text(connected_txt) if is_enabled: status_row = urwid.Columns([ (10, urwid.Text(("key", "Status: "))), (10, self.status_widget), (3, urwid.Text(" | ")), self.connection_widget, ]) else: status_row = urwid.Columns([ (10, urwid.Text(("key", "Status: "))), self.status_widget, ]) rows = [ status_row, urwid.Columns([ (10, urwid.Text(("key", "Type:"))), urwid.Text(("value", iface_type)), ]), ] if profiles_label is not None: rows.append(urwid.Columns([ (10, urwid.Text(("key", "Profiles:"))), urwid.Text(("value", profiles_label)), ])) if is_enabled: rows.append(urwid.Divider("-")) rows.append(urwid.Columns([ (5, urwid.Text(("key", "TX:"))), (14, self.tx_widget), (5, urwid.Text(("key", "RX:"))), (14, self.rx_widget), (12, urwid.Text(("key", "Announces:"))), self.announce_widget, ])) pile_contents = [title_content] + rows pile = urwid.Pile(pile_contents) padded_body = urwid.Padding(pile, left=2, right=2) box = urwid.LineBox( padded_body, title=None, #todo tlcorner="╭", tline="─", trcorner="╮", lline="│", rline="│", blcorner="╰", bline="─", brcorner="╯" ) super().__init__(box) def update_status_display(self): if self.is_enabled: self.status_widget.set_text(("connected_status", "Enabled")) else: self.status_widget.set_text(("disabled_status", "Disabled")) def selectable(self): return True def render(self, size, focus=False): self.selection_txt.set_text(self.parent.g['selected'] if focus else self.parent.g['unselected']) if focus: self.title_widget.set_text( ("interface_title_selected", f"{self.icon} {self.name}")) else: self.title_widget.set_text(("interface_title", f"{self.icon} {self.name}")) return super().render(size, focus=focus) def keypress(self, size, key): if key == "up": listbox = self.parent.box_adapter._original_widget walker = listbox.body interface_items = [i for i, item in enumerate(walker) if isinstance(item, SelectableInterfaceItem)] if interface_items and walker[listbox.focus_position] is self and \ listbox.focus_position == interface_items[0]: self.parent.app.ui.main_display.frame.focus_position = "header" return None elif key == "enter": self.parent.switch_to_show_interface(self.name) return None return key def update_stats(self, tx, rx, announce_rate=None): self.tx_widget.set_text(("value", format_bytes(tx))) self.rx_widget.set_text(("value", format_bytes(rx))) if announce_rate is not None: self.announce_widget.set_text(("value", announce_rate)) def update_connection(self, is_connected): if is_connected == self.is_connected: return self.is_connected = is_connected self.connection_widget.set_text(connection_markup(self.parent.g, is_connected)) class InterfaceOptionItem(urwid.WidgetWrap): def __init__(self, parent_display, label, value, on_select=None): self.parent_display = parent_display self.label = label self.value = value self.on_select = on_select self._selectable = True text_widget = urwid.Text(label, align="left") super().__init__(urwid.AttrMap(text_widget, "list_normal", focus_map="list_focus")) def selectable(self): return True def _activate(self): if self.on_select is not None: self.on_select() else: self.parent_display.dismiss_dialog() self.parent_display.switch_to_add_interface(self.value) def keypress(self, size, key): if key == "enter": self._activate() return None return super().keypress(size, key) def mouse_event(self, size, event, button, x, y, focus): if button == 1 and urwid.util.is_mouse_press(event): self._activate() return True return False class InterfaceBandwidthChart: def __init__(self, history_length=60, glyphset="unicode"): self.history_length = history_length self.glyphset = glyphset self.rx_rates = [0] * history_length self.tx_rates = [0] * history_length self.prev_rx = None self.prev_tx = None self.prev_time = None self.max_rx_rate = 1 self.max_tx_rate = 1 self.first_update = True self.initialization_complete = False self.stabilization_updates = 2 self.update_count = 0 self.peak_rx_for_display = 0 self.peak_tx_for_display = 0 def update(self, rx_bytes, tx_bytes): current_time = time.time() if self.prev_rx is None or self.first_update: self.prev_rx = rx_bytes self.prev_tx = tx_bytes self.prev_time = current_time self.first_update = False return time_delta = max(0.1, current_time - self.prev_time) rx_delta = max(0, rx_bytes - self.prev_rx) / time_delta tx_delta = max(0, tx_bytes - self.prev_tx) / time_delta self.prev_rx = rx_bytes self.prev_tx = tx_bytes self.prev_time = current_time self.update_count += 1 self.rx_rates.pop(0) self.tx_rates.pop(0) self.rx_rates.append(rx_delta*8) self.tx_rates.append(tx_delta*8) if self.update_count >= self.stabilization_updates: self.initialization_complete = True self.peak_rx_for_display = max(self.peak_rx_for_display, rx_delta) self.peak_tx_for_display = max(self.peak_tx_for_display, tx_delta) current_rx_max = max(self.rx_rates) current_tx_max = max(self.tx_rates) self.max_rx_rate = max(1, current_rx_max) self.max_tx_rate = max(1, current_tx_max) def get_charts(self, height=8): chart = AsciiChart(glyphset=self.glyphset) rx_data = self.rx_rates.copy() tx_data = self.tx_rates.copy() peak_rx = self.peak_rx_for_display if self.initialization_complete else 0 peak_tx = self.peak_tx_for_display if self.initialization_complete else 0 peak_rx_str = RNS.prettyspeed(peak_rx*8) peak_tx_str = RNS.prettyspeed(peak_tx*8) rx_chart = chart.plot( [rx_data], { 'height': height, 'format': RNS.prettyspeed, 'min': 0, 'max': self.max_rx_rate * 1.1, } ) tx_chart = chart.plot( [tx_data], { 'height': height, 'format': RNS.prettyspeed, 'min': 0, 'max': self.max_tx_rate * 1.1, } ) return rx_chart, tx_chart, peak_rx_str, peak_tx_str class ResponsiveChartContainer(urwid.WidgetWrap): def __init__(self, rx_box, tx_box, min_cols_for_horizontal=100): self.rx_box = rx_box self.tx_box = tx_box self.min_cols_for_horizontal = min_cols_for_horizontal self.horizontal_layout = urwid.Columns([ (urwid.WEIGHT, 1, self.rx_box), (urwid.WEIGHT, 1, self.tx_box) ]) self.vertical_layout = urwid.Pile([ self.rx_box, self.tx_box ]) self.layout = urwid.WidgetPlaceholder(self.horizontal_layout) super().__init__(self.layout) def render(self, size, focus=False): maxcol = size[0] if len(size) > 0 else 0 if maxcol >= self.min_cols_for_horizontal and self.layout.original_widget is not self.horizontal_layout: self.layout.original_widget = self.horizontal_layout elif maxcol < self.min_cols_for_horizontal and self.layout.original_widget is not self.vertical_layout: self.layout.original_widget = self.vertical_layout return super().render(size, focus) ### URWID FILLER ### class InterfaceFiller(urwid.WidgetWrap): def __init__(self, widget, app): self.app = app self.filler = urwid.Filler(widget, urwid.TOP) super().__init__(self.filler) def keypress(self, size, key): if key == "tab": self.app.ui.main_display.sub_displays.interface_display.show_profiles() return None if key == "ctrl t": self.app.ui.main_display.sub_displays.interface_display.toggle_view_mode() return None if key == "ctrl s": self.app.ui.main_display.sub_displays.interface_display.save_current_profile() return None if key == "ctrl f": self.app.ui.main_display.sub_displays.interface_display.cycle_filter() return None if key == "ctrl o": self.app.ui.main_display.sub_displays.interface_display.toggle_focused_interface() return None if key == "ctrl b": self.app.ui.main_display.sub_displays.interface_display.show_bulk_actions() return None if key == "ctrl a": # add interface self.app.ui.main_display.sub_displays.interface_display.add_interface() return elif key == "ctrl x": # remove Interface self.app.ui.main_display.sub_displays.interface_display.remove_selected_interface() return elif key == "ctrl e": # edit interface self.app.ui.main_display.sub_displays.interface_display.edit_selected_interface() return None elif key == "ctrl w": # open config file editor self.app.ui.main_display.sub_displays.interface_display.open_config_editor() return None return super().keypress(size, key) ### VIEWS ### class AddInterfaceView(urwid.WidgetWrap): def __init__(self, parent, iface_type): self.parent = parent self.iface_type = iface_type self.fields = {} self.port_pile = None self.additional_fields = {} self.additional_pile_contents = [] self.common_fields = {} self.parent.shortcuts_display.set_add_interface_shortcuts() name_field = FormEdit( config_key="name", placeholder="Enter interface name", validation_types=["required"] ) self.fields['name'] = { 'label': "Name: ", 'widget': name_field } config = INTERFACE_FIELDS.get(iface_type, INTERFACE_FIELDS["default"]) iface_fields = [field for field in config if "config_key" in field] for field in iface_fields: self._initialize_field(field) self._initialize_additional_fields(config) self._initialize_common_fields() pile_items = self._build_form_layout(iface_fields) form_pile = urwid.Pile(pile_items) form_filler = urwid.Filler(form_pile, valign="top") form_box = urwid.LineBox( form_filler, title="Add Interface", tlcorner="╭", tline="─", trcorner="╮", lline="│", rline="│", blcorner="╰", bline="─", brcorner="╯" ) background = urwid.SolidFill(" ") self.overlay = urwid.Overlay( top_w=form_box, bottom_w=background, align='center', width=('relative', 85), valign='middle', height=('relative', 85), ) super().__init__(self.overlay) def _initialize_field(self, field): if field["type"] == "dropdown": widget = FormDropdown( config_key=field["config_key"], label=field.get("label", ""), options=field["options"], default=field.get("default"), validation_types=field.get("validation", []), transform=field.get("transform") ) elif field["type"] == "checkbox": widget = FormCheckbox( config_key=field["config_key"], label=field.get("label", ""), state=field.get("default", False), validation_types=field.get("validation", []), transform=field.get("transform") ) elif field["type"] == "multilist": widget = FormMultiList( config_key=field["config_key"], placeholder=field.get("placeholder", ""), validation_types=field.get("validation", []), transform=field.get("transform") ) elif field["type"] == "multitable": widget = FormMultiTable( config_key=field["config_key"], fields=field.get("fields", {}), validation_types=field.get("validation", []), transform=field.get("transform") ) elif field["type"] == "keyvaluepairs": widget = FormKeyValuePairs( config_key=field["config_key"], validation_types=field.get("validation", []), transform=field.get("transform") ) else: widget = FormEdit( config_key=field["config_key"], caption="", edit_text=field.get("default", ""), placeholder=field.get("placeholder", ""), validation_types=field.get("validation", []), transform=field.get("transform") ) label = field.get("label", "") if not label: label = " ".join(word.capitalize() for word in field["config_key"].split('_')) + ": " self.fields[field["config_key"]] = { 'label': label, 'widget': widget } def _initialize_additional_fields(self, config): for field in config: if isinstance(field, dict) and "additional_options" in field: for option in field["additional_options"]: if option["type"] == "checkbox": widget = FormCheckbox( config_key=option["config_key"], label=option.get("label", ""), state=option.get("default", False), validation_types=option.get("validation", []), transform=option.get("transform") ) elif option["type"] == "dropdown": widget = FormDropdown( config_key=option["config_key"], label=option.get("label", ""), options=option["options"], default=option.get("default"), validation_types=option.get("validation", []), transform=option.get("transform") ) elif option["type"] == "multilist": widget = FormMultiList( config_key=option["config_key"], placeholder=option.get("placeholder", ""), validation_types=option.get("validation", []), transform=option.get("transform") ) else: widget = FormEdit( config_key=option["config_key"], caption="", edit_text=str(option.get("default", "")), placeholder=option.get("placeholder", ""), validation_types=option.get("validation", []), transform=option.get("transform") ) label = option.get("label", "") if not label: label = " ".join(word.capitalize() for word in option["config_key"].split('_')) + ": " self.additional_fields[option["config_key"]] = { 'label': label, 'widget': widget, 'type': option["type"] } def _initialize_common_fields(self): if self.parent.app.rns.transport_enabled(): # Transport mode options COMMON_INTERFACE_OPTIONS.extend([ { "config_key": "outgoing", "type": "checkbox", "label": "Allow outgoing traffic", "default": True, "validation": [], "transform": lambda x: bool(x) }, { "config_key": "mode", "type": "dropdown", "label": "Interface Mode: ", "options": ["full", "gateway", "access_point", "roaming", "boundary"], "default": "full", "validation": [], "transform": lambda x: x }, { "config_key": "announce_cap", "type": "edit", "label": "Announce Cap: ", "placeholder": "Default: 2.0", "default": "", "validation": ["float"], "transform": lambda x: float(x) if x.strip() else 2.0 } ]) for option in COMMON_INTERFACE_OPTIONS: if option["type"] == "checkbox": widget = FormCheckbox( config_key=option["config_key"], label=option["label"], state=option.get("default", False), validation_types=option.get("validation", []), transform=option.get("transform") ) elif option["type"] == "dropdown": widget = FormDropdown( config_key=option["config_key"], label=option["label"], options=option["options"], default=option.get("default"), validation_types=option.get("validation", []), transform=option.get("transform") ) else: widget = FormEdit( config_key=option["config_key"], caption="", edit_text=str(option.get("default", "")), placeholder=option.get("placeholder", ""), validation_types=option.get("validation", []), transform=option.get("transform") ) self.common_fields[option["config_key"]] = { 'label': option["label"], 'widget': widget, 'type': option["type"] } def _on_rnode_field_change(self, widget, new_value): if hasattr(self, 'rnode_calculator') and self.calculator_visible: self.rnode_calculator.update_calculation() def _build_form_layout(self, iface_fields): pile_items = [] pile_items.append(urwid.Text( ("form_title", f"Add new {_get_interface_icon(self.parent.glyphset, self.iface_type)} {self.iface_type}"), align="center")) pile_items.append(urwid.Divider("─")) for key in ["name"] + [f["config_key"] for f in iface_fields]: field = self.fields[key] widget = field["widget"] # Special case for multitable and keyvaluepairs - they already have their own layout if isinstance(widget, (FormMultiTable, FormKeyValuePairs)): pile_items.append(urwid.Text(("key", field["label"]), align="left")) pile_items.append(widget) pile_items.append(urwid.Padding(widget.error_widget, left=2)) continue field_pile = urwid.Pile([ urwid.Columns([ (26, urwid.Text(("key", field["label"]), align="right")), widget, ]), urwid.Padding(widget.error_widget, left=24) ]) if self.iface_type in ["RNodeInterface", "RNodeMultiInterface", "SerialInterface", "AX25KISSInterface", "KISSInterface"] and key == "port": refresh_btn = urwid.Button("Refresh Ports", on_press=self.refresh_ports) refresh_btn = urwid.AttrMap(refresh_btn, "button_normal", focus_map="button_focus") refresh_row = urwid.Padding(refresh_btn, left=26, width=20) field_pile.contents.append((refresh_row, field_pile.options())) self.port_pile = field_pile pile_items.append(field_pile) self.more_options_visible = False self.more_options_button = urwid.Button("Show more options", on_press=self.toggle_more_options) self.more_options_button = urwid.AttrMap(self.more_options_button, "button_normal", focus_map="button_focus") self.more_options_widget = urwid.Pile([]) self.ifac_options_visible = False self.ifac_options_button = urwid.Button("Show IFAC options", on_press=self.toggle_ifac_options) self.ifac_options_button = urwid.AttrMap(self.ifac_options_button, "button_normal", focus_map="button_focus") self.ifac_options_widget = urwid.Pile([]) if self.iface_type in ["RNodeInterface"]: self.calculator_button = urwid.Button("Show On-Air Calculations", on_press=self.toggle_calculator) self.calculator_button = urwid.AttrMap(self.calculator_button, "button_normal", focus_map="button_focus") save_btn = urwid.Button("Save", on_press=self.on_save) back_btn = urwid.Button("Cancel", on_press=self.on_back) button_row = urwid.Columns([ (urwid.WEIGHT, 0.45, save_btn), (urwid.WEIGHT, 0.1, urwid.Text("")), (urwid.WEIGHT, 0.45, back_btn), ]) pile_items.extend([ urwid.Divider(), self.more_options_button, self.more_options_widget, ]) if self.iface_type in ["RNodeInterface"]: self.rnode_calculator = RNodeCalculator(self) self.calculator_visible = False self.calculator_widget = urwid.Pile([]) pile_items.extend([ self.calculator_button, self.calculator_widget, ]) pile_items.extend(self._build_profiles_section()) pile_items.extend([ urwid.Divider("─"), button_row, ]) return pile_items def _build_profiles_section(self): self.profile_checkboxes = [] profiles = getattr(self.parent.app, "interface_profiles", None) if profiles is None or not profiles.profiles: return [] iface_name = getattr(self, "iface_name", None) current = set() if iface_name: current = set(p["id"] for p in profiles.profiles_for(iface_name)) items = [ urwid.Divider("─"), urwid.Text(("key", "Profiles"), align="left"), urwid.Text(("inactive_text", " Assign this interface to one or more profiles."), align="left"), ] for p in profiles.profiles: cb = urwid.CheckBox(p["name"] or "(unnamed)", state=(p["id"] in current)) cb.profile_id = p["id"] self.profile_checkboxes.append(cb) items.append(urwid.Padding(cb, left=2)) return items def _apply_profile_membership(self, iface_name): profiles = getattr(self.parent.app, "interface_profiles", None) if profiles is None: return checked = [cb.profile_id for cb in getattr(self, "profile_checkboxes", []) if cb.get_state()] try: profiles.set_interface_profiles(iface_name, checked) except Exception: pass def toggle_more_options(self, button): if self.more_options_visible: self.more_options_widget.contents = [] button.base_widget.set_label("Show more options") self.more_options_visible = False else: pile_contents = [] if self.additional_fields: for key, field in self.additional_fields.items(): widget = field['widget'] if field['type'] == "checkbox": centered_widget = urwid.Columns([ ('weight', 1, urwid.Text("")), ('pack', widget), ('weight', 1, urwid.Text("")) ]) field_pile = urwid.Pile([ centered_widget, urwid.Padding(widget.error_widget, left=24) ]) else: field_pile = urwid.Pile([ urwid.Columns([ (26, urwid.Text(("key", field["label"]), align="right")), widget ]), urwid.Padding(widget.error_widget, left=24) ]) pile_contents.append(field_pile) if self.additional_fields and self.common_fields: pile_contents.append(urwid.Divider("─")) if self.common_fields: for key, field in self.common_fields.items(): widget = field['widget'] if field['type'] == "checkbox": centered_widget = urwid.Columns([ ('weight', 1, urwid.Text("")), ('pack', widget), ('weight', 1, urwid.Text("")) ]) field_pile = urwid.Pile([ centered_widget, urwid.Padding(widget.error_widget, left=24) ]) else: field_pile = urwid.Pile([ urwid.Columns([ (26, urwid.Text(("key", field["label"]), align="right")), widget ]), urwid.Padding(widget.error_widget, left=24) ]) pile_contents.append(field_pile) if pile_contents: self.more_options_widget.contents = [(w, self.more_options_widget.options()) for w in pile_contents] else: self.more_options_widget.contents = [( urwid.Text("No additional options available", align="center"), self.more_options_widget.options() )] button.base_widget.set_label("Hide more options") self.more_options_visible = True def toggle_ifac_options(self, button): if self.ifac_options_visible: self.ifac_options_widget.contents = [] button.base_widget.set_label("Show IFAC options") self.ifac_options_visible = False else: dummy = urwid.Text("IFAC (Interface Access Codes)", align="left") self.ifac_options_widget.contents = [(dummy, self.more_options_widget.options())] button.base_widget.set_label("Hide IFAC options") self.ifac_options_visible = True def toggle_calculator(self, button): if self.calculator_visible: self.calculator_widget.contents = [] button.base_widget.set_label("Show On-Air Calculations") self.calculator_visible = False else: calculator_contents = [self.rnode_calculator] self.calculator_widget.contents = [(w, self.calculator_widget.options()) for w in calculator_contents] self.rnode_calculator.update_calculation() button.base_widget.set_label("Hide On-Air Calculations") self.calculator_visible = True def refresh_ports(self, button): if self.port_pile is not None: # Get fresh port config port_field = get_port_field() if port_field["type"] == "dropdown": widget = FormDropdown( config_key=port_field["config_key"], label=port_field["label"], options=port_field["options"], default=port_field.get("default"), validation_types=port_field.get("validation", []), transform=port_field.get("transform") ) else: widget = FormEdit( config_key=port_field["config_key"], caption="", edit_text=port_field.get("default", ""), placeholder=port_field.get("placeholder", ""), validation_types=port_field.get("validation", []), transform=port_field.get("transform") ) self.fields["port"] = { 'label': port_field["label"], 'widget': widget } columns = urwid.Columns([ (26, urwid.Text(("key", port_field["label"]), align="right")), widget ]) self.port_pile.contents[0] = (columns, self.port_pile.options()) self.port_pile.contents[1] = (urwid.Padding(widget.error_widget, left=24), self.port_pile.options()) def validate_all(self): all_valid = True # validate main fields for field in self.fields.values(): if not field["widget"].validate(): all_valid = False # validate additional iface fields for field in self.additional_fields.values(): if not field["widget"].validate(): all_valid = False # validate common fields for field in self.common_fields.values(): if not field["widget"].validate(): all_valid = False return all_valid def on_save(self, button): all_valid = self.validate_all() name = self.fields['name']["widget"].get_value() or "Untitled interface" existing_interfaces = self.parent.app.rns.config['interfaces'] if name in existing_interfaces: self.fields['name']["widget"].error = f"Interface name '{name}' already exists" self.fields['name']["widget"].error_widget.set_text(("error", self.fields['name']["widget"].error)) all_valid = False if not all_valid: return if self.iface_type == "CustomInterface": custom_type = self.fields.get('type', {}).get('widget').get_value() interface_config = { "type": custom_type, "interface_enabled": True } else: interface_config = { "type": self.iface_type, "interface_enabled": True } for field_key, field in self.fields.items(): if field_key not in ["name", "custom_parameters", "type"]: widget = field["widget"] value = widget.get_value() if field_key == "subinterfaces" and self.iface_type == "RNodeMultiInterface" and isinstance(value, dict): for subname, subconfig in value.items(): interface_config[f"{subname}"] = subconfig elif value is not None and value != "": interface_config[widget.config_key] = value if self.iface_type == "CustomInterface" and "custom_parameters" in self.fields: custom_params = self.fields["custom_parameters"]["widget"].get_value() if isinstance(custom_params, dict): for param_key, param_value in custom_params.items(): interface_config[param_key] = param_value for field_key, field in self.additional_fields.items(): widget = field["widget"] value = widget.get_value() if value is not None and value != "": interface_config[widget.config_key] = value for field_key, field in self.common_fields.items(): widget = field["widget"] value = widget.get_value() if value is not None and value != "": interface_config[widget.config_key] = value try: interfaces = self.parent.app.rns.config['interfaces'] interfaces[name] = interface_config write_config_atomic(self.parent.app.rns.config) self._apply_profile_membership(name) display_type = custom_type if self.iface_type == "CustomInterface" else self.iface_type new_item = SelectableInterfaceItem( parent=self.parent, name=name, is_connected=False, # will always be false until restart is_enabled=True, iface_type=display_type, tx=0, rx=0, icon=_get_interface_icon(self.parent.glyphset, display_type), iface_options=interface_config ) self.parent.interface_items.append(new_item) self.parent._rebuild_list() self.parent.mark_restart_pending() self.show_message(f"Interface {name} added. Restart NomadNet to start using this interface") except Exception as e: print(f"Error saving interface: {str(e)}") self.show_message(f"Error: {str(e)}", title="Error") def on_back(self, button): self.parent.switch_to_list() def show_message(self, message, title="Notice"): def dismiss_dialog(button): self.parent.switch_to_list() dialog = DialogLineBox( urwid.Pile([ urwid.Text(message, align="center"), urwid.Divider(), urwid.Button("OK", on_press=dismiss_dialog) ]), title=title ) overlay = urwid.Overlay( dialog, self.parent.interfaces_display, align='center', width=100, valign='middle', height=8, min_width=1, min_height=1 ) self.parent.widget = overlay self.parent.app.ui.main_display.update_active_sub_display() class EditInterfaceView(AddInterfaceView): def __init__(self, parent, iface_name): self.parent = parent self.iface_name = iface_name self.interface_config = parent.app.rns.config['interfaces'][iface_name] config_type = self.interface_config.get("type", "Unknown") # check if this is a custom interface type known_types = list(INTERFACE_FIELDS.keys()) if config_type not in known_types: self.original_type = config_type self.iface_type = "CustomInterface" else: self.original_type = None self.iface_type = config_type super().__init__(parent, self.iface_type) self.overlay.top_w.title_widget.set_text(f"Edit Interface: {iface_name}") self._populate_form_fields() def _populate_form_fields(self): self.fields['name']['widget'].edit_text = self.iface_name if self.original_type and self.iface_type == "CustomInterface": if "type" in self.fields: self.fields["type"]["widget"].edit_text = self.original_type if "custom_parameters" in self.fields: custom_params = {} standard_keys = ["type", "interface_enabled", "enabled", "description", "network_name", "bitrate", "passphrase", "ifac_size", "mode", "outgoing", "announce_cap"] for key, value in self.interface_config.items(): if key not in standard_keys: custom_params[key] = value self.fields["custom_parameters"]["widget"].set_value(custom_params) for key, field in self.fields.items(): if key not in ['name', 'type', 'custom_parameters']: widget = field['widget'] if key == "subinterfaces" and isinstance(widget, FormMultiTable): self._populate_subinterfaces(widget) continue if key in self.interface_config: value = self.interface_config[key] if key == 'frequency': value = float(value) / 1000000 value = f"{value:.6f}".rstrip('0').rstrip('.') if '.' in f"{value:.6f}" else f"{value}" self._set_field_value(widget, value) for key, field in self.additional_fields.items(): if key in self.interface_config: self._set_field_value(field['widget'], self.interface_config[key]) for key, field in self.common_fields.items(): if key in self.interface_config: self._set_field_value(field['widget'], self.interface_config[key]) def _set_field_value(self, widget, value): if hasattr(widget, 'edit_text'): widget.edit_text = str(value) elif hasattr(widget, 'set_state'): checkbox_state = value if isinstance(value, bool) else value.strip().lower() not in ( 'false', 'off', 'no', '0') widget.set_state(checkbox_state) elif isinstance(widget, FormDropdown): str_value = str(value) if str_value in widget.options: widget.selected = str_value widget.main_button.base_widget.set_text(str_value) else: # Try to match after transform for opt in widget.options: try: if widget.transform(opt) == value: widget.selected = opt widget.main_button.base_widget.set_text(opt) break except: pass elif isinstance(widget, FormMultiList): self._populate_multilist(widget, value) def _populate_multilist(self, widget, value): items = [] if isinstance(value, str): items = [item.strip() for item in value.split(',') if item.strip()] elif isinstance(value, list): items = value while len(widget.entries) > 1: widget.remove_entry(None, widget.entries[-1]) if items: first_entry = widget.entries[0] first_edit = first_entry.contents[0][0] if len(items) > 0: first_edit.edit_text = items[0] for i in range(1, len(items)): widget.add_entry(None) entry = widget.entries[i] edit_widget = entry.contents[0][0] edit_widget.edit_text = items[i] def _populate_subinterfaces(self, widget): subinterfaces = {} for key, value in self.interface_config.items(): if isinstance(value, dict): if key.startswith('[[[') and key.endswith(']]]'): clean_key = key[3:-3] # removes [[[...]]] subinterfaces[clean_key] = value elif key not in ["type", "interface_enabled", "enabled", "port", "id_callsign", "id_interval"]: subinterfaces[key] = value if subinterfaces: widget.set_value(subinterfaces) def on_save(self, button): if not self.validate_all(): return new_name = self.fields['name']["widget"].get_value() or self.iface_name if new_name != self.iface_name and new_name in self.parent.app.rns.config['interfaces']: self.fields['name']["widget"].error = f"Interface name '{new_name}' already exists" self.fields['name']["widget"].error_widget.set_text(("error", self.fields['name']["widget"].error)) return if self.iface_type == "CustomInterface": interface_type = self.fields.get('type', {}).get('widget').get_value() else: interface_type = self.iface_type updated_config = { "type": interface_type, "interface_enabled": self.interface_config.get("interface_enabled", True), } if "enabled" in self.interface_config: updated_config["enabled"] = self.interface_config["enabled"] for field_key, field in self.fields.items(): if field_key not in ["name", "custom_parameters", "type", "subinterfaces"]: widget = field["widget"] value = widget.get_value() if value is not None and value != "": updated_config[widget.config_key] = value if self.iface_type == "CustomInterface" and "custom_parameters" in self.fields: custom_params = self.fields["custom_parameters"]["widget"].get_value() if isinstance(custom_params, dict): for param_key, param_value in custom_params.items(): updated_config[param_key] = param_value elif self.iface_type == "RNodeMultiInterface" and "subinterfaces" in self.fields: subinterfaces = self.fields["subinterfaces"]["widget"].get_value() for subname, subconfig in subinterfaces.items(): updated_config[subname] = subconfig for field_key, field in self.additional_fields.items(): widget = field["widget"] value = widget.get_value() if value is not None and value != "": updated_config[widget.config_key] = value for field_key, field in self.common_fields.items(): widget = field["widget"] value = widget.get_value() if value is not None and value != "": updated_config[widget.config_key] = value try: interfaces = self.parent.app.rns.config['interfaces'] if new_name != self.iface_name: del interfaces[self.iface_name] interfaces[new_name] = updated_config try: self.parent.app.interface_profiles.rename_interface(self.iface_name, new_name) except Exception: pass for i, item in enumerate(self.parent.interface_items): if item.name == self.iface_name: self.parent.interface_items[i].name = new_name break else: interfaces[self.iface_name] = updated_config write_config_atomic(self.parent.app.rns.config) self._apply_profile_membership(new_name) display_type = interface_type for item in self.parent.interface_items: if item.name == new_name: item.iface_type = display_type break self.parent._rebuild_list() self.parent.mark_restart_pending() self.show_message(f"Interface {new_name} updated. Restart NomadNet for these changes to take effect") except Exception as e: print(f"Error saving interface: {str(e)}") self.show_message(f"Error updating interface: {str(e)}", title="Error") class ShowInterface(urwid.WidgetWrap): def __init__(self, parent, iface_name): self.parent = parent self.iface_name = iface_name self.started = False self.g = self.parent.app.ui.glyphs # get config self.interface_config = self.parent.app.rns.config['interfaces'][iface_name] iface_type = self.interface_config.get("type", "Unknown") self.parent.shortcuts_display.set_show_interface_shortcuts() self.config_rows = [] screen_cols, _ = _get_cols_rows() margin = 22 if screen_cols >= 145: self.history_length = screen_cols//2-margin else: self.history_length = screen_cols-(margin+2) # get interface stats interface_stats = self.parent.app.interface_stats stats_lookup = {iface['short_name']: iface for iface in interface_stats['interfaces']} self.stats = stats_lookup.get(iface_name, {}) self.tx = self.stats.get("txb", 0) self.rx = self.stats.get("rxb", 0) self.is_connected = self.stats.get("status", False) self.is_enabled = (str(self.interface_config.get("enabled")).lower() != 'false' and str(self.interface_config.get("interface_enabled")).lower() != 'false') header_content = [ urwid.Text(("interface_title", f"Interface: {iface_name}"), align="center"), urwid.Divider("=") ] header = urwid.Pile(header_content) self.edit_button = urwid.Button("Edit", on_press=self.on_edit) self.back_button = urwid.Button("Back", on_press=self.on_back) self.toggle_button = urwid.Button( "Disable" if self.is_enabled else "Enable", on_press=self.on_toggle_enabled ) button_row = urwid.Columns([ (urwid.WEIGHT, 0.3, self.back_button), (urwid.WEIGHT, 0.05, urwid.Text("")), (urwid.WEIGHT, 0.3, self.toggle_button), (urwid.WEIGHT, 0.05, urwid.Text("")), (urwid.WEIGHT, 0.3, self.edit_button), ]) footer_content = [ urwid.Divider("="), button_row ] footer = urwid.Pile(footer_content) # status widgets self.status_text = urwid.Text(("connected_status" if self.is_enabled else "disabled_status", "Enabled" if self.is_enabled else "Disabled")) self.status_indicator = urwid.Text(("connected_status" if self.is_enabled else "disabled_status", self.parent.g['selected'] if self.is_enabled else self.parent.g[ 'unselected'])) self.connection_text = urwid.Text(connection_markup(self.parent.g, self.is_connected)) if self.is_enabled: status_columns = urwid.Columns([ (10, urwid.Text(("key", "Status:"))), (4, self.status_indicator), (8, self.status_text), (3, urwid.Text(" | ")), self.connection_text, ]) else: status_columns = urwid.Columns([ (10, urwid.Text(("key", "Status:"))), (4, self.status_indicator), self.status_text, ]) self.info_rows = [ urwid.Columns([ (10, urwid.Text(("key", "Type:"))), urwid.Text(("value", f"{_get_interface_icon(self.parent.glyphset, iface_type)} {iface_type}")), ]), status_columns, urwid.Divider("-") ] for row in self._build_statistics(): self.info_rows.append(row) self.info_rows.append(urwid.Divider("-")) self.bandwidth_chart = InterfaceBandwidthChart(history_length=self.history_length, glyphset=self.parent.glyphset) self.bandwidth_chart.update(self.rx, self.tx) self.rx_chart_text = urwid.Text("Loading RX data...", align='left') self.tx_chart_text = urwid.Text("Loading TX data...", align='left') self.rx_peak_text = urwid.Text("Peak: 0 B/s", align='right') self.tx_peak_text = urwid.Text("Peak: 0 B/s", align='right') self.rx_box = urwid.LineBox( urwid.Pile([ urwid.AttrMap(self.rx_chart_text, "rx"), self.rx_peak_text ]), title="RX Traffic (60s)" ) self.tx_box = urwid.LineBox( urwid.Pile([ urwid.AttrMap(self.tx_chart_text, "tx"), self.tx_peak_text ]), title="TX Traffic (60s)" ) self.horizontal_charts = urwid.Columns([ (urwid.WEIGHT, 1, self.rx_box), (urwid.WEIGHT, 1, self.tx_box) ]) self.vertical_charts = urwid.Pile([ self.rx_box, self.tx_box ]) self.disconnected_message = urwid.Filler( urwid.Text(("disabled_status", "Charts not available - Interface is not connected"), align="center"), valign="top" ) self.disconnected_box = urwid.LineBox(self.disconnected_message, title="Bandwidth Charts") self.charts_widget = self.vertical_charts self.is_horizontal = False screen_cols, _ = _get_cols_rows() # RNS.log(screen_cols) if screen_cols >= 145: self.charts_widget = self.horizontal_charts self.is_horizontal = True if not self.is_connected: self.charts_widget = self.disconnected_box connection_params = [] radio_params = [] network_params = [] ifac_params = [] other_params = [] # Sort parameters into groups for key, value in self.interface_config.items(): # skip empty if value is None or value == "": continue # Skip these keys as their shown elsewhere if key in ["type", "interface_enabled", "enabled", 'selected_interface_mode', 'name']: continue # Connection parameters elif key in ["port", "listen_ip", "listen_port", "target_host", "target_port", "device"]: connection_params.append((key, value)) # Radio parameters elif key in ["frequency", "bandwidth", "spreadingfactor", "codingrate", "txpower"]: radio_params.append((key, value)) # Network parameters elif key in ["network_name", "bitrate", "peers", "group_id", "multicast_address_type", "discovery_scope", "announce_cap", "mode"]: network_params.append((key, value)) # IFAC parameters elif key in ["passphrase", "ifac_size", "ifac_netname", "ifac_netkey"]: ifac_params.append((key, value)) else: other_params.append((key, value)) def create_param_row(key, value): if isinstance(value, bool): value_str = "Yes" if value else "No" elif key == "frequency": int_value = int(value) value_str = f"{int_value / 1000000:.3f} MHz" elif key == "bandwidth": int_value = int(value) value_str = f"{int_value / 1000:.1f} kHz" elif key == "passphrase": value_str = len(value)*"*" else: value_str = str(value) # format display keys: "listen_port" => Listen Port display_key = key.replace('_', ' ').title() return urwid.Columns([ (18, urwid.Text(("key", f"{display_key}:"))), urwid.Text(("value", value_str)), ]) if connection_params: connection_params.sort(key=lambda x: x[0]) self.config_rows.append(urwid.Text(("interface_title", "Connection Parameters"), align="left")) for key, value in connection_params: self.config_rows.append(create_param_row(key, value)) self.config_rows.append(urwid.Divider("-")) if radio_params: radio_params.sort(key=lambda x: x[0]) self.config_rows.append(urwid.Text(("interface_title", "Radio Parameters"), align="left")) for key, value in radio_params: self.config_rows.append(create_param_row(key, value)) self.config_rows.append(urwid.Divider("-")) if network_params: network_params.sort(key=lambda x: x[0]) self.config_rows.append(urwid.Text(("interface_title", "Network Parameters"), align="left")) for key, value in network_params: self.config_rows.append(create_param_row(key, value)) self.config_rows.append(urwid.Divider("-")) if ifac_params: ifac_params.sort(key=lambda x: x[0]) self.config_rows.append(urwid.Text(("interface_title", "IFAC Parameters"), align="left")) for key, value in ifac_params: self.config_rows.append(create_param_row(key, value)) self.config_rows.append(urwid.Divider("-")) if other_params: other_params.sort(key=lambda x: x[0]) self.config_rows.append(urwid.Text(("interface_title", "Additional Parameters"), align="left")) for key, value in other_params: self.config_rows.append(create_param_row(key, value)) self.config_rows.append(urwid.Divider("-")) if not self.config_rows: self.config_rows.append(urwid.Text("No additional parameters", align="center")) body_content = [] body_content.extend(self.info_rows) body_content.append( self.charts_widget) body_content.append(urwid.Divider("-")) body_content.extend(self.config_rows) body_pile = urwid.Pile(body_content) body_padding = urwid.Padding(body_pile, left=2, right=2) body = urwid.ListBox(urwid.SimpleListWalker([body_padding])) self.frame = urwid.Frame( body=body, header=header, footer=footer ) self.content_box = urwid.LineBox(self.frame) super().__init__(self.content_box) def _stat_rows_spec(self, stats): # Ordered (row_id, label) pairs that apply to this interface spec = [("mode", "Mode:")] if stats.get("bitrate") is not None: spec.append(("bitrate", "Bitrate:")) spec.append(("tx", "TX:")) spec.append(("rx", "RX:")) if "rxs" in stats and "txs" in stats: spec.append(("speed", "Speed:")) if stats.get("incoming_announce_frequency") is not None: spec.append(("announce", "Announces:")) if stats.get("incoming_pr_frequency") is not None: spec.append(("pathreq", "Path Reqs:")) if stats.get("clients") is not None: spec.append(("clients", "Clients:")) if stats.get("peers") is not None: spec.append(("peers", "Peers:")) if stats.get("ifac_netname") is not None or stats.get("ifac_signature") is not None: spec.append(("ifac", "Network:")) if stats.get("announce_queue") is not None: spec.append(("queued", "Queued:")) if "held_announces" in stats: spec.append(("held", "Held:")) return spec def _radio_rows_spec(self, stats): spec = [] if "channel_load_short" in stats and "channel_load_long" in stats: spec.append(("chload", "Ch. Load:")) if "airtime_short" in stats and "airtime_long" in stats: spec.append(("airtime", "Airtime:")) if "noise_floor" in stats: spec.append(("noise", "Noise Fl.:")) if stats.get("battery_percent") is not None: spec.append(("battery", "Battery:")) if "cpu_load" in stats or "cpu_temp" in stats: spec.append(("cpu", "CPU:")) if "mem_load" in stats: spec.append(("mem", "Memory:")) return spec def _fmt_pct(self, v): try: v = float(v) except (TypeError, ValueError): return "0" if v <= 1.0: v *= 100.0 return "%d" % int(round(v)) def _fmt_pct_pair(self, short_v, long_v): return "%s%% (15s), %s%% (1h)" % (self._fmt_pct(short_v), self._fmt_pct(long_v)) def _format_stat(self, row_id, stats): if row_id == "mode": return interface_mode_str(stats.get("mode")) if row_id == "bitrate": return format_speed(stats.get("bitrate")) if row_id == "tx": return format_bytes(stats.get("txb", 0)) if row_id == "rx": return format_bytes(stats.get("rxb", 0)) if row_id == "speed": return "TX %s RX %s" % (format_speed(stats.get("txs", 0)), format_speed(stats.get("rxs", 0))) if row_id == "announce": return "In %s Out %s" % (format_announce_rate(stats.get("incoming_announce_frequency")), format_announce_rate(stats.get("outgoing_announce_frequency"))) if row_id == "pathreq": return "In %s Out %s" % (format_announce_rate(stats.get("incoming_pr_frequency")), format_announce_rate(stats.get("outgoing_pr_frequency"))) if row_id == "clients": return str(stats.get("clients")) if row_id == "peers": return "%s reachable" % stats.get("peers") if row_id == "ifac": netname = stats.get("ifac_netname") size = stats.get("ifac_size") bits = ("%d-bit IFAC" % (size * 8)) if size else "IFAC" return ("%s (%s)" % (netname, bits)) if netname else bits if row_id == "queued": n = stats.get("announce_queue") or 0 return "%d announce%s" % (n, "" if n == 1 else "s") if row_id == "held": n = stats.get("held_announces") or 0 return "%d announce%s" % (n, "" if n == 1 else "s") if row_id == "chload": return self._fmt_pct_pair(stats.get("channel_load_short"), stats.get("channel_load_long")) if row_id == "airtime": return self._fmt_pct_pair(stats.get("airtime_short"), stats.get("airtime_long")) if row_id == "noise": nf = stats.get("noise_floor") return ("%s dBm" % nf) if nf is not None else "Unknown" if row_id == "battery": try: return "%d%% (%s)" % (int(stats.get("battery_percent")), stats.get("battery_state", "Unknown")) except (TypeError, ValueError): return "Unknown" if row_id == "cpu": load = stats.get("cpu_load") temp = stats.get("cpu_temp") parts = [] if load is not None: parts.append("%s%% load" % load) if temp is not None: parts.append("%s°C" % temp) return ", ".join(parts) if parts else "Unknown" if row_id == "mem": m = stats.get("mem_load") return ("%s%%" % m) if m is not None else "Unknown" return "" def _build_statistics(self): # Builds the live statistics section and records the dynamic widgets in # self.stat_w so they can be refreshed in place each poll tick. self.stat_w = {} def stat_row(row_id, label): w = urwid.Text(("value", self._format_stat(row_id, self.stats))) self.stat_w[row_id] = w return urwid.Columns([(12, urwid.Text(("key", label))), w]) rows = [urwid.Text(("interface_title", "Statistics"), align="left")] for row_id, label in self._stat_rows_spec(self.stats): rows.append(stat_row(row_id, label)) radio = self._radio_rows_spec(self.stats) if radio: rows.append(urwid.Divider("-")) rows.append(urwid.Text(("interface_title", "Radio"), align="left")) for row_id, label in radio: rows.append(stat_row(row_id, label)) return rows def _refresh_stats(self, stats): for row_id, w in self.stat_w.items(): try: w.set_text(("value", self._format_stat(row_id, stats))) except Exception: pass def update_status_display(self): if self.is_enabled: self.status_indicator.set_text(("connected_status", self.parent.g['selected'])) self.status_text.set_text(("connected_status", "Enabled")) else: self.status_indicator.set_text(("disabled_status", self.parent.g['unselected'])) self.status_text.set_text(("disabled_status", "Disabled")) def update_connection_display(self, is_connected): old_connection_state = self.is_connected self.is_connected = is_connected self.connection_text.set_text(connection_markup(self.parent.g, self.is_connected)) if old_connection_state != self.is_connected: body_pile = self.frame.body.body[0].original_widget chart_index = None for i, (widget, options) in enumerate(body_pile.contents): if (widget == self.horizontal_charts or widget == self.vertical_charts or widget == self.disconnected_box): chart_index = i break if chart_index is not None: if self.is_connected: new_widget = self.horizontal_charts if self.is_horizontal else self.vertical_charts else: new_widget = self.disconnected_box body_pile.contents[chart_index] = (new_widget, body_pile.options()) def on_toggle_enabled(self, button): self.is_enabled = not self.is_enabled self.toggle_button.set_label("Disable" if self.is_enabled else "Enable") self.interface_config["interface_enabled"] = self.is_enabled if "enabled" in self.interface_config: self.interface_config["enabled"] = self.is_enabled try: interfaces = self.parent.app.rns.config['interfaces'] interfaces[self.iface_name] = self.interface_config write_config_atomic(self.parent.app.rns.config) self.update_status_display() for item in self.parent.interface_items: if item.name == self.iface_name: item.is_enabled = self.is_enabled item.update_status_display() if hasattr(self.parent.app.ui, 'loop') and self.parent.app.ui.loop is not None: self.parent.app.ui.loop.draw_screen() self.parent.mark_restart_pending() except Exception as e: self.show_error_message(f"Error updating interface: {str(e)}") def show_restart_required_message(self): def dismiss_dialog(button): self.parent.app.ui.main_display.frame.body = self.parent.app.ui.main_display.sub_displays.active().widget dialog = DialogLineBox( urwid.Pile([ urwid.Text( f"Interface {self.iface_name} has been " + ("enabled" if self.is_enabled else "disabled") + ".\nRestart required for changes to take effect.", align="center" ), urwid.Divider(), urwid.Button("OK", on_press=dismiss_dialog) ]), title="Notice" ) overlay = urwid.Overlay( dialog, self.parent.app.ui.main_display.frame.body, align='center', width=50, valign='middle', height=8 ) self.parent.app.ui.main_display.frame.body = overlay def show_error_message(self, message): def dismiss_dialog(button): self.parent.app.ui.main_display.frame.body = self.parent.app.ui.main_display.sub_displays.active().widget dialog = DialogLineBox( urwid.Pile([ urwid.Text(message, align="center"), urwid.Divider(), urwid.Button("OK", on_press=dismiss_dialog) ]), title="Error" ) overlay = urwid.Overlay( dialog, self.parent.app.ui.main_display.frame.body, align='center', width=50, valign='middle', height=8 ) self.parent.app.ui.main_display.frame.body = overlay def keypress(self, size, key): if key == 'tab': if self.frame.focus_position == 'body': self.frame.focus_position = 'footer' footer_pile = self.frame.footer if isinstance(footer_pile, urwid.Pile): footer_pile.focus_position = 1 # button row button_row = footer_pile.contents[1][0] if isinstance(button_row, urwid.Columns): button_row.focus_position = 0 return None elif self.frame.focus_position == 'footer': footer_pile = self.frame.footer if isinstance(footer_pile, urwid.Pile): button_row = footer_pile.contents[1][0] if isinstance(button_row, urwid.Columns): # If on first button (Back), move to Toggle button if button_row.focus_position == 0: button_row.focus_position = 2 # skip spacer return None # If on toggle button, move to Edit button elif button_row.focus_position == 2: button_row.focus_position = 4 return None # if on edit button wrap back to toggle button elif button_row.focus_position == 4: button_row.focus_position = 0 return None elif key == 'shift tab': if self.frame.focus_position == 'footer': self.frame.focus_position = 'body' return None elif self.frame.focus_position == 'footer': footer_pile = self.frame.footer if isinstance(footer_pile, urwid.Pile): button_row = footer_pile.contents[1][0] if isinstance(button_row, urwid.Columns): if button_row.focus_position == 4: # edit button button_row.focus_position = 2 # toggle button return None elif button_row.focus_position == 2: button_row.focus_position = 0 # back button return None elif button_row.focus_position == 0: # back button self.frame.focus_position = 'body' return None elif key == 'down': if self.frame.focus_position == 'body': result = super().keypress(size, key) # if the key wasn't consumed, we're at the bottom if result == 'down': self.frame.focus_position = 'footer' footer_pile = self.frame.footer if isinstance(footer_pile, urwid.Pile): footer_pile.focus_position = 1 # button row button_row = footer_pile.contents[1][0] if isinstance(button_row, urwid.Columns): button_row.focus_position = 0 # focus on back button return None return result elif key == 'up': if self.frame.focus_position == 'footer': self.frame.focus_position = 'body' listbox = self.frame.body if hasattr(listbox, 'body') and len(listbox.body) > 0: listbox.focus_position = len(listbox.body) - 1 return None elif self.frame.focus_position == 'body': result = super().keypress(size, key) # if the key wasn't consumed, we're at the top if result == 'up': pass return result elif key == "h" and self.is_connected: # horizontal layout if not self.is_horizontal: self.switch_to_horizontal() return None elif key == "v" and self.is_connected: # vertical layout if self.is_horizontal: self.switch_to_vertical() return None return super().keypress(size, key) def switch_to_horizontal(self): if not self.is_connected: return self.is_horizontal = True body_pile = self.frame.body.body[0].original_widget for i, (widget, options) in enumerate(body_pile.contents): if widget == self.vertical_charts: body_pile.contents[i] = (self.horizontal_charts, options) self.charts_widget = self.horizontal_charts break def switch_to_vertical(self): if not self.is_connected: return self.is_horizontal = False body_pile = self.frame.body.body[0].original_widget for i, (widget, options) in enumerate(body_pile.contents): if widget == self.horizontal_charts: body_pile.contents[i] = (self.vertical_charts, options) self.charts_widget = self.vertical_charts break def start(self): # Poll while the detail view is open, regardless of connection state, so # that the connection indicator and live stats keep refreshing and a # dropped link can be seen to come back up. if not self.started: self.started = True self.parent.app.ui.loop.set_alarm_in(1, self.update_bandwidth_charts) def _show_disconnect_overlay(self): # Uses its own overlay object so it never collides with the interface # list's disconnect overlay handling while both poll loops are running. if getattr(self, "_disconnect_overlay", None) is not None and self.parent.widget is self._disconnect_overlay: return dialog_text = urwid.Pile([ urwid.Text(("disconnected_status", "(!) RNS Instance Disconnected"), align="center"), urwid.Text("Waiting to Reconnect...", align="center"), ]) dialog_box = urwid.LineBox(urwid.Filler(dialog_text)) self._disconnect_overlay = urwid.Overlay( dialog_box, self, align='center', width=35, valign='middle', height=4) self.parent.widget = self._disconnect_overlay self.parent.app.ui.main_display.update_active_sub_display() def update_bandwidth_charts(self, loop, user_data): if not self.started: return try: # The interface list's poll loop owns the actual stats fetch; when the # shared RNS instance drops it flags _rns_disconnected, which we surface # here too instead of silently charting stale data. if getattr(self.parent, "_rns_disconnected", False): self._show_disconnect_overlay() else: if getattr(self, "_disconnect_overlay", None) is not None and self.parent.widget is self._disconnect_overlay: self.parent.widget = self self.parent.app.ui.main_display.update_active_sub_display() interface_stats = self.parent.app.interface_stats stats_lookup = {iface['short_name']: iface for iface in interface_stats['interfaces']} stats = stats_lookup.get(self.iface_name, {}) tx = stats.get("txb", self.tx) rx = stats.get("rxb", self.rx) new_connection_status = stats.get("status", False) if new_connection_status != self.is_connected: self.update_connection_display(new_connection_status) self._refresh_stats(stats) if self.is_connected: self.bandwidth_chart.update(rx, tx) rx_chart, tx_chart, peak_rx, peak_tx = self.bandwidth_chart.get_charts(height=8) self.rx_chart_text.set_text(rx_chart) self.tx_chart_text.set_text(tx_chart) self.rx_peak_text.set_text(f"Peak: {peak_rx}") self.tx_peak_text.set_text(f"Peak: {peak_tx}") self.tx = tx self.rx = rx except Exception as e: self._show_disconnect_overlay() finally: if self.started: loop.set_alarm_in(1, self.update_bandwidth_charts) def on_back(self, button): self.started = False self.parent.switch_to_list() def on_edit(self, button): self.started = False self.parent.switch_to_edit_interface(self.iface_name) class InterfaceTile(urwid.WidgetWrap): def __init__(self, parent, name, is_connected, is_enabled, iface_type, icon, announce_rate="0"): self.parent = parent self.name = name self.icon = icon self.g = parent.g self.dot_style = "connected_status" if is_enabled else "disabled_status" self.dot = self.g['selected'] if is_enabled else self.g['unselected'] self.title_widget = urwid.Text("", wrap="clip") typ = urwid.Text(("value", iface_type), wrap="clip") if is_enabled: status = urwid.Text([ (self.dot_style, "Enabled"), ("value", f" {self.g['sep_dot']} "), connection_markup(self.g, is_connected, "Conn." if is_connected else "Disc."), ], wrap="clip") else: status = urwid.Text([(self.dot_style, "Disabled")], wrap="clip") inner_items = [self.title_widget, typ, status] if is_enabled: inner_items.append(urwid.Text([("key", "Announces "), ("value", announce_rate)], wrap="clip")) inner = urwid.Pile(inner_items) self.box = urwid.LineBox( urwid.Padding(inner, left=1, right=1), tlcorner="╭", trcorner="╮", blcorner="╰", brcorner="╯", ) self.attr = urwid.AttrMap(self.box, None, focus_map={None: "interface_tile_focus"}) super().__init__(self.attr) self._render_title(False) def _render_title(self, focus): arrow = self.g['focus_arrow'] if focus else " " style = "interface_tile_focus" if focus else "interface_title" self.title_widget.set_text([(self.dot_style, self.dot), (style, f" {arrow} {self.icon} {self.name}")]) def selectable(self): return True def render(self, size, focus=False): self._render_title(focus) return super().render(size, focus) def keypress(self, size, key): if key == "enter": self.parent.switch_to_show_interface(self.name) return None return key def mouse_event(self, size, event, button, x, y, focus): if button == 1 and urwid.util.is_mouse_press(event): self.parent.switch_to_show_interface(self.name) return True return False ### PROFILE VIEWS ### class ProfileListRow(urwid.WidgetWrap): def __init__(self, view, pid, label, selected): self.view = view self.pid = pid self.label = label self.selected = selected self.g = view.parent.g self.text_widget = urwid.Text("") super().__init__(urwid.AttrMap(self.text_widget, None)) self._render_row(False) def _render_row(self, focus): mark = self.g['selected'] if self.selected else self.g['unselected'] arrow = self.g['focus_arrow'] if focus else " " if focus: style = "interface_tile_focus" elif self.selected: style = "connected_status" else: style = "interface_title" self.text_widget.set_text((style, f" {arrow} {mark} {self.label}")) def selectable(self): return True def render(self, size, focus=False): self._render_row(focus) return super().render(size, focus) def keypress(self, size, key): if key in (" ", "enter"): if self.pid is not None: self.view.select(self.pid) return None return key def mouse_event(self, size, event, button, x, y, focus): if button == 1 and urwid.util.is_mouse_press(event): if self.pid is not None: self.view.select(self.pid) return True return False class ProfilesView(urwid.WidgetWrap): def __init__(self, parent): self.parent = parent self.app = parent.app self.profiles = parent.app.interface_profiles self.profiles.prune() self.status = urwid.Text("", align="center") self.walker = urwid.SimpleFocusListWalker([]) self.listbox = urwid.ListBox(self.walker) body_rows = max(3, parent.list_rows - 3) self.pile = urwid.Pile([ ('pack', parent._make_tab_bar("profiles")), ('pack', self.status), ('pack', urwid.Divider()), urwid.BoxAdapter(self.listbox, body_rows), ]) super().__init__(urwid.Filler(self.pile, urwid.TOP)) self.refresh() def refresh(self, focus_pid="__keep__"): prev = None try: w, _ = self.listbox.body.get_focus() prev = getattr(w, "pid", None) except Exception: pass self.profiles.update_default_if_custom() active = self.profiles.active_profile_id() items = [ProfileListRow(self, "__default__", "Default", active is None)] for p in self.profiles.profiles: count = self.profiles.member_count(p) label = f"{p['name'] or '(unnamed)'} ({count} interface{'' if count == 1 else 's'})" items.append(ProfileListRow(self, p["id"], label, p["id"] == active)) if len(self.profiles.profiles) == 0: items.append(urwid.Text("\n No profiles yet. Press Ctrl + A to create one.", align="left")) self.walker[:] = items target = focus_pid if focus_pid != "__keep__" else prev for i, w in enumerate(self.walker): if getattr(w, "pid", "__x__") == target: try: self.listbox.focus_position = i except Exception: pass break def _focused(self): try: w, _ = self.listbox.body.get_focus() return w if isinstance(w, ProfileListRow) else None except Exception: return None def select(self, pid): if pid == "__default__": if self.profiles.default_members is None: return count = len(self.profiles.default_members) title = "Switch to Default" msg = "Restore your saved manual setup?\nThis enables %d interface%s and disables the rest." % (count, "" if count == 1 else "s") else: p = self.profiles.get(pid) if p is None: return count = self.profiles.member_count(p) title = "Switch Profile" msg = "Switch to profile '%s'?\nThis enables its %d interface%s and disables the rest." % (p["name"] or "(unnamed)", count, "" if count == 1 else "s") def yes(_b): self.dismiss_dialog() if pid == "__default__": self.profiles.select_default() else: self.profiles.select_profile(pid) self.parent.mark_restart_pending() self.refresh(focus_pid=pid) self.status.set_text(("connected_status", "Restart NomadNet for changes to take effect")) def no(_b): self.dismiss_dialog() pile = urwid.Pile([ urwid.Text(msg, align="center"), urwid.Divider(), urwid.Columns([ (urwid.WEIGHT, 0.45, urwid.Button("Yes", on_press=yes)), (urwid.WEIGHT, 0.1, urwid.Text("")), (urwid.WEIGHT, 0.45, urwid.Button("No", on_press=no)), ]), ]) self._overlay(DialogLineBox(urwid.Filler(pile, urwid.TOP), parent=self, title=title), height=10) def _overlay(self, dialog, height=8, width=50): frame = self.app.ui.main_display.frame frame.body = urwid.Overlay(dialog, self, align="center", width=width, valign="middle", height=height, min_width=20) def dismiss_dialog(self): self.app.ui.main_display.frame.body = self def new_profile(self): edit = ReadlineEdit(caption="Name: ") def ok(_b): val = edit.get_edit_text().strip() self.dismiss_dialog() if val: pid = self.profiles.create(val) self.refresh(focus_pid=pid) self._prompt("New Profile", edit, ok) def save_current(self): n = len(self.profiles.enabled_set()) edit = ReadlineEdit(caption="Name (%d enabled): " % n) def ok(_b): val = edit.get_edit_text().strip() self.dismiss_dialog() if val: pid = self.profiles.save_current_as_profile(val) self.refresh(focus_pid=pid) self.status.set_text(("connected_status", "Saved current setup (%d interface%s) as a profile" % (n, "" if n == 1 else "s"))) self._prompt("Save Current as Profile", edit, ok) def rename_profile(self): w = self._focused() if w is None or w.pid is None: return p = self.profiles.get(w.pid) if p is None: return edit = ReadlineEdit(caption="Name: ", edit_text=p["name"]) def ok(_b): val = edit.get_edit_text().strip() self.dismiss_dialog() if val: self.profiles.rename(p["id"], val) self.refresh(focus_pid=p["id"]) self._prompt("Rename Profile", edit, ok) def delete_profile(self): w = self._focused() if w is None or w.pid is None: return p = self.profiles.get(w.pid) if p is None: return def yes(_b): self.dismiss_dialog() self.profiles.delete(p["id"]) self.refresh() def no(_b): self.dismiss_dialog() pile = urwid.Pile([ urwid.Text(f"Delete profile '{p['name'] or p['id']}'?\nInterfaces keep their current enabled state.", align="center"), urwid.Divider(), urwid.Columns([ (urwid.WEIGHT, 0.45, urwid.Button("Yes", on_press=yes)), (urwid.WEIGHT, 0.1, urwid.Text("")), (urwid.WEIGHT, 0.45, urwid.Button("No", on_press=no)), ]), ]) self._overlay(DialogLineBox(urwid.Filler(pile, urwid.TOP), parent=self, title="Confirm"), height=10) def _prompt(self, title, edit, on_ok): pile = urwid.Pile([ edit, urwid.Divider(), urwid.Columns([ (urwid.WEIGHT, 0.45, urwid.Button("OK", on_press=on_ok)), (urwid.WEIGHT, 0.1, urwid.Text("")), (urwid.WEIGHT, 0.45, urwid.Button("Cancel", on_press=lambda _b: self.dismiss_dialog())), ]), ]) self._overlay(DialogLineBox(urwid.Filler(pile, urwid.TOP), parent=self, title=title)) def keypress(self, size, key): if key in ("tab", "esc"): self.parent.show_interfaces(); return None if key == "ctrl a": self.new_profile(); return None if key == "ctrl s": self.save_current(); return None if key == "ctrl e": self.rename_profile(); return None if key == "ctrl x": self.delete_profile(); return None return super().keypress(size, key) ### MAIN DISPLAY ### class InterfaceDisplay: def __init__(self, app): self.app = app self.started = False self.poll_scheduler = False self.size_scheduler = False self.interface_items = [] self.glyphset = self.app.config["textui"]["glyphs"] self.g = self.app.ui.glyphs self.terminal_cols, self.terminal_rows = _get_cols_rows() self.iface_row_offset = 4 self.list_rows = self.terminal_rows - self.iface_row_offset self.active_tab = "interfaces" self.iface_view_mode = "rows" self.iface_filter = "all" self.total_interfaces = 0 self.restart_pending = False self._notice_shown = False self._rns_disconnected = False self.profiles_view = None self._build_interface_items() self.list_walker = urwid.SimpleFocusListWalker(self._interface_list_contents()) self.list_box = urwid.ListBox(self.list_walker) self.body_rows = max(3, self.list_rows - 4) self.box_adapter = urwid.BoxAdapter(self.list_box, self.body_rows) self.tabs_pile = urwid.Pile([ ('pack', self._make_tab_bar("interfaces")), ('pack', self._make_filter_bar()), ('pack', urwid.Divider()), self.box_adapter, ]) self.interfaces_display = InterfaceFiller(self.tabs_pile, self.app) self.shortcuts_display = InterfaceDisplayShortcuts(self.app) self.widget = self.interfaces_display def _empty_message(self): if self.iface_filter != "all" and self.total_interfaces > 0: return "No %s interfaces." % self.iface_filter return "No interfaces found. Press Ctrl + A to add a new interface " def _interface_list_contents(self): if not self.interface_items: return [urwid.Text(("interface_title", self._empty_message()), align="center")] return list(self.interface_items) def _make_filter_bar(self): cols = [('weight', 1, urwid.Text(""))] cols.append(('pack', urwid.Text(("interface_title", "Filter by: ")))) for key, label in (("all", "All"), ("enabled", "Enabled"), ("disabled", "Disabled"), ("connected", "Connected"), ("disconnected", "Disconnected")): sel = self.iface_filter == key mark = self.g['selected'] if sel else self.g['unselected'] style = "interface_tile_focus" if sel else "interface_title" cols.append(('pack', ClickableIcon((style, "%s %s " % (mark, label)), on_click=lambda k=key: self._set_filter(k)))) cols.append(('weight', 1, urwid.Text(""))) return urwid.Columns(cols, dividechars=0) def _set_filter(self, key): if key == self.iface_filter: return self.iface_filter = key self._build_interface_items() self._rebuild_list() self.app.ui.main_display.update_active_sub_display() def cycle_filter(self): order = ["all", "enabled", "disabled", "connected", "disconnected"] i = order.index(self.iface_filter) if self.iface_filter in order else 0 self._set_filter(order[(i + 1) % len(order)]) def _make_tab_bar(self, active): iface = f" Interfaces ({self.total_interfaces}) " prof = " Profiles " if active == "interfaces": iface_w = ClickableIcon(("interface_title_selected", "["+iface+"]"), on_click=self.show_interfaces) prof_w = ClickableIcon(("interface_title", " "+prof+" "), on_click=self.show_profiles) else: iface_w = ClickableIcon(("interface_title", " "+iface+" "), on_click=self.show_interfaces) prof_w = ClickableIcon(("interface_title_selected", "["+prof+"]"), on_click=self.show_profiles) return urwid.Columns([ ('weight', 1, urwid.Text("")), ('pack', iface_w), ('pack', urwid.Text(" ")), ('pack', prof_w), ('weight', 1, urwid.Text("")), ], dividechars=0) def _build_grid_listbox(self): tiles = self.interface_tiles if self.interface_tiles else [urwid.Text(self._empty_message(), align="center")] self.iface_grid = urwid.GridFlow(tiles, cell_width=34, h_sep=2, v_sep=1, align="center") return urwid.ListBox(urwid.SimpleFocusListWalker([self.iface_grid])) def _set_iface_body(self): if self.iface_view_mode == "grid": self.box_adapter.original_widget = self._build_grid_listbox() else: self.list_walker = urwid.SimpleFocusListWalker(self._interface_list_contents()) self.list_box = urwid.ListBox(self.list_walker) self.box_adapter.original_widget = self.list_box def _focused_interface_name(self): try: if self.iface_view_mode == "grid": return getattr(self.iface_grid.focus, "name", None) fw, _ = self.box_adapter._original_widget.body.get_focus() return getattr(fw, "name", None) if isinstance(fw, SelectableInterfaceItem) else None except Exception: return None def toggle_view_mode(self): self.iface_view_mode = "grid" if self.iface_view_mode == "rows" else "rows" self._set_iface_body() self.app.ui.main_display.update_active_sub_display() def show_profiles(self): if getattr(self.app, "interface_profiles", None) is None: return self.active_tab = "profiles" self.profiles_view = ProfilesView(self) self.shortcuts_display.set_profiles_tab_shortcuts() self.widget = self.profiles_view self.app.ui.main_display.update_active_sub_display() def show_interfaces(self): self.active_tab = "interfaces" self._build_interface_items() self._rebuild_list() self.shortcuts_display.reset_shortcuts() self.widget = self.interfaces_display self.app.ui.main_display.update_active_sub_display() def start(self): # started from Main.py self.started = True if not self.poll_scheduler: self.app.ui.loop.set_alarm_in(1, self.poll_stats) if not self.size_scheduler: self.app.ui.loop.set_alarm_in(5, self.check_terminal_size) def switch_to_edit_interface(self, iface_name): self.edit_interface_view = EditInterfaceView(self, iface_name) self.widget = self.edit_interface_view self.app.ui.main_display.update_active_sub_display() def edit_selected_interface(self): interface_name = self._focused_interface_name() if interface_name is None: return self.switch_to_edit_interface(interface_name) def check_terminal_size(self, loop, user_data): self.size_scheduler = True new_cols, new_rows = _get_cols_rows() if new_rows != self.terminal_rows or new_cols != self.terminal_cols: self.terminal_cols, self.terminal_rows = new_cols, new_rows self.list_rows = self.terminal_rows - self.iface_row_offset self.body_rows = max(3, self.list_rows - 4 - (1 if self.restart_pending else 0)) self.box_adapter.height = self.body_rows loop.draw_screen() if self.started: loop.set_alarm_in(5, self.check_terminal_size) def _show_disconnect_overlay(self): # Surfaced when the shared RNS instance (rnsd) we are attached to goes away. if hasattr(self, 'disconnect_overlay') and self.widget is self.disconnect_overlay: return if self.widget is not self.interfaces_display: return dialog_text = urwid.Pile([ urwid.Text(("disconnected_status", "(!) RNS Instance Disconnected"), align="center"), urwid.Text("Waiting to Reconnect...", align="center"), ]) dialog_box = urwid.LineBox(urwid.Filler(dialog_text)) self.disconnect_overlay = urwid.Overlay( dialog_box, self.interfaces_display, align='center', width=35, valign='middle', height=4, ) self.widget = self.disconnect_overlay self.app.ui.main_display.update_active_sub_display() def poll_stats(self, loop, user_data): self.poll_scheduler = True try: # React to the outcome of the previous (asynchronous) stats fetch. The # fetch runs in a daemon thread so the UI never blocks on the RPC round # trip; the thread cannot raise into this loop, so it reports failure via # the _rns_disconnected flag instead. if self._rns_disconnected: self._show_disconnect_overlay() else: if hasattr(self, 'disconnect_overlay') and self.widget is self.disconnect_overlay: self.widget = self.interfaces_display self.app.ui.main_display.update_active_sub_display() interface_stats = self.app.interface_stats stats_lookup = {iface['short_name']: iface for iface in interface_stats['interfaces']} for item in self.interface_items: # use interface name as the key stats_for_interface = stats_lookup.get(item.name) if stats_for_interface: tx = stats_for_interface.get("txb", 0) rx = stats_for_interface.get("rxb", 0) announce_rate = format_announce_rate(stats_for_interface.get("incoming_announce_frequency", 0)) item.update_stats(tx, rx, announce_rate) item.update_connection(stats_for_interface.get("status", False)) else: item.update_connection(False) def job(): try: self.app.interface_stats = self.app.rns.get_interface_stats() self._rns_disconnected = False except Exception: self._rns_disconnected = True threading.Thread(target=job, daemon=True).start() except Exception as e: self._show_disconnect_overlay() finally: if self.started: loop.set_alarm_in(1, self.poll_stats) def shortcuts(self): return self.shortcuts_display def switch_to_show_interface(self, iface_name): show_interface = ShowInterface(self, iface_name) self.widget = show_interface self.app.ui.main_display.update_active_sub_display() show_interface.start() def switch_to_list(self): self.show_interfaces() def add_interface(self): dialog_widgets = [] def add_heading(txt): dialog_widgets.append(urwid.Text(("interface_title", txt), align="left")) def add_option(label, value): item = InterfaceOptionItem(self, label, value) dialog_widgets.append(item) add_heading("Paste Configuration") dialog_widgets.append(InterfaceOptionItem(self, "Paste an interface config…", None, on_select=self.show_paste_config_dialog)) dialog_widgets.append(urwid.Divider()) # Get the icons based on plain, unicode, nerdfont glyphset network_icon = _get_interface_icon(self.glyphset, "AutoInterface") rnode_icon = _get_interface_icon(self.glyphset, "RNodeInterface") serial_icon = _get_interface_icon(self.glyphset, "SerialInterface") other_icon = _get_interface_icon(self.glyphset, "PipeInterface") add_heading(f"{network_icon} IP Networks") add_option("Auto Interface", "AutoInterface") add_option("TCP Client Interface", "TCPClientInterface") add_option("TCP Server Interface", "TCPServerInterface") add_option("UDP Interface", "UDPInterface") add_option("I2P Interface", "I2PInterface") if PLATFORM_IS_LINUX: add_option("Backbone Interface", "BackboneInterface") if PYSERIAL_AVAILABLE: add_heading(f"{rnode_icon} RNodes") add_option("RNode Interface", "RNodeInterface") add_option("RNode Multi Interface", "RNodeMultiInterface") add_heading(f"{serial_icon} Hardware") add_option("Serial Interface", "SerialInterface") add_option("KISS Interface", "KISSInterface") add_option("AX.25 KISS Interface", "AX25KISSInterface") add_heading(f"{other_icon} Other") add_option("Pipe Interface", "PipeInterface") add_option("Custom Interface", "CustomInterface") listbox = urwid.ListBox(urwid.SimpleFocusListWalker(dialog_widgets)) dialog = DialogLineBox(listbox, parent=self, title="Select Interface Type") overlay = urwid.Overlay( dialog, self.interfaces_display, align='center', width=('relative', 50), valign='middle', height=('relative', 50), min_width=20, min_height=15, left=2, right=2 ) self.widget = overlay self.app.ui.main_display.update_active_sub_display() def show_paste_config_dialog(self): self.paste_edit = ReadlineEdit(caption="", multiline=True) self.paste_status = urwid.Text("") instructions = urwid.Text( "Paste an interface block, including its [[Name]] header, then Create. " "Multiple interfaces are supported.", align="left") editor = urwid.AttrMap( urwid.LineBox(urwid.BoxAdapter(urwid.ListBox(urwid.SimpleListWalker([self.paste_edit])), 12)), "list_off_focus", "list_focus") pile = urwid.Pile([ instructions, urwid.Divider(), editor, self.paste_status, urwid.Divider(), urwid.Columns([ (urwid.WEIGHT, 0.45, urwid.Button("Create", on_press=lambda _b: self._create_interfaces_from_text(self.paste_edit.get_edit_text()))), (urwid.WEIGHT, 0.1, urwid.Text("")), (urwid.WEIGHT, 0.45, urwid.Button("Cancel", on_press=lambda _b: self.dismiss_dialog())), ]), ]) dialog = DialogLineBox(urwid.Filler(pile, valign="top"), parent=self, title="Paste Interface Config") overlay = urwid.Overlay( dialog, self.interfaces_display, align="center", width=("relative", 75), valign="middle", height=("relative", 75), min_width=44, min_height=14, ) self.widget = overlay self.app.ui.main_display.update_active_sub_display() def _section_to_dict(self, sec): out = {} for k, v in sec.items(): out[k] = self._section_to_dict(v) if hasattr(v, "items") else v return out def _parse_pasted_interfaces(self, text): import RNS.vendor.configobj as configobj raw = (text or "").strip("\n") if not raw.strip(): return {} for prefix in ("", "[interfaces]\n"): try: parsed = configobj.ConfigObj(infile=(prefix + raw).split("\n")) except Exception: continue found = {} sections = [] ifsec = parsed.get("interfaces") if hasattr(ifsec, "items"): sections.extend(ifsec.items()) for name, sec in parsed.items(): if name != "interfaces" and hasattr(sec, "items"): sections.append((name, sec)) for name, sec in sections: if not hasattr(sec, "items"): continue d = self._section_to_dict(sec) if "type" in d and name not in found: found[name] = d if found: return found return {} def _create_interfaces_from_text(self, text): parsed = self._parse_pasted_interfaces(text) if not parsed: self.paste_status.set_text(("error", " Could not find an interface with a 'type'. Include the [[Name]] header.")) return interfaces = self.app.rns.config['interfaces'] existing = [n for n in parsed if n in interfaces] if existing: self.paste_status.set_text(("error", " Already exists: " + ", ".join(existing))) return try: for name, conf in parsed.items(): interfaces[name] = conf write_config_atomic(self.app.rns.config) except Exception as e: self.paste_status.set_text(("error", " Failed: " + str(e))) return self.mark_restart_pending() self.dismiss_dialog() self.show_interfaces() def switch_to_add_interface(self, iface_type): self.add_interface_view = AddInterfaceView(self, iface_type) self.widget = self.add_interface_view self.app.ui.main_display.update_active_sub_display() def remove_selected_interface(self): interface_name = self._focused_interface_name() if interface_name is None: return def on_confirm_yes(button): try: if interface_name in self.app.rns.config['interfaces']: del self.app.rns.config['interfaces'][interface_name] write_config_atomic(self.app.rns.config) try: self.app.interface_profiles.remove_interface(interface_name) except Exception: pass self.mark_restart_pending() self._build_interface_items() self._rebuild_list() self.dismiss_dialog() except Exception as e: print(e) def on_confirm_no(button): self.dismiss_dialog() confirm_text = urwid.Text(("interface_title", f"Remove interface {interface_name}?"), align="center") yes_button = urwid.Button("Yes", on_press=on_confirm_yes) no_button = urwid.Button("No", on_press=on_confirm_no) buttons_row = urwid.Columns([ (urwid.WEIGHT, 0.45, yes_button), (urwid.WEIGHT, 0.1, urwid.Text("")), (urwid.WEIGHT, 0.45, no_button), ]) pile = urwid.Pile([ confirm_text, buttons_row ]) dialog = DialogLineBox(pile, parent=self, title="?") overlay = urwid.Overlay( dialog, self.interfaces_display, align='center', width=('relative', 35), valign='middle', height=(5), min_width=5, left=2, right=2 ) dialog.original_widget.focus_position = 1 # columns row buttons_row = dialog.original_widget.contents[1][0] buttons_row.focus_position = 2 # second button "No" self.widget = overlay self.app.ui.main_display.update_active_sub_display() def dismiss_dialog(self): self.widget = self.interfaces_display self.app.ui.main_display.update_active_sub_display() def _build_interface_items(self): self.interface_items = [] self.interface_tiles = [] interfaces = self.app.rns.config['interfaces'] processed_interfaces = {} for interface_name, interface in interfaces.items(): interface_data = interface.copy() if interface_data.get("type") == "RNodeMultiInterface": sub_interfaces = [] for sub_name, sub_config in interface_data.items(): if sub_name not in {"type", "port", "interface_enabled", "selected_interface_mode", "configured_bitrate"}: if isinstance(sub_config, dict): sub_config["name"] = sub_name sub_interfaces.append(sub_config) interface_data["sub_interfaces"] = sub_interfaces for sub in sub_interfaces: del interface_data[sub["name"]] processed_interfaces[interface_name] = interface_data self.total_interfaces = len(processed_interfaces) self.app.interface_stats = self.app.rns.get_interface_stats() stats_lookup = {interface['short_name']: interface for interface in self.app.interface_stats['interfaces']} profiles = getattr(self.app, "interface_profiles", None) for interface_name, interface_data in processed_interfaces.items(): is_enabled = str(interface_data.get("enabled")).lower() not in ('false', 'off', 'no', '0') and str(interface_data.get("interface_enabled")).lower() not in ('false', 'off', 'no', '0') if self.iface_filter == "enabled" and not is_enabled: continue if self.iface_filter == "disabled" and is_enabled: continue iface_type = interface_data.get("type", "Unknown") icon = _get_interface_icon(self.glyphset, iface_type) stats_for_interface = stats_lookup.get(interface_name) if stats_for_interface: tx = stats_for_interface.get("txb", 0) rx = stats_for_interface.get("rxb", 0) is_connected = stats_for_interface["status"] announce_rate = format_announce_rate(stats_for_interface.get("incoming_announce_frequency", 0)) else: tx = 0 rx = 0 is_connected = False announce_rate = "-" if self.iface_filter == "connected" and not is_connected: continue if self.iface_filter == "disconnected" and is_connected: continue profiles_label = profiles.label_for(interface_name) if profiles is not None else None item = SelectableInterfaceItem( parent=self, name=interface_data.get("name", interface_name), is_connected=is_connected, is_enabled=is_enabled, iface_type=iface_type, tx=tx, rx=rx, icon=icon, profiles_label=profiles_label, announce_rate=announce_rate ) self.interface_items.append(item) tile = InterfaceTile( parent=self, name=interface_data.get("name", interface_name), is_connected=is_connected, is_enabled=is_enabled, iface_type=iface_type, icon=icon, announce_rate=announce_rate, ) self.interface_tiles.append(tile) def _rebuild_list(self): self._set_iface_body() self.tabs_pile.contents[0] = (self._make_tab_bar("interfaces"), self.tabs_pile.options('pack')) self.tabs_pile.contents[1] = (self._make_filter_bar(), self.tabs_pile.options('pack')) def mark_restart_pending(self): self.restart_pending = True if self._notice_shown: return try: mark = "!" if self.glyphset == "plain" else "⚠" banner = urwid.AttrMap( urwid.Text("%s Interfaces have been updated. Restart required" % mark, align="center"), "warning_text") self.tabs_pile.contents.insert(2, (banner, self.tabs_pile.options('pack'))) self._notice_shown = True self.body_rows = max(3, self.body_rows - 1) self.box_adapter.height = self.body_rows except Exception: pass def _focus_interface(self, name): try: if self.iface_view_mode == "grid": for i, t in enumerate(self.interface_tiles): if getattr(t, "name", None) == name: self.iface_grid.focus_position = i return else: body = self.box_adapter._original_widget.body for i, w in enumerate(body): if getattr(w, "name", None) == name: self.box_adapter._original_widget.focus_position = i return except Exception: pass def toggle_focused_interface(self): name = self._focused_interface_name() if name is None: return interfaces = self.app.rns.config['interfaces'] if name not in interfaces: return iface = interfaces[name] is_enabled = str(iface.get("enabled")).lower() not in ('false', 'off', 'no', '0') and \ str(iface.get("interface_enabled")).lower() not in ('false', 'off', 'no', '0') value = "false" if is_enabled else "true" iface["interface_enabled"] = value if "enabled" in iface: iface["enabled"] = value try: write_config_atomic(self.app.rns.config) except Exception: return self.mark_restart_pending() self._build_interface_items() self._rebuild_list() self._focus_interface(name) self.app.ui.main_display.update_active_sub_display() def _iface_is_enabled(self, iface): return str(iface.get("enabled")).lower() not in ('false', 'off', 'no', '0') and \ str(iface.get("interface_enabled")).lower() not in ('false', 'off', 'no', '0') def _set_iface_enabled(self, iface, enabled): value = "true" if enabled else "false" iface["interface_enabled"] = value if "enabled" in iface: iface["enabled"] = value def show_bulk_actions(self): interfaces = self.app.rns.config['interfaces'] if not interfaces: self._bulk_notice("No interfaces configured.") return dialog_widgets = [ urwid.Text(("interface_title", "Apply an action to all interfaces"), align="left"), urwid.Divider(), InterfaceOptionItem(self, "Enable all interfaces", None, on_select=lambda: self._bulk_apply("enable")), InterfaceOptionItem(self, "Disable all interfaces", None, on_select=lambda: self._bulk_apply("disable")), InterfaceOptionItem(self, "Invert (toggle each interface)", None, on_select=lambda: self._bulk_apply("invert")), InterfaceOptionItem(self, "Solo selected (disable all others)", None, on_select=lambda: self._bulk_apply("solo")), ] listbox = urwid.ListBox(urwid.SimpleFocusListWalker(dialog_widgets)) dialog = DialogLineBox(listbox, parent=self, title="Bulk Actions") overlay = urwid.Overlay( dialog, self.interfaces_display, align='center', width=('relative', 50), valign='middle', height=('relative', 50), min_width=20, min_height=10, left=2, right=2 ) self.widget = overlay self.app.ui.main_display.update_active_sub_display() def _bulk_apply(self, action): interfaces = self.app.rns.config['interfaces'] focused = self._focused_interface_name() if action == "solo" and focused is None: self._bulk_notice("Select an interface first to solo it.") return changed = 0 for name, iface in interfaces.items(): current = self._iface_is_enabled(iface) if action == "enable": target = True elif action == "disable": target = False elif action == "invert": target = not current elif action == "solo": target = (name == focused) else: target = current if target != current: self._set_iface_enabled(iface, target) changed += 1 try: write_config_atomic(self.app.rns.config) except Exception as e: self._bulk_notice(f"Error applying bulk action: {str(e)}", title="Error") return self.mark_restart_pending() self._build_interface_items() self._rebuild_list() verb = {"enable": "Enabled all", "disable": "Disabled all", "invert": "Inverted all", "solo": "Soloed selected"}[action] if changed == 0: self._bulk_notice("No interfaces needed changing.") else: self._bulk_notice(f"{verb} interfaces ({changed} changed).\nRestart required for changes to take effect.") def _bulk_notice(self, message, title="Notice"): def dismiss_dialog(button): self.dismiss_dialog() dialog = DialogLineBox( urwid.Pile([ urwid.Text(message, align="center"), urwid.Divider(), urwid.Button("OK", on_press=dismiss_dialog) ]), parent=self, title=title ) overlay = urwid.Overlay( dialog, self.interfaces_display, align='center', width=60, valign='middle', height=8, min_width=20, min_height=1 ) self.widget = overlay self.app.ui.main_display.update_active_sub_display() def save_current_profile(self): profiles = getattr(self.app, "interface_profiles", None) if profiles is None: return n = len(profiles.enabled_set()) edit = ReadlineEdit(caption="Name: ") def ok(_b): val = edit.get_edit_text().strip() self.dismiss_dialog() if val: profiles.save_current_as_profile(val) self.show_interfaces() pile = urwid.Pile([ urwid.Text("Save the %d currently enabled interface%s as a new profile." % (n, "" if n == 1 else "s"), align="left"), urwid.Divider(), edit, urwid.Divider(), urwid.Columns([ (urwid.WEIGHT, 0.45, urwid.Button("Save", on_press=ok)), (urwid.WEIGHT, 0.1, urwid.Text("")), (urwid.WEIGHT, 0.45, urwid.Button("Cancel", on_press=lambda _b: self.dismiss_dialog())), ]), ]) dialog = DialogLineBox(urwid.Filler(pile, valign="top"), parent=self, title="Save Current as Profile") overlay = urwid.Overlay( dialog, self.interfaces_display, align="center", width=("relative", 60), valign="middle", height=("relative", 45), min_width=44, min_height=9, ) self.widget = overlay self.app.ui.main_display.update_active_sub_display() def open_config_editor(self): import platform editor_cmd = self.app.config["textui"]["editor"] if platform.system() == "Darwin" and editor_cmd == "editor": editor_cmd = "nano" editor_term = urwid.Terminal( (editor_cmd, self.app.rns.configpath), encoding='utf-8', main_loop=self.app.ui.loop, ) def quit_term(*args, **kwargs): self.widget = self.interfaces_display self.app.ui.main_display.update_active_sub_display() self.app.ui.main_display.request_redraw() urwid.connect_signal(editor_term, 'closed', quit_term) editor_box = urwid.LineBox(editor_term, title="Editing RNS Config") self.widget = editor_box self.app.ui.main_display.update_active_sub_display() self.app.ui.main_display.frame.focus_position = "body" editor_term.change_focus(True) ### SHORTCUTS ### class InterfaceDisplayShortcuts: def __init__(self, app): self.app = app self.default_shortcuts = "[C-a] Add [C-e] Edit [C-x] Remove [C-o] On/Off [C-b] Bulk [Enter] Show [C-f] Filter [C-s] Save Profile [C-t] Grid/Rows [Tab] Profiles" self.current_shortcuts = self.default_shortcuts self.widget = urwid.AttrMap( urwid.Text(self.current_shortcuts), "shortcutbar" ) def update_shortcuts(self, new_shortcuts): self.current_shortcuts = new_shortcuts self.widget.original_widget.set_text(new_shortcuts) def reset_shortcuts(self): self.update_shortcuts(self.default_shortcuts) def set_show_interface_shortcuts(self): show_shortcuts = "[Up/Down] Navigate [Tab] Switch Focus [h] Horizontal Charts [v] Vertical Charts " self.update_shortcuts(show_shortcuts) def set_add_interface_shortcuts(self): add_shortcuts = "[Up/Down] Navigate Fields [Enter] Select Option" self.update_shortcuts(add_shortcuts) def set_edit_interface_shortcuts(self): edit_shortcuts = "[Up/Down] Navigate Fields [Enter] Select Option" self.update_shortcuts(edit_shortcuts) def set_profiles_tab_shortcuts(self): profiles_shortcuts = "[Space] Select [C-a] New [C-s] Save Current [C-e] Rename [C-x] Delete [Tab] Interfaces" self.update_shortcuts(profiles_shortcuts)