Add visual Config editor. Optional utilities tab with path, rncp file send utility, transport table viewer, speedtest, and identity resolver. Misc fixes

This commit is contained in:
zenith committed 2026-06-15 05:09:42 -04:00
1 parent 44594ff448
commit 94f5e7188a
11 files changed
+3658 -228

No files matched your search

+15
View File
@@ -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
+18
View File
@@ -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:
File diff suppressed because it is too large. Load diff
+24 -183
View File
@@ -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()
+5 -1
View File
@@ -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():
+195
View File
@@ -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
+125 -4
View File
@@ -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),
+32 -2
View File
@@ -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)
+503
View File
@@ -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)
File diff suppressed because it is too large. Load diff
+28 -3
View File
@@ -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)