import RNS
import urwid
import nomadnet
import time
import threading
from datetime import datetime
from nomadnet.Directory import DirectoryEntry
from nomadnet.vendor.additional_urwid_widgets import IndicativeListBox, MODIFIER_KEY
from .Browser import Browser
class NetworkDisplayShortcuts():
def __init__(self, app):
self.app = app
g = app.ui.glyphs
self.widget = urwid.AttrMap(urwid.Text("[C-l] Nodes/Announces [C-x] Remove [C-w] Disconnect [C-d] Back [C-f] Forward [C-r] Reload [C-u] URL [C-g] Fullscreen [C-s / C-b] Save Node"), "shortcutbar")
class DialogLineBox(urwid.LineBox):
def keypress(self, size, key):
if key == "esc":
self.delegate.update_conversation_list()
else:
return super(DialogLineBox, self).keypress(size, key)
class ListEntry(urwid.Text):
_selectable = True
signals = ["click"]
def keypress(self, size, key):
"""
Send 'click' signal on 'activate' command.
"""
if self._command_map[key] != urwid.ACTIVATE:
return key
self._emit('click')
def mouse_event(self, size, event, button, x, y, focus):
"""
Send 'click' signal on button 1 press.
"""
if button != 1 or not urwid.util.is_mouse_press(event):
return False
self._emit('click')
return True
class AnnounceInfo(urwid.WidgetWrap):
def keypress(self, size, key):
if key == "esc":
options = self.parent.left_pile.options(height_type=urwid.WEIGHT, height_amount=1)
self.parent.left_pile.contents[0] = (self.parent.announce_stream_display, options)
else:
return super(AnnounceInfo, self).keypress(size, key)
def __init__(self, announce, parent, app):
self.app = nomadnet.NomadNetworkApp.get_shared_instance()
self.parent = self.app.ui.main_display.sub_displays.network_display
g = self.app.ui.glyphs
source_hash = announce[1]
time_format = app.time_format
dt = datetime.fromtimestamp(announce[0])
ts_string = dt.strftime(time_format)
trust_level = self.app.directory.trust_level(source_hash)
trust_str = ""
display_str = self.app.directory.simplest_display_str(source_hash)
addr_str = "<"+RNS.hexrep(source_hash, delimit=False)+">"
info_type = announce[3]
is_node = False
is_pn = False
if info_type == "node" or info_type == True:
type_string = "Nomad Network Node " + g["node"]
is_node = True
elif info_type == "pn":
type_string = "LXMF Propagation Node " + g["sent"]
is_pn = True
elif info_type == "peer" or info_type == False:
type_string = "Peer " + g["peer"]
try:
data_str = announce[2].decode("utf-8")
data_style = ""
if trust_level != DirectoryEntry.TRUSTED and len(data_str) > 32:
data_str = data_str[:32]+" [...]"
except Exception as e:
data_str = "Decode failed"
data_style = "list_untrusted"
if trust_level == DirectoryEntry.UNTRUSTED:
trust_str = "Untrusted"
symbol = g["cross"]
style = "list_untrusted"
elif trust_level == DirectoryEntry.UNKNOWN:
trust_str = "Unknown"
symbol = g["unknown"]
style = "list_unknown"
elif trust_level == DirectoryEntry.TRUSTED:
trust_str = "Trusted"
symbol = g["check"]
style = "list_trusted"
elif trust_level == DirectoryEntry.WARNING:
trust_str = "Warning"
symbol = g["warning"]
style = "list_warning"
else:
trust_str = "Warning"
symbol = g["warning"]
style = "list_untrusted"
def show_announce_stream(sender):
options = self.parent.left_pile.options(height_type=urwid.WEIGHT, height_amount=1)
self.parent.left_pile.contents[0] = (self.parent.announce_stream_display, options)
def connect(sender):
self.parent.browser.retrieve_url(RNS.hexrep(source_hash, delimit=False))
show_announce_stream(None)
def save_node(sender):
node_entry = DirectoryEntry(source_hash, display_name=data_str, trust_level=trust_level, hosts_node=True)
self.app.directory.remember(node_entry)
self.app.ui.main_display.sub_displays.network_display.directory_change_callback()
show_announce_stream(None)
if is_node:
node_ident = RNS.Identity.recall(source_hash)
if not node_ident:
raise KeyError("Could not recall identity for selected node")
op_hash = RNS.Destination.hash_from_name_and_identity("lxmf.delivery", node_ident)
op_str = self.app.directory.simplest_display_str(op_hash)
def msg_op(sender):
show_announce_stream(None)
if is_node:
try:
existing_conversations = nomadnet.Conversation.conversation_list(self.app)
source_hash_text = RNS.hexrep(op_hash, delimit=False)
display_name = op_str
if not source_hash_text in [c[0] for c in existing_conversations]:
entry = DirectoryEntry(source_hash, display_name, trust_level)
self.app.directory.remember(entry)
new_conversation = nomadnet.Conversation(source_hash_text, nomadnet.NomadNetworkApp.get_shared_instance(), initiator=True)
self.app.ui.main_display.sub_displays.conversations_display.update_conversation_list()
self.app.ui.main_display.sub_displays.conversations_display.display_conversation(None, source_hash_text)
self.app.ui.main_display.show_conversations(None)
except Exception as e:
RNS.log("Error while starting conversation from announce. The contained exception was: "+str(e), RNS.LOG_ERROR)
def converse(sender):
show_announce_stream(None)
try:
existing_conversations = nomadnet.Conversation.conversation_list(self.app)
source_hash_text = RNS.hexrep(source_hash, delimit=False)
display_name = data_str
if not source_hash_text in [c[0] for c in existing_conversations]:
entry = DirectoryEntry(source_hash, display_name, trust_level)
self.app.directory.remember(entry)
new_conversation = nomadnet.Conversation(source_hash_text, nomadnet.NomadNetworkApp.get_shared_instance(), initiator=True)
self.app.ui.main_display.sub_displays.conversations_display.update_conversation_list()
self.app.ui.main_display.sub_displays.conversations_display.display_conversation(None, source_hash_text)
self.app.ui.main_display.show_conversations(None)
except Exception as e:
RNS.log("Error while starting conversation from announce. The contained exception was: "+str(e), RNS.LOG_ERROR)
def use_pn(sender):
show_announce_stream(None)
try:
self.app.set_user_selected_propagation_node(source_hash)
except Exception as e:
RNS.log("Error while setting active propagation node from announce. The contained exception was: "+str(e), RNS.LOG_ERROR)
if is_node:
type_button = (urwid.WEIGHT, 0.45, urwid.Button("Connect", on_press=connect))
msg_button = (urwid.WEIGHT, 0.45, urwid.Button("Msg Op", on_press=msg_op))
save_button = (urwid.WEIGHT, 0.45, urwid.Button("Save", on_press=save_node))
elif is_pn:
type_button = (urwid.WEIGHT, 0.45, urwid.Button("Use as default", on_press=use_pn))
save_button = None
else:
type_button = (urwid.WEIGHT, 0.45, urwid.Button("Converse", on_press=converse))
save_button = None
if is_node:
button_columns = urwid.Columns([
(urwid.WEIGHT, 0.45, urwid.Button("Back", on_press=show_announce_stream)),
(urwid.WEIGHT, 0.1, urwid.Text("")),
type_button,
(urwid.WEIGHT, 0.1, urwid.Text("")),
msg_button,
(urwid.WEIGHT, 0.1, urwid.Text("")),
save_button,
])
else:
button_columns = urwid.Columns([
(urwid.WEIGHT, 0.45, urwid.Button("Back", on_press=show_announce_stream)),
(urwid.WEIGHT, 0.1, urwid.Text("")),
type_button,
])
pile_widgets = []
if is_pn:
pile_widgets = [
urwid.Text("Time : "+ts_string, align=urwid.LEFT),
urwid.Text("Addr : "+addr_str, align=urwid.LEFT),
urwid.Text("Type : "+type_string, align=urwid.LEFT),
urwid.Divider(g["divider1"]),
button_columns
]
else:
pile_widgets = [
urwid.Text("Time : "+ts_string, align=urwid.LEFT),
urwid.Text("Addr : "+addr_str, align=urwid.LEFT),
urwid.Text("Type : "+type_string, align=urwid.LEFT),
urwid.Text("Name : "+display_str, align=urwid.LEFT),
urwid.Text(["Trust : ", (style, trust_str)], align=urwid.LEFT),
urwid.Divider(g["divider1"]),
urwid.Text(["Announce Data: \n", (data_style, data_str)], align=urwid.LEFT),
urwid.Divider(g["divider1"]),
button_columns
]
if is_node:
operator_entry = urwid.Text("Oprtr : "+op_str, align=urwid.LEFT)
pile_widgets.insert(4, operator_entry)
pile = urwid.Pile(pile_widgets)
self.display_widget = urwid.Filler(pile, valign=urwid.TOP, height=urwid.PACK)
super().__init__(urwid.LineBox(self.display_widget, title="Announce Info"))
class AnnounceStreamEntry(urwid.WidgetWrap):
def __init__(self, app, announce, delegate):
full_time_format = "%Y-%m-%d %H:%M:%S"
date_time_format = "%Y-%m-%d"
time_time_format = "%H:%M:%S"
short_time_format = "%Y-%m-%d %H:%M"
date_only_format = "%Y-%m-%d"
timestamp = announce[0]
source_hash = announce[1]
announce_type = announce[3]
self.app = app
self.delegate = delegate
self.timestamp = timestamp
time_format = app.time_format
dt = datetime.fromtimestamp(self.timestamp)
dtn = datetime.fromtimestamp(time.time())
g = self.app.ui.glyphs
if dt.strftime(date_time_format) == dtn.strftime(date_time_format):
ts_string = dt.strftime(time_time_format)
else:
ts_string = dt.strftime(date_only_format)
trust_level = self.app.directory.trust_level(source_hash)
display_str = self.app.directory.simplest_display_str(source_hash)
if trust_level == DirectoryEntry.UNTRUSTED:
symbol = g["cross"]
style = "list_untrusted"
focus_style = "list_focus_untrusted"
elif trust_level == DirectoryEntry.UNKNOWN:
symbol = g["unknown"]
style = "list_unknown"
focus_style = "list_focus"
elif trust_level == DirectoryEntry.TRUSTED:
symbol = g["check"]
style = "list_trusted"
focus_style = "list_focus_trusted"
elif trust_level == DirectoryEntry.WARNING:
symbol = g["warning"]
style = "list_warning"
focus_style = "list_focus"
else:
symbol = g["warning"]
style = "list_untrusted"
focus_style = "list_focus_untrusted"
if announce_type == "node" or announce_type == True:
type_symbol = g["node"]
elif announce_type == "peer" or announce_type == False:
type_symbol = g["peer"]
elif announce_type == "pn":
type_symbol = g["sent"]
widget = ListEntry(ts_string+" "+type_symbol+" "+display_str)
urwid.connect_signal(widget, "click", self.display_announce, announce)
self.display_widget = urwid.AttrMap(widget, style, focus_style)
super().__init__(self.display_widget)
def display_announce(self, event, announce):
try:
parent = self.app.ui.main_display.sub_displays.network_display
info_widget = AnnounceInfo(announce, parent, self.app)
options = parent.left_pile.options(height_type=urwid.WEIGHT, height_amount=1)
parent.left_pile.contents[0] = (info_widget, options)
except KeyError as e:
def dismiss_dialog(sender):
self.delegate.parent.close_list_dialogs()
def confirmed(sender):
def close_req(sender):
self.delegate.parent.close_list_dialogs()
dialog_pile.contents[0] = (urwid.Text("\nKeys requested from network\n", align=urwid.CENTER), options)
RNS.Transport.request_path(announce[1])
confirmed_button = urwid.Button("Request keys", on_press=confirmed)
dialog_pile = urwid.Pile([
urwid.Text(
"The keys for the announced destination could not be recalled. "
"You can wait for an announce to arrive, or request the keys from the network.\n",
align=urwid.CENTER,
),
urwid.Columns([
(urwid.WEIGHT, 0.45, confirmed_button),
(urwid.WEIGHT, 0.1, urwid.Text("")),
(urwid.WEIGHT, 0.45, urwid.Button("Close", on_press=dismiss_dialog)),
])
])
dialog = ListDialogLineBox(
dialog_pile,
title="Keys Unknown"
)
confirmed_button.dialog_pile = dialog_pile
dialog.delegate = self.delegate.parent
bottom = self.delegate
overlay = urwid.Overlay(
dialog,
bottom,
align=urwid.CENTER,
width=urwid.RELATIVE_100,
valign=urwid.MIDDLE,
height=urwid.PACK,
left=2,
right=2,
)
options = self.delegate.parent.left_pile.options(urwid.WEIGHT, 1)
self.delegate.parent.left_pile.contents[0] = (overlay, options)
def timestamp(self):
return self.timestamp
class TabButton(urwid.Button):
button_left = urwid.Text("[")
button_right = urwid.Text("]")
class AnnounceStream(urwid.WidgetWrap):
def __init__(self, app, parent):
self.app = app
self.parent = parent
self.started = False
self.timeout = self.app.config["textui"]["animation_interval"]*2
self.ilb = None
self.no_content = True
self.current_tab = "nodes"
self.added_entries = []
self.widget_list = []
self.update_widget_list()
# Create tab buttons
self.tab_nodes = TabButton("Nodes", on_press=self.show_nodes_tab)
self.tab_peers = TabButton("Peers", on_press=self.show_peers_tab)
self.tab_pn = TabButton("Propagation Nodes", on_press=self.show_pn_tab)
# Create tab bar with proportional widths
self.tab_bar = urwid.Columns([
('weight', 1, self.tab_nodes),
('weight', 1, self.tab_peers),
('weight', 3, self.tab_pn),
], dividechars=1) # Add 1 character spacing between tabs
self.ilb = ExceptionHandlingListBox(
self.widget_list,
on_selection_change=self.list_selection,
initialization_is_selection_change=False,
#modifier_key=MODIFIER_KEY.CTRL,
#highlight_offFocus="list_off_focus"
)
# Combine tab bar and list box
self.pile = urwid.Pile([
('pack', self.tab_bar),
('weight', 1, self.ilb),
])
self.display_widget = self.pile
super().__init__(urwid.LineBox(self.display_widget, title="Announce Stream"))
def keypress(self, size, key):
if key == "up" and (self.no_content or self.ilb.first_item_is_selected()):
nomadnet.NomadNetworkApp.get_shared_instance().ui.main_display.frame.focus_position = "header"
elif key == "ctrl x":
self.delete_selected_entry()
return super(AnnounceStream, self).keypress(size, key)
def delete_selected_entry(self):
if self.ilb.get_selected_item() != None:
self.app.directory.remove_announce_with_timestamp(self.ilb.get_selected_item().original_widget.timestamp)
self.rebuild_widget_list()
def rebuild_widget_list(self):
self.no_content = True
self.added_entries = []
self.widget_list = []
self.update_widget_list()
def update_widget_list(self):
self.widget_list = []
new_entries = []
for e in self.app.directory.announce_stream:
announce_type = e[3]
# Filter based on current tab
if self.current_tab == "nodes" and (announce_type == "node" or announce_type == True):
new_entries.append(e)
elif self.current_tab == "peers" and (announce_type == "peer" or announce_type == False):
new_entries.append(e)
elif self.current_tab == "pn" and announce_type == "pn":
new_entries.append(e)
for e in new_entries:
nw = AnnounceStreamEntry(self.app, e, self)
nw.timestamp = e[0]
self.widget_list.append(nw)
if len(new_entries) > 0:
self.no_content = False
else:
self.no_content = True
self.widget_list = [urwid.Text(f"No {self.current_tab} announces", align='center')]
if self.ilb:
self.ilb.set_body(self.widget_list)
def show_nodes_tab(self, button):
self.current_tab = "nodes"
self.update_widget_list()
def show_peers_tab(self, button):
self.current_tab = "peers"
self.update_widget_list()
def show_pn_tab(self, button):
self.current_tab = "pn"
self.update_widget_list()
def list_selection(self, arg1, arg2):
pass
def update(self):
self.update_widget_list()
def update_callback(self, loop=None, user_data=None):
self.update()
if self.started:
self.app.ui.loop.set_alarm_in(self.timeout, self.update_callback)
def start(self):
was_started = self.started
self.started = True
if not was_started:
self.update_callback()
def stop(self):
self.started = False
class SelectText(urwid.Text):
_selectable = True
signals = ["click"]
def keypress(self, size, key):
"""
Send 'click' signal on 'activate' command.
"""
if self._command_map[key] != urwid.ACTIVATE:
return key
self._emit('click')
def mouse_event(self, size, event, button, x, y, focus):
"""
Send 'click' signal on button 1 press.
"""
if button != 1 or not urwid.util.is_mouse_press(event):
return False
self._emit('click')
return True
class ListDialogLineBox(urwid.LineBox):
def keypress(self, size, key):
if key == "esc":
self.delegate.close_list_dialogs()
else:
return super(ListDialogLineBox, self).keypress(size, key)
class KnownNodeInfo(urwid.WidgetWrap):
def keypress(self, size, key):
if key == "esc":
options = self.parent.left_pile.options(height_type=urwid.WEIGHT, height_amount=1)
self.parent.left_pile.contents[0] = (self.parent.known_nodes_display, options)
else:
return super(KnownNodeInfo, self).keypress(size, key)
def __init__(self, node_hash):
self.app = nomadnet.NomadNetworkApp.get_shared_instance()
self.parent = self.app.ui.main_display.sub_displays.network_display
self.pn_changed = False
g = self.app.ui.glyphs
source_hash = node_hash
node_ident = RNS.Identity.recall(node_hash)
time_format = self.app.time_format
trust_level = self.app.directory.trust_level(source_hash)
trust_str = ""
node_entry = self.app.directory.find(source_hash)
sort_str = self.app.directory.sort_rank(source_hash)
if sort_str == None:
sort_str = "None"
else:
sort_str = str(sort_str)
if node_entry == None:
display_str = self.app.directory.simplest_display_str(source_hash)
else:
display_str = node_entry.display_name
addr_str = "<"+RNS.hexrep(source_hash, delimit=False)+">"
if display_str == None:
display_str = addr_str
pn_hash = RNS.Destination.hash_from_name_and_identity("lxmf.propagation", node_ident)
if node_ident != None:
lxmf_addr_str = g["sent"]+" LXMF Propagation Node Address is "+RNS.prettyhexrep(pn_hash)
else:
lxmf_addr_str = "No associated Propagation Node known"
type_string = "Nomad Network Node " + g["node"]
if trust_level == DirectoryEntry.UNTRUSTED:
trust_str = "Untrusted"
symbol = g["cross"]
style = "list_untrusted"
elif trust_level == DirectoryEntry.UNKNOWN:
trust_str = "Unknown"
symbol = g["unknown"]
style = "list_unknown"
elif trust_level == DirectoryEntry.TRUSTED:
trust_str = "Trusted"
symbol = g["check"]
style = "list_trusted"
elif trust_level == DirectoryEntry.WARNING:
trust_str = "Warning"
symbol = g["warning"]
style = "list_warning"
else:
trust_str = "Warning"
symbol = g["warning"]
style = "list_untrusted"
if trust_level == DirectoryEntry.UNTRUSTED:
untrusted_selected = True
unknown_selected = False
trusted_selected = False
elif trust_level == DirectoryEntry.UNKNOWN:
untrusted_selected = False
unknown_selected = True
trusted_selected = False
elif trust_level == DirectoryEntry.TRUSTED:
untrusted_selected = False
unknown_selected = False
trusted_selected = True
trust_button_group = []
r_untrusted = urwid.RadioButton(trust_button_group, "Untrusted", state=untrusted_selected)
r_unknown = urwid.RadioButton(trust_button_group, "Unknown", state=unknown_selected)
r_trusted = urwid.RadioButton(trust_button_group, "Trusted", state=trusted_selected)
e_name = urwid.Edit(caption="Name : ",edit_text=display_str)
e_sort = urwid.Edit(caption="Sort Rank : ",edit_text=sort_str)
node_ident = RNS.Identity.recall(source_hash)
op_hash = None
op_str = None
if node_ident != None:
op_hash = RNS.Destination.hash_from_name_and_identity("lxmf.delivery", node_ident)
op_str = self.app.directory.simplest_display_str(op_hash)
else:
op_str = "Unknown"
def show_known_nodes(sender):
options = self.parent.left_pile.options(height_type=urwid.WEIGHT, height_amount=1)
self.parent.left_pile.contents[0] = (self.parent.known_nodes_display, options)
def connect(sender):
self.parent.browser.retrieve_url(RNS.hexrep(source_hash, delimit=False))
show_known_nodes(None)
def msg_op(sender):
show_known_nodes(None)
if node_ident != None:
try:
existing_conversations = nomadnet.Conversation.conversation_list(self.app)
source_hash_text = RNS.hexrep(op_hash, delimit=False)
display_name = op_str
if not source_hash_text in [c[0] for c in existing_conversations]:
entry = DirectoryEntry(source_hash, display_name, trust_level)
self.app.directory.remember(entry)
new_conversation = nomadnet.Conversation(source_hash_text, nomadnet.NomadNetworkApp.get_shared_instance(), initiator=True)
self.app.ui.main_display.sub_displays.conversations_display.update_conversation_list()
self.app.ui.main_display.sub_displays.conversations_display.display_conversation(None, source_hash_text)
self.app.ui.main_display.show_conversations(None)
except Exception as e:
RNS.log("Error while starting conversation from node info. The contained exception was: "+str(e), RNS.LOG_ERROR)
def pn_change(sender, userdata):
self.pn_changed = True
def ident_change(sender, userdata):
pass
propagation_node_checkbox = urwid.CheckBox("Use as default propagation node", state=(self.app.get_user_selected_propagation_node() == pn_hash), on_state_change=pn_change)
connect_identify_checkbox = urwid.CheckBox("Identify when connecting", state=self.app.directory.should_identify_on_connect(source_hash), on_state_change=ident_change)
def save_node(sender):
if self.pn_changed:
if propagation_node_checkbox.get_state():
self.app.set_user_selected_propagation_node(pn_hash)
else:
self.app.set_user_selected_propagation_node(None)
trust_level = DirectoryEntry.UNTRUSTED
if r_unknown.get_state() == True:
trust_level = DirectoryEntry.UNKNOWN
if r_trusted.get_state() == True:
trust_level = DirectoryEntry.TRUSTED
display_str = e_name.get_edit_text()
sort_rank = e_sort.get_edit_text()
try:
if int(sort_rank) >= 0:
sort_rank = int(sort_rank)
else:
sort_rank = None
except:
sort_rank = None
node_entry = DirectoryEntry(source_hash, display_name=display_str, trust_level=trust_level, hosts_node=True, identify_on_connect=connect_identify_checkbox.get_state(), sort_rank=sort_rank)
self.app.directory.remember(node_entry)
self.app.ui.main_display.sub_displays.network_display.directory_change_callback()
if trust_level == DirectoryEntry.TRUSTED:
self.app.autoselect_propagation_node()
show_known_nodes(None)
back_button = (urwid.WEIGHT, 0.2, urwid.Button("Back", on_press=show_known_nodes))
connect_button = (urwid.WEIGHT, 0.2, urwid.Button("Connect", on_press=connect))
save_button = (urwid.WEIGHT, 0.2, urwid.Button("Save", on_press=save_node))
msg_button = (urwid.WEIGHT, 0.2, urwid.Button("Msg Op", on_press=msg_op))
bdiv = (urwid.WEIGHT, 0.02, urwid.Text(""))
button_columns = urwid.Columns([back_button, bdiv, connect_button, bdiv, msg_button, bdiv, save_button])
pile_widgets = [
urwid.Text("Type : "+type_string, align=urwid.LEFT),
e_name,
urwid.Text("Node Addr : "+addr_str, align=urwid.LEFT),
e_sort,
urwid.Divider(g["divider1"]),
urwid.Text(lxmf_addr_str, align=urwid.CENTER),
urwid.Divider(g["divider1"]),
propagation_node_checkbox,
connect_identify_checkbox,
urwid.Divider(g["divider1"]),
r_untrusted,
r_unknown,
r_trusted,
urwid.Divider(g["divider1"]),
button_columns
]
operator_entry = urwid.Text("Operator : "+op_str, align=urwid.LEFT)
pile_widgets.insert(3, operator_entry)
hops = RNS.Transport.hops_to(source_hash)
if hops == 1:
str_s = ""
else:
str_s = "s"
if hops != RNS.Transport.PATHFINDER_M:
hops_str = str(hops)+" hop"+str_s
else:
hops_str = "Unknown"
operator_entry = urwid.Text("Distance : "+hops_str, align=urwid.LEFT)
pile_widgets.insert(4, operator_entry)
pile = urwid.Pile(pile_widgets)
pile.focus_position = len(pile.contents)-1
button_columns.focus_position = 0
self.display_widget = urwid.Filler(pile, valign=urwid.TOP, height=urwid.PACK)
super().__init__(urwid.LineBox(self.display_widget, title="Node Info"))
# Yes, this is weird. There is a bug in Urwid/ILB that causes
# an indexing exception when the list is very small vertically.
# This mitigates it.
class ExceptionHandlingListBox(IndicativeListBox):
def keypress(self, size, key):
try:
return super(ExceptionHandlingListBox, self).keypress(size, key)
except Exception as e:
if key == "up":
nomadnet.NomadNetworkApp.get_shared_instance().ui.main_display.frame.focus_position = "header"
elif key == "down":
nomadnet.NomadNetworkApp.get_shared_instance().ui.main_display.sub_displays.network_display.left_pile.focus_position = 1
else:
RNS.log("An error occurred while processing an interface event. The contained exception was: "+str(e), RNS.LOG_ERROR)
class KnownNodes(urwid.WidgetWrap):
def __init__(self, app):
self.app = app
self.node_list = app.directory.known_nodes()
g = self.app.ui.glyphs
self.widget_list = self.make_node_widgets()
self.ilb = ExceptionHandlingListBox(
self.widget_list,
on_selection_change=self.node_list_selection,
initialization_is_selection_change=False,
highlight_offFocus="list_off_focus"
)
if len(self.node_list) > 0:
self.display_widget = self.ilb
widget_style = None
self.no_content = False
else:
self.no_content = True
widget_style = "inactive_text"
self.pile = urwid.Pile([
urwid.Text(("warning_text", g["info"]+"\n"), align=urwid.CENTER),
SelectText(
(
"warning_text",
"Currently, no nodes are saved\n\nCtrl+L to view the announce stream\n\n",
),
align=urwid.CENTER,
),
])
self.display_widget = urwid.Filler(self.pile, valign=urwid.TOP, height=urwid.PACK)
super().__init__(urwid.AttrMap(urwid.LineBox(self.display_widget, title="Saved Nodes"), widget_style))
def keypress(self, size, key):
if key == "up" and (self.no_content or self.ilb.first_item_is_selected()):
nomadnet.NomadNetworkApp.get_shared_instance().ui.main_display.frame.focus_position = "header"
elif key == "ctrl x":
self.delete_selected_entry()
return super(KnownNodes, self).keypress(size, key)
def node_list_selection(self, arg1, arg2):
pass
def connect_node(self, event, node):
source_hash = node.source_hash
trust_level = node.trust_level
trust_level = self.app.directory.trust_level(source_hash)
display_str = self.app.directory.simplest_display_str(source_hash)
parent = self.app.ui.main_display.sub_displays.network_display
def dismiss_dialog(sender):
self.delegate.close_list_dialogs()
def confirmed(sender):
self.delegate.browser.retrieve_url(RNS.hexrep(source_hash, delimit=False))
self.delegate.close_list_dialogs()
def show_info(sender):
info_widget = KnownNodeInfo(source_hash)
options = parent.left_pile.options(height_type=urwid.WEIGHT, height_amount=1)
parent.left_pile.contents[0] = (info_widget, options)
dialog = ListDialogLineBox(
urwid.Pile([
urwid.Text("Connect to node\n"+self.app.directory.simplest_display_str(source_hash)+"\n", align=urwid.CENTER),
urwid.Columns([
(urwid.WEIGHT, 0.45, urwid.Button("Yes", on_press=confirmed)),
(urwid.WEIGHT, 0.1, urwid.Text("")),
(urwid.WEIGHT, 0.45, urwid.Button("No", on_press=dismiss_dialog)),
(urwid.WEIGHT, 0.1, urwid.Text("")),
(urwid.WEIGHT, 0.45, urwid.Button("Info", on_press=show_info))])
]), title="?"
)
dialog.delegate = self.delegate
bottom = self
overlay = urwid.Overlay(dialog, bottom, align=urwid.CENTER, width=urwid.RELATIVE_100, valign=urwid.MIDDLE, height=urwid.PACK, left=2, right=2)
options = self.delegate.left_pile.options(urwid.WEIGHT, 1)
self.delegate.left_pile.contents[0] = (overlay, options)
def delete_selected_entry(self):
si = self.ilb.get_selected_item()
if si != None:
source_hash = si.original_widget.source_hash
def dismiss_dialog(sender):
self.delegate.close_list_dialogs()
def confirmed(sender):
self.app.directory.forget(source_hash)
self.rebuild_widget_list()
self.delegate.close_list_dialogs()
dialog = ListDialogLineBox(
urwid.Pile([
urwid.Text("Delete Node\n"+self.app.directory.simplest_display_str(source_hash)+"\n", align=urwid.CENTER),
urwid.Columns([
(urwid.WEIGHT, 0.45, urwid.Button("Yes", on_press=confirmed)),
(urwid.WEIGHT, 0.1, urwid.Text("")),
(urwid.WEIGHT, 0.45, urwid.Button("No", on_press=dismiss_dialog)),
])
]), title="?"
)
dialog.delegate = self.delegate
bottom = self
overlay = urwid.Overlay(
dialog,
bottom,
align=urwid.CENTER,
width=urwid.RELATIVE_100,
valign=urwid.MIDDLE,
height=urwid.PACK,
left=2,
right=2,
)
options = self.delegate.left_pile.options(urwid.WEIGHT, 1)
self.delegate.left_pile.contents[0] = (overlay, options)
def rebuild_widget_list(self):
self.node_list = self.app.directory.known_nodes()
self.widget_list = self.make_node_widgets()
self.ilb.set_body(self.widget_list)
if len(self.widget_list) > 0:
self.no_content = False
else:
self.no_content = True
self.delegate.reinit_known_nodes()
def make_node_widgets(self):
widget_list = []
for node_entry in self.node_list:
# TODO: Implement this
ne = NodeEntry(self.app, node_entry, self)
ne.source_hash = node_entry.source_hash
widget_list.append(ne)
# TODO: Sort list
return widget_list
class NodeEntry(urwid.WidgetWrap):
def __init__(self, app, node, delegate):
source_hash = node.source_hash
trust_level = node.trust_level
self.app = app
g = self.app.ui.glyphs
trust_level = self.app.directory.trust_level(source_hash)
display_str = self.app.directory.simplest_display_str(source_hash)
if trust_level == DirectoryEntry.UNTRUSTED:
symbol = g["cross"]
style = "list_untrusted"
focus_style = "list_focus_untrusted"
elif trust_level == DirectoryEntry.UNKNOWN:
symbol = g["unknown"]
style = "list_unknown"
focus_style = "list_focus"
elif trust_level == DirectoryEntry.TRUSTED:
symbol = g["check"]
style = "list_trusted"
focus_style = "list_focus_trusted"
elif trust_level == DirectoryEntry.WARNING:
symbol = g["warning"]
style = "list_warning"
focus_style = "list_focus"
else:
symbol = g["warning"]
style = "list_untrusted"
focus_style = "list_focus_untrusted"
type_symbol = g["node"]
widget = ListEntry(type_symbol+" "+display_str)
urwid.connect_signal(widget, "click", delegate.connect_node, node)
self.display_widget = urwid.AttrMap(widget, style, focus_style)
self.display_widget.source_hash = source_hash
super().__init__(self.display_widget)
class AnnounceTime(urwid.WidgetWrap):
def __init__(self, app):
self.started = False
self.app = app
self.timeout = self.app.config["textui"]["animation_interval"]
self.display_widget = urwid.Text("")
self.update_time()
super().__init__(self.display_widget)
def update_time(self):
self.last_announce_string = "Never"
if self.app.peer_settings["last_announce"] != None:
self.last_announce_string = pretty_date(int(self.app.peer_settings["last_announce"]))
self.display_widget.set_text("Announced : "+self.last_announce_string)
def update_time_callback(self, loop=None, user_data=None):
self.update_time()
if self.started:
self.app.ui.loop.set_alarm_in(self.timeout, self.update_time_callback)
def start(self):
was_started = self.started
self.started = True
if not was_started:
self.update_time_callback()
def stop(self):
self.started = False
class NodeAnnounceTime(urwid.WidgetWrap):
def __init__(self, app):
self.started = False
self.app = app
self.timeout = self.app.config["textui"]["animation_interval"]
self.display_widget = urwid.Text("")
self.update_time()
super().__init__(self.display_widget)
def update_time(self):
self.last_announce_string = "Never"
if self.app.peer_settings["node_last_announce"] != None:
self.last_announce_string = pretty_date(int(self.app.peer_settings["node_last_announce"]))
self.display_widget.set_text("Last Announce : "+self.last_announce_string)
def update_time_callback(self, loop=None, user_data=None):
self.update_time()
if self.started:
self.app.ui.loop.set_alarm_in(self.timeout, self.update_time_callback)
def start(self):
was_started = self.started
self.started = True
if not was_started:
self.update_time_callback()
def stop(self):
self.started = False
class NodeActiveConnections(urwid.WidgetWrap):
def __init__(self, app):
self.started = False
self.app = app
self.timeout = self.app.config["textui"]["animation_interval"]
self.display_widget = urwid.Text("")
self.update_stat()
super().__init__(self.display_widget)
def update_stat(self):
self.stat_string = "None"
if self.app.node != None:
self.stat_string = str(len(self.app.node.destination.links))
self.display_widget.set_text("Connected Now : "+self.stat_string)
def update_stat_callback(self, loop=None, user_data=None):
self.update_stat()
if self.started:
self.app.ui.loop.set_alarm_in(self.timeout, self.update_stat_callback)
def start(self):
was_started = self.started
self.started = True
if not was_started:
self.update_stat_callback()
def stop(self):
self.started = False
class NodeStorageStats(urwid.WidgetWrap):
def __init__(self, app):
self.started = False
self.app = app
self.timeout = self.app.config["textui"]["animation_interval"]
self.display_widget = urwid.Text("")
self.update_stat()
super().__init__(self.display_widget)
def update_stat(self):
self.stat_string = "None"
if self.app.node != None and not self.app.disable_propagation:
limit = self.app.message_router.message_storage_limit
used = self.app.message_router.message_storage_size()
if limit != None and used != None:
pct = round((used/limit)*100, 1)
pct_str = str(pct)+"%, "
limit_str = " of "+RNS.prettysize(limit)
else:
limit_str = ""
pct_str = ""
self.stat_string = pct_str+RNS.prettysize(used)+limit_str
self.display_widget.set_text("LXMF Storage : "+self.stat_string)
def update_stat_callback(self, loop=None, user_data=None):
self.update_stat()
if self.started:
self.app.ui.loop.set_alarm_in(self.timeout, self.update_stat_callback)
def start(self):
was_started = self.started
self.started = True
if not was_started:
self.update_stat_callback()
def stop(self):
self.started = False
class NodeTotalConnections(urwid.WidgetWrap):
def __init__(self, app):
self.started = False
self.app = app
self.timeout = self.app.config["textui"]["animation_interval"]
self.display_widget = urwid.Text("")
self.update_stat()
super().__init__(self.display_widget)
def update_stat(self):
self.stat_string = "None"
if self.app.node != None:
self.stat_string = str(self.app.peer_settings["node_connects"])
self.display_widget.set_text("Total Connects : "+self.stat_string)
def update_stat_callback(self, loop=None, user_data=None):
self.update_stat()
if self.started:
self.app.ui.loop.set_alarm_in(self.timeout, self.update_stat_callback)
def start(self):
was_started = self.started
self.started = True
if not was_started:
self.update_stat_callback()
def stop(self):
self.started = False
class NodeTotalPages(urwid.WidgetWrap):
def __init__(self, app):
self.started = False
self.app = app
self.timeout = self.app.config["textui"]["animation_interval"]
self.display_widget = urwid.Text("")
self.update_stat()
super().__init__(self.display_widget)
def update_stat(self):
self.stat_string = "None"
if self.app.node != None:
self.stat_string = str(self.app.peer_settings["served_page_requests"])
self.display_widget.set_text("Served Pages : "+self.stat_string)
def update_stat_callback(self, loop=None, user_data=None):
self.update_stat()
if self.started:
self.app.ui.loop.set_alarm_in(self.timeout, self.update_stat_callback)
def start(self):
was_started = self.started
self.started = True
if not was_started:
self.update_stat_callback()
def stop(self):
self.started = False
class NodeTotalFiles(urwid.WidgetWrap):
def __init__(self, app):
self.started = False
self.app = app
self.timeout = self.app.config["textui"]["animation_interval"]
self.display_widget = urwid.Text("")
self.update_stat()
super().__init__(self.display_widget)
def update_stat(self):
self.stat_string = "None"
if self.app.node != None:
self.stat_string = str(self.app.peer_settings["served_file_requests"])
self.display_widget.set_text("Served Files : "+self.stat_string)
def update_stat_callback(self, loop=None, user_data=None):
self.update_stat()
if self.started:
self.app.ui.loop.set_alarm_in(self.timeout, self.update_stat_callback)
def start(self):
was_started = self.started
self.started = True
if not was_started:
self.update_stat_callback()
def stop(self):
self.started = False
class LocalPeer(urwid.WidgetWrap):
announce_timer = None
def __init__(self, app, parent):
self.app = app
self.parent = parent
g = self.app.ui.glyphs
self.dialog_open = False
display_name = self.app.lxmf_destination.display_name
if display_name == None:
display_name = ""
t_id = urwid.Text("LXMF Addr : "+RNS.prettyhexrep(self.app.lxmf_destination.hash))
i_id = urwid.Text("Identity : "+RNS.prettyhexrep(self.app.identity.hash))
e_name = urwid.Edit(caption="Name : ", edit_text=display_name)
def save_query(sender):
def dismiss_dialog(sender):
self.dialog_open = False
self.parent.left_pile.contents[1] = (LocalPeer(self.app, self.parent), options)
self.app.set_display_name(e_name.get_edit_text())
dialog = DialogLineBox(
urwid.Pile([
urwid.Text("\n\n\nSaved\n\n", align=urwid.CENTER),
urwid.Button("OK", on_press=dismiss_dialog)
]), title=g["info"]
)
dialog.delegate = self
bottom = self
#overlay = urwid.Overlay(dialog, bottom, align=urwid.CENTER, width=urwid.RELATIVE_100, valign=urwid.MIDDLE, height=urwid.PACK, left=4, right=4)
overlay = dialog
options = self.parent.left_pile.options(height_type=urwid.PACK, height_amount=None)
self.dialog_open = True
self.parent.left_pile.contents[1] = (overlay, options)
def announce_query(sender):
def dismiss_dialog(sender):
self.dialog_open = False
options = self.parent.left_pile.options(height_type=urwid.PACK, height_amount=None)
self.parent.left_pile.contents[1] = (LocalPeer(self.app, self.parent), options)
self.app.announce_now()
dialog = DialogLineBox(
urwid.Pile([
urwid.Text("\n\n\nAnnounce Sent\n\n\n", align=urwid.CENTER),
urwid.Button("OK", on_press=dismiss_dialog)
]), title=g["info"]
)
dialog.delegate = self
bottom = self
#overlay = urwid.Overlay(dialog, bottom, align=urwid.CENTER, width=urwid.RELATIVE_100, valign=urwid.MIDDLE, height=urwid.PACK, left=4, right=4)
overlay = dialog
self.dialog_open = True
options = self.parent.left_pile.options(height_type=urwid.PACK, height_amount=None)
self.parent.left_pile.contents[1] = (overlay, options)
def node_info_query(sender):
options = self.parent.left_pile.options(height_type=urwid.PACK, height_amount=None)
self.parent.left_pile.contents[1] = (self.parent.node_info_display, options)
if LocalPeer.announce_timer == None:
self.t_last_announce = AnnounceTime(self.app)
LocalPeer.announce_timer = self.t_last_announce
else:
self.t_last_announce = LocalPeer.announce_timer
self.t_last_announce.update_time()
announce_button = urwid.Button("Announce Now", on_press=announce_query)
self.display_widget = urwid.Pile(
[
t_id,
i_id,
e_name,
urwid.Divider(g["divider1"]),
self.t_last_announce,
announce_button,
urwid.Divider(g["divider1"]),
urwid.Columns([
(urwid.WEIGHT, 0.45, urwid.Button("Save", on_press=save_query)),
(urwid.WEIGHT, 0.1, urwid.Text("")),
(urwid.WEIGHT, 0.45, urwid.Button("Node Info", on_press=node_info_query)),
])
]
)
super().__init__(urwid.LineBox(self.display_widget, title="Local Peer Info"))
def start(self):
self.t_last_announce.start()
class NodeInfo(urwid.WidgetWrap):
announce_timer = None
links_timer = None
conns_timer = None
pages_timer = None
files_timer = None
storage_timer = None
def __init__(self, app, parent):
self.app = app
self.parent = parent
g = self.app.ui.glyphs
self.dialog_open = False
widget_style = ""
def show_peer_info(sender):
options = self.parent.left_pile.options(height_type=urwid.PACK, height_amount=None)
self.parent.left_pile.contents[1] = (LocalPeer(self.app, self.parent), options)
if self.app.enable_node:
if self.app.node != None:
display_name = self.app.node.name
else:
display_name = None
if display_name == None:
display_name = ""
t_id = urwid.Text("Addr : "+RNS.hexrep(self.app.node.destination.hash, delimit=False))
e_name = urwid.Text("Name : "+display_name)
def stats_query(sender):
self.app.peer_settings["node_connects"] = 0
self.app.peer_settings["served_page_requests"] = 0
self.app.peer_settings["served_file_requests"] = 0
self.app.save_peer_settings()
def announce_query(sender):
def dismiss_dialog(sender):
self.dialog_open = False
options = self.parent.left_pile.options(height_type=urwid.PACK, height_amount=None)
self.parent.left_pile.contents[1] = (NodeInfo(self.app, self.parent), options)
self.app.node.announce()
dialog = DialogLineBox(
urwid.Pile([
urwid.Text("\n\n\nAnnounce Sent\n\n", align=urwid.CENTER),
urwid.Button("OK", on_press=dismiss_dialog)
]), title=g["info"]
)
dialog.delegate = self
bottom = self
#overlay = urwid.Overlay(dialog, bottom, align=urwid.CENTER, width=urwid.RELATIVE_100, valign=urwid.MIDDLE, height=urwid.PACK, left=4, right=4)
overlay = dialog
self.dialog_open = True
options = self.parent.left_pile.options(height_type=urwid.PACK, height_amount=None)
self.parent.left_pile.contents[1] = (overlay, options)
def connect_query(sender):
self.parent.browser.retrieve_url(RNS.hexrep(self.app.node.destination.hash, delimit=False))
if NodeInfo.announce_timer == None:
self.t_last_announce = NodeAnnounceTime(self.app)
NodeInfo.announce_timer = self.t_last_announce
else:
self.t_last_announce = NodeInfo.announce_timer
self.t_last_announce.update_time()
if NodeInfo.links_timer == None:
self.t_active_links = NodeActiveConnections(self.app)
NodeInfo.links_timer = self.t_active_links
else:
self.t_active_links = NodeInfo.links_timer
self.t_active_links.update_stat()
if NodeInfo.storage_timer == None:
self.t_storage_stats = NodeStorageStats(self.app)
NodeInfo.storage_timer = self.t_storage_stats
else:
self.t_storage_stats = NodeInfo.storage_timer
self.t_storage_stats.update_stat()
if NodeInfo.conns_timer == None:
self.t_total_connections = NodeTotalConnections(self.app)
NodeInfo.conns_timer = self.t_total_connections
else:
self.t_total_connections = NodeInfo.conns_timer
self.t_total_connections.update_stat()
if NodeInfo.pages_timer == None:
self.t_total_pages = NodeTotalPages(self.app)
NodeInfo.pages_timer = self.t_total_pages
else:
self.t_total_pages = NodeInfo.pages_timer
self.t_total_pages.update_stat()
if NodeInfo.files_timer == None:
self.t_total_files = NodeTotalFiles(self.app)
NodeInfo.files_timer = self.t_total_files
else:
self.t_total_files = NodeInfo.files_timer
self.t_total_files.update_stat()
lxmf_addr_str = g["sent"]+" LXMF Propagation Node Address is "+RNS.prettyhexrep(RNS.Destination.hash_from_name_and_identity("lxmf.propagation", self.app.node.destination.identity))
e_lxmf = urwid.Text(lxmf_addr_str, align=urwid.CENTER)
announce_button = urwid.Button("Announce", on_press=announce_query)
connect_button = urwid.Button("Browse", on_press=connect_query)
reset_button = urwid.Button("Rst Stats", on_press=stats_query)
if not self.app.disable_propagation:
pile = urwid.Pile([
t_id,
e_name,
urwid.Divider(g["divider1"]),
e_lxmf,
urwid.Divider(g["divider1"]),
self.t_last_announce,
self.t_storage_stats,
self.t_active_links,
self.t_total_connections,
self.t_total_pages,
self.t_total_files,
urwid.Divider(g["divider1"]),
urwid.Columns([
(urwid.WEIGHT, 5, urwid.Button("Back", on_press=show_peer_info)),
(urwid.WEIGHT, 0.5, urwid.Text("")),
(urwid.WEIGHT, 6, connect_button),
(urwid.WEIGHT, 0.5, urwid.Text("")),
(urwid.WEIGHT, 8, reset_button),
(urwid.WEIGHT, 0.5, urwid.Text("")),
(urwid.WEIGHT, 7, announce_button),
])
])
else:
pile = urwid.Pile([
t_id,
e_name,
urwid.Divider(g["divider1"]),
self.t_last_announce,
self.t_storage_stats,
self.t_active_links,
self.t_total_connections,
self.t_total_pages,
self.t_total_files,
urwid.Divider(g["divider1"]),
urwid.Columns([
(urwid.WEIGHT, 5, urwid.Button("Back", on_press=show_peer_info)),
(urwid.WEIGHT, 0.5, urwid.Text("")),
(urwid.WEIGHT, 6, connect_button),
(urwid.WEIGHT, 0.5, urwid.Text("")),
(urwid.WEIGHT, 8, reset_button),
(urwid.WEIGHT, 0.5, urwid.Text("")),
(urwid.WEIGHT, 7, announce_button),
])
])
else:
pile = urwid.Pile([
urwid.Text("\n"+g["info"], align=urwid.CENTER),
urwid.Text("\nThis instance is not hosting a node\n\n", align=urwid.CENTER),
urwid.Padding(urwid.Button("Back", on_press=show_peer_info), urwid.CENTER, urwid.PACK)
])
self.display_widget = pile
super().__init__(urwid.AttrMap(urwid.LineBox(self.display_widget, title="Local Node Info"), widget_style))
def start(self):
if self.app.node != None:
self.t_last_announce.start()
self.t_active_links.start()
self.t_total_connections.start()
self.t_total_pages.start()
self.t_total_files.start()
class UpdatingText(urwid.WidgetWrap):
def __init__(self, app, title, value_method, append_text=""):
self.started = False
self.app = app
self.timeout = self.app.config["textui"]["animation_interval"]*5
self.display_widget = urwid.Text("")
self.value = None
self.value_method = value_method
self.title = title
self.append_text = append_text
self.update()
super().__init__(self.display_widget)
def update(self):
self.value = self.value_method()
self.display_widget.set_text(self.title+str(self.value)+str(self.append_text))
def update_callback(self, loop=None, user_data=None):
self.update()
if self.started:
self.app.ui.loop.set_alarm_in(self.timeout, self.update_callback)
def start(self):
was_started = self.started
self.started = True
if not was_started:
self.update_callback()
def stop(self):
self.started = False
class NetworkStats(urwid.WidgetWrap):
def __init__(self, app, parent):
self.app = app
self.parent = parent
def get_num_peers():
return self.app.directory.number_of_known_peers(lookback_seconds=30*60)
def get_num_nodes():
return self.app.directory.number_of_known_nodes()
self.w_heard_peers = UpdatingText(self.app, "Heard Peers: ", get_num_peers, append_text=" (30m)")
self.w_known_nodes = UpdatingText(self.app, "Known Nodes: ", get_num_nodes)
pile = urwid.Pile([
self.w_heard_peers,
self.w_known_nodes,
])
self.display_widget = urwid.LineBox(pile, title="Network Stats")
super().__init__(self.display_widget)
def start(self):
self.w_heard_peers.start()
self.w_known_nodes.start()
class NetworkLeftPile(urwid.Pile):
def keypress(self, size, key):
if key == "ctrl l":
self.parent.toggle_list()
elif key == "ctrl g":
self.parent.toggle_fullscreen()
elif key == "ctrl e":
self.parent.selected_node_info()
elif key == "ctrl p":
self.parent.reinit_lxmf_peers()
self.parent.show_peers()
elif key == "ctrl w":
self.parent.browser.disconnect()
elif key == "ctrl u":
self.parent.browser.url_dialog()
elif key == "ctrl s":
self.parent.browser.save_node_dialog()
else:
return super(NetworkLeftPile, self).keypress(size, key)
class NetworkDisplay():
list_width = 0.33
given_list_width = 52
def __init__(self, app):
self.app = app
g = self.app.ui.glyphs
self.browser = Browser(self.app, "nomadnetwork", "node", auth_identity = self.app.identity, delegate = self)
if self.app.node != None:
self.browser.loopback = self.app.node.destination.hash
self.known_nodes_display = KnownNodes(self.app)
self.lxmf_peers_display = LXMFPeers(self.app)
self.network_stats_display = NetworkStats(self.app, self)
self.announce_stream_display = AnnounceStream(self.app, self)
self.local_peer_display = LocalPeer(self.app, self)
self.node_info_display = NodeInfo(self.app, self)
self.known_nodes_display.delegate = self
self.list_display = 1
self.left_pile = NetworkLeftPile([
(urwid.WEIGHT, 1, self.known_nodes_display),
# (urwid.PACK, self.network_stats_display),
(urwid.PACK, self.local_peer_display),
])
self.left_pile.parent = self
self.left_area = self.left_pile
self.right_area = self.browser.display_widget
self.right_area_width = 1-NetworkDisplay.list_width
self.columns = urwid.Columns(
[
# (urwid.WEIGHT, NetworkDisplay.list_width, self.left_area),
# (urwid.WEIGHT, self.right_area_width, self.right_area)
(NetworkDisplay.given_list_width, self.left_area),
(urwid.WEIGHT, 1, self.right_area)
],
dividechars=0, focus_column=0
)
self.shortcuts_display = NetworkDisplayShortcuts(self.app)
self.widget = self.columns
def toggle_list(self):
if self.list_display != 0:
options = self.left_pile.options(height_type=urwid.WEIGHT, height_amount=1)
self.left_pile.contents[0] = (self.announce_stream_display, options)
self.list_display = 0
else:
options = self.left_pile.options(height_type=urwid.WEIGHT, height_amount=1)
self.left_pile.contents[0] = (self.known_nodes_display, options)
self.list_display = 1
def toggle_fullscreen(self):
if NetworkDisplay.given_list_width != 0:
self.saved_list_width = NetworkDisplay.given_list_width
NetworkDisplay.given_list_width = 0
else:
NetworkDisplay.given_list_width = self.saved_list_width
options = self.widget.options("given", NetworkDisplay.given_list_width)
self.widget.contents[0] = (self.left_area, options)
def show_peers(self):
options = self.left_pile.options(height_type=urwid.WEIGHT, height_amount=1)
self.left_pile.contents[0] = (self.lxmf_peers_display, options)
if self.list_display != 0:
self.list_display = 0
else:
self.list_display = 1
def selected_node_info(self):
if self.list_display == 1:
parent = self.app.ui.main_display.sub_displays.network_display
selected_node_entry = parent.known_nodes_display.ilb.get_selected_item()
if selected_node_entry is not None:
selected_node_hash = selected_node_entry.base_widget.display_widget.source_hash
if selected_node_hash is not None:
info_widget = KnownNodeInfo(selected_node_hash)
options = parent.left_pile.options(height_type=urwid.WEIGHT, height_amount=1)
parent.left_pile.contents[0] = (info_widget, options)
def focus_lists(self):
self.columns.focus_position = 0
def reinit_known_nodes(self):
self.known_nodes_display = KnownNodes(self.app)
self.known_nodes_display.delegate = self
self.close_list_dialogs()
self.announce_stream_display.rebuild_widget_list()
def reinit_lxmf_peers(self):
if self.lxmf_peers_display:
si = self.lxmf_peers_display.ilb.get_selected_position()
else:
si = None
self.lxmf_peers_display = LXMFPeers(self.app)
self.lxmf_peers_display.delegate = self
self.close_list_dialogs()
if si != None:
self.lxmf_peers_display.ilb.select_item(si)
def close_list_dialogs(self):
if self.list_display == 0:
options = self.left_pile.options(height_type=urwid.WEIGHT, height_amount=1)
self.left_pile.contents[0] = (self.announce_stream_display, options)
else:
options = self.left_pile.options(height_type=urwid.WEIGHT, height_amount=1)
self.left_pile.contents[0] = (self.known_nodes_display, options)
def start(self):
self.local_peer_display.start()
self.node_info_display.start()
self.network_stats_display.start()
def shortcuts(self):
return self.shortcuts_display
def directory_change_callback(self):
self.announce_stream_display.rebuild_widget_list()
if self.known_nodes_display.no_content:
self.reinit_known_nodes()
else:
self.known_nodes_display.rebuild_widget_list()
class LXMFPeers(urwid.WidgetWrap):
def __init__(self, app):
self.app = app
self.peer_list = app.message_router.peers
# self.peer_list = {}
g = self.app.ui.glyphs
self.widget_list = self.make_peer_widgets()
self.ilb = IndicativeListBox(
self.widget_list,
on_selection_change=self.node_list_selection,
initialization_is_selection_change=False,
highlight_offFocus="list_off_focus"
)
if len(self.peer_list) > 0:
self.display_widget = self.ilb
widget_style = None
self.no_content = False
else:
self.no_content = True
widget_style = "inactive_text"
self.pile = urwid.Pile([
urwid.Text(("warning_text", g["info"]+"\n"), align=urwid.CENTER),
SelectText(("warning_text", "Currently, no LXMF nodes are peered\n\n"), align=urwid.CENTER),
])
self.display_widget = urwid.Filler(self.pile, valign=urwid.TOP, height=urwid.PACK)
if hasattr(self, "peer_list") and self.peer_list:
pl = len(self.peer_list)
else:
pl = 0
super().__init__(urwid.AttrMap(urwid.LineBox(self.display_widget, title=f"LXMF Propagation Peers ({pl})"), widget_style))
def keypress(self, size, key):
if key == "up" and (self.no_content or self.ilb.first_item_is_selected()):
nomadnet.NomadNetworkApp.get_shared_instance().ui.main_display.frame.focus_position = "header"
elif key == "ctrl x":
self.delete_selected_entry()
elif key == "ctrl r":
self.sync_selected_entry()
return super(LXMFPeers, self).keypress(size, key)
def node_list_selection(self, arg1, arg2):
pass
def delete_selected_entry(self):
si = self.ilb.get_selected_item()
if si != None:
destination_hash = si.original_widget.destination_hash
self.app.message_router.unpeer(destination_hash)
self.delegate.reinit_lxmf_peers()
self.delegate.show_peers()
def sync_selected_entry(self):
sync_grace = 10
si = self.ilb.get_selected_item()
if si != None:
destination_hash = si.original_widget.destination_hash
if destination_hash in self.app.message_router.peers:
peer = self.app.message_router.peers[destination_hash]
if time.time() > peer.last_sync_attempt+sync_grace:
peer.next_sync_attempt = time.time()-1
def job():
peer.sync()
threading.Thread(target=job, daemon=True).start()
time.sleep(0.25)
def dismiss_dialog(sender):
self.close_list_dialogs()
dialog = ListDialogLineBox(
urwid.Pile([
urwid.Text("A delivery sync of all unhandled LXMs was manually requested for the selected node\n", align=urwid.CENTER),
urwid.Columns([
(urwid.WEIGHT, 0.1, urwid.Text("")),
(urwid.WEIGHT, 0.45, urwid.Button("OK", on_press=dismiss_dialog)),
])
]),
title="!",
)
dialog.delegate = self.delegate
bottom = self
overlay = urwid.Overlay(dialog, bottom, align=urwid.CENTER, width=urwid.RELATIVE_100, valign=urwid.MIDDLE, height=urwid.PACK, left=2, right=2)
options = self.delegate.left_pile.options(urwid.WEIGHT, 1)
self.delegate.left_pile.contents[0] = (overlay, options)
def close_list_dialogs(self):
self.delegate.reinit_lxmf_peers()
self.delegate.show_peers()
def rebuild_widget_list(self):
self.peer_list = self.app.message_router.peers
self.widget_list = self.make_peer_widgets()
self.ilb.set_body(self.widget_list)
if len(self.widget_list) > 0:
self.no_content = False
else:
self.no_content = True
self.delegate.reinit_lxmf_peers()
def make_peer_widgets(self):
widget_list = []
sorted_peers = sorted(self.peer_list, key=lambda pid: (self.app.directory.pn_trust_level(pid), self.peer_list[pid].sync_transfer_rate), reverse=True)
for peer_id in sorted_peers:
peer = self.peer_list[peer_id]
trust_level = self.app.directory.pn_trust_level(peer_id)
pe = LXMFPeerEntry(self.app, peer, self, trust_level)
pe.destination_hash = peer.destination_hash
widget_list.append(pe)
return widget_list
class LXMFPeerEntry(urwid.WidgetWrap):
def __init__(self, app, peer, delegate, trust_level):
destination_hash = peer.destination_hash
self.app = app
g = self.app.ui.glyphs
node_identity = RNS.Identity.recall(destination_hash)
display_str = RNS.prettyhexrep(destination_hash)
if node_identity != None:
node_hash = RNS.Destination.hash_from_name_and_identity("nomadnetwork.node", node_identity)
display_name = self.app.directory.alleged_display_str(node_hash)
if display_name != None:
display_str = str(display_name)+"\n "+display_str
sym = g["sent"]
style = "list_unknown"
focus_style = "list_focus"
alive_string = "Unknown"
if hasattr(peer, "alive"):
if peer.alive:
alive_string = "Available"
if trust_level == DirectoryEntry.TRUSTED:
style = "list_trusted"
focus_style = "list_focus_trusted"
else:
style = "list_normal"
focus_style = "list_focus"
else:
alive_string = "Unresponsive"
style = "list_unresponsive"
focus_style = "list_focus_unresponsive"
if peer.propagation_transfer_limit:
txfer_limit = RNS.prettysize(peer.propagation_transfer_limit*1000)
else:
txfer_limit = "No"
ar = round(peer.acceptance_rate*100, 2)
peer_info_str = sym+" "+display_str+"\n "+alive_string+", last heard "+pretty_date(int(peer.last_heard))
peer_info_str += "\n "+str(peer.unhandled_message_count)+f" unhandled LXMs, {txfer_limit} sync limit\n"
peer_info_str += f" {RNS.prettyspeed(peer.sync_transfer_rate)} STR, "
peer_info_str += f"{RNS.prettyspeed(peer.link_establishment_rate)} LER, {ar}% AR\n"
widget = ListEntry(peer_info_str)
self.display_widget = urwid.AttrMap(widget, style, focus_style)
self.display_widget.destination_hash = destination_hash
super().__init__(self.display_widget)
def pretty_date(time=False):
"""
Get a datetime object or a int() Epoch timestamp and return a
pretty string like 'an hour ago', 'Yesterday', '3 months ago',
'just now', etc
"""
from datetime import datetime
now = datetime.now()
if type(time) is int:
diff = now - datetime.fromtimestamp(time)
elif isinstance(time,datetime):
diff = now - time
elif not time:
diff = now - now
second_diff = diff.seconds
day_diff = diff.days
if day_diff < 0:
return ''
if day_diff == 0:
if second_diff < 10:
return "just now"
if second_diff < 60:
return str(second_diff) + " seconds ago"
if second_diff < 120:
return "a minute ago"
if second_diff < 3600:
return str(int(second_diff / 60)) + " minutes ago"
if second_diff < 7200:
return "an hour ago"
if second_diff < 86400:
return str(int(second_diff / 3600)) + " hours ago"
if day_diff == 1:
return "Yesterday"
if day_diff < 7:
return str(day_diff) + " days ago"
if day_diff < 31:
return str(int(day_diff / 7)) + " weeks ago"
if day_diff < 365:
return str(int(day_diff / 30)) + " months ago"
return str(int(day_diff / 365)) + " years ago"