dandri/NomadNet
1
0
Fork 0
mirror of https://github.com/markqvist/NomadNet.git synced 2026-10-06 10:17:47 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
0b3f96e06a13547af6d7e446c580165df2bb68d4
NomadNet/nomadnet/ui/textui/Browser.py
T

2345 lines
100 KiB
Python
Raw Blame History

import RNS
import LXMF
import io
import os
import time
import bisect
import urwid
import shutil
import nomadnet
import subprocess
import threading
from threading import Lock
from .MicronParser import markup_to_attrmaps, make_style, default_state
from nomadnet.Directory import DirectoryEntry
from nomadnet.Conversation import ConversationMessage
from nomadnet.vendor.Scrollable import *
from nomadnet.util import strip_modifiers
from nomadnet.util import sanitize_name
from nomadnet.ui.textui.images import _termlib
from .Helpers import ClickableIcon, osc52_copy
from .ReadlineEdit import ReadlineMixin, ReadlineEdit
IMGLOAD_NEVER = 0x00
IMGLOAD_MANUAL = 0x01
IMGLOAD_AUTO = 0x02
IMGLOAD_ALWAYS = 0x03
class BrowserFrame(urwid.Frame):
def keypress(self, size, key):
if key == "ctrl w":
self.delegate.disconnect()
elif key == "ctrl d":
self.delegate.back()
elif key == "ctrl f":
self.delegate.forward()
elif key == "ctrl r":
self.delegate.reload()
elif key == "ctrl u":
self.delegate.url_dialog()
elif key == "ctrl l":
self.delegate.load_images()
elif key == "ctrl x":
self.delegate.load_images(force_reload=True)
elif key == "ctrl s":
self.delegate.save_node_dialog()
elif key == "ctrl b":
self.delegate.save_node_dialog()
elif key == "ctrl g":
nomadnet.NomadNetworkApp.get_shared_instance().ui.main_display.sub_displays.network_display.toggle_fullscreen()
elif key == "ctrl y":
if self.delegate.app.config["textui"]["clipboard_copy"]:
self.delegate.copy_url()
elif self.focus_position == "body":
if key == "down" or key == "up":
try:
if hasattr(self.delegate, "page_list") and self.delegate.page_list:
def df(loop, user_data):
st = None
if self.delegate.page_list:
nf = self.delegate.page_list.focus
if hasattr(nf, "key_timeout"):
st = nf
elif hasattr(nf, "original_widget"):
no = nf.original_widget
if hasattr(no, "original_widget"):
st = no.original_widget
else:
if hasattr(no, "key_timeout"):
st = no
if st and hasattr(st, "key_timeout") and hasattr(st, "keypress") and callable(st.keypress):
st.keypress(None, None)
nomadnet.NomadNetworkApp.get_shared_instance().ui.loop.set_alarm_in(0.25, df)
except Exception as e:
RNS.log("Error while setting up cursor timeout. The contained exception was: "+str(e), RNS.LOG_ERROR)
return super(BrowserFrame, self).keypress(size, key)
else:
return super(BrowserFrame, self).keypress(size, key)
class PageList(urwid.ListBox):
def __init__(self, body):
super().__init__(body)
self._page_size = (0, 0)
self._page_focus = None
self._page_layouts = {}
def invalidate_layout(self):
self._page_layouts = {}
self._invalidate()
def _layout(self, maxcol):
if maxcol not in self._page_layouts:
layout_started = time.time()
offsets = [0]
for widget in self.body:
offsets.append(offsets[-1] + widget.rows((maxcol,), True))
self._page_layouts[maxcol] = offsets
# RNS.log(f" rebuilt {len(self.body)} widgets, {offsets[-1]} rows, width {maxcol}, {(time.time()-layout_started)*1000:.1f}ms", RNS.LOG_DEBUG)
return self._page_layouts[maxcol]
def render(self, size, focus=False):
self._page_size = size
if self.body and self.focus is not self._page_focus:
self._page_focus = self.focus
widget = self.focus
while widget is not None:
widget._invalidate()
widget = getattr(widget, "original_widget", None)
return super().render(size, focus)
def rows_max(self, size=None, focus=False):
size = size or self._page_size
return self._layout(size[0])[-1]
def get_scrollpos(self, size=None, focus=False):
size = size or self._page_size
if not self.body or not size[1]: return 0
middle, top, bottom = self.calculate_visible(size, focus)
if middle is None: return 0
return self._layout(size[0])[self.top_position(size)] + top.trim
def set_scrollpos(self, position):
size = self._page_size
maxcol, maxrow = size
if not self.body or not maxrow: return
offsets = self._layout(maxcol)
position = max(0, min(int(position), max(0, offsets[-1] - maxrow)))
pos = bisect.bisect_left(offsets, position, 0, len(self.body))
if pos == len(self.body) or offsets[pos] - position >= maxrow: pos -= 1
self.change_focus(size, pos, offsets[pos] - position)
def top_position(self, size=None):
size = size or self._page_size
if not self.body or not size[1]: return 0
middle, top, bottom = self.calculate_visible(size, True)
if middle is None: return 0
return top.fill[-1].position if top.fill else middle.focus_pos
def scroll(self, rows):
self.set_scrollpos(self.get_scrollpos() + rows)
def focused_widget_name(self):
widget = self.focus
while getattr(widget, "original_widget", None) is not None: widget = widget.original_widget
return type(widget).__name__
def keypress(self, size, key):
self._page_size = size
key_started = time.time()
focus_before = self.focus_position if self.body else None
scroll_before = self.get_scrollpos(size, True)
unhandled_key = self.handle_key(size, key)
# RNS.log(f" {key}: {unhandled_key} focus {focus_before} -> {self.focus_position if self.body else None} ({self.focused_widget_name()}) scrollpos {scroll_before}->{self.get_scrollpos(size, True)}, {(time.time()-key_started) * 1000:.1f}ms", RNS.LOG_DEBUG)
return unhandled_key
def handle_key(self, size, key):
try:
if key in ("page up", "page down", "home", "end"):
focus_widget = self.focus
if focus_widget is not None and focus_widget.selectable():
key = focus_widget.keypress((size[0],), key)
if key is None:
self.make_cursor_visible(size)
return None
if key == "page down": self.scroll(size[1] - 1)
elif key == "page up": self.scroll(1 - size[1])
elif key == "home": self.set_scrollpos(0)
elif key == "end": self.set_scrollpos(self.rows_max(size))
return None
unhandled_key = super().keypress(size, key)
if key in ("up", "down"): return None
return unhandled_key
except Exception as e:
RNS.log(f"Error while handling browser keypress: {e}", RNS.LOG_ERROR)
RNS.trace_exception(e)
return None
def mouse_event(self, size, event, button, col, row, focus):
self._page_size = size
if button in (4, 5): return False
return super().mouse_event(size, event, button, col, row, focus)
class Browser:
DEFAULT_PATH = "/page/index.mu"
DEFAULT_TIMEOUT = 10
DEFAULT_CACHE_TIME = 12*60*60
NO_PATH = 0x00
PATH_REQUESTED = 0x01
ESTABLISHING_LINK = 0x02
LINK_TIMEOUT = 0x03
LINK_ESTABLISHED = 0x04
REQUESTING = 0x05
REQUEST_SENT = 0x06
REQUEST_FAILED = 0x07
REQUEST_TIMEOUT = 0x08
RECEIVING_RESPONSE = 0x09
DISCONECTED = 0xFE
DONE = 0xFF
def __init__(self, app, app_name, aspects, destination_hash = None, path = None, auth_identity = None, delegate = None):
self.app = app
self.g = self.app.ui.glyphs
self.delegate = delegate
self.app_name = app_name
self.aspects = aspects
self.destination_hash = destination_hash
self.path = path
self.request_data = None
self.timeout = Browser.DEFAULT_TIMEOUT
self.last_keypress = None
self.link = None
self.loopback = None
self.status = Browser.DISCONECTED
self.progress_updated_at = None
self.previous_progress = 0
self.response_progress = 0
self.response_speed = None
self.response_size = None
self.response_transfer_size = None
self.page_background_color = None
self.page_foreground_color = None
self.saved_file_name = None
self.saved_file_size = 0
self.file_saved_at = 0
self.file_save_notice_timeout = 3
self.page_data = None
self.displayed_page_data = None
self.auth_identity = auth_identity
self.display_widget = None
self.link_status_showing = False
self.link_target = None
self.frame = None
self.attr_maps = []
self.page_list = None
self.page_rows = []
self.page_partials = {}
self.page_images = {}
self.updater_running = False
self.image_updater_running = False
self.partial_updater_lock = Lock()
self.image_updater_lock = Lock()
self.image_mode = self.app.config["textui"]["image_loading"]
self.image_rendering_supported = _termlib.is_kitty_supported() and self.image_mode != IMGLOAD_NEVER
self.build_display()
self.history = []
self.history_ptr = 0
self.history_inc = False
self.history_dec = False
self.reloading = False
self.loaded_from_cache = False
if self.path == None:
self.path = Browser.DEFAULT_PATH
if self.destination_hash != None:
self.load_page()
self.clean_cache()
if self.image_rendering_supported: RNS.log(f"In-browser image rendering supported", RNS.LOG_DEBUG)
else: RNS.log(f"In-browser image rendering is not supported", RNS.LOG_DEBUG)
def current_url(self):
if self.destination_hash == None:
return ""
path = "" if self.path == None else self.path
url = RNS.hexrep(self.destination_hash, delimit=False)+":"+path
if isinstance(self.request_data, dict) and self.request_data:
parts = []
for k, v in self.request_data.items():
if not isinstance(k, str):
continue
if not k.startswith("var_"):
continue
parts.append(k[4:]+"="+str(v))
if parts:
url += "`" + "|".join(parts)
return url
def url_hash(self, url):
if url == None:
return None
else:
url = url.encode("utf-8")
return RNS.hexrep(RNS.Identity.full_hash(url), delimit=False)
def marked_link(self, link_target, link_fields=None):
if link_fields:
fields_str = "|".join(link_fields)
link_target = f"{link_target}`{fields_str}"
if self.status == Browser.DONE:
self.link_target = link_target
self.app.ui.loop.set_alarm_in(0.1, self.marked_link_job)
def marked_link_job(self, sender, event):
link_target = self.link_target
can_display = time.time() > self.file_saved_at + self.file_save_notice_timeout
if link_target == None:
if self.link_status_showing:
self.browser_footer = self.make_status_widget()
self.frame.contents["footer"] = (self.browser_footer, self.frame.options())
self.link_status_showing = False
if self.page_background_color != None or self.page_foreground_color != None:
style_name = make_style(default_state(fg=self.page_foreground_color, bg=self.page_background_color))
self.browser_footer.set_attr_map({None: style_name})
else:
if can_display:
self.link_status_showing = True
lt_str = str(link_target)
lmax = self._content_cols()
lstr = "Link to "+lt_str
if len(lstr) > lmax: lstr = lstr[:lmax-1]+"…"
self.browser_footer = urwid.AttrMap(urwid.Pile([urwid.Divider(self.g["divider1"]), urwid.Text(lstr)]), "browser_controls")
self.frame.contents["footer"] = (self.browser_footer, self.frame.options())
if self.page_background_color != None or self.page_foreground_color != None:
style_name = make_style(default_state(fg=self.page_foreground_color, bg=self.page_background_color))
self.browser_footer.set_attr_map({None: style_name})
def expand_shorthands(self, destination_type):
if destination_type == "nnn":
return "nomadnetwork.node"
elif destination_type == "lxmf":
return "lxmf.delivery"
elif destination_type == "rrc":
return "rrc.hub.session"
else:
return destination_type
def handle_link(self, link_target, link_data = None):
partial_ids = None
request_data = None
if link_data != None:
link_fields = []
request_data = {}
all_fields = True if "*" in link_data else False
for e in link_data:
if "=" in e:
c = e.split("=")
if len(c) == 2:
request_data["var_"+str(c[0])] = str(c[1])
else:
link_fields.append(e)
def recurse_down(w):
if isinstance(w, list):
for t in w:
recurse_down(t)
elif isinstance(w, tuple):
for t in w:
recurse_down(t)
elif hasattr(w, "contents"):
recurse_down(w.contents)
elif hasattr(w, "original_widget"):
recurse_down(w.original_widget)
elif hasattr(w, "_original_widget"):
recurse_down(w._original_widget)
else:
if hasattr(w, "field_name") and (all_fields or w.field_name in link_fields):
field_key = "field_" + w.field_name
if isinstance(w, urwid.Edit):
request_data[field_key] = w.edit_text
elif isinstance(w, urwid.RadioButton):
if w.state:
user_data = getattr(w, "field_value", None)
if user_data is not None:
request_data[field_key] = user_data
elif isinstance(w, urwid.CheckBox):
user_data = getattr(w, "field_value", "1")
if w.state:
existing_value = request_data.get(field_key, '')
if existing_value:
# Concatenate the new value with the existing one
request_data[field_key] = existing_value + ',' + user_data
else:
# Initialize the field with the current value
request_data[field_key] = user_data
else:
pass # do nothing if checkbox is not check
recurse_down(self.attr_maps)
RNS.log("Including request data: "+str(request_data), RNS.LOG_EXTREME)
# In-document anchor link (#name or empty #)
if link_target.startswith("#"):
def df(loop, user_data): self._jump_to_anchor(link_target[1:])
nomadnet.NomadNetworkApp.get_shared_instance().ui.loop.set_alarm_in(0.0, df)
return
# rrc://<hex>[:<dest_name>]/<room> URL form
if link_target.startswith("rrc://"):
self.handle_rrc_link(link_target[6:])
return
components = link_target.split("@")
destination_type = None
if len(components) == 2:
destination_type = self.expand_shorthands(components[0])
link_target = components[1]
elif link_target.startswith("p:"):
comps = link_target.split(":")
if len(comps) > 1: partial_ids = comps[1:]
destination_type = "partial"
else:
destination_type = "nomadnetwork.node"
link_target = components[0]
if destination_type == "nomadnetwork.node":
if self.status >= Browser.DISCONECTED:
RNS.log("Browser handling link to: "+str(link_target), RNS.LOG_DEBUG)
self.browser_footer = urwid.Text("Opening link to: "+str(link_target))
try:
self.retrieve_url(link_target, request_data)
except Exception as e:
self.browser_footer = urwid.Text("Could not open link: "+str(e))
self.frame.contents["footer"] = (self.browser_footer, self.frame.options())
else:
RNS.log("Browser already handling link, cannot handle link to: "+str(link_target), RNS.LOG_DEBUG)
elif destination_type == "lxmf.delivery":
RNS.log("Passing LXMF link to handler", RNS.LOG_DEBUG)
self.handle_lxmf_link(link_target)
elif destination_type == "rrc.hub.session":
RNS.log("Passing RRC link to handler", RNS.LOG_DEBUG)
self.handle_rrc_link(link_target)
elif destination_type == "partial":
if partial_ids != None and len(partial_ids) > 0: self.handle_partial_updates(partial_ids)
else:
RNS.log("No known handler for destination type "+str(destination_type), RNS.LOG_DEBUG)
self.browser_footer = urwid.Text("Could not open link: "+"No known handler for destination type "+str(destination_type))
self.frame.contents["footer"] = (self.browser_footer, self.frame.options())
def _jump_to_anchor(self, name):
anchors = getattr(self.attr_maps, "anchors", None) or {}
header_rows = getattr(self.attr_maps, "header_rows", None) or []
target_idx = None
if name:
target_idx = anchors.get(name)
if target_idx is None:
self.browser_footer = urwid.Text("Unknown anchor: #"+name)
self.frame.contents["footer"] = (self.browser_footer, self.frame.options())
return
else:
current = 0
try:
current = self.page_rows[self.page_list.top_position()]
except Exception:
current = 0
for hr in header_rows:
if hr > current:
target_idx = hr
break
if target_idx is None:
return
self.reveal_index(int(target_idx))
try:
self.page_list.set_focus(self.page_rows.index(int(target_idx)))
self.page_list.set_focus_valign("top")
except Exception as e: RNS.log("Anchor jump failed: "+str(e), RNS.LOG_ERROR)
def _content_cols(self):
try:
lw = self.delegate.given_list_width
screen_cols = self.app.ui.loop.screen.get_cols_rows()[0]
browser_cols = screen_cols-lw-2
cols = browser_cols
except Exception as e:
RNS.log(f"Could not calculate browser width, anchors will be inaccurate: {e}", RNS.LOG_WARNING)
RNS.trace_exception(e)
cols = 100
return max(40, cols)
def handle_lxmf_link(self, link_target):
try:
def san(name):
if self.app.config["textui"]["sanitize_names"]: return sanitize_name(name)
else: return strip_modifiers(name)
if not type(link_target) is str:
raise ValueError("Invalid data type for LXMF link")
if len(link_target) != (RNS.Reticulum.TRUNCATED_HASHLENGTH//8)*2:
raise ValueError("Invalid length for LXMF link")
try:
bytes.fromhex(link_target)
except Exception as e:
raise ValueError("Could not decode destination hash from LXMF link")
existing_conversations = nomadnet.Conversation.conversation_list(self.app)
source_hash_text = link_target
display_name_data = RNS.Identity.recall_app_data(bytes.fromhex(source_hash_text))
display_name = None
if display_name_data != None:
display_name = san(LXMF.display_name_from_app_data(display_name_data))
if not source_hash_text in [c[0] for c in existing_conversations]:
entry = DirectoryEntry(bytes.fromhex(source_hash_text), display_name=display_name)
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 link. The contained exception was: "+str(e), RNS.LOG_ERROR)
self.browser_footer = urwid.Text("Could not open LXMF link: "+str(e))
self.frame.contents["footer"] = (self.browser_footer, self.frame.options())
def handle_rrc_link(self, link_target):
try:
if not isinstance(link_target, str):
raise ValueError("invalid RRC link payload")
rest = link_target.strip()
if rest.startswith("/"):
rest = rest[1:]
hub_part, _, room = rest.partition("/")
hex_part, _, dest = hub_part.partition(":")
hex_part = hex_part.strip()
dest = dest.strip() or None
try:
hub_hash = bytes.fromhex(hex_part)
except Exception:
raise ValueError("invalid hub hash")
expected_len = RNS.Reticulum.TRUNCATED_HASHLENGTH // 8
if len(hub_hash) != expected_len:
raise ValueError("hub hash must be "+str(expected_len)+" bytes")
room = room.strip().lstrip("#").strip()
room_norm = room.lower() if room else None
existing = self.app.rrc.find_hub(hub_hash, dest_name=dest)
self.app.ui.main_display.show_channels(None)
channels = self.app.ui.main_display.sub_displays.channels_display
if existing is not None:
channels.update_list()
if room_norm:
channels._select_room(None, (existing, room_norm))
else:
channels._select_hub(None, existing)
return
channels.confirm_new_hub_dialog(hub_hash, dest, room_norm)
except Exception as e:
RNS.log("Could not open RRC link: "+str(e), RNS.LOG_ERROR)
self.browser_footer = urwid.Text("Could not open RRC link: "+str(e))
self.frame.contents["footer"] = (self.browser_footer, self.frame.options())
def micron_released_focus(self):
if self.delegate != None:
self.delegate.focus_lists()
def build_display(self):
self.browser_header = urwid.Text("")
self.browser_footer = urwid.Text("")
self.page_list = None
self.page_rows = []
self.page_partials = {}
self.page_images = {}
self.browser_body = urwid.Filler(
urwid.Text("Disconnected\n"+self.g["arrow_l"]+" "+self.g["arrow_r"], align=urwid.CENTER),
urwid.MIDDLE,
)
self.frame = BrowserFrame(self.browser_body, header=self.browser_header, footer=self.browser_footer)
self.frame.delegate = self
self.linebox = urwid.LineBox(self.frame, title="Remote Node")
self.display_widget = urwid.AttrMap(self.linebox, "browser_inactive")
def make_status_widget(self):
if self.response_progress > 0:
if self.page_background_color != None or self.page_foreground_color != None:
style_name = make_style(default_state(fg=self.page_foreground_color, bg=self.page_background_color))
style_name_inverted = make_style(default_state(bg=self.page_foreground_color, fg=self.page_background_color))
else:
style_name = "progress_empty"
style_name_inverted = "progress_full"
pb = ResponseProgressBar(style_name , style_name_inverted, current=self.response_progress, done=1.0, satt=None, owner=self)
widget = urwid.Pile([urwid.Divider(self.g["divider1"]), pb])
else:
widget = urwid.Pile([urwid.Divider(self.g["divider1"]), urwid.Text(self.status_text())])
return urwid.AttrMap(widget, "browser_controls")
def make_control_widget(self):
clipboard_copy_enabled = self.app.config["textui"]["clipboard_copy"]
copy_glyph = self.g.get("copy", "[C]")
lstr = self.g["node"]+" "+self.current_url()
if clipboard_copy_enabled:
lmax = self._content_cols()-len(copy_glyph)-len(copy_glyph)-1
else:
lmax = self._content_cols()-1
if len(lstr) > lmax: lstr = lstr[:lmax-1]+"…"
url_text = urwid.Text(lstr)
if clipboard_copy_enabled:
copy_icon = ClickableIcon(copy_glyph, on_click=lambda: self.copy_url())
copy_width = len(copy_glyph) + 2
header_row = urwid.Columns([
("weight", 1, url_text),
(copy_width, urwid.Padding(copy_icon, left=1, right=1)),
])
else:
header_row = url_text
return urwid.AttrMap(urwid.Pile([header_row, urwid.Divider(self.g["divider1"])]), "browser_controls")
def make_request_failed_widget(self):
def back_action(sender):
self.status = Browser.DONE
self.destination_hash = self.previous_destination_hash
self.path = self.previous_path
self.update_display()
columns = urwid.Columns([
(urwid.WEIGHT, 0.5, urwid.Text(" ")),
(8, urwid.Button("Back", on_press=back_action)),
(urwid.WEIGHT, 0.5, urwid.Text(" ")),
])
if len(self.attr_maps) > 0:
pile = urwid.Pile([
urwid.Text("!\n\n"+self.status_text()+"\n", align=urwid.CENTER),
columns
])
else:
pile = urwid.Pile([urwid.Text("!\n\n"+self.status_text(), align=urwid.CENTER)])
return urwid.Filler(pile, urwid.MIDDLE)
def update_display(self):
if self.status == Browser.DISCONECTED:
self.display_widget.set_attr_map({None: "browser_inactive"})
self.page_list = None
self.page_rows = []
self.page_partials = {}
self.page_images = {}
self.browser_body = urwid.Filler(
urwid.Text("Disconnected\n"+self.g["arrow_l"]+" "+self.g["arrow_r"], align=urwid.CENTER),
urwid.MIDDLE,
)
self.browser_footer = urwid.Text("")
self.browser_header = urwid.Text("")
self.linebox.set_title("Remote Node")
else:
self.display_widget.set_attr_map({None: "body_text"})
self.browser_header = self.make_control_widget()
if self.page_background_color != None or self.page_foreground_color != None:
style_name = make_style(default_state(fg=self.page_foreground_color, bg=self.page_background_color))
self.browser_header.set_attr_map({None: style_name})
if self.destination_hash != None:
remote_display_string = self.app.directory.simplest_display_str(self.destination_hash)
else:
remote_display_string = ""
if self.loopback != None and remote_display_string == RNS.prettyhexrep(self.loopback):
remote_display_string = self.app.node.name
self.linebox.set_title(remote_display_string)
if self.status == Browser.DONE:
self.browser_footer = self.make_status_widget()
self.update_page_display()
if self.page_background_color != None or self.page_foreground_color != None:
style_name = make_style(default_state(fg=self.page_foreground_color, bg=self.page_background_color))
self.browser_body.set_attr_map({None: style_name})
self.browser_footer.set_attr_map({None: style_name})
self.browser_header.set_attr_map({None: style_name})
self.display_widget.set_attr_map({None: style_name})
elif self.status == Browser.LINK_TIMEOUT:
self.browser_body = self.make_request_failed_widget()
self.browser_footer = urwid.Text("")
elif self.status <= Browser.REQUEST_SENT:
if len(self.attr_maps) == 0:
self.browser_body = urwid.Filler(
urwid.Text("Retrieving\n["+self.current_url()+"]", align=urwid.CENTER),
urwid.MIDDLE,
)
self.browser_footer = self.make_status_widget()
if self.page_background_color != None or self.page_foreground_color != None:
style_name = make_style(default_state(fg=self.page_foreground_color, bg=self.page_background_color))
self.browser_footer.set_attr_map({None: style_name})
self.browser_header.set_attr_map({None: style_name})
self.display_widget.set_attr_map({None: style_name})
elif self.status == Browser.REQUEST_FAILED:
self.browser_body = self.make_request_failed_widget()
self.browser_footer = urwid.Text("")
elif self.status == Browser.REQUEST_TIMEOUT:
self.browser_body = self.make_request_failed_widget()
self.browser_footer = urwid.Text("")
else:
pass
self.frame.contents["body"] = (self.browser_body, self.frame.options())
self.frame.contents["header"] = (self.browser_header, self.frame.options())
self.frame.contents["footer"] = (self.browser_footer, self.frame.options())
def update_page_display(self):
self._purge_page_images()
self.page_rows = list(range(len(self.attr_maps)))
self.page_list = PageList(urwid.SimpleFocusListWalker(list(self.attr_maps)))
self.page_partials = {}
self.page_images = {}
self.browser_body = urwid.AttrMap(ScrollBar(self.page_list, thumb_char="\u2503", trough_char=" "), "scrollbar")
self.detect_partials()
if self.should_load_images(): self.detect_images()
self.init_folds()
def refresh_page_layout(self):
if self.page_list is not None: self.page_list.invalidate_layout()
def image_cache_path(self, url):
url_hash = self.url_hash(url)
if not url_hash: return None
else:
path = f"{self.app.cachepath}/images/{url_hash}"
return path
def resolve_image(self, url):
cachepath = self.image_cache_path(url)
if not cachepath: return None
else:
if os.path.exists(cachepath): return cachepath
else: return None
def _purge_page_images(self):
if not self.image_rendering_supported: return
try:
screen = getattr(getattr(self.app.ui, "loop", None), "screen", None)
if hasattr(screen, "purge_images"): screen.purge_images()
except Exception: pass
def should_load_images(self):
if self.image_mode == IMGLOAD_NEVER: return False
elif self.image_mode == IMGLOAD_MANUAL: return False
elif self.image_mode == IMGLOAD_ALWAYS: return True
elif self.image_mode == IMGLOAD_AUTO:
if self.destination_hash == self.loopback: return True
rtt_limit = 1.5; edr_limit = 10000
if self.link:
rtt = self.link.rtt
edr = self.link.get_expected_rate() or self.last_response_speed()
rtt_ok = rtt and rtt < rtt_limit
edr_ok = edr and edr > edr_limit
RNS.log(f"Checking image load conditions, RTT is {RNS.prettyshorttime(rtt) if rtt else 'unknown'}, EDR is {RNS.prettyspeed(edr) if edr else 'unknown'}", RNS.LOG_DEBUG)
if rtt_ok and edr_ok: return True
else: return False
return False
def load_images(self, force_reload=False):
if not self.image_rendering_supported: return
elif self.image_mode == IMGLOAD_NEVER: return
else: self.detect_images(force_reload=force_reload)
def detect_images(self, force_reload=False):
if not self.image_rendering_supported: return
for w in self.attr_maps:
o = w._original_widget
if hasattr(o, "_contained_image"): o = o._contained_image
if hasattr(o, "image_url"):
resolved_image = self.resolve_image(o.image_url)
if resolved_image and force_reload:
RNS.log(f"Clearing image cache for {o.image_url} ({resolved_image})", RNS.LOG_DEBUG)
try: os.unlink(resolved_image)
except Exception as e: RNS.log(f"Could not clear image cache for {o.image_url}: {e}", RNS.LOG_ERROR)
resolved_image = None
if force_reload: continue
elif resolved_image and self.image_rendering_supported: o.load(resolved_image, remote_source=True)
else:
image = { "widget": o, "id": o.image_id, "url": o.image_url, "refresh": None, "updated": None,
"update_requested": None, "request_id": None, "link": None, "failed": False, "pr_throttle": 0,
"progress_updated": None, "previous_progress": 0 }
self.page_images[o.image_id] = image
self.refresh_page_layout()
if force_reload: self.reload()
elif len(self.page_images) > 0: self.start_image_updater()
def start_image_updater(self):
if not self.image_rendering_supported: return
if not self.image_updater_running: self.update_images()
def update_images(self, loop=None, user_data=None):
if not self.image_rendering_supported: return
def job():
if self.image_updater_lock.locked(): return
with self.image_updater_lock:
self.image_updater_running = True
for iid in self.page_images:
try:
image = self.page_images[iid]
if image["failed"]: continue
if not image["updated"] or (image["refresh"] != None and time.time() > image["updated"]+image["refresh"]):
image["update_requested"] = time.time()
self.__load_image(image)
while not image["updated"] and not image["failed"] and len(self.page_images) > 0: time.sleep(0.2)
except Exception as e: RNS.log(f"Error updating page image: {e}", RNS.LOG_ERROR)
if len(self.page_images) > 0: self.app.ui.loop.set_alarm_in(1, self.update_images)
else: self.image_updater_running = False
threading.Thread(target=job, daemon=True).start()
def __load_image(self, image):
if image["failed"] == True: return
w = image["widget"]
url = image["url"]
try: image_destination_hash, path = self.parse_url(image["url"])
except Exception as e:
RNS.log(f"Could not parse image URL: {e}", RNS.LOG_ERROR)
image["failed"] = True
w.notice(f"Could not load image {url}: {e}")
return
if image_destination_hash == self.loopback:
local_image = path.replace("/media/", "")
local_path = f"{self.app.pagespath}/{local_image}"
if self.image_rendering_supported:
w.load(local_path)
self.refresh_page_layout()
image["updated"] = time.time()
return
if not RNS.Transport.has_path(image_destination_hash):
if time.time() <= image["pr_throttle"]: return
else:
image["pr_throttle"] = time.time()+15
RNS.log(f"Requesting path for image: {path}", RNS.LOG_DEBUG)
RNS.Transport.request_path(image_destination_hash)
pr_time = time.time()+self.path_timeout(image_destination_hash)
while not RNS.Transport.has_path(image_destination_hash):
if time.time() > pr_time: return
time.sleep(0.25)
if not image["link"]:
if self.link and self.link.destination.hash == image_destination_hash and self.link.status == RNS.Link.ACTIVE:
image["link"] = self.link
RNS.log(f"Re-using existing page link for image {url}", RNS.LOG_EXTREME)
if not image["link"]:
for pid in self.page_images:
other_image = self.page_images[pid]
if other_image["link"]:
existing_link = other_image["link"]
if existing_link.destination.hash == image_destination_hash and existing_link.status == RNS.Link.ACTIVE:
RNS.log(f"Re-using existing link for image {url}", RNS.LOG_EXTREME)
image["link"] = existing_link
break
if not image["link"] or (image["link"] and image["link"].status) == RNS.Link.CLOSED:
RNS.log(f"Establishing link for image: {image_destination_hash} / {path}", RNS.LOG_DEBUG)
identity = RNS.Identity.recall(image_destination_hash)
destination = RNS.Destination(identity, RNS.Destination.OUT, RNS.Destination.SINGLE, self.app_name, self.aspects)
w.notice(f"Establishing link...")
def f_established(wd):
def established(link):
RNS.log(f"Link established for image: {path}", RNS.LOG_DEBUG)
wd.notice(f"Link established")
return established
def f_closed(wd):
def closed(link):
RNS.log(f"Link closed for image: {path}", RNS.LOG_DEBUG)
image["link"] = None
image["failed"] = True
wd.notice(f"Link closed during transfer")
return closed
image["link"] = RNS.Link(destination, established_callback = f_established(w), closed_callback = f_closed(w))
timeout = time.time()+max(self.timeout, image["link"].establishment_timeout)
while image["link"].status != RNS.Link.ACTIVE and time.time() < timeout: time.sleep(0.1)
if image["link"] and image["link"].status == RNS.Link.ACTIVE and image["request_id"] == None:
RNS.log(f"Sending request for image: {path}", RNS.LOG_DEBUG)
receipt = image["link"].request("/media", data=self.__get_image_request_data(image), response_callback = self.image_received,
failed_callback = self.image_failed, progress_callback = self.image_progressed)
w.notice(f"Request sent")
if receipt: image["request_id"] = receipt.request_id
else:
RNS.log(f"Image request failed", RNS.LOG_ERROR)
image["failed"] = True
w.notice(f"Request failed")
def __get_image_request_data(self, image):
destination_hash, path = self.parse_url(image["url"])
request_data = {"path": path, "key": None}
return request_data
def image_failed(self, request_receipt):
RNS.log("Loading page image failed", RNS.LOG_ERROR)
for iid in self.page_images:
image = self.page_images[iid]
if image["request_id"] == request_receipt.request_id:
try:
image["updated"] = time.time()
image["request_id"] = None
image["failed"] = True
url = image["url"]
w = image["widget"]
w.notice(f"Could not load image {url}")
except Exception as e:
RNS.log(f"Error in image failed callback: {e}", RNS.LOG_ERROR)
RNS.trace_exception(e)
def image_progressed(self, request_receipt):
for iid in self.page_images:
i = self.page_images[iid]
if i["request_id"] == request_receipt.request_id:
try:
response_progress = request_receipt.progress
response_time = request_receipt.get_response_time()
response_size = request_receipt.response_size
response_transfer_size = request_receipt.response_transfer_size
w = i["widget"]
now = time.time()
if i["progress_updated"] == None: i["progress_updated"] = now
if now > i["progress_updated"]+0.5:
td = now - i["progress_updated"]
pd = response_progress - i["previous_progress"]
bd = pd*response_transfer_size
response_speed = (bd/td)*8
i["previous_progress"] = response_progress
i["progress_updated"] = now
stats = f"{round(response_progress*100, 2)}% ({RNS.prettysize(response_progress*response_size)} of {RNS.prettysize(response_size)}) - {RNS.prettyspeed(response_speed)}"
w.notice(stats)
except Exception as e: RNS.trace_exception(e)
def image_received(self, request_receipt):
if not request_receipt.response: self.image_failed(request_receipt)
for iid in self.page_images:
image = self.page_images[iid]
if image["request_id"] == request_receipt.request_id:
try:
image["updated"] = image["update_requested"]
image["request_id"] = None
w = image["widget"]
w.notice(f"Image loaded")
except Exception as e:
RNS.log(f"Error while loading received image: {e}", RNS.LOG_ERROR)
RNS.trace_exception(e)
image["failed"] = True
try:
url = image["url"]
cachepath = self.image_cache_path(url)
if isinstance(request_receipt.response, bytes):
with open(cachepath, "wb") as f: f.write(request_receipt.response)
elif type(request_receipt.response) == io.BufferedReader:
if request_receipt.metadata != None:
file_handle = request_receipt.response
file_destination = cachepath
shutil.move(file_handle.name, file_destination)
resolved_path = self.resolve_image(url)
if resolved_path and self.image_rendering_supported:
w.load(resolved_path, remote_source=True)
self.refresh_page_layout()
except Exception as e:
RNS.log("Error while handling image response: "+str(e), RNS.LOG_ERROR)
image["failed"] = True
# Render an arbitrary markup buffer (used by the page editor preview) and
# return the content widget, with partials/folds/fields wired as usual
def render_markup_buffer(self, markup):
self.status = Browser.DONE
self.markup = markup
self.page_background_color = None
self.page_foreground_color = None
for tag, attr in (("#!bg=", "page_background_color"), ("#!fg=", "page_foreground_color")):
pos = markup.find(tag)
if pos >= 0:
endpos = markup.find("\n", pos)
if endpos < 0: endpos = len(markup)
if endpos-(pos+len(tag)) in (3, 6):
setattr(self, attr, markup[pos+len(tag):endpos])
try:
self.attr_maps = markup_to_attrmaps(strip_modifiers(markup), url_delegate=self, fg_color=self.page_foreground_color, bg_color=self.page_background_color)
except Exception as e:
self.attr_maps = [urwid.AttrMap(urwid.Text(f"Could not render preview: {e}"), "browser_inactive")]
self.update_page_display()
return self.browser_body
def init_folds(self):
self.fold_sections = {}
self.collapsed_headings = set()
attrmaps = self.attr_maps
header_levels = getattr(attrmaps, "header_levels", None) or {}
row_levels = getattr(attrmaps, "row_levels", None) or {}
collapsibles = getattr(attrmaps, "collapsibles", None) or {}
if not collapsibles:
return
n = len(attrmaps)
for idx in sorted(collapsibles.keys()):
widget = self.collapsible_widget_at(idx)
if widget is None: continue
depth = header_levels.get(idx, 1)
end = n
for i in range(idx+1, n):
if row_levels.get(i, depth) < depth or \
(i in header_levels and header_levels.get(i, 1) <= depth):
end = i; break
self.fold_sections[widget] = (idx+1, end)
if collapsibles[idx]:
self.collapsed_headings.add(widget)
try: widget.set_collapsed(True)
except Exception: pass
if self.collapsed_headings:
self.rebuild_visible()
def collapsible_widget_at(self, idx):
try:
o = self.attr_maps[idx]._original_widget
if getattr(o, "is_collapsible_heading", False): return o
except Exception: pass
return None
def hidden_indices(self):
hidden = set()
for widget in getattr(self, "collapsed_headings", ()):
rng = getattr(self, "fold_sections", {}).get(widget)
if rng: hidden.update(range(rng[0], rng[1]))
return hidden
def rebuild_visible(self):
page = getattr(self, "page_list", None)
if page is None: return
hidden = self.hidden_indices()
self.page_rows = [i for i in range(len(self.attr_maps)) if i not in hidden]
focus = page.focus_position if page.body else 0
page.body[:] = [self.attr_maps[i] for i in self.page_rows]
page.invalidate_layout()
if page.body: page.set_focus(min(focus, len(page.body)-1))
def fold_changed(self, heading_widget):
if getattr(heading_widget, "collapsed", False):
self.collapsed_headings.add(heading_widget)
else:
self.collapsed_headings.discard(heading_widget)
self.rebuild_visible()
try:
for pos, w in enumerate(self.page_list.body):
if getattr(w, "_original_widget", None) is heading_widget:
self.page_list.set_focus(pos)
break
except Exception: pass
def reveal_index(self, idx):
changed = False
for widget in list(getattr(self, "collapsed_headings", ())):
rng = getattr(self, "fold_sections", {}).get(widget)
if rng and rng[0] <= idx < rng[1]:
self.collapsed_headings.discard(widget)
try: widget.set_collapsed(False)
except Exception: pass
changed = True
if changed: self.rebuild_visible()
def parse_url(self, url):
path = None
destination_hash = None
components = url.split(":")
if len(components) == 1:
if len(components[0]) == (RNS.Reticulum.TRUNCATED_HASHLENGTH//8)*2:
try: destination_hash = bytes.fromhex(components[0])
except Exception as e: raise ValueError("Malformed URL")
path = Browser.DEFAULT_PATH
else: raise ValueError("Malformed URL")
elif len(components) == 2:
if len(components[0]) == (RNS.Reticulum.TRUNCATED_HASHLENGTH//8)*2:
try: destination_hash = bytes.fromhex(components[0])
except Exception as e: raise ValueError("Malformed URL")
path = components[1]
if len(path) == 0: path = Browser.DEFAULT_PATH
else:
if len(components[0]) == 0:
if self.destination_hash != None:
destination_hash = self.destination_hash
path = components[1]
if len(path) == 0: path = Browser.DEFAULT_PATH
else: raise ValueError("Malformed URL")
else: raise ValueError("Malformed URL")
else: raise ValueError("Malformed URL")
return destination_hash, path
def detect_partials(self):
for w in self.attr_maps:
o = w._original_widget
if hasattr(o, "partial_hash"):
RNS.log(f"Found partial: {o.partial_hash} / {o.partial_url} / {o.partial_refresh}")
partial = {"hash": o.partial_hash, "id": o.partial_id, "url": o.partial_url, "fields": o.partial_fields,
"refresh": o.partial_refresh, "content": None, "updated": None, "update_requested": None, "request_id": None,
"destination": None, "link": None, "pile": o, "attr_maps": None, "failed": False, "pr_throttle": 0}
self.page_partials[o.partial_hash] = partial
if len(self.page_partials) > 0: self.start_partial_updater()
def partial_failed(self, request_receipt):
RNS.log("Loading page partial failed", RNS.LOG_ERROR)
for pid in self.page_partials:
partial = self.page_partials[pid]
if partial["request_id"] == request_receipt.request_id:
try:
partial["updated"] = time.time()
partial["request_id"] = None
partial["content"] = None
partial["attr_maps"] = None
url = partial["url"]
pile = partial["pile"]
pile.contents = [(urwid.Text(f"Could not load partial {url}: The resource transfer failed"), pile.options())]
self.refresh_page_layout()
except Exception as e:
RNS.log(f"Error in partial failed callback: {e}", RNS.LOG_ERROR)
RNS.trace_exception(e)
def partial_progressed(self, request_receipt):
pass
def partial_received(self, request_receipt):
for pid in self.page_partials:
partial = self.page_partials[pid]
if partial["request_id"] == request_receipt.request_id:
try:
partial["updated"] = partial["update_requested"]
partial["request_id"] = None
partial["content"] = request_receipt.response.decode("utf-8").rstrip()
partial["attr_maps"] = markup_to_attrmaps(strip_modifiers(partial["content"]), url_delegate=self, fg_color=self.page_foreground_color, bg_color=self.page_background_color)
pile = partial["pile"]
pile.contents = [(e, pile.options()) for e in partial["attr_maps"]]
self.refresh_page_layout()
except Exception as e:
RNS.trace_exception(e)
def __load_partial(self, partial):
if partial["failed"] == True: return
try: partial_destination_hash, path = self.parse_url(partial["url"])
except Exception as e:
RNS.log(f"Could not parse partial URL: {e}", RNS.LOG_ERROR)
partial["failed"] = True
pile = partial["pile"]
url = partial["url"]
pile.contents = [(urwid.Text(f"Could not load partial {url}: {e}"), pile.options())]
self.refresh_page_layout()
return
if partial_destination_hash != self.loopback and not RNS.Transport.has_path(partial_destination_hash):
if time.time() <= partial["pr_throttle"]: return
else:
partial["pr_throttle"] = time.time()+15
RNS.log(f"Requesting path for partial: {partial_destination_hash} / {path}", RNS.LOG_EXTREME)
RNS.Transport.request_path(partial_destination_hash)
pr_time = time.time()+self.path_timeout(partial_destination_hash)
while not RNS.Transport.has_path(partial_destination_hash):
if time.time() > pr_time: return
time.sleep(0.25)
for pid in self.page_partials:
other_partial = self.page_partials[pid]
if other_partial["link"]:
existing_link = other_partial["link"]
if existing_link.destination.hash == partial_destination_hash and existing_link.status == RNS.Link.ACTIVE:
RNS.log(f"Re-using existing link: {existing_link}", RNS.LOG_EXTREME)
partial["link"] = existing_link
break
if not partial["link"] or partial["link"].status == RNS.Link.CLOSED:
RNS.log(f"Establishing link for partial: {partial_destination_hash} / {path}", RNS.LOG_EXTREME)
identity = RNS.Identity.recall(partial_destination_hash)
destination = RNS.Destination(identity, RNS.Destination.OUT, RNS.Destination.SINGLE, self.app_name, self.aspects)
def established(link):
RNS.log(f"Link established for partial: {partial_destination_hash} / {path}", RNS.LOG_EXTREME)
def closed(link):
RNS.log(f"Link closed for partial: {partial_destination_hash} / {path}", RNS.LOG_EXTREME)
partial["link"] = None
partial["link"] = RNS.Link(destination, established_callback = established, closed_callback = closed)
timeout = time.time()+max(self.timeout, partial["link"].establishment_timeout)
while partial["link"].status != RNS.Link.ACTIVE and time.time() < timeout: time.sleep(0.1)
if partial["link"] and partial["link"].status == RNS.Link.ACTIVE and partial["request_id"] == None:
RNS.log(f"Sending request for partial: {partial_destination_hash} / {path}", RNS.LOG_EXTREME)
receipt = partial["link"].request(path, data=self.__get_partial_request_data(partial), response_callback = self.partial_received,
failed_callback = self.partial_failed, progress_callback = self.partial_progressed)
if receipt: partial["request_id"] = receipt.request_id
else: RNS.log(f"Partial request failed", RNS.LOG_ERROR)
def __get_partial_request_data(self, partial):
request_data = None
if partial["fields"] != None:
link_data = partial["fields"]
link_fields = []
request_data = {}
all_fields = True if "*" in link_data else False
for e in link_data:
if "=" in e:
c = e.split("=")
if len(c) == 2:
request_data["var_"+str(c[0])] = str(c[1])
else:
link_fields.append(e)
def recurse_down(w):
if isinstance(w, list):
for t in w:
recurse_down(t)
elif isinstance(w, tuple):
for t in w:
recurse_down(t)
elif hasattr(w, "contents"):
recurse_down(w.contents)
elif hasattr(w, "original_widget"):
recurse_down(w.original_widget)
elif hasattr(w, "_original_widget"):
recurse_down(w._original_widget)
else:
if hasattr(w, "field_name") and (all_fields or w.field_name in link_fields):
field_key = "field_" + w.field_name
if isinstance(w, urwid.Edit):
request_data[field_key] = w.edit_text
elif isinstance(w, urwid.RadioButton):
if w.state:
user_data = getattr(w, "field_value", None)
if user_data is not None:
request_data[field_key] = user_data
elif isinstance(w, urwid.CheckBox):
user_data = getattr(w, "field_value", "1")
if w.state:
existing_value = request_data.get(field_key, '')
if existing_value:
# Concatenate the new value with the existing one
request_data[field_key] = existing_value + ',' + user_data
else:
# Initialize the field with the current value
request_data[field_key] = user_data
else:
pass # do nothing if checkbox is not check
recurse_down(self.attr_maps)
RNS.log("Including request data: "+str(request_data), RNS.LOG_EXTREME)
return request_data
def start_partial_updater(self):
if not self.updater_running: self.update_partials()
def handle_partial_updates(self, partial_ids):
RNS.log(f"Update partials: {partial_ids}")
def job():
for pid in self.page_partials:
try:
partial = self.page_partials[pid]
if partial["id"] in partial_ids:
partial["update_requested"] = time.time()
self.__load_partial(partial)
except Exception as e: RNS.log(f"Error updating page partial: {e}", RNS.LOG_ERROR)
threading.Thread(target=job, daemon=True).start()
def update_partials(self, loop=None, user_data=None):
with self.partial_updater_lock:
def job():
for pid in self.page_partials:
try:
partial = self.page_partials[pid]
if partial["failed"]: continue
if not partial["updated"] or (partial["refresh"] != None and time.time() > partial["updated"]+partial["refresh"]):
partial["update_requested"] = time.time()
self.__load_partial(partial)
except Exception as e: RNS.log(f"Error updating page partial: {e}", RNS.LOG_ERROR)
threading.Thread(target=job, daemon=True).start()
if len(self.page_partials) > 0:
self.updater_running = True
self.app.ui.loop.set_alarm_in(1, self.update_partials)
else:
self.updater_running = False
def identify(self):
if self.link != None:
if self.link.status == RNS.Link.ACTIVE:
self.link.identify(self.auth_identity)
def disconnect(self):
if self.link != None:
self.link.teardown()
self.request_data = None
self.attr_maps = []
self.status = Browser.DISCONECTED
self.response_progress = 0
self.response_speed = None
self.progress_updated_at = None
self.previous_progress = 0
self.response_size = None
self.response_transfer_size = None
self.history = []
self.history_ptr = 0
self.history_inc = False
self.history_dec = False
self.update_display()
def retrieve_url(self, url, request_data = None):
components = url.split("`")
if len(components) == 2:
url = components[0]
try:
link_fields_str = components[1]
if link_fields_str != "":
if not request_data: request_data = {}
for e in link_fields_str.split("|"):
if "=" in e:
c = e.split("=")
if len(c) == 2:
request_data["var_"+str(c[0])] = str(c[1])
except Exception as e:
RNS.log(f"Malformed URL: {e}", RNS.LOG_WARNING)
raise ValueError(f"Malformed URL: {e}")
self.previous_destination_hash = self.destination_hash
self.previous_path = self.path
destination_hash = None
path = None
components = url.split(":")
if len(components) == 1:
if len(components[0]) == (RNS.Reticulum.TRUNCATED_HASHLENGTH//8)*2:
try: destination_hash = bytes.fromhex(components[0])
except Exception as e: raise ValueError("Malformed URL")
path = Browser.DEFAULT_PATH
else: raise ValueError("Malformed URL")
elif len(components) == 2:
if len(components[0]) == (RNS.Reticulum.TRUNCATED_HASHLENGTH//8)*2:
try: destination_hash = bytes.fromhex(components[0])
except Exception as e: raise ValueError("Malformed URL")
path = components[1]
if len(path) == 0: path = Browser.DEFAULT_PATH
else:
if len(components[0]) == 0:
if self.destination_hash != None:
destination_hash = self.destination_hash
path = components[1]
if len(path) == 0: path = Browser.DEFAULT_PATH
else: raise ValueError("Malformed URL")
else: raise ValueError("Malformed URL")
else: raise ValueError("Malformed URL")
if destination_hash != None and path != None:
if path.startswith("/file/"):
if destination_hash != self.loopback:
if destination_hash == self.destination_hash:
self.download_file(destination_hash, path, data=request_data)
else:
RNS.log("Cannot request file download from a node that is not currently connected.", RNS.LOG_ERROR)
RNS.log("The requested URL was: "+str(url), RNS.LOG_ERROR)
else:
self.download_local_file(path)
else:
self.set_destination_hash(destination_hash)
self.set_path(path)
self.set_request_data(request_data)
self.load_page()
def set_destination_hash(self, destination_hash):
if len(destination_hash) == RNS.Identity.TRUNCATED_HASHLENGTH//8:
self.destination_hash = destination_hash
return True
else:
return False
def set_path(self, path):
self.path = path
def set_request_data(self, request_data):
self.request_data = request_data
def set_timeout(self, timeout):
self.timeout = timeout
def path_timeout(self, destination_hash):
return max(self.timeout + self.app.rns.get_first_hop_timeout(destination_hash), self.app.rns.get_medium_path_timeout())
def download_local_file(self, path):
try:
file_path = self.app.filespath+path.replace("/file", "", 1)
if os.path.isfile(file_path):
file_name = os.path.basename(file_path)
file_destination = self.app.downloads_path+"/"+file_name
counter = 0
while os.path.isfile(file_destination):
counter += 1
file_destination = self.app.downloads_path+"/"+file_name+"."+str(counter)
fs = open(file_path, "rb")
fd = open(file_destination, "wb")
fd.write(fs.read())
fd.close()
fs.close()
self.saved_file_name = file_destination.replace(self.app.downloads_path+"/", "", 1)
try: self.saved_file_size = os.path.getsize(file_destination)
except: self.saved_file_size = 0
self.update_display()
else:
RNS.log("The requested local download file does not exist: "+str(file_path), RNS.LOG_ERROR)
self.status = Browser.DONE
self.response_progress = 0
self.response_speed = None
self.progress_updated_at = None
self.previous_progress = 0
except Exception as e:
RNS.log("An error occurred while handling file response. The contained exception was: "+str(e), RNS.LOG_ERROR)
def download_file(self, destination_hash, path, data=None):
if self.link == None or self.link.destination.hash != self.destination_hash:
if not RNS.Transport.has_path(self.destination_hash):
self.status = Browser.NO_PATH
self.update_display()
RNS.Transport.request_path(self.destination_hash)
self.status = Browser.PATH_REQUESTED
self.update_display()
pr_time = time.time()+self.path_timeout(self.destination_hash)
while not RNS.Transport.has_path(self.destination_hash):
if time.time() > pr_time:
self.request_timeout()
return
time.sleep(0.25)
self.status = Browser.ESTABLISHING_LINK
self.update_display()
identity = RNS.Identity.recall(self.destination_hash)
destination = RNS.Destination(
identity,
RNS.Destination.OUT,
RNS.Destination.SINGLE,
self.app_name,
self.aspects
)
self.link = RNS.Link(destination, established_callback = self.link_established, closed_callback = self.link_closed)
while self.status == Browser.ESTABLISHING_LINK:
time.sleep(0.1)
if self.status != Browser.LINK_ESTABLISHED:
return
self.update_display()
if self.link != None and self.link.destination.hash == self.destination_hash:
# Send the request
self.status = Browser.REQUESTING
self.response_progress = 0
self.response_speed = None
self.progress_updated_at = None
self.previous_progress = 0
self.response_size = None
self.response_transfer_size = None
self.saved_file_name = None
self.saved_file_size = 0
self.update_display()
receipt = self.link.request(
path,
data = data,
response_callback = self.file_received,
failed_callback = self.request_failed,
progress_callback = self.response_progressed
)
if receipt:
self.last_request_receipt = receipt
self.last_request_id = receipt.request_id
self.status = Browser.REQUEST_SENT
self.update_display()
else:
self.link.teardown()
def write_history(self):
entry = [self.destination_hash, self.path, self.request_data]
self.history.insert(self.history_ptr, entry)
self.history_ptr += 1
if len(self.history) > self.history_ptr:
self.history = self.history[:self.history_ptr]
def back(self):
target_ptr = self.history_ptr-1
if not self.history_inc and not self.history_dec:
if target_ptr > 0:
self.history_dec = True
entry = self.history[target_ptr-1]
url = RNS.hexrep(entry[0], delimit=False)+":"+entry[1]
self.history_ptr = target_ptr
self.retrieve_url(url, request_data=entry[2])
def forward(self):
target_ptr = self.history_ptr+1
if not self.history_inc and not self.history_dec:
if target_ptr <= len(self.history):
self.history_dec = True
entry = self.history[target_ptr-1]
url = RNS.hexrep(entry[0], delimit=False)+":"+entry[1]
self.history_ptr = target_ptr
self.retrieve_url(url, request_data=entry[2])
def reload(self):
if not self.reloading and self.status == Browser.DONE:
self.reloading = True
self.uncache_page(self.current_url())
self.load_page()
def copy_url(self):
url = self.current_url()
if not url:
return
if not osc52_copy(url):
return
try:
notice = urwid.AttrMap(urwid.Pile([urwid.Divider(self.g["divider1"]), urwid.Text("Copied URL to clipboard")]), "browser_controls")
if self.page_background_color != None or self.page_foreground_color != None:
style_name = make_style(default_state(fg=self.page_foreground_color, bg=self.page_background_color))
notice.set_attr_map({None: style_name})
self.browser_footer = notice
self.frame.contents["footer"] = (self.browser_footer, self.frame.options())
def restore_footer(loop, user_data):
if self.status == Browser.DONE:
self.browser_footer = self.make_status_widget()
else:
self.browser_footer = urwid.Text("")
if self.frame is not None:
self.frame.contents["footer"] = (self.browser_footer, self.frame.options())
self.app.ui.loop.set_alarm_in(2.0, restore_footer)
except Exception as e:
RNS.log("Could not update footer after URL copy: "+str(e), RNS.LOG_ERROR)
def close_dialogs(self):
options = self.delegate.columns.options(urwid.WEIGHT, self.delegate.right_area_width)
self.delegate.columns.contents[1] = (self.display_widget, options)
def url_dialog(self):
e_url = UrlEdit(caption="URL : ", edit_text=self.current_url())
def dismiss_dialog(sender):
self.close_dialogs()
def confirmed(sender):
try:
entered_url = e_url.get_edit_text().strip()
if not "`" in entered_url and ":" in entered_url:
pos = entered_url.find("|")
if pos > 0: entered_url = entered_url[:pos]+"`"+entered_url[pos+1:]
self.retrieve_url(entered_url)
except Exception as e:
self.browser_footer = urwid.Text("Could not open link: "+str(e))
self.frame.contents["footer"] = (self.browser_footer, self.frame.options())
self.close_dialogs()
dialog = UrlDialogLineBox(
urwid.Pile([
e_url,
urwid.Columns([
(urwid.WEIGHT, 0.45, urwid.Button("Cancel", on_press=dismiss_dialog)),
(urwid.WEIGHT, 0.1, urwid.Text("")),
(urwid.WEIGHT, 0.45, urwid.Button("Go", on_press=confirmed)),
])
]), title="Enter URL"
)
e_url.confirmed = confirmed
dialog.confirmed = confirmed
dialog.delegate = self
bottom = self.display_widget
overlay = urwid.Overlay(
dialog,
bottom,
align=urwid.CENTER,
width=(urwid.RELATIVE, 65),
valign=urwid.MIDDLE,
height=urwid.PACK,
left=2,
right=2,
)
options = self.delegate.columns.options(urwid.WEIGHT, self.delegate.right_area_width)
self.delegate.columns.contents[1] = (overlay, options)
self.delegate.columns.focus_position = 1
def save_node_dialog(self):
if self.destination_hash != None:
try:
def dismiss_dialog(sender):
self.close_dialogs()
display_name = RNS.Identity.recall_app_data(self.destination_hash)
disp_str = ""
if display_name != None:
display_name = display_name.decode("utf-8")
disp_str = " \""+display_name+"\""
def confirmed(sender):
node_entry = DirectoryEntry(self.destination_hash, display_name=display_name, hosts_node=True)
self.app.directory.remember(node_entry)
self.app.ui.main_display.sub_displays.network_display.directory_change_callback()
self.close_dialogs()
dialog = UrlDialogLineBox(
urwid.Pile([
urwid.Text("Save connected node"+disp_str+" "+RNS.prettyhexrep(self.destination_hash)+" to Known Nodes?\n"),
urwid.Columns([
(urwid.WEIGHT, 0.45, urwid.Button("Cancel", on_press=dismiss_dialog)),
(urwid.WEIGHT, 0.1, urwid.Text("")),
(urwid.WEIGHT, 0.45, urwid.Button("Save", on_press=confirmed)),
])
]), title="Save Node"
)
dialog.confirmed = confirmed
dialog.delegate = self
bottom = self.display_widget
overlay = urwid.Overlay(
dialog,
bottom,
align=urwid.CENTER,
width=(urwid.RELATIVE, 50),
valign=urwid.MIDDLE,
height=urwid.PACK,
left=2,
right=2,
)
options = self.delegate.columns.options(urwid.WEIGHT, self.delegate.right_area_width)
self.delegate.columns.contents[1] = (overlay, options)
self.delegate.columns.focus_position = 1
except Exception as e:
pass
def load_page(self):
if self.request_data == None:
cached = self.get_cached(self.current_url())
else:
cached = None
if cached:
self.status = Browser.DONE
self.page_data = cached
self.markup = self.page_data.decode("utf-8")
self.page_background_color = None
bgpos = self.markup.find("#!bg=")
if bgpos >= 0:
endpos = self.markup.find("\n", bgpos)
if endpos-(bgpos+5) == 3:
bg = self.markup[bgpos+5:endpos]
self.page_background_color = bg
elif endpos-(bgpos+5) == 6:
bg = self.markup[bgpos+5:endpos]
self.page_background_color = bg
self.page_foreground_color = None
fgpos = self.markup.find("#!fg=")
if fgpos >= 0:
endpos = self.markup.find("\n", fgpos)
if endpos-(fgpos+5) == 3:
fg = self.markup[fgpos+5:endpos]
self.page_foreground_color = fg
elif endpos-(fgpos+5) == 6:
fg = self.markup[fgpos+5:endpos]
self.page_foreground_color = fg
try: self.attr_maps = markup_to_attrmaps(strip_modifiers(self.markup), url_delegate=self, fg_color=self.page_foreground_color, bg_color=self.page_background_color)
except Exception as e: self.attr_maps = [urwid.AttrMap(urwid.Text(f"Could not render page: {e}"), "browser_inactive")]
self.response_progress = 0
self.response_speed = None
self.progress_updated_at = None
self.previous_progress = 0
self.response_size = None
self.response_transfer_size = None
self.saved_file_name = None
self.saved_file_size = 0
self.loaded_from_cache = True
self.update_display()
if not self.history_inc and not self.history_dec and not self.reloading:
self.write_history()
else:
self.history_dec = False
self.history_inc = False
self.reloading = False
if self.request_data and "var_anchor" in self.request_data:
def df(loop, user_data): self._jump_to_anchor(self.request_data["var_anchor"])
nomadnet.NomadNetworkApp.get_shared_instance().ui.loop.set_alarm_in(0.0, df)
else:
if self.destination_hash != self.loopback:
load_thread = threading.Thread(target=self.__load)
load_thread.setDaemon(True)
load_thread.start()
else:
RNS.log("Browser handling local page: "+str(self.path), RNS.LOG_DEBUG)
page_path = self.app.pagespath+self.path.replace("/page", "", 1)
page_data = b"The requested local page did not exist in the file system"
if os.path.isfile(page_path):
if os.access(page_path, os.X_OK):
if self.request_data != None:
env_map = self.request_data
else:
env_map = {}
if "PATH" in os.environ:
env_map["PATH"] = os.environ["PATH"]
generated = subprocess.run([page_path], stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, env=env_map)
page_data = generated.stdout
else:
file = open(page_path, "rb")
page_data = file.read()
file.close()
self.status = Browser.DONE
self.page_data = page_data
self.markup = self.page_data.decode("utf-8")
self.page_background_color = None
bgpos = self.markup.find("#!bg=")
if bgpos >= 0:
endpos = self.markup.find("\n", bgpos)
if endpos-(bgpos+5) == 3:
bg = self.markup[bgpos+5:endpos]
self.page_background_color = bg
elif endpos-(bgpos+5) == 6:
bg = self.markup[bgpos+5:endpos]
self.page_background_color = bg
self.page_foreground_color = None
fgpos = self.markup.find("#!fg=")
if fgpos >= 0:
endpos = self.markup.find("\n", fgpos)
if endpos-(fgpos+5) == 3:
fg = self.markup[fgpos+5:endpos]
self.page_foreground_color = fg
elif endpos-(fgpos+5) == 6:
fg = self.markup[fgpos+5:endpos]
self.page_foreground_color = fg
try: self.attr_maps = markup_to_attrmaps(strip_modifiers(self.markup), url_delegate=self, fg_color=self.page_foreground_color, bg_color=self.page_background_color)
except Exception as e: self.attr_maps = [urwid.AttrMap(urwid.Text(f"Could not render page: {e}"), "browser_inactive")]
self.response_progress = 0
self.response_speed = None
self.progress_updated_at = None
self.previous_progress = 0
self.response_size = None
self.response_transfer_size = None
self.saved_file_name = None
self.saved_file_size = 0
self.loaded_from_cache = False
self.update_display()
if not self.history_inc and not self.history_dec and not self.reloading:
self.write_history()
else:
self.history_dec = False
self.history_inc = False
self.reloading = False
if self.request_data and "var_anchor" in self.request_data:
def df(loop, user_data): self._jump_to_anchor(self.request_data["var_anchor"])
nomadnet.NomadNetworkApp.get_shared_instance().ui.loop.set_alarm_in(0.0, df)
def __load(self):
# If an established link exists, but it doesn't match the target
# destination, we close and clear it.
if self.link != None and (self.link.destination.hash != self.destination_hash or self.link.status != RNS.Link.ACTIVE):
self.link.teardown()
self.link = None
# If no link to the destination exists, we create one.
if self.link == None:
if not RNS.Transport.has_path(self.destination_hash):
self.status = Browser.NO_PATH
self.update_display()
RNS.Transport.request_path(self.destination_hash)
self.status = Browser.PATH_REQUESTED
self.update_display()
pr_time = time.time()+self.path_timeout(self.destination_hash)
while not RNS.Transport.has_path(self.destination_hash):
if time.time() > pr_time:
self.request_timeout()
return
time.sleep(0.25)
self.status = Browser.ESTABLISHING_LINK
self.update_display()
identity = RNS.Identity.recall(self.destination_hash)
destination = RNS.Destination(
identity,
RNS.Destination.OUT,
RNS.Destination.SINGLE,
self.app_name,
self.aspects
)
self.link = RNS.Link(destination, established_callback = self.link_established, closed_callback = self.link_closed)
while self.status == Browser.ESTABLISHING_LINK:
time.sleep(0.1)
if self.status != Browser.LINK_ESTABLISHED:
return
self.update_display()
# Send the request
self.status = Browser.REQUESTING
self.response_progress = 0
self.response_speed = None
self.progress_updated_at = None
self.previous_progress = 0
self.response_size = None
self.response_transfer_size = None
self.saved_file_name = None
self.saved_file_size = 0
self.update_display()
receipt = self.link.request(
self.path,
data = self.request_data,
response_callback = self.response_received,
failed_callback = self.request_failed,
progress_callback = self.response_progressed
)
if receipt:
self.last_request_receipt = receipt
self.last_request_id = receipt.request_id
self.status = Browser.REQUEST_SENT
self.update_display()
else:
self.link.teardown()
def link_established(self, link):
self.status = Browser.LINK_ESTABLISHED
if self.app.directory.should_identify_on_connect(self.destination_hash):
RNS.log("Link established, identifying to remote system...", RNS.LOG_VERBOSE)
self.link.identify(self.app.identity)
def link_closed(self, link):
if self.status == Browser.DISCONECTED or self.status == Browser.DONE:
self.link = None
elif self.status == Browser.ESTABLISHING_LINK:
self.link_establishment_timeout()
else:
self.link = None
self.status = Browser.REQUEST_FAILED
self.update_display()
def link_establishment_timeout(self):
self.status = Browser.LINK_TIMEOUT
self.response_progress = 0
self.response_speed = None
self.progress_updated_at = None
self.previous_progress = 0
self.response_size = None
self.response_transfer_size = None
self.link = None
self.update_display()
def response_received(self, request_receipt):
try:
self.status = Browser.DONE
self.page_data = request_receipt.response
self.markup = self.page_data.decode("utf-8")
self.page_background_color = None
bgpos = self.markup.find("#!bg=")
if bgpos >= 0:
endpos = self.markup.find("\n", bgpos)
if endpos-(bgpos+5) == 3:
bg = self.markup[bgpos+5:endpos]
self.page_background_color = bg
elif endpos-(bgpos+5) == 6:
bg = self.markup[bgpos+5:endpos]
self.page_background_color = bg
self.page_foreground_color = None
fgpos = self.markup.find("#!fg=")
if fgpos >= 0:
endpos = self.markup.find("\n", fgpos)
if endpos-(fgpos+5) == 3:
fg = self.markup[fgpos+5:endpos]
self.page_foreground_color = fg
elif endpos-(fgpos+5) == 6:
fg = self.markup[fgpos+5:endpos]
self.page_foreground_color = fg
try: self.attr_maps = markup_to_attrmaps(strip_modifiers(self.markup), url_delegate=self, fg_color=self.page_foreground_color, bg_color=self.page_background_color)
except Exception as e: self.attr_maps = [urwid.AttrMap(urwid.Text(f"Could not render page: {e}"), "browser_inactive")]
self.response_progress = 0
self.response_speed = None
self.progress_updated_at = None
self.previous_progress = 0
self.loaded_from_cache = False
# Simple header handling. Should be expanded when more
# header tags are added.
cache_time = Browser.DEFAULT_CACHE_TIME
if self.markup[:4] == "#!c=":
try:
endpos = self.markup.find("\n")
if endpos == -1:
endpos = len(self.markup)
cache_time = int(self.markup[4:endpos])
except Exception as e: RNS.log(f"Invalid page cache time header: {e}", RNS.LOG_DEBUG)
self.update_display()
if not self.history_inc and not self.history_dec and not self.reloading:
self.write_history()
else:
self.history_dec = False
self.history_inc = False
self.reloading = False
if cache_time == 0:
RNS.log("Received page "+str(self.current_url())+", not caching due to header.", RNS.LOG_DEBUG)
else:
RNS.log("Received page "+str(self.current_url())+", caching for %.3f hours." % (cache_time/60/60), RNS.LOG_DEBUG)
self.cache_page(cache_time)
if self.request_data and "var_anchor" in self.request_data:
def df(loop, user_data): self._jump_to_anchor(self.request_data["var_anchor"])
nomadnet.NomadNetworkApp.get_shared_instance().ui.loop.set_alarm_in(0.0, df)
except Exception as e:
RNS.log("An error occurred while handling response. The contained exception was: "+str(e))
def uncache_page(self, url):
url_hash = self.url_hash(url)
files = os.listdir(self.app.cachepath)
for file in files:
if file.startswith(url_hash):
cachefile = self.app.cachepath+"/"+file
os.unlink(cachefile)
RNS.log("Removed "+str(cachefile)+" from cache.", RNS.LOG_DEBUG)
def get_cached(self, url):
url_hash = self.url_hash(url)
files = os.listdir(self.app.cachepath)
for file in files:
cachepath = self.app.cachepath+"/"+file
try:
components = file.split("_")
if len(components) == 2 and len(components[0]) == 64 and len(components[1]) > 0:
expires = float(components[1])
if time.time() > expires:
RNS.log("Removing stale cache entry "+str(file), RNS.LOG_DEBUG)
os.unlink(cachepath)
else:
if file.startswith(url_hash):
RNS.log("Found "+str(file)+" in cache.", RNS.LOG_DEBUG)
RNS.log("Returning cached page", RNS.LOG_DEBUG)
file = open(cachepath, "rb")
data = file.read()
file.close()
return data
except Exception as e:
RNS.log("Error while parsing cache entry "+str(cachepath)+", removing it.", RNS.LOG_ERROR)
RNS.log("The contained exception was: "+str(e), RNS.LOG_ERROR)
try:
os.unlink(cachepath)
except Exception as e:
RNS.log("Additionally, an exception occurred while unlinking the entry: "+str(e), RNS.LOG_ERROR)
RNS.log("You will probably need to remove this entry manually by deleting the file: "+str(cachepath), RNS.LOG_ERROR)
return None
def clean_cache(self):
files = os.listdir(self.app.cachepath)
for file in files:
cachepath = self.app.cachepath+"/"+file
try:
components = file.split("_")
if len(components) == 2 and len(components[0]) == 64 and len(components[1]) > 0:
expires = float(components[1])
if time.time() > expires:
RNS.log("Removing stale cache entry "+str(file), RNS.LOG_DEBUG)
os.unlink(cachepath)
except Exception as e:
pass
def cache_page(self, cache_time):
url_hash = self.url_hash(self.current_url())
if url_hash == None:
RNS.log("Could not cache page "+str(self.current_url()), RNS.LOG_ERROR)
else:
try:
self.uncache_page(self.current_url())
cache_expires = time.time()+cache_time
filename = url_hash+"_"+str(cache_expires)
cachefile = self.app.cachepath+"/"+filename
file = open(cachefile, "wb")
file.write(self.page_data)
file.close()
RNS.log("Cached page "+str(self.current_url())+" to "+str(cachefile), RNS.LOG_DEBUG)
except Exception as e:
RNS.log("Could not write cache file for page "+str(self.current_url()), RNS.LOG_ERROR)
RNS.log("The contained exception was: "+str(e), RNS.LOG_ERROR)
def file_received(self, request_receipt):
try:
if isinstance(request_receipt.response, bytes):
markup = request_receipt.response.decode("utf-8")
self.attr_maps = markup_to_attrmaps(strip_modifiers(markup), url_delegate=self, fg_color=self.page_foreground_color, bg_color=self.page_background_color)
elif type(request_receipt.response) == io.BufferedReader:
if request_receipt.metadata != None:
file_name = ConversationMessage.safe_attachment_name(request_receipt.metadata["name"])
file_handle = request_receipt.response
file_destination = self.app.downloads_path+"/"+file_name
counter = 0
while os.path.isfile(file_destination):
counter += 1
file_destination = self.app.downloads_path+"/"+file_name+"."+str(counter)
shutil.move(file_handle.name, file_destination)
self.saved_file_name = file_destination.replace(self.app.downloads_path+"/", "", 1)
try: self.saved_file_size = os.path.getsize(file_destination)
except: self.saved_file_size = 0
self.file_saved_at = time.time()
else:
file_name = request_receipt.response[0]
file_data = request_receipt.response[1]
file_destination_name = ConversationMessage.safe_attachment_name(file_name)
file_destination = self.app.downloads_path+"/"+file_destination_name
counter = 0
while os.path.isfile(file_destination):
counter += 1
file_destination = self.app.downloads_path+"/"+file_destination_name+"."+str(counter)
fh = open(file_destination, "wb")
fh.write(file_data)
fh.close()
self.saved_file_name = file_destination.replace(self.app.downloads_path+"/", "", 1)
try: self.saved_file_size = os.path.getsize(file_destination)
except: self.saved_file_size = 0
self.status = Browser.DONE
self.response_progress = 0
self.response_speed = None
self.progress_updated_at = None
self.previous_progress = 0
self.update_display()
except Exception as e:
RNS.log("An error occurred while handling file response. The contained exception was: "+str(e), RNS.LOG_ERROR)
def request_failed(self, request_receipt=None):
if request_receipt != None:
if request_receipt.request_id == self.last_request_id:
self.status = Browser.REQUEST_FAILED
self.response_progress = 0
self.response_speed = None
self.progress_updated_at = None
self.previous_progress = 0
self.response_size = None
self.response_transfer_size = None
self.update_display()
if self.link != None:
try: self.link.teardown()
except Exception as e: pass
else:
self.status = Browser.REQUEST_FAILED
self.response_progress = 0
self.response_speed = None
self.progress_updated_at = None
self.previous_progress = 0
self.response_size = None
self.response_transfer_size = None
self.update_display()
if self.link != None:
try:
self.link.teardown()
except Exception as e:
pass
def request_timeout(self, request_receipt=None):
self.status = Browser.REQUEST_TIMEOUT
self.response_progress = 0
self.response_speed = None
self.progress_updated_at = None
self.previous_progress = 0
self.response_size = None
self.response_transfer_size = None
self.update_display()
if self.link != None:
try:
self.link.teardown()
except Exception as e:
pass
def response_progressed(self, request_receipt):
self.response_progress = request_receipt.progress
self.response_time = request_receipt.get_response_time()
self.response_size = request_receipt.response_size
self.response_transfer_size = request_receipt.response_transfer_size
now = time.time()
if self.progress_updated_at == None: self.progress_updated_at = now
if now > self.progress_updated_at+0.5:
td = now - self.progress_updated_at
pd = self.response_progress - self.previous_progress
bd = pd*self.response_transfer_size
self.response_speed = (bd/td)*8
self.previous_progress = self.response_progress
self.progress_updated_at = now
self.update_display()
def last_response_speed(self):
if self.response_transfer_size and self.response_time:
return (self.response_transfer_size/max(self.response_time, 0.0001))*8
else: return None
def status_text(self):
if self.status == Browser.DONE and self.response_transfer_size != None:
if self.response_time != None: response_time_str = "{:.2f}".format(self.response_time)
else: response_time_str = "None"
if self.saved_file_name == None: resp_size = self.response_size
else: resp_size = self.saved_file_size
if resp_size: stats_string = " "+self.g["page"]+size_str(resp_size)
else: stats_string = " "
if self.response_transfer_size:
stats_string += " "+self.g["arrow_d"]+size_str(self.response_transfer_size)+" in "+response_time_str
if self.response_transfer_size and self.response_time:
stats_string += "s "+self.g["speed"]+size_str(self.response_transfer_size/self.response_time, suffix="b")+"/s"
elif self.loaded_from_cache:
stats_string = " (cached)"
else:
stats_string = ""
if self.status == Browser.NO_PATH:
return "No path to destination known"
elif self.status == Browser.PATH_REQUESTED:
return "Path requested, waiting for path..."
elif self.status == Browser.ESTABLISHING_LINK:
return "Establishing link..."
elif self.status == Browser.LINK_TIMEOUT:
return "Link establishment timed out"
elif self.status == Browser.LINK_ESTABLISHED:
return "Link established"
elif self.status == Browser.REQUESTING:
return "Sending request..."
elif self.status == Browser.REQUEST_SENT:
return "Request sent, awaiting response..."
elif self.status == Browser.REQUEST_FAILED:
return "Request failed"
elif self.status == Browser.REQUEST_TIMEOUT:
return "Request timed out"
elif self.status == Browser.RECEIVING_RESPONSE:
return "Receiving response..."
elif self.status == Browser.DONE:
if self.saved_file_name == None: return "Done"+stats_string
else: return "Saved "+str(self.saved_file_name)+stats_string
elif self.status == Browser.DISCONECTED:
return "Disconnected"
else:
return "Browser Status Unknown"
class ResponseProgressBar(urwid.ProgressBar):
def __init__(self, empty, full, current=None, done=None, satt=None, owner=None):
super().__init__(empty, full, current=current, done=done, satt=satt)
self.owner = owner
def get_text(self):
if self.owner.response_speed: speed_str = " "+RNS.prettyspeed(self.owner.response_speed)
else: speed_str = ""
return "Receiving response "+super().get_text().replace(" %", "%")+speed_str
# A convenience function for printing a human-
# readable file size
def size_str(num, suffix='B'):
units = ['','K','M','G','T','P','E','Z']
last_unit = 'Y'
if suffix == 'b':
num *= 8
units = ['','K','M','G','T','P','E','Z']
last_unit = 'Y'
for unit in units:
if abs(num) < 1000.0:
if unit == "":
return "%.0f%s%s" % (num, unit, suffix)
else:
return "%.2f%s%s" % (num, unit, suffix)
num /= 1000.0
return "%.2f%s%s" % (num, last_unit, suffix)
class UrlDialogLineBox(urwid.LineBox):
def keypress(self, size, key):
if key == "esc":
self.delegate.close_dialogs()
else:
return super(UrlDialogLineBox, self).keypress(size, key)
class UrlEdit(ReadlineMixin, urwid.Edit):
def keypress(self, size, key):
if key == "enter":
self.confirmed(self)
else:
return super(UrlEdit, self).keypress(size, key)
Reference in New Issue View Git Blame Copy Permalink