From 94f5e7188a234f5aae3638ca3fa1ba92e086fc52 Mon Sep 17 00:00:00 2001 From: zenith <157907903+RFnexus@users.noreply.github.com> Date: Mon, 15 Jun 2026 05:09:42 -0400 Subject: [PATCH] Add visual Config editor. Optional utilities tab with path, rncp file send utility, transport table viewer, speedtest, and identity resolver. Misc fixes --- nomadnet/NomadNetworkApp.py | 15 + nomadnet/ui/TextUI.py | 18 + nomadnet/ui/textui/Config.py | 1225 ++++++++++++- nomadnet/ui/textui/Conversations.py | 207 +-- nomadnet/ui/textui/Editor.py | 6 +- nomadnet/ui/textui/FileBrowser.py | 195 +++ nomadnet/ui/textui/Interfaces.py | 129 +- nomadnet/ui/textui/Main.py | 34 +- nomadnet/ui/textui/Transport.py | 503 ++++++ nomadnet/ui/textui/Utilities.py | 1523 +++++++++++++++++ .../additional_urwid_widgets/FormWidgets.py | 31 +- 11 files changed, 3658 insertions(+), 228 deletions(-) create mode 100644 nomadnet/ui/textui/FileBrowser.py create mode 100644 nomadnet/ui/textui/Transport.py create mode 100644 nomadnet/ui/textui/Utilities.py diff --git a/nomadnet/NomadNetworkApp.py b/nomadnet/NomadNetworkApp.py index 24a75a7..6bf3be2 100644 --- a/nomadnet/NomadNetworkApp.py +++ b/nomadnet/NomadNetworkApp.py @@ -984,6 +984,13 @@ class NomadNetworkApp: if value == "web": self.uimode = nomadnet.ui.UI_WEB + if not "utilities" in self.config: + self.config["utilities"] = {} + if not "enable_utilities" in self.config["utilities"]: + self.config["utilities"]["enable_utilities"] = False + else: + self.config["utilities"]["enable_utilities"] = self.config["utilities"].as_bool("enable_utilities") + if "rrc" in self.config: for option in self.config["rrc"]: if option == "history_per_room_cap": @@ -1383,6 +1390,14 @@ sanitize_names = yes # escape sequences. clipboard_copy = no +[utilities] + +# You can enable a Utilities section in the +# menu, where you can view the path and announce +# rate tables and probe destinations. Disabled +# by default. +enable_utilities = no + [rrc] # Maximum number of messages retained per diff --git a/nomadnet/ui/TextUI.py b/nomadnet/ui/TextUI.py index 058228e..50d65d1 100644 --- a/nomadnet/ui/TextUI.py +++ b/nomadnet/ui/TextUI.py @@ -128,6 +128,23 @@ THEMES = { } } +CONFIG_SECTION_COLORS = { + "logging": ("yellow", "#fd3", "brown", "#a80"), + "client": ("light cyan", "#3cd", "dark cyan", "#077"), + "textui": ("light green", "#6c5", "dark green", "#2a0"), + "utilities": ("light gray", "#bbb", "dark gray", "#555"), + "rrc": ("light magenta", "#d7f", "dark magenta", "#a2a"), + "node": ("light blue", "#5af", "dark blue", "#06c"), + "printing": ("brown", "#b86", "brown", "#852"), + "reticulum": ("light magenta", "#a6f", "dark magenta", "#63c"), +} + +for _section, (_d16, _dhi, _l16, _lhi) in CONFIG_SECTION_COLORS.items(): + THEMES[THEME_DARK]["urwid_theme"].append(("config_title_"+_section, _d16, "default", "default", _dhi, "default")) + THEMES[THEME_DARK]["urwid_theme"].append(("config_title_"+_section+"_selected", _d16+",bold", "default", "bold", _dhi+",bold", "default")) + THEMES[THEME_LIGHT]["urwid_theme"].append(("config_title_"+_section, _l16, "default", "default", _lhi, "default")) + THEMES[THEME_LIGHT]["urwid_theme"].append(("config_title_"+_section+"_selected", _l16+",bold", "default", "bold", _lhi+",bold", "default")) + GLYPHSETS = { "plain": 1, "unicode": 2, @@ -181,6 +198,7 @@ GLYPHS = { ("folder_open", "[-]", "\u25be", "\uf07c"), ("fold_open", "-", "▾", "▾"), ("fold_closed", "+", "▸", "▸"), + ("dropdown", " [v]", " ▾", " "), } class TextUI: diff --git a/nomadnet/ui/textui/Config.py b/nomadnet/ui/textui/Config.py index 36d92a5..84de154 100644 --- a/nomadnet/ui/textui/Config.py +++ b/nomadnet/ui/textui/Config.py @@ -1,13 +1,705 @@ +import RNS import nomadnet import urwid import platform +from RNS.vendor.configobj import ConfigObj +from nomadnet.vendor.additional_urwid_widgets.FormWidgets import * + +CONFIG_SECTION_GLYPHS = { + ("logging", "(L)", "≡", ""), + ("client", "(C)", "✉", ""), + ("utilities", "(Y)", "⚒", ""), + ("textui", "(U)", "▤", ""), + ("rrc", "(R)", "❖", ""), + ("node", "(N)", "Ⓝ", "\U000f0002"), + ("printing", "(P)", "⎙", ""), + ("reticulum", "(Rt)", "⇄", ""), + ("editor", "(E)", "✎", ""), +} + +def _get_section_icon(glyphset, name): + glyphset_index = 1 + if glyphset == "plain": + glyphset_index = 0 + elif glyphset == "nerdfont": + glyphset_index = 2 + + for glyph_tuple in CONFIG_SECTION_GLYPHS: + if glyph_tuple[0] == name: + return glyph_tuple[glyphset_index + 1] + + return "(#)" if glyphset == "plain" else "⚙" if glyphset == "unicode" else "" + +def _get_cols_rows(): + return nomadnet.NomadNetworkApp.get_shared_instance().ui.screen.get_cols_rows() + +def section_title_attr(section_key): + return "config_title_" + section_key + +class ConfigCheckbox(FormCheckbox): + def __init__(self, config_key, on_text="Enabled", off_text="Disabled", state=False, validation_types=None, transform=None, **kwargs): + self.on_text = on_text + self.off_text = off_text + super().__init__(config_key, label="", state=state, validation_types=validation_types, transform=transform, **kwargs) + self._refresh_label(self.get_state()) + urwid.connect_signal(self, "change", self._on_change) + + def _on_change(self, widget, new_state): + self._refresh_label(new_state) + + def _refresh_label(self, state): + if state: + self.set_label(("connected_status", " " + self.on_text)) + else: + self.set_label(("inactive_text", " " + self.off_text)) + +CONFIG_SECTIONS = [ + { + "key": "logging", + "name": "Logging", + "glyph": "logging", + "description": "Log verbosity and where log output is written", + "fields": [ + { + "config_key": "loglevel", + "label": "Log level: ", + "type": "dropdown", + "options": ["0", "1", "2", "3", "4", "5", "6", "7"], + "default": "4", + "help": "0 = critical, 4 = info, 7 = extreme verbosity", + }, + { + "config_key": "destination", + "label": "Destination: ", + "type": "dropdown", + "options": ["file", "stdout"], + "default": "file", + "help": "Write log to a file or to standard output", + }, + { + "config_key": "logfile", + "label": "Log file: ", + "type": "edit", + "placeholder": "Defaults to the config directory", + "help": "Path to the log file (when destination is file)", + }, + ], + }, + { + "key": "client", + "name": "LXMF client", + "glyph": "client", + "description": "Client behaviour, announces, syncing and messages", + "fields": [ + { + "config_key": "downloads_path", + "label": "Downloads path: ", + "type": "edit", + "default": "~/Downloads", + "placeholder": "~/Downloads", + "help": "Where downloaded files are saved", + }, + { + "config_key": "attachment_save_path", + "label": "Attachment path: ", + "type": "edit", + "placeholder": "Defaults to the downloads path", + "help": "Where message attachments are saved", + }, + { + "config_key": "notify_on_new_message", + "label": "Notify on message: ", + "type": "checkbox", + "default": True, + "help": "Show a notification when a new message arrives", + }, + { + "config_key": "announce_at_start", + "label": "Announce at start: ", + "type": "checkbox", + "default": True, + "help": "Announce the local peer when starting", + }, + { + "config_key": "announce_interval", + "label": "Announce interval: ", + "type": "edit", + "default": "360", + "validation": ["number"], + "help": "Minutes between peer announces (minimum 30)", + }, + { + "config_key": "try_propagation_on_send_fail", + "label": "Propagate on fail: ", + "type": "checkbox", + "default": True, + "help": "Use a propagation node if direct delivery fails", + }, + { + "config_key": "periodic_lxmf_sync", + "label": "Periodic sync: ", + "type": "checkbox", + "default": True, + "help": "Automatically sync messages from a propagation node", + }, + { + "config_key": "lxmf_sync_interval", + "label": "Sync interval: ", + "type": "edit", + "default": "360", + "validation": ["number"], + "help": "Minutes between automatic syncs", + }, + { + "config_key": "lxmf_sync_limit", + "label": "Sync limit: ", + "type": "edit", + "default": "8", + "validation": ["number"], + "help": "Max messages fetched per sync (0 = no limit)", + }, + { + "config_key": "required_stamp_cost", + "label": "Required stamp cost: ", + "type": "edit", + "default": "None", + "placeholder": "None", + "help": "Required inbound stamp cost 1-255, or None to disable", + }, + { + "config_key": "accept_invalid_stamps", + "label": "Accept invalid stamps: ", + "type": "checkbox", + "default": False, + "help": "Accept messages carrying an invalid stamp", + }, + { + "config_key": "max_accepted_size", + "label": "Max message size: ", + "type": "edit", + "default": "500", + "validation": ["float"], + "help": "Maximum accepted incoming message size, in KB", + }, + { + "config_key": "compact_announce_stream", + "label": "Compact announces: ", + "type": "checkbox", + "default": True, + "help": "Show only one announce per destination", + }, + { + "config_key": "compose_in_markdown", + "label": "Compose in markdown: ", + "type": "checkbox", + "default": True, + "help": "Author messages using markdown", + }, + ], + }, + { + "key": "textui", + "name": "Text UI", + "glyph": "textui", + "description": "Appearance and behaviour of the text interface", + "note": "Switch between the light and dark theme, glyphs and other appearance options here.", + "fields": [ + { + "config_key": "intro_time", + "label": "Show intro: ", + "type": "checkbox", + "default": True, + "on_value": "1.0", + "off_value": "0", + "help": "Show the intro screen at startup", + }, + { + "config_key": "intro_text", + "label": "Intro text: ", + "type": "edit", + "default": "Nomad Network", + "placeholder": "Nomad Network", + "help": "Text shown on the intro screen", + }, + { + "config_key": "theme", + "label": "Theme: ", + "type": "dropdown", + "options": ["dark", "light"], + "default": "dark", + "help": "Colour theme", + }, + { + "config_key": "colormode", + "label": "Color mode: ", + "type": "dropdown", + "options": ["monochrome", "16", "88", "256", "24bit"], + "default": "24bit", + "help": "Colour capability of your terminal", + }, + { + "config_key": "glyphs", + "label": "Glyphs: ", + "type": "dropdown", + "options": ["plain", "unicode", "nerdfont"], + "default": "nerdfont", + "help": "Glyph set used for icons", + }, + { + "config_key": "mouse_enabled", + "label": "Mouse enabled: ", + "type": "checkbox", + "default": True, + "help": "Allow mouse interaction", + }, + { + "config_key": "editor", + "label": "Editor: ", + "type": "edit", + "default": "nano", + "placeholder": "nano", + "help": "Command used to edit text", + }, + { + "config_key": "hide_guide", + "label": "Hide guide: ", + "type": "checkbox", + "default": False, + "help": "Hide the Guide entry in the menu", + }, + { + "config_key": "sanitize_names", + "label": "Sanitize names: ", + "type": "checkbox", + "default": True, + "help": "Strip control characters from announced names", + }, + { + "config_key": "clipboard_copy", + "label": "Clipboard copy: ", + "type": "checkbox", + "default": False, + "help": "Enable copying to the system clipboard", + }, + { + "config_key": "animation_interval", + "label": "Animation interval: ", + "type": "edit", + "default": "1", + "validation": ["number"], + "help": "Interval between UI animation frames", + }, + ], + }, + { + "key": "utilities", + "name": "Utilities", + "glyph": "utilities", + "description": "Optional utility views and tools", + "note": "Adds a Utilities menu where you can view path/announce tables and probe destinations.", + "fields": [ + { + "config_key": "enable_utilities", + "label": "Enable utilities: ", + "type": "checkbox", + "default": False, + "help": "Show the Utilities section (path tables, probe) in the menu", + }, + ], + }, + { + "key": "rrc", + "name": "RRC Client Settings", + "glyph": "rrc", + "description": "Rendering and history for the rooms client", + "fields": [ + { + "config_key": "history_per_room_cap", + "label": "History cap: ", + "type": "edit", + "default": "500", + "validation": ["number"], + "help": "Messages kept in memory per room", + }, + { + "config_key": "filter_loaded_history", + "label": "Filter history: ", + "type": "checkbox", + "default": True, + "help": "Filter system messages out of loaded history", + }, + { + "config_key": "ephemeral_notices", + "label": "Ephemeral notices: ", + "type": "edit", + "default": "10", + "validation": ["float"], + "help": "Minutes to keep notices (0 keeps them)", + }, + { + "config_key": "color_mention_timestamps", + "label": "Color mention time: ", + "type": "checkbox", + "default": True, + "help": "Colour the timestamp when you are mentioned", + }, + { + "config_key": "render_markdown", + "label": "Render markdown: ", + "type": "checkbox", + "default": True, + "help": "Render markdown formatting in messages", + }, + { + "config_key": "render_micron", + "label": "Render micron: ", + "type": "checkbox", + "default": True, + "help": "Render micron markup in messages", + }, + { + "config_key": "nick_colors", + "label": "Nick colors: ", + "type": "checkbox", + "default": True, + "help": "Colour user nicknames", + }, + { + "config_key": "justify_msgs", + "label": "Justify messages: ", + "type": "checkbox", + "default": True, + "help": "Justify message text", + }, + { + "config_key": "space_msgs", + "label": "Space messages: ", + "type": "checkbox", + "default": False, + "help": "Add extra spacing between messages", + }, + { + "config_key": "show_gutters", + "label": "Show gutters: ", + "type": "checkbox", + "default": True, + "help": "Show gutters alongside messages", + }, + { + "config_key": "mention_color", + "label": "Mention color: ", + "type": "edit", + "placeholder": "6-digit hex, e.g. ff8800", + "help": "Hex colour used for mentions of your nick", + }, + { + "config_key": "nick_colors_theme", + "label": "Nick color theme: ", + "type": "list", + "default": [], + "placeholder": "6-digit hex colour", + "help": "Custom palette of hex colours used for nicknames", + }, + { + "config_key": "enable_esoterics", + "label": "Enable esoterics: ", + "type": "checkbox", + "default": False, + "help": "Enable esoteric features", + }, + ], + }, + { + "key": "node", + "name": "Node", + "glyph": "node", + "description": "Node hosting, propagation and announces", + "note": "Enable this to host a node on the network, serving pages and files to others.", + "fields": [ + { + "config_key": "enable_node", + "label": "Enable node: ", + "type": "checkbox", + "default": False, + "help": "Host a node on the network", + }, + { + "config_key": "node_name", + "label": "Node name: ", + "type": "edit", + "default": "None", + "placeholder": "None", + "help": "Display name for the node, or None", + }, + { + "config_key": "announce_at_start", + "label": "Announce at start: ", + "type": "checkbox", + "default": True, + "help": "Announce the node when starting", + }, + { + "config_key": "announce_interval", + "label": "Announce interval: ", + "type": "edit", + "default": "360", + "validation": ["number"], + "help": "Minutes between node announces (minimum 1)", + }, + { + "config_key": "disable_propagation", + "label": "Disable propagation: ", + "type": "checkbox", + "default": True, + "help": "Disable the propagation node functionality", + }, + { + "config_key": "propagation_cost", + "label": "Propagation cost: ", + "type": "edit", + "default": "16", + "validation": ["number"], + "help": "Stamp cost for propagation (minimum 13)", + }, + { + "config_key": "max_transfer_size", + "label": "Max transfer size: ", + "type": "edit", + "default": "256", + "validation": ["float"], + "help": "Maximum single message size on propagation, in KB", + }, + { + "config_key": "max_sync_size", + "label": "Max sync size: ", + "type": "edit", + "default": "10240", + "validation": ["float"], + "help": "Maximum sync batch size, in KB", + }, + { + "config_key": "pages_path", + "label": "Pages path: ", + "type": "edit", + "placeholder": "Default node pages directory", + "help": "Directory of pages served by the node", + }, + { + "config_key": "page_refresh_interval", + "label": "Page refresh: ", + "type": "edit", + "default": "0", + "validation": ["number"], + "help": "Minutes between page rescans (0 = never)", + }, + { + "config_key": "files_path", + "label": "Files path: ", + "type": "edit", + "placeholder": "Default node files directory", + "help": "Directory of files served by the node", + }, + { + "config_key": "file_refresh_interval", + "label": "File refresh: ", + "type": "edit", + "default": "0", + "validation": ["number"], + "help": "Minutes between file rescans (0 = never)", + }, + { + "config_key": "prioritise_destinations", + "label": "Priority dests: ", + "type": "list", + "default": [], + "placeholder": "Destination hash", + "help": "Destination hashes given storage priority", + }, + { + "config_key": "static_peers", + "label": "Static peers: ", + "type": "list", + "default": [], + "placeholder": "Destination hash", + "help": "Propagation peers always kept connected", + }, + { + "config_key": "max_peers", + "label": "Max peers: ", + "type": "edit", + "placeholder": "No limit", + "validation": ["number"], + "help": "Maximum number of propagation peers", + }, + { + "config_key": "message_storage_limit", + "label": "Storage limit: ", + "type": "edit", + "default": "2000", + "validation": ["float"], + "help": "Maximum propagation storage, in MB", + }, + ], + }, + { + "key": "printing", + "name": "Printing", + "glyph": "printing", + "description": "Printing of received messages", + "note": "The rendered message file is appended to this command, so it runs as e.g. 'lp '.", + "fields": [ + { + "config_key": "print_messages", + "label": "Print messages: ", + "type": "checkbox", + "default": False, + "help": "Print received messages", + }, + { + "config_key": "print_command", + "label": "Print command: ", + "type": "edit", + "default": "lp", + "placeholder": "lp", + "help": ("Command to print with; the message file is appended as the last argument. Examples: " + "lp | " + "lp -d NAME (named printer, list with lpstat -p) | " + "lp -d NAME -o cpi=16 -o lpi=8 (small thermal-roll printers) | " + "lp -d NAME -o media=Custom.58x210mm (58mm thermal roll)"), + }, + { + "config_key": "print_from", + "label": "Print from: ", + "type": "edit", + "placeholder": "everywhere, trusted, or hashes", + "help": "Which senders may trigger automatic printing", + }, + { + "config_key": "message_template", + "label": "Message template: ", + "type": "edit", + "placeholder": "Path to template file", + "help": "Template file used when printing messages", + }, + ], + }, + { + "key": "reticulum", + "name": "Reticulum (global)", + "glyph": "reticulum", + "configfile": "reticulum", + "description": "Global Reticulum settings (shared instance, transport)", + "fields": [ + { + "config_key": "enable_transport", + "label": "Enable transport: ", + "type": "checkbox", + "default": False, + "help": "Route traffic for other peers (transport node)", + }, + { + "config_key": "share_instance", + "label": "Share instance: ", + "type": "checkbox", + "default": True, + "help": "Run a shared instance other local programs connect to", + }, + { + "config_key": "instance_name", + "label": "Instance name: ", + "type": "edit", + "default": "default", + "placeholder": "default", + "help": "Name used to isolate multiple shared instances", + }, + { + "config_key": "shared_instance_type", + "label": "Instance type: ", + "type": "edit", + "placeholder": "unix or tcp (blank for auto)", + "help": "Transport used for shared instance communication", + }, + { + "config_key": "shared_instance_port", + "label": "Instance port: ", + "type": "edit", + "placeholder": "37428", + "validation": ["number"], + "help": "TCP port for the shared instance", + }, + { + "config_key": "instance_control_port", + "label": "Control port: ", + "type": "edit", + "placeholder": "37429", + "validation": ["number"], + "help": "TCP control port for the shared instance", + }, + { + "config_key": "discover_interfaces", + "label": "Discover interfaces: ", + "type": "checkbox", + "default": False, + "help": "Discover interfaces advertised by transport instances", + }, + { + "config_key": "enable_remote_management", + "label": "Remote management: ", + "type": "checkbox", + "default": False, + "help": "Allow remote management of this instance", + }, + { + "config_key": "remote_management_allowed", + "label": "Remote allowed: ", + "type": "list", + "default": [], + "placeholder": "Identity hash", + "help": "Identity hashes allowed to manage this instance", + }, + { + "config_key": "respond_to_probes", + "label": "Respond to probes: ", + "type": "checkbox", + "default": False, + "help": "Answer reachability probes from other nodes", + }, + { + "config_key": "use_implicit_proof", + "label": "Implicit proof: ", + "type": "checkbox", + "default": True, + "help": "Use implicit proofs for packet delivery", + }, + { + "config_key": "panic_on_interface_error", + "label": "Panic on iface error: ", + "type": "checkbox", + "default": False, + "help": "Exit if an interface has an unrecoverable error", + }, + ], + }, +] + +_SECTION_ORDER = {"node": 0, "client": 1, "logging": 2, "textui": 3, "utilities": 4, "rrc": 5, "printing": 6, "reticulum": 7} +CONFIG_SECTIONS.sort(key=lambda s: _SECTION_ORDER.get(s["key"], 99)) + class ConfigDisplayShortcuts(): def __init__(self, app): - import urwid self.app = app + self.default_shortcuts = "[Enter] Edit Section [C-w] Text Editor [Up/Down] Navigate" + self.widget = urwid.AttrMap(urwid.Text(self.default_shortcuts), "shortcutbar") - self.widget = urwid.AttrMap(urwid.Text(""), "shortcutbar") + def update_shortcuts(self, new_shortcuts): + self.widget.original_widget.set_text(new_shortcuts) + + def reset_shortcuts(self): + self.update_shortcuts(self.default_shortcuts) + + def set_section_shortcuts(self): + self.update_shortcuts("[Up/Down] Navigate Fields [Enter] Select Option [C-s] Save [C-r] Reset [Esc] Cancel") class ConfigFiller(urwid.WidgetWrap): def __init__(self, widget, app): @@ -15,41 +707,509 @@ class ConfigFiller(urwid.WidgetWrap): self.filler = urwid.Filler(widget, urwid.TOP) super().__init__(self.filler) + def keypress(self, size, key): + if key == "ctrl w": + self.app.ui.main_display.sub_displays.config_display.open_text_editor() + return None + return super(ConfigFiller, self).keypress(size, key) + +class SelectableConfigSection(urwid.WidgetWrap): + def __init__(self, parent, section_key, name, description, icon, note=None): + self.parent = parent + self.section_key = section_key + self.name = name + self.icon = icon + self._selectable = True + + self.title_attr = section_title_attr(section_key) + self.title_attr_selected = self.title_attr + "_selected" + + self.selection_txt = urwid.Text(" ") + self.title_widget = urwid.Text((self.title_attr, f"{icon} {name}")) + + title_content = urwid.Columns([ + (4, self.selection_txt), + self.title_widget, + ]) + + description_row = urwid.Columns([ + (4, urwid.Text(" ")), + urwid.Text(("value", description)), + ]) + + pile_rows = [title_content, description_row] + if note: + pile_rows.append(urwid.Columns([ + (4, urwid.Text(" ")), + urwid.Text(("inactive_text", note)), + ])) + + pile = urwid.Pile(pile_rows) + padded_body = urwid.Padding(pile, left=2, right=2) + + box = urwid.LineBox( + padded_body, + title=None, + tlcorner="╭", tline="─", + trcorner="╮", lline="│", + rline="│", blcorner="╰", + bline="─", brcorner="╯" + ) + + super().__init__(box) + + 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((self.title_attr_selected, f"{self.icon} {self.name}")) + else: + self.title_widget.set_text((self.title_attr, f"{self.icon} {self.name}")) + + return super().render(size, focus=focus) def keypress(self, size, key): if key == "up": - self.app.ui.main_display.frame.focus_position = "header" - - return super(ConfigFiller, self).keypress(size, key) + listbox = self.parent.list_box + if listbox.focus_position == 0: + self.parent.app.ui.main_display.frame.focus_position = "header" + return None + elif key == "enter": + self.parent.switch_to_section(self.section_key) + return None + return key + +class ConfigActionItem(urwid.WidgetWrap): + def __init__(self, parent, label, on_select): + self.parent = parent + self.on_select = on_select + self._selectable = True + + text_widget = urwid.Text(label, align="left") + super().__init__(urwid.AttrMap(text_widget, "list_normal", focus_map="list_focus")) + + def selectable(self): + return True + + def keypress(self, size, key): + if key == "enter": + self.on_select() + return None + return super().keypress(size, key) + + def mouse_event(self, size, event, button, x, y, focus): + if button == 1 and urwid.util.is_mouse_press(event): + self.on_select() + return True + return False + +class ConfigSectionView(urwid.WidgetWrap): + LABEL_WIDTH = 24 + + def __init__(self, parent, section): + self.parent = parent + self.section = section + self.section_key = section["key"] + self.fields = {} + + self.parent.shortcuts_display.set_section_shortcuts() + + self.config_path = self._resolve_config_path(section) + self.file_config = ConfigObj(self.config_path) + section_config = self.file_config.get(self.section_key, {}) + + cols, rows = _get_cols_rows() + self.content_width = min(80, max(40, cols - 6)) + + icon = _get_section_icon(self.parent.glyphset, section["glyph"]) + + field_items = [] + for field in section["fields"]: + widget = self._build_field_widget(field) + current = section_config.get(field["config_key"], field.get("default")) + self._populate_field(widget, field, current) + self.fields[field["config_key"]] = {"field": field, "widget": widget} + field_items.append(self._build_field_row(field, widget)) + + self.list_walker = urwid.SimpleFocusListWalker(field_items) + self.form_listbox = urwid.ListBox(self.list_walker) + + header = urwid.Pile([ + urwid.Text((section_title_attr(self.section_key) + "_selected", f"{icon} {section['name']}"), align=urwid.CENTER), + urwid.Divider("─"), + ]) + + save_btn = urwid.AttrMap(urwid.Button("Save", on_press=self.on_save), "button_normal", focus_map="button_focus") + reset_btn = urwid.AttrMap(urwid.Button("Reset", on_press=self.on_reset), "button_normal", focus_map="button_focus") + back_btn = urwid.AttrMap(urwid.Button("Cancel", on_press=self.on_back), "button_normal", focus_map="button_focus") + button_row = urwid.Columns([ + (urwid.WEIGHT, 0.3, save_btn), + (urwid.WEIGHT, 0.05, urwid.Text("")), + (urwid.WEIGHT, 0.3, reset_btn), + (urwid.WEIGHT, 0.05, urwid.Text("")), + (urwid.WEIGHT, 0.3, back_btn), + ]) + self._footer = urwid.Pile([ + urwid.Divider("─"), + urwid.Padding(button_row, align=urwid.CENTER, width=self.content_width), + ]) + try: + self._footer.focus_position = 1 + except Exception: + pass + + self._frame = urwid.Frame(self.form_listbox, header=header, footer=self._footer) + super().__init__(self._frame) + + def _resolve_config_path(self, section): + if section.get("configfile") == "reticulum": + try: + return self.parent.app.rns.config.filename + except Exception: + return RNS.Reticulum.configpath + return self.parent.app.configpath + + def _build_field_row(self, field, widget): + row = urwid.Columns([ + (self.LABEL_WIDTH, urwid.Text(("key", field["label"]), align=urwid.RIGHT)), + widget, + ]) + contents = [row] + help_text = self._field_help(field) + if help_text: + contents.append(urwid.Padding(urwid.Text(("inactive_text", help_text)), left=self.LABEL_WIDTH + 2)) + contents.append(urwid.Padding(widget.error_widget, left=self.LABEL_WIDTH + 2)) + contents.append(urwid.Divider()) + field_pile = urwid.Pile(contents) + return urwid.Padding(field_pile, align=urwid.CENTER, width=self.content_width) + + def _field_help(self, field): + parts = [] + if field.get("help"): + parts.append(field["help"]) + default = field.get("default", None) + if field["type"] == "checkbox": + default_text = "yes" if default else "no" + elif field["type"] == "list": + if not default: + default_text = "none" + elif isinstance(default, (list, tuple)): + default_text = ", ".join(str(item) for item in default) + else: + default_text = str(default) + else: + default_text = str(default) if default not in (None, "") else "none" + parts.append(f"(default: {default_text})") + return " ".join(parts) + + def _build_field_widget(self, field): + if field["type"] == "dropdown": + return FormDropdown( + config_key=field["config_key"], + label=field.get("label", ""), + options=field["options"], + default=str(field.get("default", field["options"][0])), + validation_types=field.get("validation", []), + ) + elif field["type"] == "checkbox": + return ConfigCheckbox( + config_key=field["config_key"], + on_text=field.get("on_text", "Enabled"), + off_text=field.get("off_text", "Disabled"), + state=bool(field.get("default", False)), + validation_types=field.get("validation", []), + ) + elif field["type"] == "list": + return FormMultiList( + config_key=field["config_key"], + placeholder=field.get("placeholder", ""), + validation_types=field.get("validation", []), + ) + else: + return FormEdit( + config_key=field["config_key"], + caption="", + edit_text=str(field.get("default", "")), + placeholder=field.get("placeholder", ""), + validation_types=field.get("validation", []), + ) + + def _populate_field(self, widget, field, value): + if value is None: + return + + if field["type"] == "checkbox": + if isinstance(value, bool): + state = value + else: + s = str(value).strip().lower() + if s in ("", "false", "off", "no"): + state = False + else: + try: + state = float(s) != 0 + except ValueError: + state = True + widget.set_state(state) + elif field["type"] == "dropdown": + str_value = str(value) + if str_value in widget.options: + widget.selected = str_value + widget.main_button.base_widget.set_text(str_value) + elif field["type"] == "list": + if value: + widget.set_value(value if isinstance(value, list) else [value]) + else: + widget.edit_text = str(value) + + def validate_all(self): + all_valid = True + for entry in self.fields.values(): + if not entry["widget"].validate(): + all_valid = False + return all_valid + + def on_save(self, button): + if not self.validate_all(): + return + + if self.section_key not in self.file_config: + self.file_config[self.section_key] = {} + + section_config = self.file_config[self.section_key] + + for config_key, entry in self.fields.items(): + field = entry["field"] + widget = entry["widget"] + + if field["type"] == "checkbox": + on_value = field.get("on_value", "yes") + off_value = field.get("off_value", "no") + section_config[config_key] = on_value if widget.get_state() else off_value + elif field["type"] == "list": + value = widget.get_value() + if value: + section_config[config_key] = value + elif config_key in section_config: + del section_config[config_key] + else: + value = widget.get_value() + if value != "": + section_config[config_key] = value + elif config_key in section_config: + del section_config[config_key] + + try: + self.file_config.write() + self.parent.mark_restart_pending() + self.show_message(f"{self.section['name']} configuration saved.\nRestart Nomad Network for changes to take effect.") + except Exception as e: + self.show_message(f"Error saving configuration: {str(e)}", title="Error") + + def on_reset(self, button=None): + self._confirm( + f"Reset all {self.section['name']} settings to their defaults?\nYou will still need to Save to apply.", + self._do_reset, + ) + + def _do_reset(self): + for entry in self.fields.values(): + self._reset_field(entry["widget"], entry["field"]) + + def _reset_field(self, widget, field): + default = field.get("default") + if field["type"] == "checkbox": + widget.set_state(bool(default)) + elif field["type"] == "dropdown": + default_str = str(default if default is not None else widget.options[0]) + if default_str in widget.options: + widget.selected = default_str + widget.main_button.base_widget.set_text(default_str) + elif field["type"] == "list": + widget.set_value(default if isinstance(default, list) else []) + else: + widget.edit_text = "" if default is None else str(default) + widget.error_widget.set_text("") + + def on_back(self, button): + self.parent.switch_to_list() + + def dismiss_dialog(self): + self._dismiss_overlay() + + def _confirm(self, message, on_yes): + def yes(button): + self._dismiss_overlay() + on_yes() + + def no(button): + self._dismiss_overlay() + + dialog = DialogLineBox( + urwid.Pile([ + urwid.Text(message, align=urwid.CENTER), + urwid.Divider(), + urwid.Columns([ + (urwid.WEIGHT, 0.45, urwid.Button("Yes", on_press=yes)), + (urwid.WEIGHT, 0.1, urwid.Text("")), + (urwid.WEIGHT, 0.45, urwid.Button("No", on_press=no)), + ]), + ]), + parent=self, + title="Confirm", + ) + self._show_overlay(dialog, height=9, width=64) + + def _show_overlay(self, dialog, height=8, width=60): + overlay = urwid.Overlay( + dialog, + self, + align=urwid.CENTER, + width=width, + valign=urwid.MIDDLE, + height=height, + min_width=1, + min_height=1, + ) + self.parent.widget = overlay + self.parent.app.ui.main_display.update_active_sub_display() + + def _dismiss_overlay(self): + self.parent.widget = self + self.parent.app.ui.main_display.update_active_sub_display() + + def show_message(self, message, title="Notice"): + def dismiss_dialog(button): + self.parent.switch_to_list() + + dialog = DialogLineBox( + urwid.Pile([ + urwid.Text(message, align=urwid.CENTER), + urwid.Divider(), + urwid.Button("OK", on_press=dismiss_dialog) + ]), + title=title + ) + + overlay = urwid.Overlay( + dialog, + self.parent.list_view, + align=urwid.CENTER, + width=60, + valign=urwid.MIDDLE, + height=10, + min_width=1, + min_height=1 + ) + + self.parent.widget = overlay + self.parent.app.ui.main_display.update_active_sub_display() + + def keypress(self, size, key): + if key == "ctrl s": + self.on_save(None) + return None + if key == "ctrl r": + self.on_reset(None) + return None + if key == "esc": + self.parent.switch_to_list() + return None + + result = super().keypress(size, key) + if result is None: + return None + + if result in ("down", "tab") and self._frame.focus_position != "footer": + self._frame.focus_position = "footer" + try: + self._footer.focus_position = 1 + except Exception: + pass + return None + if result == "up": + if self._frame.focus_position == "footer": + self._frame.focus_position = "body" + else: + self.parent.app.ui.main_display.frame.focus_position = "header" + return None + + return result class ConfigDisplay(): def __init__(self, app): - import urwid self.app = app + self.glyphset = self.app.config["textui"]["glyphs"] + self.g = self.app.ui.glyphs + self.restart_pending = False + self.editor_term = None - def open_editor(sender): - self.editor_term = EditorTerminal(self.app, self) - self.widget = urwid.LineBox(self.editor_term) - self.app.ui.main_display.update_active_sub_display() - self.app.ui.main_display.frame.focus_position = "body" - self.editor_term.term.change_focus(True) - - pile = urwid.Pile([ - urwid.Text( - ( - "body_text", - "\nTo change the configuration, edit the config file located at:\n\n" - +self.app.configpath - +"\n\nRestart Nomad Network for changes to take effect\n", - ), - align=urwid.CENTER, - ), - urwid.Padding(urwid.Button("Open Editor", on_press=open_editor), width=15, align=urwid.CENTER), - ]) - - self.config_explainer = ConfigFiller(pile, self.app) self.shortcuts_display = ConfigDisplayShortcuts(self.app) - self.widget = self.config_explainer + + self._build_list_view() + self.widget = self.list_view + + def _build_list_view(self): + self.terminal_cols, self.terminal_rows = _get_cols_rows() + + self.section_items = [] + for section in CONFIG_SECTIONS: + icon = _get_section_icon(self.glyphset, section["glyph"]) + self.section_items.append( + SelectableConfigSection(self, section["key"], section["name"], section["description"], icon, note=section.get("note")) + ) + + editor_icon = _get_section_icon(self.glyphset, "editor") + self.editor_item = ConfigActionItem(self, f"{editor_icon} Open config file in text editor", self.open_text_editor) + + list_contents = list(self.section_items) + [urwid.Divider(), self.editor_item] + self.list_walker = urwid.SimpleFocusListWalker(list_contents) + self.list_box = urwid.ListBox(self.list_walker) + + self.list_rows = max(5, self.terminal_rows - 7) + self.box_adapter = urwid.BoxAdapter(self.list_box, self.list_rows) + + self.header_text = urwid.Text("") + self.list_divider = urwid.Divider("─") + self.list_pile = urwid.Pile([self.box_adapter]) + self._update_header() + + self.list_view = ConfigFiller(self.list_pile, self.app) + + def _update_header(self): + contents = [] + if self.restart_pending: + self.header_text.set_text(("warning_text", "Changes saved. Restart NomadNet for them to take effect")) + contents.append((self.header_text, self.list_pile.options('pack'))) + contents.append((self.list_divider, self.list_pile.options('pack'))) + contents.append((self.box_adapter, self.list_pile.options('weight', 1))) + self.list_pile.contents = contents + + def mark_restart_pending(self): + self.restart_pending = True + self._update_header() + + def switch_to_section(self, section_key): + section = next(s for s in CONFIG_SECTIONS if s["key"] == section_key) + self.section_view = ConfigSectionView(self, section) + self.widget = self.section_view + self.app.ui.main_display.update_active_sub_display() + + def switch_to_list(self): + self.shortcuts_display.reset_shortcuts() + self._update_header() + self.widget = self.list_view + self.app.ui.main_display.update_active_sub_display() + + def open_text_editor(self): + self.editor_term = EditorTerminal(self.app, self) + self.widget = urwid.LineBox(self.editor_term) + self.app.ui.main_display.update_active_sub_display() + self.app.ui.main_display.frame.focus_position = "body" + self.editor_term.term.change_focus(True) def shortcuts(self): return self.shortcuts_display @@ -72,9 +1232,7 @@ class EditorTerminal(urwid.WidgetWrap): ) def quit_term(*args, **kwargs): - self.parent.widget = self.parent.config_explainer - self.app.ui.main_display.update_active_sub_display() - self.app.ui.main_display.show_config(None) + self.parent.switch_to_list() self.app.ui.main_display.request_redraw() urwid.connect_signal(self.term, 'closed', quit_term) @@ -83,7 +1241,4 @@ class EditorTerminal(urwid.WidgetWrap): def keypress(self, size, key): - # TODO: Decide whether there should be a way to get out while editing - #if key == "up": - # nomadnet.NomadNetworkApp.get_shared_instance().ui.main_display.frame.focus_position = "header" - return super(EditorTerminal, self).keypress(size, key) \ No newline at end of file + return super(EditorTerminal, self).keypress(size, key) diff --git a/nomadnet/ui/textui/Conversations.py b/nomadnet/ui/textui/Conversations.py index 4d30068..bb0fe21 100644 --- a/nomadnet/ui/textui/Conversations.py +++ b/nomadnet/ui/textui/Conversations.py @@ -13,6 +13,7 @@ from datetime import datetime, timedelta from nomadnet.Directory import DirectoryEntry from LXMF import pn_announce_data_is_valid, PN_META_NAME from nomadnet.Conversation import ConversationMessage +from .FileBrowser import FileBrowser from nomadnet.util import strip_modifiers from nomadnet.util import sanitize_name @@ -2422,7 +2423,29 @@ class ConversationWidget(urwid.WidgetWrap): def attach_file(self): self.dialog_active = True - browser = FileBrowserDialog(self) + + def on_change(selected): + self.pending_attachments[:] = selected + self.frame.contents["footer"] = (self._build_footer(), None) + + def on_done(selected): + self.pending_attachments[:] = selected + self.frame.contents["footer"] = (self._build_footer(), None) + self.file_browser_closed() + + def on_cancel(): + self.pending_attachments.clear() + self.frame.contents["footer"] = (self._build_footer(), None) + self.file_browser_closed() + + browser = FileBrowser( + on_done=on_done, + on_cancel=on_cancel, + on_change=on_change, + title="Attach File", + multi_select=True, + selected=self.pending_attachments, + ) bottom = self.messagelist overlay = urwid.Overlay(browser, bottom, align=urwid.CENTER, width=("relative", 90), valign=urwid.MIDDLE, height=("relative", 80), left=2, right=2) self.frame.contents["body"] = (overlay, self.frame.options()) @@ -2886,188 +2909,6 @@ def _save_attachment_to_disk(filename, data): return save_path -class FileBrowserEntry(urwid.WidgetWrap): - signals = ["click"] - - def __init__(self, name, full_path, is_dir=False, is_parent=False, selected=False): - self.full_path = full_path - self.name = name - self.is_dir = is_dir - self.is_parent = is_parent - self.selected = selected - g = nomadnet.NomadNetworkApp.get_shared_instance().ui.glyphs - if is_parent: - display = g["arrow_l"]+" .." - elif is_dir: - display = g["arrow_r"]+" "+name+"/" - elif selected: - display = g["check"]+" "+name - else: - display = " "+name - self.text_widget = urwid.SelectableIcon(display, 0) - if is_dir or is_parent: - style = "list_trusted" - focus_style = "list_focus" - elif selected: - style = "list_trusted" - focus_style = "list_focus_trusted" - else: - style = "list_unknown" - focus_style = "list_focus" - display_widget = urwid.AttrMap(self.text_widget, style, focus_style) - super().__init__(display_widget) - - def keypress(self, size, key): - if key == "enter": - self._emit("click") - else: - return key - - def mouse_event(self, size, event, button, x, y, focus): - if button == 1 and urwid.util.is_mouse_press(event): - self._emit("click") - return True - return False - - -class FileBrowserDialog(urwid.WidgetWrap): - def __init__(self, delegate): - self.delegate = delegate - app = nomadnet.NomadNetworkApp.get_shared_instance() - self.g = app.ui.glyphs - self.current_path = os.path.expanduser("~") - - self.path_label = urwid.Text("") - self.status_label = urwid.Text("") - self.file_walker = urwid.SimpleFocusListWalker([]) - self.file_listbox = urwid.ListBox(self.file_walker) - - self.button_columns = urwid.Columns([ - (urwid.WEIGHT, 0.45, urwid.Button("Done", on_press=self._dismiss)), - (urwid.WEIGHT, 0.1, urwid.Text("")), - (urwid.WEIGHT, 0.45, urwid.Button("Cancel", on_press=self._cancel)), - ]) - - header_pile = urwid.Pile([ - self.path_label, - self.status_label, - urwid.Divider(self.g["divider1"]), - ]) - - footer_pile = urwid.Pile([ - urwid.Divider(self.g["divider1"]), - self.button_columns, - ]) - - self._populate() - - self.browser_frame = urwid.Frame( - self.file_listbox, - header=header_pile, - footer=footer_pile, - ) - - linebox = urwid.LineBox(self.browser_frame, title="Attach File") - super().__init__(linebox) - - def _update_status(self): - pending = self.delegate.pending_attachments - if pending: - names = [os.path.basename(p) for p in pending] - self.status_label.set_text(" "+self.g["file"]+" "+str(len(pending))+" selected: "+", ".join(names)) - else: - self.status_label.set_text(" No files selected") - - def _populate(self): - self.path_label.set_text(" "+self.current_path) - self._update_status() - - focus_pos = None - try: - focus_pos = self.file_listbox.focus_position - except Exception: - pass - - entries = [] - parent = os.path.dirname(self.current_path) - if parent != self.current_path: - entry = FileBrowserEntry("..", parent, is_parent=True) - urwid.connect_signal(entry, "click", self._entry_clicked, entry) - entries.append(entry) - - try: - items = sorted(os.listdir(self.current_path)) - except PermissionError: - entries.append(urwid.Text(("error_text", " Permission denied"))) - self.file_walker[:] = entries - return - - dirs = [] - files = [] - for item in items: - if item.startswith("."): - continue - full = os.path.join(self.current_path, item) - if os.path.isdir(full): - dirs.append((item, full)) - elif os.path.isfile(full): - files.append((item, full)) - - for name, full in dirs: - entry = FileBrowserEntry(name, full, is_dir=True) - urwid.connect_signal(entry, "click", self._entry_clicked, entry) - entries.append(entry) - - for name, full in files: - is_selected = full in self.delegate.pending_attachments - entry = FileBrowserEntry(name, full, selected=is_selected) - urwid.connect_signal(entry, "click", self._entry_clicked, entry) - entries.append(entry) - - if not dirs and not files: - entries.append(urwid.Text(("inactive_text", " (empty)"))) - - self.file_walker[:] = entries - if focus_pos is not None and focus_pos < len(entries): - self.file_listbox.set_focus(focus_pos) - elif entries: - self.file_listbox.set_focus(0) - - def _entry_clicked(self, entry_widget, user_data=None): - entry = user_data if user_data else entry_widget - if entry.is_dir or entry.is_parent: - self.current_path = entry.full_path - self._populate() - else: - if entry.full_path in self.delegate.pending_attachments: - self.delegate.pending_attachments.remove(entry.full_path) - else: - self.delegate.pending_attachments.append(entry.full_path) - self.delegate.frame.contents["footer"] = (self.delegate._build_footer(), None) - self._populate() - - def _dismiss(self, sender): - self.delegate.file_browser_closed() - - def _cancel(self, sender): - self.delegate.pending_attachments.clear() - self.delegate.frame.contents["footer"] = (self.delegate._build_footer(), None) - self.delegate.file_browser_closed() - - def keypress(self, size, key): - if key == "esc": - self.delegate.file_browser_closed() - return - result = super().keypress(size, key) - if result == "down" and self.browser_frame.focus_position == "body": - self.browser_frame.focus_position = "footer" - return - elif result == "up" and self.browser_frame.focus_position == "footer": - self.browser_frame.focus_position = "body" - return - return result - - class SyncProgressBar(urwid.ProgressBar): def get_text(self): status = nomadnet.NomadNetworkApp.get_shared_instance().get_sync_status() diff --git a/nomadnet/ui/textui/Editor.py b/nomadnet/ui/textui/Editor.py index fd66046..a035d0d 100644 --- a/nomadnet/ui/textui/Editor.py +++ b/nomadnet/ui/textui/Editor.py @@ -400,7 +400,11 @@ class EditorColumns(urwid.Columns): if key == "esc": nomadnet.NomadNetworkApp.get_shared_instance().ui.main_display.frame.focus_position = "header" return None - return super().keypress(size, key) + result = super().keypress(size, key) + if key == "up" and result == "up": + nomadnet.NomadNetworkApp.get_shared_instance().ui.main_display.frame.focus_position = "header" + return None + return result class PageEditorShortcuts(): diff --git a/nomadnet/ui/textui/FileBrowser.py b/nomadnet/ui/textui/FileBrowser.py new file mode 100644 index 0000000..5a9863a --- /dev/null +++ b/nomadnet/ui/textui/FileBrowser.py @@ -0,0 +1,195 @@ +import os +import nomadnet +import urwid + +class FileBrowserEntry(urwid.WidgetWrap): + signals = ["click"] + + def __init__(self, name, full_path, is_dir=False, is_parent=False, selected=False): + self.full_path = full_path + self.name = name + self.is_dir = is_dir + self.is_parent = is_parent + self.selected = selected + g = nomadnet.NomadNetworkApp.get_shared_instance().ui.glyphs + if is_parent: + display = g["arrow_l"]+" .." + elif is_dir: + display = g["arrow_r"]+" "+name+"/" + elif selected: + display = g["check"]+" "+name + else: + display = " "+name + self.text_widget = urwid.SelectableIcon(display, 0) + if is_dir or is_parent: + style = "list_trusted" + focus_style = "list_focus" + elif selected: + style = "list_trusted" + focus_style = "list_focus_trusted" + else: + style = "list_unknown" + focus_style = "list_focus" + display_widget = urwid.AttrMap(self.text_widget, style, focus_style) + super().__init__(display_widget) + + def keypress(self, size, key): + if key == "enter": + self._emit("click") + else: + return key + + def mouse_event(self, size, event, button, x, y, focus): + if button == 1 and urwid.util.is_mouse_press(event): + self._emit("click") + return True + return False + +class FileBrowser(urwid.WidgetWrap): + def __init__(self, on_done=None, on_cancel=None, on_change=None, title="Select File", multi_select=True, start_path=None, selected=None): + self.on_done_cb = on_done + self.on_cancel_cb = on_cancel + self.on_change_cb = on_change + self.multi_select = multi_select + self.selected = list(selected) if selected else [] + + app = nomadnet.NomadNetworkApp.get_shared_instance() + self.g = app.ui.glyphs + self.current_path = os.path.expanduser(start_path) if start_path else os.path.expanduser("~") + + self.path_label = urwid.Text("") + self.status_label = urwid.Text("") + self.file_walker = urwid.SimpleFocusListWalker([]) + self.file_listbox = urwid.ListBox(self.file_walker) + + self.button_columns = urwid.Columns([ + (urwid.WEIGHT, 0.45, urwid.Button("Done", on_press=self._dismiss)), + (urwid.WEIGHT, 0.1, urwid.Text("")), + (urwid.WEIGHT, 0.45, urwid.Button("Cancel", on_press=self._cancel)), + ]) + + header_pile = urwid.Pile([ + self.path_label, + self.status_label, + urwid.Divider(self.g["divider1"]), + ]) + footer_pile = urwid.Pile([ + urwid.Divider(self.g["divider1"]), + self.button_columns, + ]) + + self._populate() + + self.browser_frame = urwid.Frame(self.file_listbox, header=header_pile, footer=footer_pile) + super().__init__(urwid.LineBox(self.browser_frame, title=title)) + + def _update_status(self): + if self.selected: + names = [os.path.basename(p) for p in self.selected] + self.status_label.set_text(" "+self.g["file"]+" "+str(len(self.selected))+" selected: "+", ".join(names)) + elif self.multi_select: + self.status_label.set_text(" No files selected") + else: + self.status_label.set_text(" Select a file") + + def _populate(self): + self.path_label.set_text(" "+self.current_path) + self._update_status() + + focus_pos = None + try: + focus_pos = self.file_listbox.focus_position + except Exception: + pass + + entries = [] + parent = os.path.dirname(self.current_path) + if parent != self.current_path: + entry = FileBrowserEntry("..", parent, is_parent=True) + urwid.connect_signal(entry, "click", self._entry_clicked, entry) + entries.append(entry) + + try: + items = sorted(os.listdir(self.current_path)) + except PermissionError: + entries.append(urwid.Text(("error_text", " Permission denied"))) + self.file_walker[:] = entries + return + except Exception: + entries.append(urwid.Text(("error_text", " Cannot read directory"))) + self.file_walker[:] = entries + return + + dirs = [] + files = [] + for item in items: + if item.startswith("."): + continue + full = os.path.join(self.current_path, item) + if os.path.isdir(full): + dirs.append((item, full)) + elif os.path.isfile(full): + files.append((item, full)) + + for name, full in dirs: + entry = FileBrowserEntry(name, full, is_dir=True) + urwid.connect_signal(entry, "click", self._entry_clicked, entry) + entries.append(entry) + + for name, full in files: + entry = FileBrowserEntry(name, full, selected=(full in self.selected)) + urwid.connect_signal(entry, "click", self._entry_clicked, entry) + entries.append(entry) + + if not dirs and not files: + entries.append(urwid.Text(("inactive_text", " (empty)"))) + + self.file_walker[:] = entries + if focus_pos is not None and focus_pos < len(entries): + self.file_listbox.set_focus(focus_pos) + elif entries: + self.file_listbox.set_focus(0) + + def _entry_clicked(self, entry_widget, user_data=None): + entry = user_data if user_data else entry_widget + if entry.is_dir or entry.is_parent: + self.current_path = entry.full_path + self._populate() + return + + if self.multi_select: + if entry.full_path in self.selected: + self.selected.remove(entry.full_path) + else: + self.selected.append(entry.full_path) + if self.on_change_cb is not None: + self.on_change_cb(list(self.selected)) + self._populate() + else: + self.selected = [entry.full_path] + if self.on_change_cb is not None: + self.on_change_cb(list(self.selected)) + if self.on_done_cb is not None: + self.on_done_cb(list(self.selected)) + + def _dismiss(self, sender): + if self.on_done_cb is not None: + self.on_done_cb(list(self.selected)) + + def _cancel(self, sender): + self.selected = [] + if self.on_cancel_cb is not None: + self.on_cancel_cb() + + def keypress(self, size, key): + if key == "esc": + self._cancel(None) + return None + result = super().keypress(size, key) + if result == "down" and self.browser_frame.focus_position == "body": + self.browser_frame.focus_position = "footer" + return None + elif result == "up" and self.browser_frame.focus_position == "footer": + self.browser_frame.focus_position = "body" + return None + return result diff --git a/nomadnet/ui/textui/Interfaces.py b/nomadnet/ui/textui/Interfaces.py index 6d49738..0a74333 100644 --- a/nomadnet/ui/textui/Interfaces.py +++ b/nomadnet/ui/textui/Interfaces.py @@ -1488,6 +1488,9 @@ class InterfaceFiller(urwid.WidgetWrap): if key == "ctrl o": self.app.ui.main_display.sub_displays.interface_display.toggle_focused_interface() return None + if key == "ctrl b": + self.app.ui.main_display.sub_displays.interface_display.show_bulk_actions() + return None if key == "ctrl a": # add interface self.app.ui.main_display.sub_displays.interface_display.add_interface() @@ -3401,11 +3404,12 @@ class InterfaceDisplay: def _make_filter_bar(self): cols = [('weight', 1, urwid.Text(""))] - for key, label in (("all", "All"), ("enabled", "Enabled"), ("disabled", "Disabled")): + cols.append(('pack', urwid.Text(("interface_title", "Filter by: ")))) + for key, label in (("all", "All"), ("enabled", "Enabled"), ("disabled", "Disabled"), ("connected", "Connected"), ("disconnected", "Disconnected")): sel = self.iface_filter == key mark = self.g['selected'] if sel else self.g['unselected'] style = "interface_tile_focus" if sel else "interface_title" - cols.append(('pack', ClickableIcon((style, "%s %s " % (mark, label)), on_click=lambda k=key: self._set_filter(k)))) + cols.append(('pack', ClickableIcon((style, "%s %s " % (mark, label)), on_click=lambda k=key: self._set_filter(k)))) cols.append(('weight', 1, urwid.Text(""))) return urwid.Columns(cols, dividechars=0) @@ -3418,7 +3422,7 @@ class InterfaceDisplay: self.app.ui.main_display.update_active_sub_display() def cycle_filter(self): - order = ["all", "enabled", "disabled"] + order = ["all", "enabled", "disabled", "connected", "disconnected"] i = order.index(self.iface_filter) if self.iface_filter in order else 0 self._set_filter(order[(i + 1) % len(order)]) @@ -3870,6 +3874,11 @@ class InterfaceDisplay: is_connected = False announce_rate = "-" + if self.iface_filter == "connected" and not is_connected: + continue + if self.iface_filter == "disconnected" and is_connected: + continue + profiles_label = profiles.label_for(interface_name) if profiles is not None else None item = SelectableInterfaceItem( @@ -3959,6 +3968,118 @@ class InterfaceDisplay: self._focus_interface(name) self.app.ui.main_display.update_active_sub_display() + def _iface_is_enabled(self, iface): + return str(iface.get("enabled")).lower() not in ('false', 'off', 'no', '0') and \ + str(iface.get("interface_enabled")).lower() not in ('false', 'off', 'no', '0') + + def _set_iface_enabled(self, iface, enabled): + value = "true" if enabled else "false" + iface["interface_enabled"] = value + if "enabled" in iface: + iface["enabled"] = value + + def show_bulk_actions(self): + interfaces = self.app.rns.config['interfaces'] + if not interfaces: + self._bulk_notice("No interfaces configured.") + return + + dialog_widgets = [ + urwid.Text(("interface_title", "Apply an action to all interfaces"), align="left"), + urwid.Divider(), + InterfaceOptionItem(self, "Enable all interfaces", None, on_select=lambda: self._bulk_apply("enable")), + InterfaceOptionItem(self, "Disable all interfaces", None, on_select=lambda: self._bulk_apply("disable")), + InterfaceOptionItem(self, "Invert (toggle each interface)", None, on_select=lambda: self._bulk_apply("invert")), + InterfaceOptionItem(self, "Solo selected (disable all others)", None, on_select=lambda: self._bulk_apply("solo")), + ] + + listbox = urwid.ListBox(urwid.SimpleFocusListWalker(dialog_widgets)) + dialog = DialogLineBox(listbox, parent=self, title="Bulk Actions") + + overlay = urwid.Overlay( + dialog, + self.interfaces_display, + align='center', + width=('relative', 50), + valign='middle', + height=('relative', 50), + min_width=20, + min_height=10, + left=2, + right=2 + ) + self.widget = overlay + self.app.ui.main_display.update_active_sub_display() + + def _bulk_apply(self, action): + interfaces = self.app.rns.config['interfaces'] + focused = self._focused_interface_name() + + if action == "solo" and focused is None: + self._bulk_notice("Select an interface first to solo it.") + return + + changed = 0 + for name, iface in interfaces.items(): + current = self._iface_is_enabled(iface) + if action == "enable": + target = True + elif action == "disable": + target = False + elif action == "invert": + target = not current + elif action == "solo": + target = (name == focused) + else: + target = current + + if target != current: + self._set_iface_enabled(iface, target) + changed += 1 + + try: + self.app.rns.config.write() + except Exception as e: + self._bulk_notice(f"Error applying bulk action: {str(e)}", title="Error") + return + + self.mark_restart_pending() + self._build_interface_items() + self._rebuild_list() + + verb = {"enable": "Enabled all", "disable": "Disabled all", "invert": "Inverted all", "solo": "Soloed selected"}[action] + if changed == 0: + self._bulk_notice("No interfaces needed changing.") + else: + self._bulk_notice(f"{verb} interfaces ({changed} changed).\nRestart required for changes to take effect.") + + def _bulk_notice(self, message, title="Notice"): + def dismiss_dialog(button): + self.dismiss_dialog() + + dialog = DialogLineBox( + urwid.Pile([ + urwid.Text(message, align="center"), + urwid.Divider(), + urwid.Button("OK", on_press=dismiss_dialog) + ]), + parent=self, + title=title + ) + + overlay = urwid.Overlay( + dialog, + self.interfaces_display, + align='center', + width=60, + valign='middle', + height=8, + min_width=20, + min_height=1 + ) + self.widget = overlay + self.app.ui.main_display.update_active_sub_display() + def save_current_profile(self): profiles = getattr(self.app, "interface_profiles", None) if profiles is None: @@ -4024,7 +4145,7 @@ class InterfaceDisplay: class InterfaceDisplayShortcuts: def __init__(self, app): self.app = app - self.default_shortcuts = "[C-a] Add [C-e] Edit [C-x] Remove [C-o] On/Off [Enter] Show [C-f] Filter [C-s] Save Profile [C-t] Grid/Rows [Tab] Profiles" + self.default_shortcuts = "[C-a] Add [C-e] Edit [C-x] Remove [C-o] On/Off [C-b] Bulk [Enter] Show [C-f] Filter [C-s] Save Profile [C-t] Grid/Rows [Tab] Profiles" self.current_shortcuts = self.default_shortcuts self.widget = urwid.AttrMap( urwid.Text(self.current_shortcuts), diff --git a/nomadnet/ui/textui/Main.py b/nomadnet/ui/textui/Main.py index 12fad81..9eb8ed1 100644 --- a/nomadnet/ui/textui/Main.py +++ b/nomadnet/ui/textui/Main.py @@ -6,6 +6,8 @@ from .Channels import * from .Directory import * from .Config import * from .Interfaces import * +from .Transport import * +from .Utilities import * from .Map import * from .Log import * from .Guide import * @@ -21,6 +23,7 @@ class SubDisplays(): self.directory_display = DirectoryDisplay(self.app) self.config_display = ConfigDisplay(self.app) self.interface_display = InterfaceDisplay(self.app) + self.utilities_display = UtilitiesDisplay(self.app) if self._utilities_enabled(app) else None self.map_display = MapDisplay(self.app) self.log_display = LogDisplay(self.app) self.guide_display = GuideDisplay(self.app) @@ -31,6 +34,12 @@ class SubDisplays(): else: self.active_display = self.conversations_display + def _utilities_enabled(self, app): + try: + return app.config["utilities"].as_bool("enable_utilities") + except Exception: + return False + def active(self): return self.active_display @@ -141,6 +150,13 @@ class MainDisplay(): self.update_active_sub_display() self.sub_displays.interface_display.start() + def show_utilities(self, user_data): + if self.sub_displays.utilities_display is None: + return + self.sub_displays.active_display = self.sub_displays.utilities_display + self.update_active_sub_display() + self.sub_displays.utilities_display.start() + def show_log(self, user_data): self.sub_displays.active_display = self.sub_displays.log_display self.sub_displays.log_display.show() @@ -184,6 +200,11 @@ class MainDisplay(): if getattr(self, "_quit_dialog_open", False): return + active = self.sub_displays.active_display + if active in (self.sub_displays.log_display, self.sub_displays.config_display): + self.do_quit() + return + def confirm(button=None): self.do_quit() @@ -218,7 +239,7 @@ class MainDisplay(): ) self._quit_dialog_open = True - + self.app.ui.loop.widget = urwid.Overlay( dialog, self.frame, align="center", width=44, @@ -264,6 +285,7 @@ class MenuDisplay(): 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_utilities = (13, MenuButton("Utilities", on_press=handler.show_utilities)) button_guide = (9, MenuButton("Guide", on_press=handler.show_guide)) button_mynode = (11, MenuButton("My Node", on_press=handler.show_page_editor)) button_quit = (8, MenuButton("Quit", on_press=handler.quit)) @@ -272,7 +294,15 @@ class MenuDisplay(): buttons = [menu_text, button_conversations, button_network] if self.app.enable_node: buttons.append(button_mynode) - buttons += [button_channels, button_log, button_interfaces, button_config] + buttons += [button_channels, button_log, button_interfaces] + enable_utilities = False + try: + enable_utilities = self.app.config["utilities"].as_bool("enable_utilities") + except Exception: + enable_utilities = False + if enable_utilities: + buttons.append(button_utilities) + buttons.append(button_config) if not self.app.config["textui"]["hide_guide"]: buttons.append(button_guide) buttons.append(button_quit) diff --git a/nomadnet/ui/textui/Transport.py b/nomadnet/ui/textui/Transport.py new file mode 100644 index 0000000..dbc7c69 --- /dev/null +++ b/nomadnet/ui/textui/Transport.py @@ -0,0 +1,503 @@ +import RNS +import time +import nomadnet +import urwid + +def _get_cols_rows(): + return nomadnet.NomadNetworkApp.get_shared_instance().ui.screen.get_cols_rows() + +def _fmt_secs(secs): + secs = int(secs) + if secs < 0: + secs = 0 + if secs < 60: + return "%ds" % secs + if secs < 3600: + return "%dm" % (secs // 60) + if secs < 86400: + return "%dh" % (secs // 3600) + return "%dd" % (secs // 86400) + +def _rel_future(ts): + if not ts: + return "-" + remaining = ts - time.time() + if remaining <= 0: + return "expired" + return _fmt_secs(remaining) + +def _rel_past(ts): + if not ts: + return "-" + return _fmt_secs(time.time() - ts) + " ago" + +def _fmt_duration(secs): + if not secs: + return "n/a" + secs = int(secs) + if secs < 0: + secs = 0 + days = secs // 86400 + hours = (secs % 86400) // 3600 + minutes = (secs % 3600) // 60 + if days: + return "%dd %dh" % (days, hours) + if hours: + return "%dh %dm" % (hours, minutes) + if minutes: + return "%dm" % minutes + return "%ds" % secs + +def _fmt_rate(hz): + try: + hz = float(hz or 0) + except Exception: + return "0" + if hz <= 0: + return "0" + per_hour = hz * 3600 + if per_hour >= 1: + return "%.0f/h" % per_hour + return "%.1f/d" % (hz * 86400) + +def _mode_str(mode): + return {1: "Full", 2: "Point-to-Point", 3: "Access Point", 4: "Roaming", 5: "Boundary", 6: "Gateway"}.get(mode, "Full") + +class TransportRow(urwid.WidgetWrap): + def __init__(self, columns): + self._selectable = True + super().__init__(urwid.AttrMap(columns, "list_normal", focus_map="list_focus")) + + def selectable(self): + return True + + def keypress(self, size, key): + return key + +class LazyTableWalker(urwid.ListWalker): + def __init__(self): + self.items = [] + self.factory = None + self.empty_widget = None + self.focus = 0 + + def configure(self, items, factory, empty_widget=None): + self.items = items or [] + self.factory = factory + self.empty_widget = empty_widget + if self.focus >= len(self.items): + self.focus = max(0, len(self.items) - 1) + self._modified() + + def _build(self, position): + return self.factory(self.items[position], position) + + def get_focus(self): + if not self.items: + if self.empty_widget is not None: + return (self.empty_widget, 0) + return (None, None) + if self.focus >= len(self.items): + self.focus = len(self.items) - 1 + if self.focus < 0: + self.focus = 0 + return (self._build(self.focus), self.focus) + + def set_focus(self, position): + self.focus = position + self._modified() + + def get_next(self, position): + n = position + 1 + if not self.items or n >= len(self.items): + return (None, None) + return (self._build(n), n) + + def get_prev(self, position): + p = position - 1 + if not self.items or p < 0: + return (None, None) + return (self._build(p), p) + + def positions(self, reverse=False): + count = len(self.items) + if count == 0: + return [0] if self.empty_widget is not None else [] + if reverse: + return range(count - 1, -1, -1) + return range(count) + + def __len__(self): + return len(self.items) + + def __getitem__(self, position): + if not self.items: + if position == 0 and self.empty_widget is not None: + return self.empty_widget + raise IndexError(position) + if position < 0 or position >= len(self.items): + raise IndexError(position) + return self._build(position) + +class TransportFiller(urwid.WidgetWrap): + def __init__(self, widget, display): + self.display = display + super().__init__(urwid.Filler(widget, urwid.TOP)) + + def keypress(self, size, key): + if key == "tab": + self.display.cycle_tab() + return None + if key == "ctrl r" or key == "f5": + self.display.refresh() + return None + if key == "esc": + self.display.go_back() + return None + result = super().keypress(size, key) + if key == "up" and result == "up": + self.display.go_back_to_header() + return None + return result + +class TransportDisplayShortcuts: + def __init__(self, app): + self.app = app + self.widget = urwid.AttrMap( + urwid.Text("[Tab] Switch Table [Up] Filter [C-r] Refresh [Esc] Back"), + "shortcutbar" + ) + +class TransportDisplay: + POLL_INTERVAL = 5 + + HASH_W = 34 + HOPS_W = 6 + VIA_W = 20 + EXP_W = 12 + LAST_W = 16 + VIOL_W = 11 + STATUS_W = 10 + PATHS_W = 8 + HELD_W = 8 + QUEUE_W = 8 + RATE_W = 10 + MODE_W = 12 + + def __init__(self, app, parent=None): + self.app = app + self.parent = parent + self.started = False + self.poll_scheduler = False + self.glyphset = self.app.config["textui"]["glyphs"] + self.g = self.app.ui.glyphs + + self.active_tab = "paths" + self.paths = [] + self.rates = [] + self.interfaces = [] + self.paths_per_interface = {} + self.link_count = 0 + self.transport_id = None + self.network_id = None + self.transport_uptime = None + self.held_announces = 0 + self.filter_text = "" + + self.summary_text = urwid.Text("") + self.list_walker = LazyTableWalker() + self.list_box = urwid.ListBox(self.list_walker) + + self.filter_edit = urwid.Edit("", "") + urwid.connect_signal(self.filter_edit, "change", self._on_filter_change) + self.filter_row = urwid.Columns([ + (9, urwid.Text(("key", "Filter:"))), + urwid.AttrMap(self.filter_edit, "list_normal", focus_map="list_focus"), + ], dividechars=1) + + _cols, rows = _get_cols_rows() + self.box_adapter = urwid.BoxAdapter(self.list_box, max(3, rows - 10)) + + self.pile = urwid.Pile([ + ('pack', self.summary_text), + ('pack', self._tab_bar()), + ('pack', self.filter_row), + ('pack', self._header_row()), + ('pack', urwid.Divider("─")), + self.box_adapter, + ]) + + self.transport_view = TransportFiller(self.pile, self) + self.shortcuts_display = TransportDisplayShortcuts(self.app) + self.widget = self.transport_view + + self._refresh_data() + self._rebuild() + + def shortcuts(self): + return self.shortcuts_display + + def _tab_bar(self): + def tab(label, key): + attr = "list_focus" if self.active_tab == key else "list_normal" + return (urwid.PACK, urwid.Text((attr, " %s " % label))) + + return urwid.Columns([ + tab("Paths", "paths"), + tab("Announce Rates", "rates"), + tab("Interfaces", "interfaces"), + (urwid.WEIGHT, 1, urwid.Text("")), + ], dividechars=1) + + def _header_row(self): + if self.active_tab == "paths": + cols = [ + (self.HASH_W, urwid.Text(("key", "Destination"), wrap="clip")), + (self.HOPS_W, urwid.Text(("key", "Hops"), wrap="clip")), + (urwid.WEIGHT, 1, urwid.Text(("key", "Interface"), wrap="clip")), + (self.VIA_W, urwid.Text(("key", "Via"), wrap="clip")), + (self.EXP_W, urwid.Text(("key", "Expires"), wrap="clip")), + ] + elif self.active_tab == "interfaces": + cols = [ + (urwid.WEIGHT, 1, urwid.Text(("key", "Interface"), wrap="clip")), + (self.STATUS_W, urwid.Text(("key", "Status"), wrap="clip")), + (self.PATHS_W, urwid.Text(("key", "Paths"), wrap="clip")), + (self.HELD_W, urwid.Text(("key", "Held"), wrap="clip")), + (self.QUEUE_W, urwid.Text(("key", "Queue"), wrap="clip")), + (self.RATE_W, urwid.Text(("key", "Ann in"), wrap="clip")), + (self.MODE_W, urwid.Text(("key", "Mode"), wrap="clip")), + ] + else: + cols = [ + (self.HASH_W, urwid.Text(("key", "Destination"), wrap="clip")), + (self.LAST_W, urwid.Text(("key", "Last announce"), wrap="clip")), + (self.VIOL_W, urwid.Text(("key", "Violations"), wrap="clip")), + (urwid.WEIGHT, 1, urwid.Text(("key", "Blocked"), wrap="clip")), + ] + return urwid.Columns(cols, dividechars=1) + + def _refresh_data(self): + try: + self.paths = self.app.rns.get_path_table() or [] + except Exception: + self.paths = [] + try: + self.rates = self.app.rns.get_rate_table() or [] + except Exception: + self.rates = [] + try: + self.link_count = self.app.rns.get_link_count() + except Exception: + self.link_count = 0 + + try: + stats = self.app.rns.get_interface_stats() or {} + except Exception: + stats = {} + self.transport_id = stats.get("transport_id") + self.network_id = stats.get("network_id") + self.transport_uptime = stats.get("transport_uptime") + self.interfaces = stats.get("interfaces", []) or [] + try: + self.held_announces = sum(int(i.get("held_announces", 0) or 0) for i in self.interfaces) + except Exception: + self.held_announces = 0 + + self.paths_per_interface = {} + for entry in self.paths: + iface = entry.get("interface", "") + self.paths_per_interface[iface] = self.paths_per_interface.get(iface, 0) + 1 + + try: + self.interfaces.sort(key=lambda i: self.paths_per_interface.get(i.get("name", ""), 0), reverse=True) + except Exception: + pass + + try: + self.paths.sort(key=lambda e: (e.get("interface", ""), e.get("hops", 0))) + except Exception: + pass + try: + self.rates.sort(key=lambda e: e.get("last", 0), reverse=True) + except Exception: + pass + + def _make_path_row(self, entry, position): + dest = RNS.prettyhexrep(entry["hash"]) if entry.get("hash") else "?" + via = RNS.prettyhexrep(entry["via"]) if entry.get("via") else "?" + hops = entry.get("hops", 0) + iface = entry.get("interface", "") + expires = _rel_future(entry.get("expires", 0)) + + columns = urwid.Columns([ + (self.HASH_W, urwid.Text(("list_normal", dest), wrap="clip")), + (self.HOPS_W, urwid.Text(str(hops), wrap="clip")), + (urwid.WEIGHT, 1, urwid.Text(iface, wrap="clip")), + (self.VIA_W, urwid.Text(via, wrap="clip")), + (self.EXP_W, urwid.Text(expires, wrap="clip")), + ], dividechars=1) + return TransportRow(columns) + + def _make_rate_row(self, entry, position): + now = time.time() + dest = RNS.prettyhexrep(entry["hash"]) if entry.get("hash") else "?" + last = _rel_past(entry.get("last", 0)) + violations = str(entry.get("rate_violations", 0)) + blocked_until = entry.get("blocked_until", 0) or 0 + if blocked_until > now: + blocked = ("disconnected_status", "blocked %s" % _rel_future(blocked_until)) + else: + blocked = ("connected_status", "no") + + columns = urwid.Columns([ + (self.HASH_W, urwid.Text(("list_normal", dest), wrap="clip")), + (self.LAST_W, urwid.Text(last, wrap="clip")), + (self.VIOL_W, urwid.Text(violations, wrap="clip")), + (urwid.WEIGHT, 1, urwid.Text(blocked, wrap="clip")), + ], dividechars=1) + return TransportRow(columns) + + def _make_iface_row(self, entry, position): + name = entry.get("short_name") or entry.get("name") or "?" + paths = self.paths_per_interface.get(entry.get("name", ""), 0) + held = entry.get("held_announces", 0) or 0 + queue = entry.get("announce_queue") + queue_str = "-" if queue is None else str(queue) + rate = _fmt_rate(entry.get("incoming_announce_frequency", 0)) + mode = _mode_str(entry.get("mode")) + + if entry.get("status"): + status = ("connected_status", "up") + else: + status = ("disconnected_status", "down") + + columns = urwid.Columns([ + (urwid.WEIGHT, 1, urwid.Text(("list_normal", name), wrap="clip")), + (self.STATUS_W, urwid.Text(status, wrap="clip")), + (self.PATHS_W, urwid.Text(str(paths), wrap="clip")), + (self.HELD_W, urwid.Text(str(held), wrap="clip")), + (self.QUEUE_W, urwid.Text(queue_str, wrap="clip")), + (self.RATE_W, urwid.Text(rate, wrap="clip")), + (self.MODE_W, urwid.Text(mode, wrap="clip")), + ], dividechars=1) + return TransportRow(columns) + + def _rebuild(self): + blocked_now = 0 + try: + now = time.time() + blocked_now = sum(1 for r in self.rates if (r.get("blocked_until", 0) or 0) > now) + except Exception: + pass + + transport_id = RNS.prettyhexrep(self.transport_id) if self.transport_id else "n/a" + network_id = RNS.prettyhexrep(self.network_id) if self.network_id else "n/a" + uptime = _fmt_duration(self.transport_uptime) + + summary = [ + ("key", "Transport ID: "), ("value", transport_id), + ("key", " Network ID: "), ("value", network_id), + ("key", " Uptime: "), ("value", uptime), "\n", + ("key", "Active links: "), ("value", str(self.link_count)), + ("key", " Known paths: "), ("value", str(len(self.paths))), + ("key", " Held announces: "), ("value", str(self.held_announces)), + ("key", " Announce sources: "), ("value", str(len(self.rates))), + ("key", " Rate-limited: "), ("value", str(blocked_now)), + ] + + if self.active_tab == "paths": + data = self._filtered(self.paths) + factory = self._make_path_row + empty_msg = "No paths in the routing table yet." + elif self.active_tab == "interfaces": + data = self._filtered(self.interfaces) + factory = self._make_iface_row + empty_msg = "No interface statistics available." + else: + data = self._filtered(self.rates) + factory = self._make_rate_row + empty_msg = "No announce rate data available." + + if self.filter_text.strip(): + summary.append(("inactive_text", " [filter: %d shown]" % len(data))) + self.summary_text.set_text(summary) + + self.pile.contents[1] = (self._tab_bar(), self.pile.options('pack')) + self.pile.contents[3] = (self._header_row(), self.pile.options('pack')) + + empty = urwid.Text(("inactive_text", empty_msg), align=urwid.CENTER) + self.list_walker.configure(data, factory, empty) + + try: + _cols, rows = _get_cols_rows() + self.box_adapter.height = max(3, rows - 10) + except Exception: + pass + + def cycle_tab(self): + order = ["paths", "rates", "interfaces"] + i = order.index(self.active_tab) if self.active_tab in order else 0 + self.active_tab = order[(i + 1) % len(order)] + self._rebuild() + + def _on_filter_change(self, widget, new_text): + self.filter_text = new_text + self._rebuild() + + def _haystack(self, entry): + parts = [] + for key in ("hash", "via"): + value = entry.get(key) + if value: + parts.append(RNS.prettyhexrep(value)) + for key in ("interface", "name", "short_name"): + value = entry.get(key) + if value: + parts.append(str(value)) + return " ".join(parts).lower() + + def _filtered(self, items): + needle = self.filter_text.strip().lower() + if not needle: + return items + return [entry for entry in items if needle in self._haystack(entry)] + + def go_back(self): + if self.parent is not None: + self.parent.show_landing() + + def go_back_to_header(self): + try: + self.app.ui.main_display.frame.focus_position = "header" + except Exception: + pass + + def _is_visible(self): + md = self.app.ui.main_display + if self.parent is not None: + return md.sub_displays.active_display is self.parent and getattr(self.parent, "active", None) == "transport" + return md.sub_displays.active_display is self + + def refresh(self): + self._refresh_data() + self._rebuild() + + def start(self): + self.started = True + self._refresh_data() + self._rebuild() + if not self.poll_scheduler and self.app.ui.loop is not None: + self.poll_scheduler = True + self.app.ui.loop.set_alarm_in(self.POLL_INTERVAL, self._poll) + + def _poll(self, loop, user_data): + if self._is_visible(): + self._refresh_data() + self._rebuild() + try: + loop.draw_screen() + except Exception: + pass + if self.started: + loop.set_alarm_in(self.POLL_INTERVAL, self._poll) diff --git a/nomadnet/ui/textui/Utilities.py b/nomadnet/ui/textui/Utilities.py new file mode 100644 index 0000000..088f5a4 --- /dev/null +++ b/nomadnet/ui/textui/Utilities.py @@ -0,0 +1,1523 @@ +import RNS +import os +import time +import nomadnet +import urwid + +from .Transport import TransportDisplay, _get_cols_rows +from .ReadlineEdit import ReadlineEdit +from .FileBrowser import FileBrowser + +DEFAULT_PROBE_SIZE = 16 +DEFAULT_PROBE_TIMEOUT = 12 + +def _interface_enabled(iface): + try: + return str(iface.get("enabled")).lower() not in ("false", "off", "no", "0") and \ + str(iface.get("interface_enabled")).lower() not in ("false", "off", "no", "0") + except Exception: + return True + +def _has_active_interfaces(app): + try: + interfaces = app.rns.config["interfaces"] + except Exception: + return True + try: + return any(_interface_enabled(interfaces[name]) for name in interfaces) + except Exception: + return True + +class UtilityItem(urwid.WidgetWrap): + def __init__(self, parent, name, description, on_select): + self.parent = parent + self.name = name + self.on_select = on_select + self._selectable = True + + self.selection_txt = urwid.Text(" ") + self.title_widget = urwid.Text(("interface_title", name)) + + title_row = urwid.Columns([ + (4, self.selection_txt), + self.title_widget, + ]) + description_row = urwid.Columns([ + (4, urwid.Text(" ")), + urwid.Text(("value", description)), + ]) + + box = urwid.LineBox( + urwid.Padding(urwid.Pile([title_row, description_row]), left=2, right=2), + title=None, + tlcorner="╭", tline="─", + trcorner="╮", lline="│", + rline="│", blcorner="╰", + bline="─", brcorner="╯" + ) + super().__init__(box) + + 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']) + attr = "interface_title_selected" if focus else "interface_title" + self.title_widget.set_text((attr, self.name)) + return super().render(size, focus=focus) + + def keypress(self, size, key): + if key == "up": + if self.parent.landing_listbox.focus_position == 0: + self.parent.app.ui.main_display.frame.focus_position = "header" + return None + elif key == "enter": + self.on_select() + return None + return key + +class UtilitiesDisplayShortcuts: + def __init__(self, app): + self.app = app + self.widget = urwid.AttrMap( + urwid.Text("[Enter] Open [Up/Down] Navigate"), + "shortcutbar" + ) + +class UtilitiesDisplay: + def __init__(self, app): + self.app = app + self.glyphset = self.app.config["textui"]["glyphs"] + self.g = self.app.ui.glyphs + self.active = "landing" + self.transport = None + self.probe = None + self.path = None + self.identity = None + self.send = None + self.speedtest = None + + self.shortcuts_display = UtilitiesDisplayShortcuts(self.app) + self._build_landing() + self.widget = self.landing_view + + def _build_landing(self): + items = [ + UtilityItem(self, "Path Tables", "View routing, announce rate and per-interface tables", self.open_transport), + UtilityItem(self, "Path Resolver", "Look up or drop the path to a destination", self.open_path), + UtilityItem(self, "Probe", "Send probes to a destination and measure round trip", self.open_probe), + UtilityItem(self, "Identity", "Show your addresses and look up a destination's identity", self.open_identity), + UtilityItem(self, "Send File", "Send a file to a destination over Reticulum to a rncp listener", self.open_send), + UtilityItem(self, "Speed Test", "Measure throughput to a peer also running this screen", self.open_speedtest), + ] + self.landing_listbox = urwid.ListBox(urwid.SimpleFocusListWalker(items)) + + _cols, rows = _get_cols_rows() + box = urwid.BoxAdapter(self.landing_listbox, max(4, rows - 6)) + + pile = urwid.Pile([ + ('pack', urwid.Text(("body_text", "Select a utility to open."), align=urwid.CENTER)), + ('pack', urwid.Divider("─")), + box, + ]) + self.landing_view = urwid.Filler(pile, urwid.TOP) + + def shortcuts(self): + if self.active == "transport" and self.transport is not None: + return self.transport.shortcuts_display + if self.active == "probe" and self.probe is not None: + return self.probe.shortcuts_display + if self.active == "path" and self.path is not None: + return self.path.shortcuts_display + if self.active == "identity" and self.identity is not None: + return self.identity.shortcuts_display + if self.active == "send" and self.send is not None: + return self.send.shortcuts_display + if self.active == "speedtest" and self.speedtest is not None: + return self.speedtest.shortcuts_display + return self.shortcuts_display + + def _leave_speedtest(self): + if self.speedtest is not None and self.active == "speedtest": + self.speedtest.stop_listening() + + def start(self): + self.show_landing() + + def show_landing(self): + self._leave_speedtest() + self.active = "landing" + self.widget = self.landing_view + self.app.ui.main_display.update_active_sub_display() + self.app.ui.main_display.frame.focus_position = "body" + + def open_transport(self): + if self.transport is None: + self.transport = TransportDisplay(self.app, parent=self) + self.active = "transport" + self.widget = self.transport.widget + self.app.ui.main_display.update_active_sub_display() + self.app.ui.main_display.frame.focus_position = "body" + self.transport.start() + + def open_probe(self): + if self.probe is None: + self.probe = ProbeView(self.app, parent=self) + self.probe.update_interface_status() + self.active = "probe" + self.widget = self.probe.widget + self.app.ui.main_display.update_active_sub_display() + self.app.ui.main_display.frame.focus_position = "body" + + def open_path(self): + if self.path is None: + self.path = PathView(self.app, parent=self) + self.path.update_interface_status() + self.active = "path" + self.widget = self.path.widget + self.app.ui.main_display.update_active_sub_display() + self.app.ui.main_display.frame.focus_position = "body" + + def open_identity(self): + if self.identity is None: + self.identity = IdentityView(self.app, parent=self) + self.active = "identity" + self.widget = self.identity.widget + self.app.ui.main_display.update_active_sub_display() + self.app.ui.main_display.frame.focus_position = "body" + + def open_send(self): + if self.send is None: + self.send = SendFileView(self.app, parent=self) + self.send.update_interface_status() + self.active = "send" + self.widget = self.send.widget + self.app.ui.main_display.update_active_sub_display() + self.app.ui.main_display.frame.focus_position = "body" + + def open_speedtest(self): + if self.speedtest is None: + self.speedtest = SpeedTestView(self.app, parent=self) + self.active = "speedtest" + self.widget = self.speedtest.widget + self.app.ui.main_display.update_active_sub_display() + self.app.ui.main_display.frame.focus_position = "body" + self.speedtest.update_interface_status() + self.speedtest.start_listening() + +class ProbeShortcuts: + def __init__(self, app): + self.app = app + self.widget = urwid.AttrMap( + urwid.Text("[Up/Down] Navigate [Enter] Probe / Activate [Esc] Back"), + "shortcutbar" + ) + +class UtilityViewFiller(urwid.WidgetWrap): + def __init__(self, body, view): + self.view = view + super().__init__(body) + + def keypress(self, size, key): + if key == "esc": + self.view.go_back() + return None + result = super().keypress(size, key) + if key == "up" and result == "up": + self.view.go_back_to_header() + return None + return result + +class ProbeView: + def __init__(self, app, parent=None): + self.app = app + self.parent = parent + self.glyphset = self.app.config["textui"]["glyphs"] + self.g = self.app.ui.glyphs + + self.shortcuts_display = ProbeShortcuts(self.app) + + self._probing = False + self._concluded = False + self._remaining = 0 + self._sent = 0 + self._replies = 0 + self._cur_receipt = None + + self.name_edit = ReadlineEdit("", "rnstransport.probe") + self.hash_edit = ReadlineEdit("", "") + self.size_edit = ReadlineEdit("", str(DEFAULT_PROBE_SIZE)) + self.count_edit = ReadlineEdit("", "1") + + self.probe_button = urwid.AttrMap(urwid.Button("Probe", on_press=self.on_probe), "button_normal", focus_map="button_focus") + self.back_button = urwid.AttrMap(urwid.Button("Back", on_press=lambda b: self.go_back()), "button_normal", focus_map="button_focus") + button_row = urwid.Columns([ + (urwid.WEIGHT, 0.45, self.probe_button), + (urwid.WEIGHT, 0.1, urwid.Text("")), + (urwid.WEIGHT, 0.45, self.back_button), + ]) + + self.status_text = urwid.Text("") + + rows = [ + urwid.Text(("form_title", "Probe a destination"), align=urwid.CENTER), + urwid.Divider("─"), + self.status_text, + self._field_row("Destination name:", self.name_edit), + urwid.Padding(urwid.Text(("inactive_text", "Full dotted name for example rnstransport.probe")), left=22), + self._field_row("Destination hash:", self.hash_edit), + self._field_row("Payload size:", self.size_edit), + self._field_row("Probe count:", self.count_edit), + urwid.Divider(), + button_row, + urwid.Divider("─"), + urwid.Text(("key", "Output")), + ] + self.walker = urwid.SimpleFocusListWalker(rows) + self.listbox = urwid.ListBox(self.walker) + self.widget = UtilityViewFiller(self.listbox, self) + self.update_interface_status() + + def _field_row(self, label, edit): + return urwid.Columns([ + (20, urwid.Text(("key", label), align=urwid.RIGHT)), + urwid.AttrMap(edit, "list_normal", focus_map="list_focus"), + ], dividechars=1) + + def update_interface_status(self): + if _has_active_interfaces(self.app): + self.status_text.set_text("") + else: + self.status_text.set_text(("warning_text", "(!) No interfaces are active")) + + def go_back(self): + if self.parent is not None: + self.parent.show_landing() + + def go_back_to_header(self): + try: + self.app.ui.main_display.frame.focus_position = "header" + except Exception: + pass + + def _log(self, message, attr="body_text"): + self.walker.append(urwid.Text((attr, message))) + try: + self.listbox.set_focus(len(self.walker) - 1) + except Exception: + pass + + def _draw(self): + try: + self.app.ui.loop.draw_screen() + except Exception: + pass + + def _schedule(self, fn): + try: + self.app.ui.loop.set_alarm_in(0, lambda *a: (fn(), self._draw())) + except Exception: + try: + fn() + except Exception: + pass + + def on_probe(self, button=None): + if self._probing: + return + + self.update_interface_status() + + full_name = self.name_edit.edit_text.strip() + hexhash = self.hash_edit.edit_text.strip() + + if not full_name: + self._log("Enter a destination name (e.g. rnstransport.probe)", "error_text") + return + + try: + app_name, aspects = RNS.Destination.app_and_aspects_from_name(full_name) + except Exception as e: + self._log("Invalid destination name: %s" % str(e), "error_text") + return + + dest_len = (RNS.Reticulum.TRUNCATED_HASHLENGTH // 8) * 2 + if len(hexhash) != dest_len: + self._log("Destination hash must be %d hex characters" % dest_len, "error_text") + return + try: + destination_hash = bytes.fromhex(hexhash) + except Exception: + self._log("Invalid destination hash", "error_text") + return + + try: + size = int(self.size_edit.edit_text.strip() or str(DEFAULT_PROBE_SIZE)) + except ValueError: + size = DEFAULT_PROBE_SIZE + try: + count = int(self.count_edit.edit_text.strip() or "1") + except ValueError: + count = 1 + if count < 1: + count = 1 + + self._app_name = app_name + self._aspects = aspects + self._dest_hash = destination_hash + self._size = max(1, size) + self._remaining = count + self._sent = 0 + self._replies = 0 + self._probing = True + self.probe_button.original_widget.set_label("Probing…") + self._begin_probe() + + def _begin_probe(self): + dest_hash = self._dest_hash + if not RNS.Transport.has_path(dest_hash): + try: + RNS.Transport.request_path(dest_hash) + except Exception: + pass + self._log("Requesting path to %s …" % RNS.prettyhexrep(dest_hash)) + self._path_deadline = time.time() + self._timeout() + self.app.ui.loop.set_alarm_in(0.25, self._await_path) + else: + self._send_one() + + def _timeout(self): + try: + return DEFAULT_PROBE_TIMEOUT + self.app.rns.get_first_hop_timeout(self._dest_hash) + except Exception: + return DEFAULT_PROBE_TIMEOUT + + def _await_path(self, loop, user_data): + if RNS.Transport.has_path(self._dest_hash): + self._send_one() + elif time.time() > self._path_deadline: + self._log("Path request timed out", "warning_text") + self._finish() + else: + loop.set_alarm_in(0.25, self._await_path) + + def _send_one(self): + dest_hash = self._dest_hash + identity = RNS.Identity.recall(dest_hash) + if identity is None: + self._log("Could not recall identity for destination", "error_text") + self._finish() + return + + try: + request_destination = RNS.Destination(identity, RNS.Destination.OUT, RNS.Destination.SINGLE, self._app_name, *self._aspects) + probe = RNS.Packet(request_destination, os.urandom(self._size)) + probe.pack() + except OSError: + self._log("Probe payload of %d bytes exceeds the MTU" % self._size, "error_text") + self._finish() + return + except Exception as e: + self._log("Could not create probe: %s" % str(e), "error_text") + self._finish() + return + + self._concluded = False + receipt = probe.send() + self._cur_receipt = receipt + self._sent += 1 + self._log("Sent probe %d (%d bytes) to %s …" % (self._sent, self._size, RNS.prettyhexrep(dest_hash))) + + try: + receipt.set_delivery_callback(self._on_delivered) + except Exception: + pass + self.app.ui.loop.set_alarm_in(self._timeout(), self._on_timeout) + + def _on_delivered(self, receipt): + self._schedule(lambda: self._conclude(receipt, True)) + + def _on_timeout(self, loop, user_data): + self._conclude(self._cur_receipt, False) + + def _conclude(self, receipt, delivered): + if self._concluded: + return + self._concluded = True + + is_delivered = delivered + if not is_delivered and receipt is not None: + try: + is_delivered = receipt.status == RNS.PacketReceipt.DELIVERED + except Exception: + is_delivered = False + + if is_delivered and receipt is not None: + self._replies += 1 + try: + hops = RNS.Transport.hops_to(self._dest_hash) + except Exception: + hops = 0 + try: + rtt = receipt.get_rtt() + rtt_str = "%.0f ms" % (rtt * 1000) if rtt < 1 else "%.3f s" % rtt + except Exception: + rtt_str = "?" + hop_str = "%d hop%s" % (hops, "" if hops == 1 else "s") + self._log("Reply from %s %s over %s%s" % ( + RNS.prettyhexrep(receipt.destination.hash), rtt_str, hop_str, self._reception_stats(receipt)), + "connected_status") + else: + self._log("Probe %d timed out" % self._sent, "warning_text") + + self._next() + + def _reception_stats(self, receipt): + stats = "" + try: + proof_packet = getattr(receipt, "proof_packet", None) + if proof_packet is None: + return "" + if getattr(self.app.rns, "is_connected_to_shared_instance", False): + packet_hash = proof_packet.packet_hash + rssi = self.app.rns.get_packet_rssi(packet_hash) + snr = self.app.rns.get_packet_snr(packet_hash) + q = self.app.rns.get_packet_q(packet_hash) + else: + rssi = getattr(proof_packet, "rssi", None) + snr = getattr(proof_packet, "snr", None) + q = None + if rssi is not None: + stats += " [RSSI %s dBm]" % rssi + if snr is not None: + stats += " [SNR %s dB]" % snr + if q is not None: + stats += " [Q %s%%]" % q + except Exception: + pass + return stats + + def _next(self): + self._remaining -= 1 + if self._remaining > 0: + self._send_one() + else: + self._finish() + + def _finish(self): + if self._sent > 0: + loss = round((1 - (self._replies / self._sent)) * 100, 1) + self._log("Sent %d, received %d, packet loss %s%%" % (self._sent, self._replies, loss), "key") + self._probing = False + self.probe_button.original_widget.set_label("Probe") + self._draw() + +class PathShortcuts: + def __init__(self, app): + self.app = app + self.widget = urwid.AttrMap( + urwid.Text("[Up/Down] Navigate [Enter] Resolve / Activate [Esc] Back"), + "shortcutbar" + ) + +class PathView: + def __init__(self, app, parent=None): + self.app = app + self.parent = parent + self.glyphset = self.app.config["textui"]["glyphs"] + self.g = self.app.ui.glyphs + + self.shortcuts_display = PathShortcuts(self.app) + + self._resolving = False + + self.hash_edit = ReadlineEdit("", "") + + self.resolve_button = urwid.AttrMap(urwid.Button("Resolve", on_press=self.on_resolve), "button_normal", focus_map="button_focus") + self.drop_button = urwid.AttrMap(urwid.Button("Drop path", on_press=self.on_drop), "button_normal", focus_map="button_focus") + self.back_button = urwid.AttrMap(urwid.Button("Back", on_press=lambda b: self.go_back()), "button_normal", focus_map="button_focus") + button_row = urwid.Columns([ + (urwid.WEIGHT, 0.3, self.resolve_button), + (urwid.WEIGHT, 0.05, urwid.Text("")), + (urwid.WEIGHT, 0.3, self.drop_button), + (urwid.WEIGHT, 0.05, urwid.Text("")), + (urwid.WEIGHT, 0.3, self.back_button), + ]) + + self.status_text = urwid.Text("") + + rows = [ + urwid.Text(("form_title", "Resolve a path"), align=urwid.CENTER), + urwid.Divider("─"), + self.status_text, + urwid.Columns([ + (20, urwid.Text(("key", "Destination hash:"), align=urwid.RIGHT)), + urwid.AttrMap(self.hash_edit, "list_normal", focus_map="list_focus"), + ], dividechars=1), + urwid.Padding(urwid.Text(("inactive_text", "Requests the path from the network if it is not already known.")), left=22), + urwid.Divider(), + button_row, + urwid.Divider("─"), + urwid.Text(("key", "Output")), + ] + self.walker = urwid.SimpleFocusListWalker(rows) + self.listbox = urwid.ListBox(self.walker) + self.widget = UtilityViewFiller(self.listbox, self) + self.update_interface_status() + + def update_interface_status(self): + if _has_active_interfaces(self.app): + self.status_text.set_text("") + else: + self.status_text.set_text(("warning_text", "(!) No interfaces are active")) + + def go_back(self): + if self.parent is not None: + self.parent.show_landing() + + def go_back_to_header(self): + try: + self.app.ui.main_display.frame.focus_position = "header" + except Exception: + pass + + def _log(self, message, attr="body_text"): + self.walker.append(urwid.Text((attr, message))) + try: + self.listbox.set_focus(len(self.walker) - 1) + except Exception: + pass + + def _draw(self): + try: + self.app.ui.loop.draw_screen() + except Exception: + pass + + def _validate_hash(self): + hexhash = self.hash_edit.edit_text.strip() + dest_len = (RNS.Reticulum.TRUNCATED_HASHLENGTH // 8) * 2 + if len(hexhash) != dest_len: + self._log("Destination hash must be %d hex characters" % dest_len, "error_text") + return None + try: + return bytes.fromhex(hexhash) + except Exception: + self._log("Invalid destination hash", "error_text") + return None + + def _timeout(self): + try: + return DEFAULT_PROBE_TIMEOUT + self.app.rns.get_first_hop_timeout(self._dest_hash) + except Exception: + return DEFAULT_PROBE_TIMEOUT + + def on_resolve(self, button=None): + if self._resolving: + return + self.update_interface_status() + dest_hash = self._validate_hash() + if dest_hash is None: + return + self._dest_hash = dest_hash + if RNS.Transport.has_path(dest_hash): + self._show_path() + return + try: + RNS.Transport.request_path(dest_hash) + except Exception: + pass + self._resolving = True + self.resolve_button.original_widget.set_label("Resolving…") + self._log("Requesting path to %s …" % RNS.prettyhexrep(dest_hash)) + self._deadline = time.time() + self._timeout() + self.app.ui.loop.set_alarm_in(0.25, self._await_path) + + def _await_path(self, loop, user_data): + if RNS.Transport.has_path(self._dest_hash): + self._resolving = False + self.resolve_button.original_widget.set_label("Resolve") + self._show_path() + self._draw() + elif time.time() > self._deadline: + self._resolving = False + self.resolve_button.original_widget.set_label("Resolve") + self._log("Path not found", "warning_text") + self._draw() + else: + loop.set_alarm_in(0.25, self._await_path) + + def _show_path(self): + dest_hash = self._dest_hash + try: + hops = RNS.Transport.hops_to(dest_hash) + next_hop_bytes = self.app.rns.get_next_hop(dest_hash) + except Exception: + next_hop_bytes = None + hops = 0 + if not next_hop_bytes: + self._log("Path data unavailable", "warning_text") + return + try: + interface = self.app.rns.get_next_hop_if_name(dest_hash) + except Exception: + interface = "?" + hop_str = "%d hop%s" % (hops, "" if hops == 1 else "s") + self._log("Path found: %s is %s away via %s on %s" % ( + RNS.prettyhexrep(dest_hash), hop_str, RNS.prettyhexrep(next_hop_bytes), interface), + "connected_status") + + def on_drop(self, button=None): + dest_hash = self._validate_hash() + if dest_hash is None: + return + try: + dropped = self.app.rns.drop_path(dest_hash) + except Exception as e: + self._log("Could not drop path: %s" % str(e), "error_text") + return + if dropped: + self._log("Dropped path to %s" % RNS.prettyhexrep(dest_hash)) + else: + self._log("No path to %s to drop" % RNS.prettyhexrep(dest_hash), "warning_text") + +class IdentityShortcuts: + def __init__(self, app): + self.app = app + self.widget = urwid.AttrMap( + urwid.Text("[Up/Down] Navigate [Enter] Look up / Activate [Esc] Back"), + "shortcutbar" + ) + +class IdentityView: + def __init__(self, app, parent=None): + self.app = app + self.parent = parent + self.glyphset = self.app.config["textui"]["glyphs"] + self.g = self.app.ui.glyphs + self.shortcuts_display = IdentityShortcuts(self.app) + self._resolving = False + + self.hash_edit = ReadlineEdit("", "") + self.lookup_button = urwid.AttrMap(urwid.Button("Look up", on_press=self.on_lookup), "button_normal", focus_map="button_focus") + self.back_button = urwid.AttrMap(urwid.Button("Back", on_press=lambda b: self.go_back()), "button_normal", focus_map="button_focus") + button_row = urwid.Columns([ + (urwid.WEIGHT, 0.45, self.lookup_button), + (urwid.WEIGHT, 0.1, urwid.Text("")), + (urwid.WEIGHT, 0.45, self.back_button), + ]) + + rows = [ + urwid.Text(("form_title", "Identity"), align=urwid.CENTER), + urwid.Divider("─"), + urwid.Text(("key", "Your addresses")), + ] + rows += self._address_rows() + rows += [ + urwid.Divider(), + urwid.Text(("key", "Look up a destination")), + urwid.Columns([ + (20, urwid.Text(("key", "Destination hash:"), align=urwid.RIGHT)), + urwid.AttrMap(self.hash_edit, "list_normal", focus_map="list_focus"), + ], dividechars=1), + urwid.Padding(urwid.Text(("inactive_text", "Recalls the identity, requesting it from the network if needed.")), left=22), + urwid.Divider(), + button_row, + urwid.Divider("─"), + urwid.Text(("key", "Output")), + ] + self.walker = urwid.SimpleFocusListWalker(rows) + self.listbox = urwid.ListBox(self.walker) + self.widget = UtilityViewFiller(self.listbox, self) + + def _addr(self, obj): + try: + return RNS.prettyhexrep(obj.hash) if obj is not None else "" + except Exception: + return "" + + def _address_rows(self): + rows = [] + + def row(label, value): + return urwid.Columns([ + (18, urwid.Text(("key", label), align=urwid.RIGHT)), + urwid.Text(("value", value)), + ], dividechars=1) + + rows.append(row("Identity:", self._addr(getattr(self.app, "identity", None)))) + rows.append(row("LXMF address:", self._addr(getattr(self.app, "lxmf_destination", None)))) + node = getattr(self.app, "node", None) + if node is not None: + rows.append(row("Node:", self._addr(getattr(node, "destination", None)))) + return rows + + def go_back(self): + if self.parent is not None: + self.parent.show_landing() + + def go_back_to_header(self): + try: + self.app.ui.main_display.frame.focus_position = "header" + except Exception: + pass + + def _log(self, message, attr="body_text"): + self.walker.append(urwid.Text((attr, message))) + try: + self.listbox.set_focus(len(self.walker) - 1) + except Exception: + pass + + def _draw(self): + try: + self.app.ui.loop.draw_screen() + except Exception: + pass + + def on_lookup(self, button=None): + if self._resolving: + return + hexhash = self.hash_edit.edit_text.strip() + dest_len = (RNS.Reticulum.TRUNCATED_HASHLENGTH // 8) * 2 + if len(hexhash) != dest_len: + self._log("Destination hash must be %d hex characters" % dest_len, "error_text") + return + try: + dest_hash = bytes.fromhex(hexhash) + except Exception: + self._log("Invalid destination hash", "error_text") + return + + self._dest_hash = dest_hash + identity = RNS.Identity.recall(dest_hash) + if identity is not None: + self._show_identity(identity) + return + + try: + RNS.Transport.request_path(dest_hash) + except Exception: + pass + self._resolving = True + self.lookup_button.original_widget.set_label("Looking up…") + self._log("Identity unknown, requesting from the network …") + try: + self._deadline = time.time() + DEFAULT_PROBE_TIMEOUT + self.app.rns.get_first_hop_timeout(dest_hash) + except Exception: + self._deadline = time.time() + DEFAULT_PROBE_TIMEOUT + self.app.ui.loop.set_alarm_in(0.25, self._await_identity) + + def _await_identity(self, loop, user_data): + identity = RNS.Identity.recall(self._dest_hash) + if identity is not None: + self._resolving = False + self.lookup_button.original_widget.set_label("Look up") + self._show_identity(identity) + self._draw() + elif time.time() > self._deadline: + self._resolving = False + self.lookup_button.original_widget.set_label("Look up") + self._log("Identity not found", "warning_text") + self._draw() + else: + loop.set_alarm_in(0.25, self._await_identity) + + def _show_identity(self, identity): + self._log("kno wn %s resolves to identity %s" % ( + RNS.prettyhexrep(self._dest_hash), RNS.prettyhexrep(identity.hash)), "connected_status") + +RNCP_APP_NAME = "rncp" + +class SendFileShortcuts: + def __init__(self, app): + self.app = app + self.widget = urwid.AttrMap( + urwid.Text("[Up/Down] Navigate [Enter] Choose / Send [Esc] Back"), + "shortcutbar" + ) + +class SendFileView: + def __init__(self, app, parent=None): + self.app = app + self.parent = parent + self.glyphset = self.app.config["textui"]["glyphs"] + self.g = self.app.ui.glyphs + self.shortcuts_display = SendFileShortcuts(self.app) + + self._sending = False + self._concluded = False + self._link = None + self._resource = None + self.file_path = None + + self.hash_edit = ReadlineEdit("", "") + self.file_label = urwid.Text(("inactive_text", "No file selected")) + self.status_text = urwid.Text("") + self.progress_text = urwid.Text("") + + self.choose_button = urwid.AttrMap(urwid.Button("Choose file…", on_press=self.choose_file), "button_normal", focus_map="button_focus") + self.send_button = urwid.AttrMap(urwid.Button("Send", on_press=self.on_send), "button_normal", focus_map="button_focus") + self.back_button = urwid.AttrMap(urwid.Button("Back", on_press=lambda b: self.go_back()), "button_normal", focus_map="button_focus") + button_row = urwid.Columns([ + (urwid.WEIGHT, 0.45, self.send_button), + (urwid.WEIGHT, 0.1, urwid.Text("")), + (urwid.WEIGHT, 0.45, self.back_button), + ]) + + rows = [ + urwid.Text(("form_title", "Send a file"), align=urwid.CENTER), + urwid.Divider("─"), + self.status_text, + urwid.Columns([ + (20, urwid.Text(("key", "File:"), align=urwid.RIGHT)), + urwid.AttrMap(self.choose_button, "button_normal", focus_map="button_focus"), + ], dividechars=1), + urwid.Padding(self.file_label, left=22), + urwid.Columns([ + (20, urwid.Text(("key", "Destination hash:"), align=urwid.RIGHT)), + urwid.AttrMap(self.hash_edit, "list_normal", focus_map="list_focus"), + ], dividechars=1), + urwid.Padding(urwid.Text(("inactive_text", "The receiver must be running 'rncp --listen' and have your identity hash in the allow list.")), left=22), + urwid.Divider(), + button_row, + urwid.Divider("─"), + self.progress_text, + urwid.Text(("key", "Output")), + ] + self.walker = urwid.SimpleFocusListWalker(rows) + self.listbox = urwid.ListBox(self.walker) + self.widget = UtilityViewFiller(self.listbox, self) + self.update_interface_status() + + def update_interface_status(self): + if _has_active_interfaces(self.app): + self.status_text.set_text("") + else: + self.status_text.set_text(("warning_text", "(!) No interfaces are active")) + + def go_back(self): + if self.parent is not None: + self.parent.show_landing() + + def go_back_to_header(self): + try: + self.app.ui.main_display.frame.focus_position = "header" + except Exception: + pass + + def _log(self, message, attr="body_text"): + self.walker.append(urwid.Text((attr, message))) + try: + self.listbox.set_focus(len(self.walker) - 1) + except Exception: + pass + + def _draw(self): + try: + self.app.ui.loop.draw_screen() + except Exception: + pass + + def _schedule(self, fn): + try: + self.app.ui.loop.set_alarm_in(0, lambda *a: (fn(), self._draw())) + except Exception: + try: + fn() + except Exception: + pass + + def choose_file(self, button=None): + def on_done(selected): + if selected: + self.file_path = selected[0] + self.file_label.set_text(("value", os.path.basename(self.file_path))) + self._close_browser() + + def on_cancel(): + self._close_browser() + + browser = FileBrowser(on_done=on_done, on_cancel=on_cancel, title="Select File to Send", multi_select=False) + overlay = urwid.Overlay(browser, self.widget, align=urwid.CENTER, width=("relative", 90), valign=urwid.MIDDLE, height=("relative", 80), left=2, right=2) + self.parent.widget = overlay + self.app.ui.main_display.update_active_sub_display() + self.app.ui.main_display.frame.focus_position = "body" + + def _close_browser(self): + self.parent.widget = self.widget + self.app.ui.main_display.update_active_sub_display() + self.app.ui.main_display.frame.focus_position = "body" + self._draw() + + def _timeout(self): + try: + return DEFAULT_PROBE_TIMEOUT + self.app.rns.get_first_hop_timeout(self._dest_hash) + except Exception: + return DEFAULT_PROBE_TIMEOUT + + def on_send(self, button=None): + if self._sending: + return + self.update_interface_status() + + if not self.file_path or not os.path.isfile(self.file_path): + self._log("Choose a file to send first", "error_text") + return + + hexhash = self.hash_edit.edit_text.strip() + dest_len = (RNS.Reticulum.TRUNCATED_HASHLENGTH // 8) * 2 + if len(hexhash) != dest_len: + self._log("Destination hash must be %d hex characters" % dest_len, "error_text") + return + try: + dest_hash = bytes.fromhex(hexhash) + except Exception: + self._log("Invalid destination hash", "error_text") + return + + self._dest_hash = dest_hash + self._sending = True + self._concluded = False + self._resource = None + self.send_button.original_widget.set_label("Sending…") + self.progress_text.set_text("") + + if RNS.Transport.has_path(dest_hash): + self._establish_link() + else: + try: + RNS.Transport.request_path(dest_hash) + except Exception: + pass + self._log("Requesting path to %s …" % RNS.prettyhexrep(dest_hash)) + self._deadline = time.time() + self._timeout() + self.app.ui.loop.set_alarm_in(0.25, self._await_path) + + def _await_path(self, loop, user_data): + if RNS.Transport.has_path(self._dest_hash): + self._establish_link() + elif time.time() > self._deadline: + self._log("Path not found", "warning_text") + self._finish() + else: + loop.set_alarm_in(0.25, self._await_path) + + def _establish_link(self): + identity = RNS.Identity.recall(self._dest_hash) + if identity is None: + self._log("Could not recall identity for destination", "error_text") + self._finish() + return + try: + destination = RNS.Destination(identity, RNS.Destination.OUT, RNS.Destination.SINGLE, RNCP_APP_NAME, "receive") + self._log("Establishing link with %s …" % RNS.prettyhexrep(self._dest_hash)) + self._link = RNS.Link(destination, established_callback=self._on_link_up, closed_callback=self._on_link_closed) + except Exception as e: + self._log("Could not establish link: %s" % str(e), "error_text") + self._finish() + return + self._link_deadline = time.time() + self._timeout() + self.app.ui.loop.set_alarm_in(0.5, self._check_link) + + def _check_link(self, loop, user_data): + if not self._sending or self._resource is not None: + return + try: + active = self._link is not None and self._link.status == RNS.Link.ACTIVE + except Exception: + active = False + if active: + return + if time.time() > self._link_deadline: + self._log("Link establishment timed out", "warning_text") + try: + self._link.teardown() + except Exception: + pass + self._finish() + else: + loop.set_alarm_in(0.5, self._check_link) + + def _on_link_up(self, link): + self._schedule(lambda: self._start_resource(link)) + + def _start_resource(self, link): + if not self._sending or self._resource is not None: + return + try: + link.identify(self.app.identity) + metadata = {"name": os.path.basename(self.file_path).encode("utf-8")} + self._resource = RNS.Resource(open(self.file_path, "rb"), link, metadata=metadata, + callback=self._on_resource_done, progress_callback=self._on_progress, auto_compress=True) + self._log("Advertising %s …" % os.path.basename(self.file_path)) + except Exception as e: + self._log("Could not start transfer: %s" % str(e), "error_text") + self._finish() + + def _on_progress(self, resource): + self._schedule(lambda: self._show_progress(resource)) + + def _show_progress(self, resource): + try: + pct = int(resource.get_progress() * 100) + except Exception: + pct = 0 + self.progress_text.set_text(("body_text", "Transferring… %d%%" % pct)) + + def _on_resource_done(self, resource): + self._schedule(lambda: self._resource_done(resource)) + + def _resource_done(self, resource): + if self._concluded: + return + self._concluded = True + try: + complete = resource.status == RNS.Resource.COMPLETE + except Exception: + complete = False + if complete: + self.progress_text.set_text(("connected_status", "Transferring… 100%")) + self._log("Transfer complete", "connected_status") + else: + self._log("Transfer failed or the file was not accepted ,", "warning_text") + try: + if self._link is not None: + self._link.teardown() + except Exception: + pass + self._finish() + + def _on_link_closed(self, link): + self._schedule(self._link_closed) + + def _link_closed(self): + if self._sending and self._resource is None and not self._concluded: + self._log("Link closed before the transfer started", "warning_text") + self._finish() + + def _finish(self): + self._sending = False + self.send_button.original_widget.set_label("Send") + self._draw() + +SPEEDTEST_APP_NAME = "nomadnetwork" +SPEEDTEST_ASPECTS = ("utilities", "speedtest") +DEFAULT_SPEEDTEST_MB = 1 +MAX_SPEEDTEST_MB = 50 + +def _fmt_bitrate(bytes_per_sec): + bits = float(bytes_per_sec) * 8 + for unit in ("bps", "Kbps", "Mbps", "Gbps"): + if bits < 1000: + return "%.1f %s" % (bits, unit) + bits /= 1000 + return "%.1f Tbps" % bits + +class SpeedTestShortcuts: + def __init__(self, app): + self.app = app + self.widget = urwid.AttrMap( + urwid.Text("[Up/Down] Navigate [Enter] Start / Activate [Esc] Back"), + "shortcutbar" + ) + +class SpeedTestView: + def __init__(self, app, parent=None): + self.app = app + self.parent = parent + self.glyphset = self.app.config["textui"]["glyphs"] + self.g = self.app.ui.glyphs + self.shortcuts_display = SpeedTestShortcuts(self.app) + + self.server_destination = None + self._listening = False + self._incoming_link = None + self._recv_link = None + self._recv_started = None + self._recv_bytes = 0 + + self._sending = False + self._client_link = None + self._client_resource = None + self._send_started = None + self._send_bytes = 0 + + self.hash_edit = ReadlineEdit("", "") + self.size_edit = ReadlineEdit("", str(DEFAULT_SPEEDTEST_MB)) + self.address_text = urwid.Text(("value", "—")) + self.status_text = urwid.Text("") + self.progress_text = urwid.Text("") + + self.start_button = urwid.AttrMap(urwid.Button("Start test", on_press=self.on_start), "button_normal", focus_map="button_focus") + self.back_button = urwid.AttrMap(urwid.Button("Back", on_press=lambda b: self.go_back()), "button_normal", focus_map="button_focus") + button_row = urwid.Columns([ + (urwid.WEIGHT, 0.45, self.start_button), + (urwid.WEIGHT, 0.1, urwid.Text("")), + (urwid.WEIGHT, 0.45, self.back_button), + ]) + + rows = [ + urwid.Text(("form_title", "Speed test"), align=urwid.CENTER), + urwid.Divider("─"), + self.status_text, + urwid.Columns([ + (22, urwid.Text(("key", "Your speedtest hash:"), align=urwid.RIGHT)), + self.address_text, + ], dividechars=1), + urwid.Padding(urwid.Text(("inactive_text", "Both sides open this screen and enter each other's hash; then one presses Start.")), left=24), + urwid.Columns([ + (22, urwid.Text(("key", "Peer hash:"), align=urwid.RIGHT)), + urwid.AttrMap(self.hash_edit, "list_normal", focus_map="list_focus"), + ], dividechars=1), + urwid.Columns([ + (22, urwid.Text(("key", "Test size (MB):"), align=urwid.RIGHT)), + urwid.AttrMap(self.size_edit, "list_normal", focus_map="list_focus"), + ], dividechars=1), + urwid.Divider(), + button_row, + urwid.Divider("─"), + self.progress_text, + urwid.Text(("key", "Output")), + ] + self.walker = urwid.SimpleFocusListWalker(rows) + self.listbox = urwid.ListBox(self.walker) + self.widget = UtilityViewFiller(self.listbox, self) + + def update_interface_status(self): + if _has_active_interfaces(self.app): + self.status_text.set_text("") + else: + self.status_text.set_text(("warning_text", "(!) No interfaces are active")) + + def _update_address(self): + if self.server_destination is not None: + self.address_text.set_text(("value", RNS.prettyhexrep(self.server_destination.hash))) + else: + self.address_text.set_text(("value", "—")) + + def start_listening(self): + if self.server_destination is None: + try: + self.server_destination = RNS.Destination(self.app.identity, RNS.Destination.IN, RNS.Destination.SINGLE, SPEEDTEST_APP_NAME, *SPEEDTEST_ASPECTS) + self.server_destination.set_link_established_callback(self._on_incoming_link) + try: + self.server_destination.accepts_links(True) + except Exception: + pass + except Exception as e: + self._log("Could not start listener: %s" % str(e), "error_text") + self.server_destination = None + self._listening = True + self._update_address() + + def stop_listening(self): + self._listening = False + try: + if self._incoming_link is not None: + self._incoming_link.teardown() + except Exception: + pass + self._incoming_link = None + self._recv_link = None + try: + if self.server_destination is not None: + RNS.Transport.deregister_destination(self.server_destination) + except Exception: + pass + self.server_destination = None + self._update_address() + + def _is_active(self): + try: + md = self.app.ui.main_display + return self._listening and md.sub_displays.active_display is self.parent and getattr(self.parent, "active", None) == "speedtest" + except Exception: + return False + + def _peer_identity(self): + hexhash = self.hash_edit.edit_text.strip() + dest_len = (RNS.Reticulum.TRUNCATED_HASHLENGTH // 8) * 2 + if len(hexhash) != dest_len: + return None + try: + return RNS.Identity.recall(bytes.fromhex(hexhash)) + except Exception: + return None + + def go_back(self): + self.stop_listening() + if self.parent is not None: + self.parent.show_landing() + + def go_back_to_header(self): + try: + self.app.ui.main_display.frame.focus_position = "header" + except Exception: + pass + + def _log(self, message, attr="body_text"): + self.walker.append(urwid.Text((attr, message))) + try: + self.listbox.set_focus(len(self.walker) - 1) + except Exception: + pass + + def _draw(self): + try: + self.app.ui.loop.draw_screen() + except Exception: + pass + + def _schedule(self, fn): + try: + self.app.ui.loop.set_alarm_in(0, lambda *a: (fn(), self._draw())) + except Exception: + try: + fn() + except Exception: + pass + + def _on_incoming_link(self, link): + if not self._is_active(): + try: + link.teardown() + except Exception: + pass + return + self._incoming_link = link + link.set_remote_identified_callback(self._on_remote_identified) + self.app.ui.loop.set_alarm_in(8, lambda l, u: self._incoming_timeout(link)) + + def _incoming_timeout(self, link): + if self._recv_link is link: + return + try: + link.teardown() + except Exception: + pass + + def _on_remote_identified(self, link, identity): + self._schedule(lambda: self._verify_incoming(link, identity)) + + def _verify_incoming(self, link, identity): + peer_identity = self._peer_identity() + if peer_identity is None or peer_identity.hash != identity.hash: + self._log("Rejected speed test from unauthorized peer", "warning_text") + try: + link.teardown() + except Exception: + pass + return + self._recv_link = link + self._recv_started = None + self._recv_bytes = 0 + link.set_resource_strategy(RNS.Link.ACCEPT_ALL) + link.set_resource_started_callback(self._on_recv_started) + link.set_resource_concluded_callback(self._on_recv_concluded) + self._log("Incoming speed test from %s" % RNS.prettyhexrep(identity.hash), "connected_status") + + def _on_recv_started(self, resource): + self._schedule(self._recv_started_now) + + def _recv_started_now(self): + self._recv_started = time.time() + self.progress_text.set_text(("body_text", "Receiving…")) + + def _on_recv_concluded(self, resource): + self._schedule(lambda: self._recv_concluded(resource)) + + def _recv_concluded(self, resource): + try: + complete = resource.status == RNS.Resource.COMPLETE + except Exception: + complete = False + if not complete: + self._log("Incoming transfer failed", "warning_text") + return + size = self._resource_size(resource) + elapsed = max(0.001, time.time() - (self._recv_started or time.time())) + self._log("Download: %s (%s in %.2fs)" % (_fmt_bitrate(size / elapsed), _size_str(size), elapsed), "connected_status") + self.progress_text.set_text("") + + def _resource_size(self, resource): + for getter in ("get_transfer_size", "getTransferSize"): + try: + fn = getattr(resource, getter, None) + if fn: + v = fn() + if v: + return v + except Exception: + pass + try: + return getattr(resource, "size", 0) or len(resource.data) + except Exception: + return 0 + + def on_start(self, button=None): + if self._sending: + return + self.update_interface_status() + + peer_identity = self._peer_identity() + dest_len = (RNS.Reticulum.TRUNCATED_HASHLENGTH // 8) * 2 + hexhash = self.hash_edit.edit_text.strip() + if len(hexhash) != dest_len: + self._log("Peer hash must be %d hex characters" % dest_len, "error_text") + return + try: + self._dest_hash = bytes.fromhex(hexhash) + except Exception: + self._log("Invalid peer hash", "error_text") + return + + try: + mb = float(self.size_edit.edit_text.strip() or str(DEFAULT_SPEEDTEST_MB)) + except ValueError: + mb = DEFAULT_SPEEDTEST_MB + mb = max(0.01, min(MAX_SPEEDTEST_MB, mb)) + self._send_bytes = int(mb * 1024 * 1024) + + self._sending = True + self.start_button.original_widget.set_label("Testing…") + self.progress_text.set_text("") + + if RNS.Transport.has_path(self._dest_hash): + self._establish_link() + else: + try: + RNS.Transport.request_path(self._dest_hash) + except Exception: + pass + self._log("Requesting path to %s …" % RNS.prettyhexrep(self._dest_hash)) + self._deadline = time.time() + self._timeout() + self.app.ui.loop.set_alarm_in(0.25, self._await_path) + + def _timeout(self): + try: + return DEFAULT_PROBE_TIMEOUT + self.app.rns.get_first_hop_timeout(self._dest_hash) + except Exception: + return DEFAULT_PROBE_TIMEOUT + + def _await_path(self, loop, user_data): + if RNS.Transport.has_path(self._dest_hash): + self._establish_link() + elif time.time() > self._deadline: + self._log("Path not found", "warning_text") + self._finish() + else: + loop.set_alarm_in(0.25, self._await_path) + + def _establish_link(self): + identity = RNS.Identity.recall(self._dest_hash) + if identity is None: + self._log("Could not recall identity for peer", "error_text") + self._finish() + return + try: + destination = RNS.Destination(identity, RNS.Destination.OUT, RNS.Destination.SINGLE, SPEEDTEST_APP_NAME, *SPEEDTEST_ASPECTS) + self._log("Establishing link with %s …" % RNS.prettyhexrep(self._dest_hash)) + self._client_link = RNS.Link(destination, established_callback=self._on_client_link_up, closed_callback=self._on_client_link_closed) + except Exception as e: + self._log("Could not establish link: %s" % str(e), "error_text") + self._finish() + return + self._link_deadline = time.time() + self._timeout() + self.app.ui.loop.set_alarm_in(0.5, self._check_link) + + def _check_link(self, loop, user_data): + if not self._sending or self._client_resource is not None: + return + try: + active = self._client_link is not None and self._client_link.status == RNS.Link.ACTIVE + except Exception: + active = False + if active: + return + if time.time() > self._link_deadline: + self._log("Link establishment timed out", "warning_text") + try: + self._client_link.teardown() + except Exception: + pass + self._finish() + else: + loop.set_alarm_in(0.5, self._check_link) + + def _on_client_link_up(self, link): + self._schedule(lambda: self._client_ready(link)) + + def _client_ready(self, link): + if not self._sending: + return + try: + link.identify(self.app.identity) + except Exception: + pass + self.app.ui.loop.set_alarm_in(0.6, lambda l, u: self._send_resource(link)) + + def _send_resource(self, link): + if not self._sending or self._client_resource is not None: + return + try: + data = os.urandom(self._send_bytes) + self._send_started = time.time() + self._client_resource = RNS.Resource(data, link, callback=self._on_send_done, progress_callback=self._on_send_progress, auto_compress=False) + self._log("Sending %s test payload …" % _size_str(self._send_bytes)) + except Exception as e: + self._log("Could not start transfer: %s" % str(e), "error_text") + self._finish() + + def _on_send_progress(self, resource): + self._schedule(lambda: self._send_progress(resource)) + + def _send_progress(self, resource): + try: + pct = int(resource.get_progress() * 100) + except Exception: + pct = 0 + self.progress_text.set_text(("body_text", "Uploading… %d%%" % pct)) + + def _on_send_done(self, resource): + self._schedule(lambda: self._send_done(resource)) + + def _send_done(self, resource): + try: + complete = resource.status == RNS.Resource.COMPLETE + except Exception: + complete = False + if complete: + elapsed = max(0.001, time.time() - (self._send_started or time.time())) + self._log("Upload: %s (%s in %.2fs)" % (_fmt_bitrate(self._send_bytes / elapsed), _size_str(self._send_bytes), elapsed), "connected_status") + else: + self._log("Transfer failed or was not accepted", "warning_text") + try: + if self._client_link is not None: + self._client_link.teardown() + except Exception: + pass + self.progress_text.set_text("") + self._finish() + + def _on_client_link_closed(self, link): + self._schedule(self._client_closed) + + def _client_closed(self): + if self._sending and self._client_resource is None: + self._log("Link closed before the test started", "warning_text") + self._finish() + + def _finish(self): + self._sending = False + self.start_button.original_widget.set_label("Start test") + self._draw() + +def _size_str(num, suffix="B"): + units = ["", "K", "M", "G", "T"] + num = float(num) + for unit in units: + if abs(num) < 1024.0: + return "%.1f %s%s" % (num, unit, suffix) + num /= 1024.0 + return "%.1f P%s" % (num, suffix) diff --git a/nomadnet/vendor/additional_urwid_widgets/FormWidgets.py b/nomadnet/vendor/additional_urwid_widgets/FormWidgets.py index 38a72df..fd69aa8 100644 --- a/nomadnet/vendor/additional_urwid_widgets/FormWidgets.py +++ b/nomadnet/vendor/additional_urwid_widgets/FormWidgets.py @@ -1,7 +1,14 @@ import urwid +import nomadnet from nomadnet.ui.textui.ReadlineEdit import ReadlineMixin, ReadlineEdit +def dropdown_indicator(): + try: + return nomadnet.NomadNetworkApp.get_shared_instance().ui.glyphs["dropdown"] + except Exception: + return " ▾" + class DialogLineBox(urwid.LineBox): def __init__(self, body, parent=None, title="?"): super().__init__(body, title=title) @@ -38,6 +45,13 @@ class Dropdown(urwid.WidgetWrap): self.main_button = urwid.SelectableIcon(self.main_text, 0) self.main_button = urwid.AttrMap(self.main_button, "button_normal", "button_focus") + self.indicator_widget = urwid.Text(dropdown_indicator()) + self.main_row = urwid.Columns([ + ('pack', self.main_button), + ('pack', self.indicator_widget), + ('weight', 1, urwid.Text("")), + ]) + self.option_widgets = [] for opt in options: icon = urwid.SelectableIcon(opt, 0) @@ -48,7 +62,7 @@ class Dropdown(urwid.WidgetWrap): 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.pile = urwid.Pile([self.main_row]) self.dropdown_visible = False super().__init__(self.pile) @@ -241,14 +255,25 @@ class FormMultiList(urwid.Pile, FormField): urwid.Pile.__init__(self, pile_widgets) FormField.__init__(self, config_key, transform) - def create_entry_row(self): - edit = ReadlineEdit("", "") + def create_entry_row(self, text=""): + edit = ReadlineEdit("", text) entry_row = urwid.Columns([ ('weight', 1, edit), (3, urwid.Button("×", on_press=lambda button: self.remove_entry(button, entry_row))), ]) return entry_row + def set_value(self, value): + self.entries = [] + if not value: + self.entries.append(self.create_entry_row()) + else: + if isinstance(value, str): + value = [value] + for item in value: + self.entries.append(self.create_entry_row(str(item))) + self.contents = [(w, self.options()) for w in self.get_pile_widgets()] + def remove_entry(self, button, entry_row): if len(self.entries) > 1: self.entries.remove(entry_row)