From c280e36a84c5cb3ac6bb60e285b9c53676533a1a Mon Sep 17 00:00:00 2001 From: RFNexus Date: Sun, 9 Mar 2025 15:20:36 -0400 Subject: [PATCH] Interface Management (1/2) --- nomadnet/ui/TextUI.py | 22 +- nomadnet/ui/textui/Guide.py | 6 + nomadnet/ui/textui/Interfaces.py | 2865 +++++++++++++++++ nomadnet/ui/textui/Main.py | 30 +- nomadnet/vendor/AsciiChart.py | 87 + .../additional_urwid_widgets/FormWidgets.py | 283 ++ 6 files changed, 3280 insertions(+), 13 deletions(-) create mode 100644 nomadnet/ui/textui/Interfaces.py create mode 100644 nomadnet/vendor/AsciiChart.py create mode 100644 nomadnet/vendor/additional_urwid_widgets/FormWidgets.py diff --git a/nomadnet/ui/TextUI.py b/nomadnet/ui/TextUI.py index 3e4b3cf..11dd2c7 100644 --- a/nomadnet/ui/TextUI.py +++ b/nomadnet/ui/TextUI.py @@ -51,7 +51,15 @@ THEMES = { ("browser_controls", "light gray", "default", "default", "#bbb", "default"), ("progress_full", "black", "light gray", "standout", "#111", "#bbb"), ("progress_empty", "light gray", "default", "default", "#ddd", "default"), - ], + ("interface_title", "", "", "default", "", ""), + ("interface_title_selected", "bold", "", "bold", "", ""), + ("connected_status", "dark green", "default", "default", "dark green", "default"), + ("disconnected_status", "dark red", "default", "default", "dark red", "default"), + ("placeholder", "dark gray", "default", "default", "dark gray", "default"), + ("placeholder_text", "dark gray", "default", "default", "dark gray", "default"), + ("error", "light red,blink", "default", "blink", "#f44,blink", "default"), + + ], }, THEME_LIGHT: { "urwid_theme": [ @@ -69,6 +77,7 @@ THEMES = { ("msg_header_ok", "black", "dark green", "standout", "#111", "#6b2"), ("msg_header_caution", "black", "yellow", "standout", "#111", "#fd3"), ("msg_header_sent", "black", "dark gray", "standout", "#111", "#ddd"), + ("msg_header_propagated", "black", "light blue", "standout", "#111", "#28b"), ("msg_header_delivered", "black", "light blue", "standout", "#111", "#28b"), ("msg_header_failed", "black", "dark gray", "standout", "#000", "#777"), ("msg_warning_untrusted", "black", "dark red", "standout", "#111", "dark red"), @@ -86,6 +95,13 @@ THEMES = { ("browser_controls", "dark gray", "default", "default", "#444", "default"), ("progress_full", "black", "dark gray", "standout", "#111", "#bbb"), ("progress_empty", "dark gray", "default", "default", "#ddd", "default"), + ("interface_title", "dark gray", "default", "default", "#444", "default"), + ("interface_title_selected", "dark gray,bold", "default", "bold", "#444,bold", "default"), + ("connected_status", "dark green", "default", "default", "#4a0", "default"), + ("disconnected_status", "dark red", "default", "default", "#a22", "default"), + ("placeholder", "light gray", "default", "default", "#999", "default"), + ("placeholder_text", "light gray", "default", "default", "#999", "default"), + ("error", "dark red,blink", "default", "blink", "#a22,blink", "default"), ], } } @@ -128,6 +144,8 @@ GLYPHS = { ("sent", "/\\", "\u2191", "\U000f0cd8"), ("papermsg", "P", "\u25a4", "\uf719"), ("qrcode", "QR", "\u25a4", "\uf029"), + ("selected", "[*] ", "\u25CF", "\u25CF"), + ("unselected", "[ ] ", "\u25CB", "\u25CB"), } class TextUI: @@ -163,7 +181,7 @@ class TextUI: if self.app.config["textui"]["glyphs"] == "plain": glyphset = "plain" - elif self.app.config["textui"]["glyphs"] == "unicoode": + elif self.app.config["textui"]["glyphs"] == "unicode": glyphset = "unicode" elif self.app.config["textui"]["glyphs"] == "nerdfont": glyphset = "nerdfont" diff --git a/nomadnet/ui/textui/Guide.py b/nomadnet/ui/textui/Guide.py index 6308edc..94f83d0 100644 --- a/nomadnet/ui/textui/Guide.py +++ b/nomadnet/ui/textui/Guide.py @@ -109,6 +109,7 @@ class TopicList(urwid.WidgetWrap): self.topic_list = [ GuideEntry(self.app, self, guide_display, "Introduction"), GuideEntry(self.app, self, guide_display, "Concepts & Terminology"), + GuideEntry(self.app, self, guide_display, "Interfaces"), GuideEntry(self.app, self, guide_display, "Hosting a Node"), GuideEntry(self.app, self, guide_display, "Configuration Options"), GuideEntry(self.app, self, guide_display, "Keyboard Shortcuts"), @@ -386,6 +387,10 @@ Links can be inserted into micron documents. See the `*Markup`* section of this ''' +TOPIC_INTERFACES = ''' +>TODO +''' + TOPIC_CONVERSATIONS = '''>Conversations Conversations in Nomad Network @@ -1247,6 +1252,7 @@ TOPICS = { "Introduction": TOPIC_INTRODUCTION, "Concepts & Terminology": TOPIC_CONCEPTS, "Conversations": TOPIC_CONVERSATIONS, + "Interfaces": TOPIC_INTERFACES, "Hosting a Node": TOPIC_HOSTING, "Configuration Options": TOPIC_CONFIG, "Keyboard Shortcuts": TOPIC_SHORTCUTS, diff --git a/nomadnet/ui/textui/Interfaces.py b/nomadnet/ui/textui/Interfaces.py new file mode 100644 index 0000000..281e505 --- /dev/null +++ b/nomadnet/ui/textui/Interfaces.py @@ -0,0 +1,2865 @@ +import RNS +import time +from math import log10, pow + +from nomadnet.vendor.additional_urwid_widgets.FormWidgets import * +from nomadnet.vendor.AsciiChart import AsciiChart + +### GYLPHS ### +INTERFACE_GLYPHS = { + # Glyph name # Plain # Unicode # Nerd Font + ("NetworkInterfaceType", "(IP)", "\U0001f5a7", "\U000f0200"), + ("SerialInterfaceType", "(<->)", "\u2194", "\U000f065c"), + ("RNodeInterfaceType", "(R)" , "\u16b1", "\U000f043a"), + ("OtherInterfaceType", "(#)" , "\U0001f67e", "\ued95"), +} + +### HELPER ### +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 = { + "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 "🙾" if glyphset == "unicode" else " " + +def format_bytes(bytes_value): + units = ['bytes', 'KB/s', 'MB/s', 'GB/s', 'TB/s'] + 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]}" + +### 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 = urwid.Edit("", "0") + self.antenna_gain_edit = urwid.Edit("", "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 = { + "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": ["required", "number"], + "transform": lambda x: int(x.strip()) if x.strip() else 4242 + }, + { + "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() + }, + { + "config_key": "port", + "type": "edit", + "label": "Port: ", + "default": "", + "placeholder": "e.g., 4242", + "validation": ["number"], + "transform": lambda x: int(x.strip()) if x.strip() else 4242 + }, + ] + } + ], + "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., 8080", + "validation": ["required", "number"], + "transform": lambda x: int(x.strip()) if x.strip() else 4242 + }, + { + "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": ["required", "number"], + "transform": lambda x: int(x.strip()) if x.strip() else 4242 + }, + { + "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 4242 + }, + { + "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 4242 + }, + ] + } + ], + "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) + }, + ] + } + ], + "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) + }, + ] + } + ], + "default": [ + { + + }, + ] +} + +### INTERFACE WIDGETS #### +class SelectableInterfaceItem(urwid.WidgetWrap): + def __init__(self, parent, name, is_connected, is_enabled, iface_type, tx, rx, icon="?", iface_options=None): + 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 = ("disconnected_status", "Disabled") + + if is_connected: + connected_txt = ("connected_status", "Connected") + else: + connected_txt = ("disconnected_status", "Disconnected") + + 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.status_widget = urwid.Text(enabled_txt) + self.connection_widget = urwid.Text(connected_txt) + + rows = [ + urwid.Columns([ + (10, urwid.Text(("key", "Status: "))), + (10, self.status_widget), + (3, urwid.Text(" | ")), + self.connection_widget, + ]), + + urwid.Columns([ + (10, urwid.Text(("key", "Type:"))), + urwid.Text(("value", iface_type)), + ]), + + urwid.Divider("-"), + + urwid.Columns([ + (10, urwid.Text(("key", "TX:"))), + (15, self.tx_widget), + (10, urwid.Text(("key", "RX:"))), + self.rx_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(("disconnected_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 == "enter": + self.parent.switch_to_show_interface(self.name) + return None + return key + + def update_stats(self, tx, rx): + self.tx_widget.set_text(("value", format_bytes(tx))) + self.rx_widget.set_text(("value", format_bytes(rx))) + +class InterfaceOptionItem(urwid.WidgetWrap): + def __init__(self, parent_display, label, value): + self.parent_display = parent_display + self.label = label + self.value = value + 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 keypress(self, size, key): + if key == "enter": + self.parent_display.dismiss_dialog() + self.parent_display.switch_to_add_interface(self.value) + return None + return super().keypress(size, key) + +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) + self.tx_rates.append(tx_delta) + + 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 = format_bytes(peak_rx) + peak_tx_str = format_bytes(peak_tx) + + rx_chart = chart.plot( + [rx_data], + { + 'height': height, + 'format': '{:8.1f}', + 'min': 0, + 'max': self.max_rx_rate * 1.1, + } + ) + + tx_chart = chart.plot( + [tx_data], + { + 'height': height, + 'format': '{:8.1f}', + '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 == "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 == "tab": + # navigation + self.app.ui.main_display.frame.focus_position = "header" + return + + 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") + #todo + 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["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["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") + ) + 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") + ) + + self.fields[field["config_key"]] = { + 'label': field["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["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") + ) + 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") + ) + + self.additional_fields[option["config_key"]] = { + 'label': option["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"] + + 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", "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 == "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 == "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([ + urwid.Divider("─"), + button_row, + ]) + + return pile_items + + 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: + # pile_contents.append(urwid.Text(("interface_title", "Common "), align="center")) + 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 + + interface_config = { + "type": self.iface_type, + "interface_enabled": True + } + + for field_key, field in self.fields.items(): + if field_key != "name": + 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.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 + + # Add interface to RNS config + try: + interfaces = self.parent.app.rns.config['interfaces'] + + interfaces[name] = interface_config + + self.parent.app.rns.config.write() + print(self.parent.glyphset) + new_item = SelectableInterfaceItem( + parent=self.parent, + name=name, + is_connected=False, # will always be false until restart + is_enabled=True, + iface_type=self.iface_type, + tx=0, + rx=0, + icon=_get_interface_icon(self.parent.glyphset, self.iface_type), + iface_options=interface_config + ) + + self.parent.interface_items.append(new_item) + self.parent._rebuild_list() + + 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] + self.iface_type = self.interface_config.get("type", "Unknown") + + 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 + + for key, field in self.fields.items(): + if key != 'name' and key in self.interface_config: + widget = field['widget'] + value = self.interface_config[key] + + if key == 'frequency': + # convert Hz back to MHz + value = float(value) / 1000000 + # decimal format + value = f"{value:.6f}".rstrip('0').rstrip('.') if '.' in f"{value:.6f}" else f"{value}" + + if hasattr(widget, 'edit_text'): + # For text input fields, set edit_text + widget.edit_text = str(value) + elif hasattr(widget, 'set_state'): + # For checkboxes + widget.set_state(bool(value)) + elif isinstance(widget, FormDropdown): + # For dropdowns - update selected and update display + str_value = str(value) + if str_value in widget.options: + widget.selected = str_value + widget.main_button.base_widget.set_text(str_value) + else: + 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): + if isinstance(value, str): + items = [item.strip() for item in value.split(',') if item.strip()] + self._populate_multilist(widget, items) + elif isinstance(value, list): + self._populate_multilist(widget, value) + + for key, field in self.additional_fields.items(): + if key in self.interface_config: + widget = field['widget'] + value = self.interface_config[key] + + if hasattr(widget, 'edit_text'): + widget.edit_text = str(value) + elif hasattr(widget, 'set_state'): + # RNS.log(f"KEY: {key} VAL: {value}") + 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): + if isinstance(value, str): + items = [item.strip() for item in value.split(',') if item.strip()] + self._populate_multilist(widget, items) + elif isinstance(value, list): + self._populate_multilist(widget, value) + + for key, field in self.common_fields.items(): + if key in self.interface_config: + widget = field['widget'] + value = self.interface_config[key] + + if hasattr(widget, 'edit_text'): + widget.edit_text = str(value) + elif hasattr(widget, 'set_state'): + widget.set_state(bool(value)) + 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): + if isinstance(value, str): + items = [item.strip() for item in value.split(',') if item.strip()] + self._populate_multilist(widget, items) + elif isinstance(value, list): + self._populate_multilist(widget, value) + + def _populate_multilist(self, widget, items): + 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 on_save(self, button): + if not self.validate_all(): + return + + new_name = self.fields['name']["widget"].get_value() or self.iface_name + + updated_config = { + "type": self.iface_type, + "interface_enabled": True + } + + for field_key, field in self.fields.items(): + if field_key != "name": + 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.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 + + 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 + + self.parent.app.rns.config.write() + + for item in self.parent.interface_items: + if item.name == new_name: + item.iface_type = self.iface_type + break + + self.parent._rebuild_list() + + 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 = [] + + # get interface stats + interface_stats = self.parent.app.rns.get_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 "disconnected_status", + "Enabled" if self.is_enabled else "Disabled")) + + self.status_indicator = urwid.Text(("connected_status" if self.is_enabled else "disconnected_status", + self.parent.g['selected'] if self.is_enabled else self.parent.g[ + 'unselected'])) + + self.connection_text = urwid.Text(("connected_status" if self.is_connected else "disconnected_status", + "Connected" if self.is_connected else "Disconnected")) + + self.info_rows = [ + urwid.Columns([ + (10, urwid.Text(("key", "Type:"))), + urwid.Text(("value", f"{_get_interface_icon(self.parent.glyphset, iface_type)} {iface_type}")), + ]), + urwid.Columns([ + (10, urwid.Text(("key", "Status:"))), + (4, self.status_indicator), + (8, self.status_text), + (3, urwid.Text(" | ")), + self.connection_text, + ]), + urwid.Divider("-") + ] + + self.tx_text = urwid.Text(("value", format_bytes(self.tx))) + self.rx_text = urwid.Text(("value", format_bytes(self.rx))) + + self.stat_row = urwid.Columns([ + (10, urwid.Text(("key", "TX:"))), + (15, self.tx_text), + (10, urwid.Text(("key", "RX:"))), + self.rx_text, + ]) + + self.info_rows.append(self.stat_row) + self.info_rows.append(urwid.Divider("-")) + + self.bandwidth_chart = InterfaceBandwidthChart(history_length=60, 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(("disconnected_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 = urwid.raw_display.Screen() + screen_cols, _ = screen.get_cols_rows() + if screen_cols >= 100: + 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" + 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 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(("disconnected_status", self.parent.g['unselected'])) + self.status_text.set_text(("disconnected_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(("connected_status" if self.is_connected else "disconnected_status", + "Connected" if self.is_connected else "Disconnected")) + + 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 + if not self.started: + self.start() + else: + new_widget = self.disconnected_box + self.started = False + + body_pile.contents[chart_index] = (new_widget, body_pile.options()) + + def on_toggle_enabled(self, button): + action = "disable" if self.is_enabled else "enable" + + def on_confirm_yes(confirm_button): + self.parent.app.ui.main_display.frame.body = self.parent.app.ui.main_display.sub_displays.active().widget + + self.is_enabled = not self.is_enabled + + self.toggle_button.set_label("Disable" if self.is_enabled else "Enable") + + if "interface_enabled" in self.interface_config: + self.interface_config["interface_enabled"] = self.is_enabled + else: + self.interface_config["enabled"] = self.is_enabled + + try: + interfaces = self.parent.app.rns.config['interfaces'] + + interfaces[self.iface_name] = self.interface_config + + self.parent.app.rns.config.write() + + 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.show_restart_required_message() + + except Exception as e: + self.show_error_message(f"Error updating interface: {str(e)}") + + def on_confirm_no(confirm_button): + self.parent.app.ui.main_display.frame.body = self.parent.app.ui.main_display.sub_displays.active().widget + + confirm_text = urwid.Text(( + "interface_title", + f"Are you sure you want to {action} the {self.iface_name} interface?" + ), 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, + urwid.Divider(), + buttons_row + ]) + + dialog = DialogLineBox(pile, title="Confirm") + + overlay = urwid.Overlay( + dialog, + self.parent.app.ui.main_display.frame.body, + align='center', + width=50, + valign='middle', + height=7 + ) + + self.parent.app.ui.main_display.frame.body = overlay + + 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 + # If we're currently in footer change the focus + 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 == "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): + if not self.started and self.is_connected: + self.started = True + self.parent.app.ui.loop.set_alarm_in(1, self.update_bandwidth_charts) + + def update_bandwidth_charts(self, loop, user_data): + if not self.started: + return + + interface_stats = self.parent.app.rns.get_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) + + if not self.is_connected: + return + + self.tx_text.set_text(("value", format_bytes(tx))) + self.rx_text.set_text(("value", format_bytes(rx))) + + 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 + + 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) + +### MAIN DISPLAY ### +class InterfaceDisplay: + def __init__(self, app): + self.app = app + self.started = False + self.interface_items = [] + self.glyphset = self.app.config["textui"]["glyphs"] + self.g = self.app.ui.glyphs + + self.terminal_cols, self.terminal_rows = urwid.raw_display.Screen().get_cols_rows() + self.iface_row_offset = 4 + self.list_rows = self.terminal_rows - self.iface_row_offset + + interfaces = app.rns.config['interfaces'] + processed_interfaces = {} + + for interface_name, interface in interfaces.items(): + interface_data = interface.copy() + + # handle sub-interfaces for RNodeMultiInterface + 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) + + # add sub-interfaces to the main interface data + interface_data["sub_interfaces"] = sub_interfaces + + for sub in sub_interfaces: + del interface_data[sub["name"]] + + processed_interfaces[interface_name] = interface_data + + interface_stats = app.rns.get_interface_stats() + stats_lookup = {interface['short_name']: interface for interface in interface_stats['interfaces']} + # print(stats_lookup) + for interface_name, interface_data in processed_interfaces.items(): + # configobj false values + 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') + + 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"] + else: + tx = 0 + rx = 0 + is_connected = False + + 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 + ) + + self.interface_items.append(item) + + interface_header = urwid.Text(("interface_title", "Interfaces"), align="center") + if len(self.interface_items) == 0: + interface_header = urwid.Text( + ("interface_title", "No interfaces found. Press Ctrl + A to add a new interface "), align="center") + + + list_contents = [ + interface_header, + urwid.Divider(), + ] + self.interface_items + + self.list_walker = urwid.SimpleFocusListWalker(list_contents) + self.list_box = urwid.ListBox(self.list_walker) + + self.box_adapter = urwid.BoxAdapter(self.list_box, self.list_rows) + + + pile = urwid.Pile([self.box_adapter]) + self.interfaces_display = InterfaceFiller(pile, self.app) + self.shortcuts_display = InterfaceDisplayShortcuts(self.app) + self.widget = self.interfaces_display + + def start(self): + # started from Main.py + self.started = True + self.app.ui.loop.set_alarm_in(1, self.poll_stats) + 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): + focus_widget, focus_position = self.box_adapter._original_widget.body.get_focus() + + if not isinstance(focus_widget, SelectableInterfaceItem): + return + + selected_item = focus_widget + interface_name = selected_item.name + + self.switch_to_edit_interface(interface_name) + + def check_terminal_size(self, loop, user_data): + new_cols, new_rows = loop.screen.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.box_adapter.height = self.list_rows + + loop.draw_screen() + + if self.started: + loop.set_alarm_in(5, self.check_terminal_size) + + def poll_stats(self, loop, user_data): + interface_stats = self.app.rns.get_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) + item.update_stats(tx, rx) + + 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.shortcuts_display.reset_shortcuts() + self.widget = self.interfaces_display + self._rebuild_list() + self.app.ui.main_display.update_active_sub_display() + + 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) + + # 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") + + add_heading(f"{rnode_icon} RNodes") + add_option("RNode Interface", "RNodeInterface") + # add_option("RNode Multi Interface", "RNodeMultiInterface") TODO + + 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") TODO + + 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 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): + focus_widget, focus_position = self.box_adapter._original_widget.body.get_focus() + if not isinstance(focus_widget, SelectableInterfaceItem): + return + + selected_item = focus_widget + interface_name = selected_item.name + + def on_confirm_yes(button): + try: + if interface_name in self.app.rns.config['interfaces']: + del self.app.rns.config['interfaces'][interface_name] + self.app.rns.config.write() + + if selected_item in self.interface_items: + self.interface_items.remove(selected_item) + + 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 _rebuild_list(self): + interface_header = urwid.Text(("interface_title", f"Interfaces ({len(self.interface_items)})"), align="center") + if len(self.interface_items) == 0: + interface_header = urwid.Text(("interface_title", "No interfaces found. Press Ctrl + A to add a new interface "), align="center") + + new_list = [ + interface_header, + urwid.Divider(), + ] + self.interface_items + # RNS.log(f"items: {self.interface_items}") + + walker = urwid.SimpleFocusListWalker(new_list) + self.box_adapter._original_widget.body = walker + self.box_adapter._original_widget.focus_position = len(new_list) - 1 + +### SHORTCUTS ### +class InterfaceDisplayShortcuts: + def __init__(self, app): + self.app = app + self.default_shortcuts = "[C-a] Add Interface [C-e] Edit Interface [C-x] Remove Interface [Enter] Show Interface" + 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 = "[Back] Return to List [Tab] Navigate [Shift-tab] Change Focus [h] Horizontal Charts [v] Vertical Charts [Up/Down] Scroll" + 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) \ No newline at end of file diff --git a/nomadnet/ui/textui/Main.py b/nomadnet/ui/textui/Main.py index 303024a..34d88c3 100644 --- a/nomadnet/ui/textui/Main.py +++ b/nomadnet/ui/textui/Main.py @@ -4,6 +4,7 @@ from .Network import * from .Conversations import * from .Directory import * from .Config import * +from .Interfaces import * from .Map import * from .Log import * from .Guide import * @@ -16,6 +17,7 @@ class SubDisplays(): self.conversations_display = ConversationsDisplay(self.app) self.directory_display = DirectoryDisplay(self.app) self.config_display = ConfigDisplay(self.app) + self.interface_display = InterfaceDisplay(self.app) self.map_display = MapDisplay(self.app) self.log_display = LogDisplay(self.app) self.guide_display = GuideDisplay(self.app) @@ -113,6 +115,11 @@ class MainDisplay(): self.sub_displays.active_display = self.sub_displays.config_display self.update_active_sub_display() + def show_interfaces(self, user_data): + self.sub_displays.active_display = self.sub_displays.interface_display + self.update_active_sub_display() + self.sub_displays.interface_display.start() + def show_log(self, user_data): self.sub_displays.active_display = self.sub_displays.log_display self.sub_displays.log_display.show() @@ -171,21 +178,22 @@ class MenuDisplay(): self.menu_indicator = urwid.Text("") - menu_text = (urwid.PACK, self.menu_indicator) - button_network = (11, MenuButton("Network", on_press=handler.show_network)) - button_conversations = (17, MenuButton("Conversations", on_press=handler.show_conversations)) - button_directory = (13, MenuButton("Directory", on_press=handler.show_directory)) - button_map = (7, MenuButton("Map", on_press=handler.show_map)) - button_log = (7, MenuButton("Log", on_press=handler.show_log)) - button_config = (10, MenuButton("Config", on_press=handler.show_config)) - button_guide = (9, MenuButton("Guide", on_press=handler.show_guide)) - button_quit = (8, MenuButton("Quit", on_press=handler.quit)) + menu_text = (urwid.PACK, self.menu_indicator) + button_network = (11, MenuButton("Network", on_press=handler.show_network)) + button_conversations = (17, MenuButton("Conversations", on_press=handler.show_conversations)) + button_directory = (13, MenuButton("Directory", on_press=handler.show_directory)) + button_map = (7, MenuButton("Map", on_press=handler.show_map)) + button_log = (7, MenuButton("Log", on_press=handler.show_log)) + button_config = (10, MenuButton("Config", on_press=handler.show_config)) + button_interfaces = (14, MenuButton("Interfaces", on_press=handler.show_interfaces)) + button_guide = (9, MenuButton("Guide", on_press=handler.show_guide)) + button_quit = (8, MenuButton("Quit", on_press=handler.quit)) # buttons = [menu_text, button_conversations, button_node, button_directory, button_map] if self.app.config["textui"]["hide_guide"]: - buttons = [menu_text, button_conversations, button_network, button_log, button_config, button_quit] + buttons = [menu_text, button_conversations, button_network, button_log, button_interfaces, button_config, button_quit] else: - buttons = [menu_text, button_conversations, button_network, button_log, button_config, button_guide, button_quit] + buttons = [menu_text, button_conversations, button_network, button_log, button_interfaces, button_config, button_guide, button_quit] columns = MenuColumns(buttons, dividechars=1) columns.handler = handler diff --git a/nomadnet/vendor/AsciiChart.py b/nomadnet/vendor/AsciiChart.py new file mode 100644 index 0000000..4689590 --- /dev/null +++ b/nomadnet/vendor/AsciiChart.py @@ -0,0 +1,87 @@ +from __future__ import division +from math import ceil, floor, isnan +# Derived from asciichartpy | https://github.com/kroitor/asciichart/blob/master/asciichartpy/__init__.py +class AsciiChart: + def __init__(self, glyphset="unicode"): + self.symbols = ['┼', '┤', '╶', '╴', '─', '╰', '╭', '╮', '╯', '│'] + if glyphset == "plain": + self.symbols = ['+', '|', '-', '-', '-', '\'', ',', '.', '`', '|'] + def plot(self, series, cfg=None): + if len(series) == 0: + return '' + if not isinstance(series[0], list): + if all(isnan(n) for n in series): + return '' + else: + series = [series] + cfg = cfg or {} + minimum = cfg.get('min', min(filter(lambda n: not isnan(n), [j for i in series for j in i]))) + maximum = cfg.get('max', max(filter(lambda n: not isnan(n), [j for i in series for j in i]))) + symbols = cfg.get('symbols', self.symbols) + if minimum > maximum: + raise ValueError('The min value cannot exceed the max value.') + interval = maximum - minimum + offset = cfg.get('offset', 3) + height = cfg.get('height', interval) + ratio = height / interval if interval > 0 else 1 + + min2 = int(floor(minimum * ratio)) + max2 = int(ceil(maximum * ratio)) + + def clamp(n): + return min(max(n, minimum), maximum) + + def scaled(y): + return int(round(clamp(y) * ratio) - min2) + + rows = max2 - min2 + + width = 0 + for i in range(0, len(series)): + width = max(width, len(series[i])) + width += offset + + placeholder = cfg.get('format', '{:8.2f} ') + + result = [[' '] * width for i in range(rows + 1)] + + for y in range(min2, max2 + 1): + label = placeholder.format(maximum - ((y - min2) * interval / (rows if rows else 1))) + result[y - min2][max(offset - len(label), 0)] = label + result[y - min2][offset - 1] = symbols[0] if y == 0 else symbols[1] + + d0 = series[0][0] + if not isnan(d0): + result[rows - scaled(d0)][offset - 1] = symbols[0] + + for i in range(0, len(series)): + for x in range(0, len(series[i]) - 1): + d0 = series[i][x + 0] + d1 = series[i][x + 1] + + if isnan(d0) and isnan(d1): + continue + + if isnan(d0) and not isnan(d1): + result[rows - scaled(d1)][x + offset] = symbols[2] + continue + + if not isnan(d0) and isnan(d1): + result[rows - scaled(d0)][x + offset] = symbols[3] + continue + + y0 = scaled(d0) + y1 = scaled(d1) + if y0 == y1: + result[rows - y0][x + offset] = symbols[4] + continue + + result[rows - y1][x + offset] = symbols[5] if y0 > y1 else symbols[6] + result[rows - y0][x + offset] = symbols[7] if y0 > y1 else symbols[8] + + start = min(y0, y1) + 1 + end = max(y0, y1) + for y in range(start, end): + result[rows - y][x + offset] = symbols[9] + + return '\n'.join([''.join(row).rstrip() for row in result]) \ No newline at end of file diff --git a/nomadnet/vendor/additional_urwid_widgets/FormWidgets.py b/nomadnet/vendor/additional_urwid_widgets/FormWidgets.py new file mode 100644 index 0000000..4e4809a --- /dev/null +++ b/nomadnet/vendor/additional_urwid_widgets/FormWidgets.py @@ -0,0 +1,283 @@ +import urwid + +class DialogLineBox(urwid.LineBox): + def __init__(self, body, parent=None, title="?"): + super().__init__(body, title=title) + self.parent = parent + + def keypress(self, size, key): + if key == "esc": + if self.parent and hasattr(self.parent, "dismiss_dialog"): + self.parent.dismiss_dialog() + return None + return super().keypress(size, key) + +class Placeholder(urwid.Edit): + def __init__(self, caption="", edit_text="", placeholder="", **kwargs): + super().__init__(caption, edit_text, **kwargs) + self.placeholder = placeholder + + def render(self, size, focus=False): + if not self.edit_text and not focus: + placeholder_widget = urwid.Text(("placeholder", self.placeholder)) + return placeholder_widget.render(size, focus) + else: + return super().render(size, focus) + +class Dropdown(urwid.WidgetWrap): + signals = ['change'] # emit for urwid.connect_signal fn + + def __init__(self, label, options, default=None): + self.label = label + self.options = options + self.selected = default if default is not None else options[0] + + self.main_text = f"{self.selected}" + self.main_button = urwid.SelectableIcon(self.main_text, 0) + self.main_button = urwid.AttrMap(self.main_button, "button_normal", "button_focus") + + self.option_widgets = [] + for opt in options: + icon = urwid.SelectableIcon(opt, 0) + icon = urwid.AttrMap(icon, "list_normal", "list_focus") + self.option_widgets.append(icon) + + self.options_walker = urwid.SimpleFocusListWalker(self.option_widgets) + self.options_listbox = urwid.ListBox(self.options_walker) + self.dropdown_box = None # will be created on open_dropdown + + self.pile = urwid.Pile([self.main_button]) + self.dropdown_visible = False + + super().__init__(self.pile) + + def open_dropdown(self): + if not self.dropdown_visible: + height = len(self.options) + self.dropdown_box = urwid.BoxAdapter(self.options_listbox, height) + self.pile.contents.append((self.dropdown_box, self.pile.options())) + self.dropdown_visible = True + self.pile.focus_position = 1 + self.options_walker.set_focus(0) + + def close_dropdown(self): + if self.dropdown_visible: + self.pile.contents.pop() # remove the dropdown_box + self.dropdown_visible = False + self.pile.focus_position = 0 + self.dropdown_box = None + + def keypress(self, size, key): + if not self.dropdown_visible: + if key == "enter": + self.open_dropdown() + return None + return self.main_button.keypress(size, key) + else: + if key == "enter": + focus_result = self.options_walker.get_focus() + if focus_result is not None: + focus_widget = focus_result[0] + new_val = focus_widget.base_widget.text + old_val = self.selected + self.selected = new_val + self.main_button.base_widget.set_text(f"{self.selected}") + + if old_val != new_val: + self._emit('change', new_val) + + self.close_dropdown() + return None + return self.dropdown_box.keypress(size, key) + + def get_value(self): + return self.selected + +class ValidationError(urwid.Text): + def __init__(self, message=""): + super().__init__(("error", message)) + +class FormField: + def __init__(self, config_key, transform=None): + self.config_key = config_key + self.transform = transform or (lambda x: x) + +class FormEdit(Placeholder, FormField): + def __init__(self, config_key, caption="", edit_text="", placeholder="", validation_types=None, transform=None, **kwargs): + Placeholder.__init__(self, caption, edit_text, placeholder, **kwargs) + FormField.__init__(self, config_key, transform) + self.validation_types = validation_types or [] + self.error_widget = urwid.Text("") + self.error = None + + def get_value(self): + return self.transform(self.edit_text.strip()) + + def validate(self): + value = self.edit_text.strip() + self.error = None + + for validation in self.validation_types: + if validation == "required": + if not value: + self.error = "This field is required" + break + elif validation == "number": + if value and not value.replace('-', '').replace('.', '').isdigit(): + self.error = "This field must be a number" + break + elif validation == "float": + try: + if value: + float(value) + except ValueError: + self.error = "This field must be decimal number" + break + + self.error_widget.set_text(("error", self.error or "")) + return self.error is None + +class FormCheckbox(urwid.CheckBox, FormField): + def __init__(self, config_key, label="", state=False, validation_types=None, transform=None, **kwargs): + urwid.CheckBox.__init__(self, label, state, **kwargs) + FormField.__init__(self, config_key, transform) + self.validation_types = validation_types or [] + self.error_widget = urwid.Text("") + self.error = None + + def get_value(self): + return self.transform(self.get_state()) + + def validate(self): + + value = self.get_state() + self.error = None + + for validation in self.validation_types: + if validation == "required": + if not value: + self.error = "This field is required" + break + + self.error_widget.set_text(("error", self.error or "")) + return self.error is None + +class FormDropdown(Dropdown, FormField): + signals = ['change'] + + def __init__(self, config_key, label, options, default=None, validation_types=None, transform=None): + self.options = [str(opt) for opt in options] + + if default is not None: + default_str = str(default) + if default_str in self.options: + default = default_str + elif transform: + try: + default_transformed = transform(default_str) + for opt in self.options: + if transform(opt) == default_transformed: + default = opt + break + except: + default = self.options[0] + else: + default = self.options[0] + else: + default = self.options[0] + + Dropdown.__init__(self, label, self.options, default) + FormField.__init__(self, config_key, transform) + + self.validation_types = validation_types or [] + self.error_widget = urwid.Text("") + self.error = None + + if hasattr(self, 'main_button'): + self.main_button.base_widget.set_text(str(default)) + + def get_value(self): + return self.transform(self.selected) + + def validate(self): + value = self.get_value() + self.error = None + + for validation in self.validation_types: + if validation == "required": + if not value: + self.error = "This field is required" + break + + self.error_widget.set_text(("error", self.error or "")) + return self.error is None + + def open_dropdown(self): + if not self.dropdown_visible: + super().open_dropdown() + try: + current_index = self.options.index(self.selected) + self.options_walker.set_focus(current_index) + except ValueError: + pass + +class FormMultiList(urwid.Pile, FormField): + def __init__(self, config_key, placeholder="", validation_types=None, transform=None, **kwargs): + self.entries = [] + self.error_widget = urwid.Text("") + self.error = None + self.placeholder = placeholder + self.validation_types = validation_types or [] + + first_entry = self.create_entry_row() + self.entries.append(first_entry) + + self.add_button = urwid.Button("+ Add Another", on_press=self.add_entry) + add_button_padded = urwid.Padding(self.add_button, left=2, right=2) + + pile_widgets = [first_entry, add_button_padded] + urwid.Pile.__init__(self, pile_widgets) + FormField.__init__(self, config_key, transform) + + def create_entry_row(self): + edit = urwid.Edit("", "") + entry_row = urwid.Columns([ + ('weight', 1, edit), + (3, urwid.Button("×", on_press=lambda button: self.remove_entry(button, entry_row))), + ]) + return entry_row + + def remove_entry(self, button, entry_row): + if len(self.entries) > 1: + self.entries.remove(entry_row) + self.contents = [(w, self.options()) for w in self.get_pile_widgets()] + + def add_entry(self, button): + new_entry = self.create_entry_row() + self.entries.append(new_entry) + + self.contents = [(w, self.options()) for w in self.get_pile_widgets()] + + def get_pile_widgets(self): + return self.entries + [urwid.Padding(self.add_button, left=2, right=2)] + + def get_value(self): + values = [] + for entry in self.entries: + edit_widget = entry.contents[0][0] + value = edit_widget.edit_text.strip() + if value: + values.append(value) + return self.transform(values) + + def validate(self): + values = self.get_value() + self.error = None + + for validation in self.validation_types: + if validation == "required" and not values: + self.error = "At least one entry is required" + break + + self.error_widget.set_text(("error", self.error or "")) + return self.error is None \ No newline at end of file