diff --git a/nomadnet/Node.py b/nomadnet/Node.py index 622d8aa..714f270 100644 --- a/nomadnet/Node.py +++ b/nomadnet/Node.py @@ -32,6 +32,7 @@ class Node: self.register_pages() self.register_files() + self.register_media() self.destination.set_link_established_callback(self.peer_connected) @@ -73,6 +74,10 @@ class Node: response_generator = self.serve_page, allow = RNS.Destination.ALLOW_ALL) + def register_media(self): + self.destination.register_request_handler("/media", response_generator = self.serve_media, + allow = RNS.Destination.ALLOW_ALL) + def register_files(self): # TODO: Deregister previously registered files # that no longer exist. @@ -153,6 +158,39 @@ class Node: RNS.log("The contained exception was: "+str(e), RNS.LOG_ERROR) return None + def serve_media(self, path, data, request_id, link_id, remote_identity, requested_at): + if not type(data) == dict: return None + if not "path" in data: return None + if not "key" in data: return None + media_path = data["path"].replace("/media/", "").lstrip("/").replace("../", "") + media_path = f"{self.app.pagespath}/{media_path}" + if not media_path.startswith(self.app.pagespath): + RNS.log(f"Invalid media request path: {media_path}", RNS.LOG_DEBUG) + return None + + RNS.log(f"Media request {RNS.prettyhexrep(request_id)} for: {media_path}", RNS.LOG_VERBOSE) + try: + self.app.peer_settings["served_media_requests"] += 1 + self.settings_dirty = True + except Exception as e: RNS.log("Could not increase served page request count", RNS.LOG_ERROR) + + request_allowed = self.request_allowed(media_path, remote_identity) + try: + if request_allowed: + file_path = media_path + file_name = os.path.basename(file_path) + RNS.log(f"Serving media: {file_path}", RNS.LOG_VERBOSE) + return [open(file_path, "rb"), {"name": file_name.encode("utf-8")}] + + else: + RNS.log("Request denied", RNS.LOG_VERBOSE) + return None + + except Exception as e: + RNS.log("Error occurred while handling request "+RNS.prettyhexrep(request_id)+" for: "+str(media_path), RNS.LOG_ERROR) + RNS.log("The contained exception was: "+str(e), RNS.LOG_ERROR) + return None + def request_allowed(self, file_path, remote_identity): allowed_path = file_path+".allowed" if not os.path.isfile(allowed_path): @@ -194,7 +232,7 @@ class Node: RNS.log("Could not increase served file request count", RNS.LOG_ERROR) file_path = path.replace("/file", self.app.filespath, 1) - file_name = path.replace("/file/", "", 1) + file_name = os.path.basename(file_path) if not self.request_allowed(file_path, remote_identity): RNS.log("Request denied", RNS.LOG_VERBOSE) return DEFAULT_NOTALLOWED.encode("utf-8") diff --git a/nomadnet/NomadNetworkApp.py b/nomadnet/NomadNetworkApp.py index 27bdaa6..1319e5f 100644 --- a/nomadnet/NomadNetworkApp.py +++ b/nomadnet/NomadNetworkApp.py @@ -112,6 +112,7 @@ class NomadNetworkApp: self.storagepath = self.configdir+"/storage" self.identitypath = self.configdir+"/storage/identity" self.cachepath = self.configdir+"/storage/cache" + self.imagecachepath = self.configdir+"/storage/cache/images" self.resourcepath = self.configdir+"/storage/resources" self.conversationpath = self.configdir+"/storage/conversations" self.directorypath = self.configdir+"/storage/directory" @@ -121,7 +122,6 @@ class NomadNetworkApp: self.pagespath = self.configdir+"/storage/pages" self.filespath = self.configdir+"/storage/files" - self.cachepath = self.configdir+"/storage/cache" self.examplespath = self.configdir+"/examples" self.downloads_path = os.path.expanduser("~/Downloads") @@ -176,6 +176,9 @@ class NomadNetworkApp: if not os.path.isdir(self.cachepath): os.makedirs(self.cachepath) + if not os.path.isdir(self.imagecachepath): + os.makedirs(self.imagecachepath) + if not os.path.isdir(self.resourcepath): os.makedirs(self.resourcepath) @@ -274,9 +277,15 @@ class NomadNetworkApp: if not "served_page_requests" in self.peer_settings: self.peer_settings["served_page_requests"] = 0 + if not "served_media_requests" in self.peer_settings: + self.peer_settings["served_media_requests"] = 0 + if not "served_file_requests" in self.peer_settings: self.peer_settings["served_file_requests"] = 0 + if not "quit_confirm" in self.peer_settings: + self.peer_settings["quit_confirm"] = True + self.peer_settings["announce_interval"] = self.announce_interval if path != self.peersettingspath: self.peer_settings_notice = "Peer settings were restored from the last backup.\nCheck your display name and propagation node." diff --git a/nomadnet/ui/TextUI.py b/nomadnet/ui/TextUI.py index bbaa73a..bdc13bb 100644 --- a/nomadnet/ui/TextUI.py +++ b/nomadnet/ui/TextUI.py @@ -7,6 +7,7 @@ import platform import nomadnet from nomadnet.ui import THEME_DARK, THEME_LIGHT from nomadnet.ui.textui import * +from nomadnet.ui.textui.images import ImageScreen from nomadnet import NomadNetworkApp COLORMODE_MONO = 1 @@ -245,7 +246,7 @@ class TextUI: for glyph in GLYPHS: self.glyphs[glyph[0]] = glyph[GLYPHSETS[glyphset]] - self.screen = urwid.raw_display.Screen() + self.screen = ImageScreen() self.screen.register_palette(self.palette) self.main_display = Main.MainDisplay(self, self.app) @@ -291,10 +292,8 @@ class TextUI: self.restore_palette = True def unhandled_input(self, key): - if key == "ctrl q": - self.main_display.show_quit_dialog() - elif key == "ctrl e": - pass + if key == "ctrl q": self.main_display.quit() + elif key == "ctrl e": pass def display_main(self, loop, user_data): self.loop.widget = self.main_display.widget diff --git a/nomadnet/ui/textui/Browser.py b/nomadnet/ui/textui/Browser.py index 5f92244..2ed21c0 100644 --- a/nomadnet/ui/textui/Browser.py +++ b/nomadnet/ui/textui/Browser.py @@ -125,8 +125,11 @@ class Browser: self.attr_maps = [] self.page_pile = None 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.build_display() self.history = [] @@ -481,6 +484,7 @@ class Browser: self.page_pile = None 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, @@ -556,6 +560,7 @@ class Browser: self.display_widget.set_attr_map({None: "browser_inactive"}) self.page_pile = None 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, @@ -626,15 +631,240 @@ class Browser: self.frame.contents["footer"] = (self.browser_footer, self.frame.options()) def update_page_display(self): + self._purge_page_images() pile = urwid.Pile(self.attr_maps) pile.automove_cursor_on_scroll = True self.page_pile = pile self.page_partials = {} + self.page_images = {} self.browser_body = urwid.AttrMap(ScrollBar(Scrollable(pile, force_forward_keypress=True), thumb_char="\u2503", trough_char=" "), "scrollbar") self.detect_partials() + self.detect_images() self.init_folds() - # render an arbitrary markup buffer (used by the page editor preview) and + 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): + try: + screen = getattr(getattr(self.app.ui, "loop", None), "screen", None) + if hasattr(screen, "purge_images"): screen.purge_images() + except Exception: pass + + def detect_images(self): + 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: o.load(resolved_image) + 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 + + if len(self.page_images) > 0: self.start_image_updater() + + def start_image_updater(self): + if not self.image_updater_running: self.update_images() + + def update_images(self, loop=None, user_data=None): + 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}" + w.load(local_path) + 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"].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: w.load(resolved_path) + + 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 @@ -1805,10 +2035,8 @@ class Browser: self.update_display() if self.link != None: - try: - self.link.teardown() - except Exception as e: - pass + try: self.link.teardown() + except Exception as e: pass else: self.status = Browser.REQUEST_FAILED self.response_progress = 0 diff --git a/nomadnet/ui/textui/Log.py b/nomadnet/ui/textui/Log.py index c301d63..8472809 100644 --- a/nomadnet/ui/textui/Log.py +++ b/nomadnet/ui/textui/Log.py @@ -57,7 +57,7 @@ class LogTerminal(urwid.WidgetWrap): def keypress(self, size, key): if key == "ctrl q": - nomadnet.NomadNetworkApp.get_shared_instance().ui.main_display.show_quit_dialog() + nomadnet.NomadNetworkApp.get_shared_instance().ui.main_display.quit() return None if key == "up": diff --git a/nomadnet/ui/textui/Main.py b/nomadnet/ui/textui/Main.py index cf1c959..e439db2 100644 --- a/nomadnet/ui/textui/Main.py +++ b/nomadnet/ui/textui/Main.py @@ -194,11 +194,11 @@ class MainDisplay(): self.menu_display.start() def quit(self, sender=None): - self.show_quit_dialog() + if self.app.peer_settings["quit_confirm"]: self.show_quit_dialog() + else: self.do_quit() def show_quit_dialog(self, sender=None): - if getattr(self, "_quit_dialog_open", False): - return + if getattr(self, "_quit_dialog_open", False): return active = self.sub_displays.active_display if active is self.sub_displays.log_display: @@ -208,7 +208,14 @@ class MainDisplay(): self.do_quit() return + dont_ask_again = urwid.CheckBox("Don't ask again") + def confirm(button=None): + if dont_ask_again.get_state(): + try: + self.app.peer_settings["quit_confirm"] = False + self.app.save_peer_settings() + except Exception: pass self.do_quit() def cancel(button=None): @@ -220,19 +227,15 @@ class MainDisplay(): (urwid.WEIGHT, 0.1, urwid.Text("")), (urwid.WEIGHT, 0.45, urwid.Button("No", on_press=cancel)), ]) - try: - - - buttons.focus_position = 2 # default to "No" - - - except Exception: - pass + try: buttons.focus_position = 2 # default to "No" + except Exception: pass pile = urwid.Pile([ urwid.Text("Are you sure you want to quit?", align="center"), urwid.Divider(), buttons, + urwid.Divider(), + dont_ask_again, ]) dialog = QuitDialogLineBox( @@ -246,7 +249,7 @@ class MainDisplay(): self.app.ui.loop.widget = urwid.Overlay( dialog, self.frame, align="center", width=44, - valign="middle", height=8, min_width=20, + valign="middle", height=7, min_width=20, ) def show_notice(self, text): diff --git a/nomadnet/ui/textui/MicronParser.py b/nomadnet/ui/textui/MicronParser.py index ff946c4..331a9a2 100644 --- a/nomadnet/ui/textui/MicronParser.py +++ b/nomadnet/ui/textui/MicronParser.py @@ -9,6 +9,8 @@ from urwid.text_layout import calc_coords from .ReadlineEdit import ReadlineEdit from RNS.Utilities.rngit.util import MarkdownToMicron from nomadnet.util import STRIP_CONTROL_RE +try: from .images import ImageWidget +except Exception: ImageWidget = None DEFAULT_FG_DARK = "ddd" DEFAULT_FG_LIGHT = "222" @@ -216,6 +218,72 @@ def parse_partial(line): except Exception as e: return None +def parse_image(line, state, url_delegate): + try: + endpos = line.rfind(")") + if endpos <= 0: return None + image_data = line[:endpos] + fields = image_data.split("`") + if len(fields) < 2: return None + + alt_text = fields[0].strip() + image_url = fields[-1].strip() + properties = fields[1:-1] + + width = None + height = None + align = None + for prop in properties: + if "=" not in prop: continue + key, _, value = prop.partition("=") + key = key.strip() + value = value.strip() + if key == "w": width = value + elif key == "h": height = value + elif key == "a": align = value + + widget = None + if image_url: + try: path = url_delegate.resolve_image(image_url) + except Exception as e: + RNS.trace_exception(e) # TODO: Remove + path = None + + if ImageWidget is not None: + try: + if not ImageWidget.placeholder_glyph: ImageWidget.placeholder_glyph = f"{url_delegate.g['image']} " + + widget = ImageWidget(path, name=alt_text) + for prop_name, prop_value in ( ("width", width), ("height", height), ("align", align) ): + if prop_value is not None: + try: + if isinstance(prop_value, str) and prop_value.isdigit(): prop_value = int(prop_value) + setattr(widget, prop_name, prop_value) + except Exception as e: + RNS.trace_exception(e) # TODO: Remove + pass + + except Exception as e: + RNS.trace_exception(e) # TODO: Remove + widget = None + + if widget is None: + placeholder = alt_text if alt_text else (image_url or "image") + widget = urwid.Text(f"[{url_delegate.g['warning']} Error loading image: {placeholder}]") + + widget.image_url = image_url + widget.image_alt = alt_text + if state["depth"] == 0: return [widget] + else: + padded = urwid.Padding(widget, left=left_indent(state), right=right_indent(state)) + padded._contained_image = widget + return [padded] + + except Exception as e: + # TODO: Remove + RNS.trace_exception(e) + return None + def render_table(lines, state, url_delegate): if len(lines) < 2: return None if state["table_maxwidth"]: max_width = state["table_maxwidth"] @@ -482,6 +550,10 @@ def parse_line(line, state, url_delegate): elif line.startswith("`{"): return parse_partial(line[2:]) + # Check for images + elif line.startswith("`("): + return parse_image(line[2:], state, url_delegate) + # Check for section heading reset elif first_char == "<": state["depth"] = 0 diff --git a/nomadnet/ui/textui/images/__init__.py b/nomadnet/ui/textui/images/__init__.py new file mode 100644 index 0000000..619be17 --- /dev/null +++ b/nomadnet/ui/textui/images/__init__.py @@ -0,0 +1,18 @@ +# A minimal, dependency-free image widget for urwid, rendering images through +# the Kitty terminal graphics protocol. + +from ._termlib import get_cell_size, is_kitty_supported, reset_caches, diagnostics +from ._imagedata import ImageData, parse_image_header +from .widget import ImageWidget, ImageCanvas, UI_MARGIN +from .screen import ImageScreen + +__all__ = ( "ImageWidget", + "ImageCanvas", + "ImageScreen", + "ImageData", + "parse_image_header", + "is_kitty_supported", + "get_cell_size", + "diagnostics", + "reset_caches", + "UI_MARGIN" ) diff --git a/nomadnet/ui/textui/images/_ctlseqs.py b/nomadnet/ui/textui/images/_ctlseqs.py new file mode 100644 index 0000000..564b7d2 --- /dev/null +++ b/nomadnet/ui/textui/images/_ctlseqs.py @@ -0,0 +1,78 @@ +# Kitty Terminal Graphics Protocol control sequences +# +# Minimal set of control sequences needed for the Kitty graphics protocol +# side-channel image rendering used by the urwid image widget. + +from base64 import standard_b64encode +import re + +# Basic building blocks +APC = b"\x1b_" # Application Program Command +ST = b"\x1b\\" # String Terminator +CSI = b"\x1b[" + +# Kitty graphics: transmit-and-display with chunking support. +# ESC _ G ; ESC \ +KITTY_TRANSMISSION = b"\x1b_G%s;%s\x1b\\" + +# Delete images: d=A (all placements visible on screen), d=i (one +# placement, data kept), d=I (whole image, data freed). +KITTY_DELETE_ALL = b"\x1b_Ga=d,d=A\x1b\\" +KITTY_DELETE_PLACEMENT = b"\x1b_Ga=d,d=i,i=%d,p=%d\x1b\\" +KITTY_DELETE_IMAGE = b"\x1b_Ga=d,d=I,i=%d\x1b\\" + +# Support query (payload is base64 of 4 zero bytes; expects "i=31;OK") +KITTY_QUERY = b"\x1b_Ga=q,t=d,i=31,f=24,s=1,v=1,C=1,c=1,r=1;AAAA\x1b\\" +KITTY_QUERY_RESPONSE_RE = re.compile(b"\x1b_Gi=31;(?P[^\x1b]*)\x1b\\\\") + +# Cursor: absolute positioning, erase characters, cursor forward +CURSOR_POSITION = b"\x1b[%d;%dH" +ERASE_CHARS = b"\x1b[%dX" + +# Synchronized output (DEC 2026) - reduces flicker/tearing when redrawing +BEGIN_SYNCED_UPDATE = b"\x1b[?2026h" +END_SYNCED_UPDATE = b"\x1b[?2026l" + +# Terminal queries: cell size in pixels (XTWINOPS 16) + primary DA +CELL_SIZE_PX = b"\x1b[16t" +DA1 = b"\x1b[c" +# Response: CSI 4 ; ; t +CELL_SIZE_RESPONSE_RE = re.compile(b"\x1b\\[4;(?P\\d+);(?P\\d+)t") + +# Maximum payload (base64) per transmission chunk, as recommended by the +# protocol documentation +CHUNK_SIZE = 4096 + +# Control data for a transmit-only (a=t) with a quiet mode (q=1) +def kitty_transmit_control(image_id): return b"a=t,i=%d,f=100,t=d,q=1" % image_id + +# A placement command (a=p) for a previously transmitted image. +# Re-sending the same (image id, placement id) pair replaces the +# placement in place - used to move/resize/slice without flicker. +# y/h select the source rectangle (image pixels) when a slice of +# the image is displayed. +def kitty_placement(image_id, placement_id, cols, rows, z, y=None, h=None): + control = b"a=p,i=%d,p=%d,c=%d,r=%d,z=%d,C=1,q=1" % (image_id, placement_id, cols, rows, z) + if y is not None and h is not None: control += b",y=%d,h=%d" % (y, h) + return KITTY_TRANSMISSION % (control, b"") + +# Encodes a kitty graphics transmission, chunking the base64-encoded +# payload into ``chunk_size``-byte pieces. +def kitty_transmission(control, payload, chunk_size=CHUNK_SIZE): + encoded = standard_b64encode(payload) + chunks = [encoded[i:i + chunk_size] for i in range(0, len(encoded), chunk_size)] + if not chunks: + chunks = [b""] + + def control_for(index, n_chunks): + if n_chunks == 1: + return control + if index == 0: + # The control data appears only in the first chunk; the chunk + # marker says more chunks are coming. + return control + b",m=1" + if index == n_chunks - 1: + return b"m=0" + return b"m=1" + + return b"".join(KITTY_TRANSMISSION % (control_for(i, len(chunks)), chunk) for i, chunk in enumerate(chunks)) diff --git a/nomadnet/ui/textui/images/_imagedata.py b/nomadnet/ui/textui/images/_imagedata.py new file mode 100644 index 0000000..294f72c --- /dev/null +++ b/nomadnet/ui/textui/images/_imagedata.py @@ -0,0 +1,137 @@ +# Image file handling for the side-channel renderer +# +# The Kitty graphics protocol can decode and scale images in the terminal +# itself (data format ``f=100``), which means the widget needs no image +# decoding library at all. It only needs the *dimensions* of the image to lay +# it out correctly in the urwid canvas; those are trivially parseable from the +# file headers of the supported formats. +# +# Supported source formats: +# - WebP (VP8, VP8L and VP8X) +# - PNG +# +# The payload transmitted to the terminal is always a PNG file: PNG sources +# are passed through verbatim, and WebP sources are converted transparently +# by the multi-backend conversion chain in :mod:`._webp` (no required +# dependencies). The conversion result is cached on disk, so each unique +# WebP is converted at most once per system. + +import hashlib +import os + +MAX_PAYLOAD_BYTES = 8 * 1024 * 1024 +MAX_CONVERTED_PAYLOAD_BYTES = 16 * 1024 * 1024 +_PNG_SIGNATURE = b"\x89PNG\r\n\x1a\n" + +# Returns (width, height) from a PNG file. +def _parse_png(data): + if not data.startswith(_PNG_SIGNATURE): raise ValueError("not a PNG") + if len(data) < 24 or data[12:16] != b"IHDR": raise ValueError("PNG missing IHDR chunk") + width = int.from_bytes(data[16:20], "big") + height = int.from_bytes(data[20:24], "big") + if not (width > 0 and height > 0): raise ValueError("PNG has null dimensions") + return width, height + +# Returns (width, height) from a WebP file (VP8, VP8L or VP8X). +def _parse_webp(data): + if not data.startswith(b"RIFF") or data[8:12] != b"WEBP": raise ValueError("not a WebP") + if len(data) < 30: raise ValueError("WebP file too short") + fourcc = data[12:16] + if fourcc == b"VP8X": + # 10-byte chunk: flags(1) reserved(3) canvas-w-1(3 LE) canvas-h-1(3 LE) + width = int.from_bytes(data[24:27], "little") + 1 + height = int.from_bytes(data[27:30], "little") + 1 + elif fourcc == b"VP8 ": + # frame tag(3) start code 0x9d012a(3) width(2 LE, 14 bits) height(2 LE, 14 bits) + width = int.from_bytes(data[26:28], "little") & 0x3FFF + height = int.from_bytes(data[28:30], "little") & 0x3FFF + elif fourcc == b"VP8L": + # signature(1) then 4 bytes: w-1 (14 bits) | h-1 (14 bits) | ... + bits = int.from_bytes(data[21:25], "little") + width = (bits & 0x3FFF) + 1 + height = ((bits >> 14) & 0x3FFF) + 1 + else: raise ValueError("unsupported WebP variant %r" % fourcc) + if not (width > 0 and height > 0): raise ValueError("WebP has null dimensions") + return width, height + + +def parse_image_header(data): + if data[:8] == _PNG_SIGNATURE: + w, h = _parse_png(data) + return "PNG", w, h + if data[:4] == b"RIFF" and data[8:12] == b"WEBP": + w, h = _parse_webp(data) + return "WebP", w, h + raise ValueError("unsupported image format (only PNG and WebP)") + +# Loads an image file and exposes its PNG bytes and dimensions. +# The payload is always a PNG file, transmitted verbatim to the terminal +# (Kitty's f=100 data format). The dimensions are used only for layout. +# WebP sources are converted transparently to PNG on load. +class ImageData(object): + def __init__(self, path, max_bytes=MAX_PAYLOAD_BYTES): + self.path = path + self.error = None + self.format = None + self.source_format = None + self.conversion_backend = None + self.width = 0 + self.height = 0 + self.data = None + try: + with open(path, "rb") as f: data = f.read() + except OSError as e: + self.error = "could not read file: %s" % e + return + + if len(data) > max_bytes: + self.error = "file too large (%.1f MiB > %d MiB)" % (len(data) / (1024 * 1024), max_bytes // (1024 * 1024)) + return + + if not data: + self.error = "empty file" + return + + try: self.format, self.width, self.height = parse_image_header(data) + except ValueError as e: + self.error = str(e) + return + + self.source_format = self.format + if self.format == "WebP": + from . import _webp + png = _webp.convert_webp_to_png(data) + if png is None: + self.error = "webp conversion failed (no working backend available)" + return + if len(png) > MAX_CONVERTED_PAYLOAD_BYTES: + self.error = "converted image too large (%.1f MiB > %d MiB)" % (len(png) / (1024 * 1024), MAX_CONVERTED_PAYLOAD_BYTES // (1024 * 1024)) + return + + self.data = png + self.format = "PNG" + self.conversion_backend = _webp.last_backend() + + else: self.data = data + + # Content key for de-duplication: identical bytes = same key, so + # the same image transmitted once can be placed many times. + self._key = hashlib.sha256(self.data).hexdigest() + + @property + def ok(self): return self.data is not None + + @property + def key(self): return getattr(self, "_key", None) + + @property + def size(self): return (self.width, self.height) + + def description(self): + if self.ok: + if self.source_format == "WebP": + return "WebP %dx%d -> PNG (%s)" % (self.width, self.height, self.conversion_backend or "cached") + else: + return "%s %dx%d" % (self.format, self.width, self.height) + + return self.error diff --git a/nomadnet/ui/textui/images/_imagestore.py b/nomadnet/ui/textui/images/_imagestore.py new file mode 100644 index 0000000..78935b5 --- /dev/null +++ b/nomadnet/ui/textui/images/_imagestore.py @@ -0,0 +1,90 @@ +# Registry of terminal-side image data +# +# Tracks which image payloads have been transmitted to the terminal (Kitty +# image ids), de-duplicates identical image data across widgets (one payload +# transmitted once, placed many times), and counts live widget references so +# that orphaned data can be evicted from the terminal's buffers. +# +# This module only tracks *state*; all protocol output is performed by +# ImageScreen, which reads and updates the store. +# +# Eviction: +# +# - Automatic, widgets unregister when closed/destroyed; the screen evicts +# (deletes from the terminal) any registered data with no live references. +# - Explicit, the browser can call ImageScreen.purge_images() on page +# navigation to drop everything at once. + +DEFAULT_IMAGE_ID_MAX = 2 ** 32 - 1 # Protocol limit for image ids + +class ImageStore(object): + def __init__(self): + self._entries = {} # content key -> {"image_id", "data", "refs", "transmitted"} + self._next_image_id = 1 + + ############# + # Lifecycle # + ############# + + # Registers image data and returns (key, is_new). + # If identical data is already registered, the existing entry is + # reused (de-duplication) and its reference count is incremented. + def register(self, key, data): + entry = self._entries.get(key) + if entry is None: + image_id = self._next_image_id + self._next_image_id += 1 + if image_id > DEFAULT_IMAGE_ID_MAX: raise OverflowError("image id space exhausted") + entry = { "image_id": image_id, "data": data, + "refs": 0, "transmitted": False } + self._entries[key] = entry + is_new = True + else: is_new = False + + entry["refs"] += 1 + return key, is_new + + # Decrements the reference count of registered data + def unregister(self, key): + entry = self._entries.get(key) + if entry is None: return + entry["refs"] = max(0, entry["refs"] - 1) + + ########### + # Queries # + ########### + + def entry_for(self, key): return self._entries.get(key) + + def transmitted(self): return [key for key, entry in self._entries.items() if entry["transmitted"]] + + # Returns [(key, image_id), ...] for entries with no live + # widget references that are still present in the terminal. + def evict_orphans(self): + return [ (key, entry["image_id"]) + for key, entry in self._entries.items() + if entry["refs"] == 0 and entry["transmitted"] ] + + ############ + # Mutation # + ############ + + def mark_transmitted(self, key): + entry = self._entries.get(key) + if entry is not None: + entry["transmitted"] = True + + def mark_all_untransmitted(self): + for entry in self._entries.values(): + entry["transmitted"] = False + + # Drops the entry for evicted data (frees the payload bytes) + def evict(self, key): self._entries.pop(key, None) + + # Forgets all state (used by tests) + def reset(self): + self._entries.clear() + self._next_image_id = 1 + +# Module-level singleton shared by widgets and the screen. +image_store = ImageStore() diff --git a/nomadnet/ui/textui/images/_selftest.py b/nomadnet/ui/textui/images/_selftest.py new file mode 100644 index 0000000..e1755bb --- /dev/null +++ b/nomadnet/ui/textui/images/_selftest.py @@ -0,0 +1,681 @@ +#!/usr/bin/env python3 +""" +.. Images - headless self test + +Exercises the widget + screen machinery without a terminal: image header +parsing, display sizing, shard scanning, APC construction/chunking and the +transmit/delete state machine at various scroll positions. + +Run:: + + python3 -m nomadnet.ui.textui.images._selftest [image-file] +""" + +import base64 +import io +import os +import shutil +import struct +import sys +import tempfile + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "..", "..")) + +import urwid + +from nomadnet.ui.textui.images import _ctlseqs +from nomadnet.ui.textui.images import _imagestore +from nomadnet.ui.textui.images import _termlib +from nomadnet.ui.textui.images._imagedata import ImageData, parse_image_header +from nomadnet.ui.textui.images.screen import ImageScreen +from nomadnet.ui.textui.images.widget import ImageWidget + +DEFAULT_IMAGE = os.path.expanduser("~/Scratchpad/test.png") +VIEWPORT = (80, 24) +CELL_SIZE = (1, 2) # forced fallback for deterministic layout + +PASS = 0 +FAIL = 0 + + +def check(cond, label): + global PASS, FAIL + if cond: + PASS += 1 + print(" ok - %s" % label) + else: + FAIL += 1 + print(" FAIL - %s" % label) + + +def build_pile(image_path, wrap_attrmap=False): + widgets = [] + for i in range(12): + widgets.append(urwid.Text("[%02d]\nline one\nline two\n" % i)) # 3 rows each + img = ImageWidget(image_path) + if wrap_attrmap: + img = urwid.AttrMap(img, "body") + widgets.append(urwid.Divider(" ")) + widgets.append(img) + widgets.append(urwid.Divider(" ")) + for i in range(12): + widgets.append(urwid.Text("[%02d]\nline one\nline two\n" % (13 + i))) + return urwid.Pile(widgets) + + +class RecordingScreen(ImageScreen): + def __init__(self): + super().__init__() + self.log = [] + + def write(self, data): + if isinstance(data, bytes): + data = data.decode("ascii") + self.log.append(data) + return data + + def flush(self): + pass + + +def extract_transmit(log): + """Returns (image_id, joined_b64_payload) of the a=t transmission, or + None if no transmission is present.""" + image_id = None + payloads = [] + for item in log: + if item.startswith("\x1b_Ga=t"): + for piece in item.split("\x1b_G")[1:]: + body = piece[:-2] # strip the trailing ST (ESC \) + head, sep, payload = body.partition(";") + if not sep: + continue + if image_id is None: + image_id = int(head.split("i=")[1].split(",")[0]) + payloads.append(payload) + if image_id is None: + return None + return image_id, "".join(payloads) + + +def extract_placements(log): + """Returns the control data of every a=p placement command.""" + return [ + item[3:-2] + for item in log + if item.startswith("\x1b_Ga=p") + ] + + +def extract_deletes(log): + """Returns the control data of every a=d delete command.""" + return [ + item[3:-2] + for item in log + if item.startswith("\x1b_Ga=d") + ] + + +def screen_probe_placement_key(widget, row, col, canvas_cols, trim_top, vis_rows): + """Computes an ImageScreen placement key without a screen instance.""" + return ImageScreen._placement_key( + None, widget, row, col, canvas_cols, trim_top, vis_rows) + + +def test_header_parse(image_path): + print("== header parsing ==") + with open(image_path, "rb") as f: + data = f.read() + fmt, w, h = parse_image_header(data) + check(fmt == "PNG", "detected PNG") + check(w == 1280 and h == 960, "dimensions %dx%d" % (w, h)) + imgdata = ImageData(image_path) + check(imgdata.ok, "ImageData loads without error") + check(imgdata.size == (w, h), "ImageData size matches") + + +def test_transmission_encoding(image_path): + print("== APC encoding/chunking ==") + small = b"\x89PNG\r\n\x1a\n" + b"x" * 100 + single = _ctlseqs.kitty_transmission(b"a=T,f=100,c=80,r=30,z=1", small) + check(single.count(b"\x1b_G") == 1, "single-chunk transmission has one APC") + check(single.startswith(b"\x1b_Ga=T,f=100,c=80,r=30,z=1;"), "control data correct") + check(single.endswith(b"\x1b\\"), "ends with ST") + payload = single.split(b";", 1)[1][:-2] + check(base64.b64decode(payload) == small, "payload round-trips to original file") + + with open(image_path, "rb") as f: + data = f.read() + chunks = _ctlseqs.kitty_transmission(b"a=T,f=100", data, chunk_size=16) + apcs = chunks.split(b"\x1b\\")[:-1] + check(len(apcs) > 1, "multi-chunk transmission splits into %d APCs" % len(apcs)) + first, last = apcs[0], apcs[-1] + heads = [a.split(b";", 1)[0][3:] for a in apcs] + check(b",m=1;" in first, "first chunk carries control + m=1") + check(last.startswith(b"\x1b_Gm=0;"), "last chunk carries m=0") + check(sum(b"a=T" in h for h in heads) == 1, "control data appears only once") + check(all(h == b"m=1" or h == b"m=0" for h in heads[1:]), "continuation chunks carry only the chunk marker") + joined_b64 = b"".join(a.split(b";", 1)[1] for a in apcs) + check(base64.b64decode(joined_b64) == data, "multi-chunk payload reassembles") + + +_TINY_WEBP = base64.b64decode( + "UklGRiQAAABXRUJQVlA4IBgAAAAwAQCdASoBAAEAAUAmJaQAA3AA/vuUAAA=") + + +def _webp_fixture(): + """Returns (webp_bytes, (width, height)): a 4x4 WebP generated with + Pillow when available (more representative), else a 1x1 constant.""" + try: + from PIL import Image + if _webp_module_available(): + image = Image.new("RGBA", (4, 4)) + for y in range(4): + for x in range(4): + image.putpixel((x, y), (x * 60, y * 60, x * y * 10, 255)) + buf = io.BytesIO() + image.save(buf, format="WEBP") + data = buf.getvalue() + if parse_image_header(data)[0] == "WebP": + return data, (4, 4) + except Exception: + pass + return _TINY_WEBP, (1, 1) + + +def _webp_module_available(): + try: + from PIL import features + return bool(features.check("webp")) + except Exception: + return False + + +def test_webp_conversion(): + print("== webp conversion ==") + from nomadnet.ui.textui.images import _webp + from nomadnet.ui.textui.images._imagedata import MAX_CONVERTED_PAYLOAD_BYTES + + fmt, w, h = parse_image_header(_TINY_WEBP) + check((fmt, w, h) == ("WebP", 1, 1), "embedded fixture is a 1x1 WebP") + + fixture, dims = _webp_fixture() + check(parse_image_header(fixture)[1:] == dims, + "webp fixture has expected dimensions") + + cache_tmp = tempfile.mkdtemp(prefix="imgw-test-cache-") + prev_cache_env = os.environ.get("NOMADNET_IMAGE_CACHE") + os.environ["NOMADNET_IMAGE_CACHE"] = cache_tmp + try: + # Each available backend must convert the fixture to a valid PNG + # with matching dimensions. Backends are invoked directly, since + # the on-disk cache would otherwise mask all but the first one. + _webp.reset() + present = [name for name, avail in _webp.available_backends() if avail] + converted = 0 + for name in present: + png = _webp._BACKENDS[name](fixture, dims) + ok = (png is not None and png.startswith(b"\x89PNG") + and _webp._png_dims(png) == dims) + check(ok, "%s backend converts to a valid PNG" % name) + if ok: + converted += 1 + check(converted >= 1, "at least one backend converts webp (got %d)" % converted) + + # Repeat conversion is served from the on-disk cache: the backend + # table is disabled and the conversion still succeeds. + _webp.reset() + png1 = _webp.convert_webp_to_png(fixture) + check(png1 is not None, "first conversion succeeds") + orig_backends = dict(_webp._BACKENDS) + _webp._BACKENDS = {name: (lambda *a, **k: None) for name in _webp.BACKENDS} + png2 = _webp.convert_webp_to_png(fixture) + check(png1 == png2 and _webp.stats()["cache_hits"] >= 1, + "repeat conversion served from the on-disk cache") + _webp._BACKENDS = orig_backends + + # ImageData-level transparency: a webp file loads as a PNG. + _webp.reset() + fd, webp_path = tempfile.mkstemp(suffix=".webp") + with os.fdopen(fd, "wb") as f: + f.write(fixture) + d = ImageData(webp_path) + check(d.ok and d.format == "PNG" and d.size == dims, + "ImageData converts webp transparently to PNG") + check(d.data.startswith(b"\x89PNG\r\n\x1a\n") and d.key is not None, + "converted payload is a keyed PNG") + check("-> PNG" in d.description(), "description notes the conversion") + os.unlink(webp_path) + + # Converted-payload cap. + _webp.reset() + orig_convert = _webp.convert_webp_to_png + _webp.convert_webp_to_png = lambda data: b"x" * (MAX_CONVERTED_PAYLOAD_BYTES + 1) + fd, webp_path = tempfile.mkstemp(suffix=".webp") + with os.fdopen(fd, "wb") as f: + f.write(fixture) + d = ImageData(webp_path) + check(not d.ok and "converted image too large" in d.error, + "converted payload capped at %d MiB" + % (MAX_CONVERTED_PAYLOAD_BYTES // (1024 * 1024))) + os.unlink(webp_path) + _webp.convert_webp_to_png = orig_convert + + # Undecodable webp (valid header/dimensions, garbage bitstream). + _webp.reset() + junk = (b"RIFF" + struct.pack(" (10, 20)") + check(_termlib._cell_size_from_winsize([24, 80, 960, 480]) == (12, 20), + "correct winsize -> (12, 20)") + check(_termlib._cell_size_from_winsize([24, 80, 0, 0]) is None, + "zero pixel fields -> None") + check(_termlib._cell_size_from_winsize([40, 100, 800, 1000]) == (8, 25), + "portrait window cell keeps valid aspect") + check(_termlib._cell_size_from_winsize([30, 80, 1200, 240]) is None, + "implausible aspect (broken reporting) -> None") + check(_termlib._cell_size_from_winsize([0, 0, 0, 0]) is None, + "empty winsize -> None") + + +def test_widget_layout(image_path): + print("== widget layout ==") + _termlib.kitty_supported = True + _termlib.cell_size = CELL_SIZE + # Force a deterministic terminal size for the height-percent cases. + _orig_terminal_size = _termlib.get_terminal_size + _termlib.get_terminal_size = lambda: (80, 24) + _orig_ui_margin = None + from nomadnet.ui.textui.images import widget as _widget_mod + _orig_ui_margin = _widget_mod.UI_MARGIN + _widget_mod.UI_MARGIN = 8 + try: + imgdata = ImageData(image_path) + w = ImageWidget(image_path) + iw, ih = imgdata.size + expected_rows = max(1, round(80 * CELL_SIZE[0] * ih / (CELL_SIZE[1] * iw))) + rows = w.rows((80,)) + check(rows == expected_rows, + "default: rows((80,)) == %d, got %d" % (expected_rows, rows)) + canvas = w.render((80,)) + check(isinstance(canvas, urwid.SolidCanvas), "render returns a solid canvas") + check(canvas.rows() == rows and canvas.cols() == 80, "canvas sized 80x%d" % rows) + check(w._ti_rows == rows and w._ti_cols == 80, "widget stores display size") + w.close() + + # width as absolute columns (aspect preserved); the canvas fills the + # allocated width while the image rect is the resolved size. + w = ImageWidget(image_path, width=40) + c, r = w._display_size(80) + check((c, r) == (40, 15), "width=40 -> (40, 15), got (%d, %d)" % (c, r)) + canvas = w.render((80,)) + check((canvas.cols(), canvas.rows()) == (80, 15), + "width=40 canvas fills the row (80, 15), got (%d, %d)" + % (canvas.cols(), canvas.rows())) + check((w._ti_cols, w._ti_rows) == (40, 15), + "width=40 image rect stored (40, 15)") + w.close() + + # width as percent of the layout width + w = ImageWidget(image_path, width="50%") + c, r = w._display_size(80) + check((c, r) == (40, 15), "width=50%% -> (40, 15), got (%d, %d)" % (c, r)) + w.close() + + # height as absolute rows (width derived, aspect preserved) + w = ImageWidget(image_path, height=20) + c, r = w._display_size(80) + check(r == 20 and c == 53, "height=20 -> (53, 20), got (%d, %d)" % (c, r)) + w.close() + + # height as percent of the visible page height + # page = terminal lines (24) - UI_MARGIN (8) = 16; 50% -> 8 rows + w = ImageWidget(image_path, height="50%") + c, r = w._display_size(80) + check(r == 8 and c == 21, "height=50%% -> (21, 8), got (%d, %d)" % (c, r)) + w.close() + + # both given: stretch to the exact rectangle + w = ImageWidget(image_path, width=40, height=30) + c, r = w._display_size(80) + check((c, r) == (40, 30), "width=40,height=30 stretches -> (40, 30), got (%d, %d)" % (c, r)) + w.close() + + # stretching is clamped uniformly when overflowing the space + w = ImageWidget(image_path, width=400, height=200) + c, r = w._display_size(80) + check((c, r) == (80, 40), "overflowing stretch clamped -> (80, 40), got (%d, %d)" % (c, r)) + w.close() + + # BOX layout: the canvas fills the box; the image rect is clamped + # and re-derived (width from the clamped height). + w = ImageWidget(image_path, name="box") + canvas = w.render((80, 10)) + check((canvas.cols(), canvas.rows()) == (80, 10), + "BOX render((80,10)) canvas fills the box, got (%d, %d)" + % (canvas.cols(), canvas.rows())) + check((w._ti_cols, w._ti_rows) == (27, 10), + "BOX image rect clamped -> (27, 10), got (%d, %d)" + % (w._ti_cols, w._ti_rows)) + w.close() + + # invalid specifications + def raises(ctor, want): + try: + ctor() + except want: + return True + except Exception: + return False + return False + check(raises(lambda: ImageWidget(image_path, width="xx"), ValueError), + "width='xx' raises ValueError") + check(raises(lambda: ImageWidget(image_path, height=0), ValueError), + "height=0 raises ValueError") + check(raises(lambda: ImageWidget(image_path, width=1.5), TypeError), + "width=1.5 raises TypeError") + check(raises(lambda: ImageWidget(image_path, height="50"), ValueError), + "height='50' raises ValueError") + + # settable properties re-resolve and invalidate + w = ImageWidget(image_path) + w.width = "50%" + c, r = w._display_size(80) + check((c, r) == (40, 15), "set width=50%% property -> (40, 15)") + w.height = 10 + c, r = w._display_size(80) + check((c, r) == (40, 10), "width+height set -> stretch (40, 10), got (%d, %d)" % (c, r)) + w.width = None + w.height = 10 + c, r = w._display_size(80) + check((c, r) == (27, 10), "height-only after reset -> (27, 10), got (%d, %d)" % (c, r)) + w.close() + + # alignment: placement column of a narrowed image inside its canvas + w = ImageWidget(image_path, width=40) # default center + w.render((80,)) # sets _ti_cols/_ti_rows + check(w._ti_align == "|", "default alignment is center") + key = screen_probe_placement_key(w, 3, 0, 80, 0, 15) + check(key == (3, 20, 40, 15, None, None), + "centered placement col -> %s" % (key,)) + w.align = "left" + key = screen_probe_placement_key(w, 3, 0, 80, 0, 15) + check(key == (3, 0, 40, 15, None, None), "left placement col -> 0") + w.align = "right" + key = screen_probe_placement_key(w, 3, 0, 80, 0, 15) + check(key == (3, 40, 40, 15, None, None), "right placement col -> 40") + check(raises(lambda: ImageWidget(image_path, align="top"), ValueError), + "align='top' raises ValueError") + w.close() + finally: + _termlib.get_terminal_size = _orig_terminal_size + if _orig_ui_margin is not None: + _widget_mod.UI_MARGIN = _orig_ui_margin + + +def find_image_widget(pile): + for widget, _opts in pile.contents: + child = widget + if isinstance(child, urwid.AttrMap): + child = child.original_widget + if isinstance(child, ImageWidget): + return child + return None + + +def run_screen_test(image_path, wrap_attrmap, label): + print("== screen state machine (%s) ==" % label) + _termlib.kitty_supported = True + _termlib.cell_size = CELL_SIZE + _imagestore.image_store.reset() + + pile = build_pile(image_path, wrap_attrmap=wrap_attrmap) + scrollable = urwid.Scrollable(pile, force_forward_keypress=True) + screen = RecordingScreen() + + img_widget = find_image_widget(pile) + check(img_widget is not None, "image widget found in pile") + z = img_widget._ti_z + entry = _imagestore.image_store.entry_for(img_widget._ti_key) + check(entry is not None and entry["refs"] == 1, "widget registered with image store") + image_id = entry["image_id"] + b64 = base64.b64encode(img_widget._ti_data.data).decode("ascii") + + def entries_at(offset): + scrollable.set_scrollpos(offset) + canvas = scrollable.render(VIEWPORT) + return screen._ti_scan(canvas) + + # Build a timeline of image visibility across all scroll offsets. + # state = (trim_top, row, visible_rows) + timeline = [] + max_offset = 200 + for off in range(max_offset + 1): + entries = entries_at(off) + if not entries: + timeline.append((off, None)) + continue + widget, row, col, canvas_cols, trim_top, vis_rows = entries[0] + timeline.append((off, (trim_top, row, vis_rows))) + image_rows = img_widget._ti_rows + visible = [t for t in timeline if t[1] is not None] + check(len(visible) > 0, "image is visible at some scroll offset") + if not visible: + return + + def first_offset(pred): + for off, state in timeline: + if state is not None and pred(state): + return off + return None + + off_enter = first_offset(lambda s: s[0] == 0) # trim_top == 0 + off_clip = first_offset(lambda s: s[0] > 0) # top-clipped + check(off_enter is not None, "found offset with image fully entered") + check(off_clip is not None, "found offset with image top-clipped") + if off_enter is None or off_clip is None: + return + # State at the entry offset: (trim_top, row, visible_rows) + off_enter_vis_rows = timeline[off_enter][1][2] + + last_visible_off = visible[-1][0] + off_away = last_visible_off + 1 if last_visible_off < max_offset else None + check(off_away is not None, "found offset past the image") + + def update(offset): + screen.log = [] + scrollable.set_scrollpos(offset) + canvas = scrollable.render(VIEWPORT) + screen._ti_update(canvas) + return list(screen.log) + + # 1. before the image is visible: nothing + log = update(0) + check(not any("a=t" in l or "a=p" in l for l in log), + "nothing written while out of view") + check(not screen._ti_images, "nothing tracked while out of view") + + # 2. image entered (bottom-clipped): transmit once, place the visible + # slice (y=0 crop) at the bottom. + log = update(off_enter) + trans = extract_transmit(log) + placements = extract_placements(log) + check(trans is not None, "image data transmitted on entry") + if trans: + trans_id, pb64 = trans + check(trans_id == image_id, "transmit uses the store's image id") + check(pb64 == b64, "transmitted payload is the original file") + check(len(placements) == 1, "one placement on entry, got %d" % len(placements)) + if placements: + ctrl = placements[0] + check("i=%d" % image_id in ctrl and "p=%d" % z in ctrl, + "placement references image id and placement id") + check("c=80" in ctrl and "r=%d" % off_enter_vis_rows in ctrl, + "placement has display size") + check("y=0," in ctrl and "h=" in ctrl, "entry slice uses a y=0 crop") + check("z=%d" % z in ctrl and "C=1" in ctrl and "q=1" in ctrl, + "placement has z-index, C=1 and quiet mode") + check(len(screen._ti_images) == 1 and + screen._ti_images[id(img_widget)]["drawn"] is not None, + "image tracked as drawn") + check(entry["transmitted"], "store entry marked transmitted") + + # 3. move while visible: in-place placement update, no re-transmit, + # no delete. + log = update(off_enter + 1) + check(extract_transmit(log) is None, "position change does not re-transmit") + check(len(extract_placements(log)) == 1, "position change re-places in place") + check(not extract_deletes(log), "position change does not delete") + + # 4. top-clipped: placement with a y>0 crop. + log = update(off_clip) + placements = extract_placements(log) + check(len(placements) == 1, "top-clip places a slice, got %d" % len(placements)) + if placements: + ctrl = placements[0] + y_px = int(ctrl.split("y=")[1].split(",")[0].split(";")[0]) + check(",y=" in ctrl and "h=" in ctrl and y_px > 0, + "top-clipped placement has y>0 crop (y=%d)" % y_px) + + # 5. scrolled past: placement deleted, image data kept. + log = update(off_away) + deletes = extract_deletes(log) + check(deletes == ["a=d,d=i,i=%d,p=%d" % (image_id, z)], + "scroll-away deletes the placement only: %s" % deletes) + check(extract_transmit(log) is None, "scroll-away does not transmit") + check(id(img_widget) not in screen._ti_images, "widget untracked after scroll-away") + + # 6. scroll back: re-placed from the kept data, no re-transmission. + log = update(off_enter) + check(extract_transmit(log) is None, "scroll-back does not re-transmit") + check(len(extract_placements(log)) == 1, "scroll-back re-places the image") + + # 7. still visible, unchanged position: nothing written. + log = update(off_enter) + check(not extract_placements(log) and not extract_deletes(log), + "unchanged position writes nothing") + + # 8. widget closed -> orphaned data evicted from the terminal. + img_widget.close() + log = update(off_away) + deletes = extract_deletes(log) + check("a=d,d=I,i=%d" % image_id in deletes, + "orphaned image data evicted with d=I") + check(_imagestore.image_store.entry_for(img_widget._ti_key) is None, + "evicted entry dropped from the store") + + +def test_eviction_and_purge(image_path): + print("== eviction and purge ==") + _termlib.kitty_supported = True + _termlib.cell_size = CELL_SIZE + _imagestore.image_store.reset() + + # Shared source: two widgets with the same file share one store entry. + w1 = ImageWidget(image_path) + w2 = ImageWidget(image_path) + entry = _imagestore.image_store.entry_for(w1._ti_key) + check(entry["refs"] == 2, "identical sources share one registry entry (refs=%d)" + % entry["refs"]) + check(w2._ti_key == w1._ti_key, "identical sources have the same content key") + + w1.close() + check(entry["refs"] == 1, "closing one widget decrements refcount") + check(entry["transmitted"] is False, "no data transmitted yet") + + # Place w2 via a screen update and verify orphan eviction after close. + pile = urwid.Pile([urwid.Text("x"), urwid.Divider(" "), w2, + urwid.Divider(" "), urwid.Text("y")]) + scrollable = urwid.Scrollable(pile, force_forward_keypress=True) + screen = RecordingScreen() + image_id = entry["image_id"] + + def update(offset): + screen.log = [] + scrollable.set_scrollpos(offset) + screen._ti_update(scrollable.render(VIEWPORT)) + return list(screen.log) + + log = update(2) + check(extract_transmit(log) is not None, "image transmitted when widget visible") + check(entry["transmitted"], "entry marked transmitted") + + w2.close() + log = update(0) + deletes = extract_deletes(log) + check("a=d,d=I,i=%d" % image_id in deletes, "last reference closed -> d=I eviction") + check(_imagestore.image_store.entry_for(w2._ti_key) is None, + "entry dropped after eviction") + + # Purge: after transmitting, purge_images() deletes everything and the + # next redraw re-transmits for still-visible widgets. + _imagestore.image_store.reset() + w3 = ImageWidget(image_path) + pile = urwid.Pile([urwid.Text("x"), urwid.Divider(" "), w3, + urwid.Divider(" "), urwid.Text("y")]) + scrollable = urwid.Scrollable(pile, force_forward_keypress=True) + screen2 = RecordingScreen() + + def update2(offset): + screen2.log = [] + scrollable.set_scrollpos(offset) + screen2._ti_update(scrollable.render(VIEWPORT)) + return list(screen2.log) + + log = update2(2) + check(extract_transmit(log) is not None, "purge-test: image transmitted") + screen2.log = [] + screen2.purge_images() + check(any("a=d,d=A" in l for l in screen2.log), "purge deletes all (d=A)") + check(_imagestore.image_store.transmitted() == [], "purge marks all untransmitted") + + log = update2(2) + check(extract_transmit(log) is not None, "visible image re-transmitted after purge") + check(len(extract_placements(log)) == 1, "visible image re-placed after purge") + w3.close() + + +def main(): + image_path = sys.argv[1] if len(sys.argv) > 1 else DEFAULT_IMAGE + print("Using image: %s" % image_path) + _termlib.reset_caches() + print("ambient cell size (env): %s" % (_termlib._query_cell_size(),)) + + test_header_parse(image_path) + test_transmission_encoding(image_path) + test_webp_conversion() + test_winsize_parsing() + test_widget_layout(image_path) + run_screen_test(image_path, wrap_attrmap=False, label="bare widget") + run_screen_test(image_path, wrap_attrmap=True, label="attrmap-wrapped") + test_eviction_and_purge(image_path) + + print("\n%d passed, %d failed" % (PASS, FAIL)) + return 1 if FAIL else 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/nomadnet/ui/textui/images/_termlib.py b/nomadnet/ui/textui/images/_termlib.py new file mode 100644 index 0000000..abe7575 --- /dev/null +++ b/nomadnet/ui/textui/images/_termlib.py @@ -0,0 +1,196 @@ +# Terminal queries for capability detection and layout metrics. +# +# All values are resolved once and cached; subsequent calls are instant. +# Queries should be performed *before* starting the urwid main loop, because +# they read directly from the terminal and would otherwise race with urwid's +# own input handling. If no terminal is available, or the queries time out, +# sensible fallbacks are used. + +import os +import select +import shutil +import sys +import termios +import time + +from . import _ctlseqs + +QUERY_TIMEOUT = 0.3 + +kitty_supported = None +cell_size = None + +# Last raw terminal query response (bytes), for diagnostics. +last_query_response = bytearray() +last_kitty_response = bytearray() + +# Last TIOCGWINSZ buffer (rows, cols, xpixel, ypixel), for diagnostics. +winsize = [0, 0, 0, 0] + +# Number of times a query attempt has been made; used to avoid repeated +# slow queries in a broken environment. +_query_failed = False + +def _open_tty(): + # Opens the active terminal for reading and writing. + for stream in ("out", "in", "err"): + try: return os.open(os.ttyname(getattr(sys, "__std%s__" % stream).fileno()), os.O_RDWR) + except (OSError, AttributeError): continue + try: return os.open("/dev/tty", os.O_RDWR | os.O_NOCTTY) + except OSError: return -1 + + +def query_tty(request, timeout=QUERY_TIMEOUT): + # Sends a query to the terminal and returns the raw response. + global _query_failed + if _query_failed: return b"" + + fd = _open_tty() + if fd == -1: + _query_failed = True + return b"" + + # Never block indefinitely, even on a broken pseudo-terminal, + # select has a deadline and reads are non-blocking. + try: os.set_blocking(fd, False) + except OSError: pass + + old_attr = termios.tcgetattr(fd) + new_attr = termios.tcgetattr(fd) + # Raw-ish mode; disable canonical mode and echo. In canonical mode the + # terminal buffers input until a newline, which terminal query responses + # (for example. kitty's "i=31;OK") never contain. The read would block until + # timeout and the query would always "fail". + new_attr[3] &= ~(termios.ECHO | termios.ICANON) + new_attr[6][termios.VMIN] = 0 + new_attr[6][termios.VTIME] = 0 + try: + # TCSANOW (not TCSAFLUSH): TCSAFLUSH drains the output queue first, + # which can block indefinitely on some pseudo-terminals. + termios.tcsetattr(fd, termios.TCSANOW, new_attr) + try: os.write(fd, request) + except (BlockingIOError, OSError): return b"" + response = bytearray() + deadline = time.monotonic() + timeout + while True: + remaining = deadline - time.monotonic() + if remaining <= 0: break + try: ready, _, _ = select.select([fd], [], [], remaining) + except (OSError, ValueError): break + if not ready: break + try: chunk = os.read(fd, 4096) + except (BlockingIOError, OSError): continue + if not chunk: break + response += chunk + if response.endswith(b"\x1b\\") or response.endswith(b"c"): break + last_query_response[:] = response + return bytes(response) + + finally: + try: termios.tcsetattr(fd, termios.TCSANOW, old_attr) + except termios.error: pass + finally: os.close(fd) + +def is_kitty_supported(): + global kitty_supported + if kitty_supported is None: + response = query_tty(_ctlseqs.KITTY_QUERY) + last_kitty_response[:] = response + kitty_supported = bool(response and _ctlseqs.KITTY_QUERY_RESPONSE_RE.search(response)) + return kitty_supported + +def force_kitty_supported(status): + # Forces the support status (used by tests). + global kitty_supported + kitty_supported = bool(status) + +def _cell_size_from_winsize(buf): + # Interprets a TIOCGWINSZ buffer as (cell_width_px, cell_height_px). + # The struct layout is (rows, cols, xpixel, ypixel). Returns None if the + # terminal does not report pixel dimensions, or if the result is + # implausible (e.g. several times wider than tall - a sign of broken + # winsize reporting). + rows, cols, xpix, ypix = buf + if not (cols and rows and xpix and ypix): return None + cell_w, cell_h = max(1, xpix // cols), max(1, ypix // rows) + if not (0.1 <= cell_w / cell_h <= 1.5): + # Real terminal cells should be as wide as, or narrower than, + # they are tall; anything far outside that range is bogus. + return None + return (cell_w, cell_h) + +def _query_cell_size(): + # Determines the current cell size in pixels, or None if unknown. + fd = _open_tty() + if fd != -1: + # First, try the TIOCGWINSZ pixel fields (fast, no query) + import array + import fcntl + try: + buf = array.array("H", [0, 0, 0, 0]) + fcntl.ioctl(fd, termios.TIOCGWINSZ, buf) + winsize[:] = buf + cell = _cell_size_from_winsize(buf) + if cell is not None: return cell + + except OSError: pass + finally: os.close(fd) + + # Fall back to XTWINOPS 16 (cell size in pixels); DA1 is appended to + # speed up the response (terminals treat queries as FIFO). + response = query_tty(_ctlseqs.CELL_SIZE_PX + _ctlseqs.DA1) + if response: + match = _ctlseqs.CELL_SIZE_RESPONSE_RE.search(response) + if match: return (int(match.group("w")), int(match.group("h"))) + return None + +def get_cell_size(): + # Returns the terminal cell size in pixels as (width, height), or None + # if it could not be determined. + global cell_size + if cell_size is None: cell_size = _query_cell_size() + return cell_size + + +def get_terminal_size(): + # Returns the active terminal size as (columns, lines). + # Uses the tty directly when available (correct even with redirected + # output), falling back to shutil.get_terminal_size. + + fd = _open_tty() + if fd != -1: + try: + size = os.get_terminal_size(fd) + return (size.columns, size.lines) + except OSError: pass + finally: os.close(fd) + size = shutil.get_terminal_size(fallback=(80, 24)) + return (size.columns, size.lines) + +def reset_caches(): + # Clears all cached capability/metric values (used by tests). + global kitty_supported, cell_size, _query_failed + kitty_supported = None + cell_size = None + _query_failed = False + last_query_response[:] = b"" + last_kitty_response[:] = b"" + winsize[:] = (0, 0, 0, 0) + +def diagnostics(): + # Returns a list of diagnostic strings about the terminal environment, + # for display/debugging. + import os + lines = [ "TERM=%r" % os.environ.get("TERM"), + "TERM_PROGRAM=%r TERM_PROGRAM_VERSION=%r" % (os.environ.get("TERM_PROGRAM"), os.environ.get("TERM_PROGRAM_VERSION")), + "kitty graphics protocol supported: %s" % is_kitty_supported(), + "kitty query response: %r" % bytes(last_kitty_response), + "cell size (px): %s" % (get_cell_size(),), + "cell size query response: %r" % bytes(last_query_response), + "winsize (rows, cols, xpix, ypix): %r" % (tuple(winsize),) ] + try: + from . import _webp + lines.append("webp conversion: %s" % _webp.describe()) + except Exception: pass + + return lines diff --git a/nomadnet/ui/textui/images/_webp.py b/nomadnet/ui/textui/images/_webp.py new file mode 100644 index 0000000..e47743c --- /dev/null +++ b/nomadnet/ui/textui/images/_webp.py @@ -0,0 +1,367 @@ +# Multi-backend WebP decoding +# +# Kitty's graphics protocol decodes PNG natively but not WebP, so WebP +# sources are converted transparently at the ImageData layer. This module +# provides a conversion chain with that *should* work almost anywhere with +# zero extra dependencies: +# +# - Pillow/PIL, if installed with WebP support +# - libwebp via ctypes, decoding to RGBA, paired with a pure-Python PNG +# encoder. The system libwebp is present on most Linux/macOS systems +# as a transitive dependency of common graphics stacks and browsers +# - sips - the macOS built-in image tool +# - dwebp - from the system PATH (the "official" WebP tool) +# - ImageMagick / GraphicsMagick (magick, convert, gm convert) +# - ffmpeg / avconv +# - gdk-pixbuf-thumbnailer (GTK/GNOME systems) +# +# Every backend's output is validated (it must be a valid PNG with the same +# dimensions as the source WebP), so missing tools, or tools built without +# WebP support, are simply skipped. The first backend that produces a valid +# conversion is remembered and preferred for subsequent conversions, but the +# preference is replaced whenever a different backend wins for a particular +# image. +# +# Environment overrides: +# +# NOMADNET_IMAGE_BACKEND force a specific backend (strict) +# NOMADNET_IMAGE_CACHE override the conversion cache directory + +import ctypes +import ctypes.util +import hashlib +import io +import os +import shutil +import struct +import subprocess +import tempfile +import zlib + +MAX_CACHE_BYTES = 192 * 1024 * 1024 +MAX_CACHE_FILES = 512 +CONVERSION_TIMEOUT = 5 + +# libwebp soname candidates, in order of preference. +_LIBWEBP_NAMES = [ "libwebp.so.7", + "libwebp.so.6", + "libwebp.so.5", + "libwebp.so", + "libwebp.dylib", + "libwebp-7.dll", + "libwebp.dll" ] + +# The built-in backend chain, in preference order. +BACKENDS = ( "pil", + "libwebp-ctypes", + "sips", + "dwebp", + "magick", + "convert", + "gm", + "ffmpeg", + "avconv", + "gdk-pixbuf" ) + +_ENV_BACKEND = os.environ.get("NOMADNET_IMAGE_BACKEND") +_winner = None # name of the currently preferred backend +_last_backend = None # backend used for the most recent conversion +_convert_count = 0 +_cache_hits = 0 + +_libwebp_cache = {"lib": None, "probed": False} + +############################### +# PNG validation and encoding # +############################### + +# Returns (width, height) if png is a valid PNG, else None +def _png_dims(png): + if not png.startswith(b"\x89PNG\r\n\x1a\n") or len(png) < 24: + return None + if png[12:16] != b"IHDR": + return None + width = int.from_bytes(png[16:20], "big") + height = int.from_bytes(png[20:24], "big") + if not (width > 0 and height > 0): + return None + return (width, height) + + +# Encodes RGBA pixel data as a PNG (8-bit RGBA, no filtering) +def _encode_png_rgba(rgba, width, height): + def chunk(tag, payload): + return (struct.pack(">I", len(payload)) + tag + payload + + struct.pack(">I", zlib.crc32(tag + payload) & 0xFFFFFFFF)) + + ihdr = struct.pack(">IIBBBBB", width, height, 8, 6, 0, 0, 0) + stride = width * 4 + raw = bytearray(height * (stride + 1)) + pos = 0 + for row in range(height): + raw[pos] = 0 # filter type: none + pos += 1 + start = row * stride + raw[pos:pos + stride] = rgba[start:start + stride] + pos += stride + + return (b"\x89PNG\r\n\x1a\n" + + chunk(b"IHDR", ihdr) + + chunk(b"IDAT", zlib.compress(bytes(raw), 9)) + + chunk(b"IEND", b"")) + + +#################### +# Conversion cache # +#################### + +def cache_dir(): + base = os.environ.get("NOMADNET_IMAGE_CACHE") + if not base: + base = os.environ.get("XDG_CACHE_HOME") or os.path.join(os.path.expanduser("~"), ".cache") + base = os.path.join(base, "nomadnet", "image-cache") + return base + +def _cache_path(source_data): + return os.path.join(cache_dir(), hashlib.sha256(source_data).hexdigest() + ".png") + +def _cache_get(source_data): + try: + with open(_cache_path(source_data), "rb") as f: png = f.read() + if _png_dims(png) is not None: return png + except OSError: pass + return None + +def _cache_put(source_data, png): + try: + directory = cache_dir() + os.makedirs(directory, exist_ok=True) + fd, tmp_path = tempfile.mkstemp(prefix=".tmp-", dir=directory) + try: + with os.fdopen(fd, "wb") as f: f.write(png) + os.replace(tmp_path, _cache_path(source_data)) + except OSError: + try: os.unlink(tmp_path) + except OSError: pass + _prune_cache() + except OSError: pass + +# Evicts least-recently-used entries beyond the size/file limits +def _prune_cache(): + try: + entries = [] + for name in os.listdir(cache_dir()): + path = os.path.join(cache_dir(), name) + try: + st = os.stat(path) + if st.st_size > 0: + entries.append((st.st_mtime, st.st_size, path)) + except OSError: pass + + total = sum(e[1] for e in entries) + entries.sort() + while len(entries) > MAX_CACHE_FILES or total > MAX_CACHE_BYTES: + _, size, path = entries.pop(0) + total -= size + try: os.unlink(path) + except OSError: pass + except OSError: pass + +####################### +# Individual backends # +####################### + +# Runs a conversion CLI over temp files +def _run_cli(argv, data): + if shutil.which(argv[0]) is None: return None + try: + with tempfile.TemporaryDirectory(prefix="nomadnet-img-") as td: + src = os.path.join(td, "source.webp") + dst = os.path.join(td, "output.png") + with open(src, "wb") as f: f.write(data) + args = [a.replace("{src}", src).replace("{dst}", dst) for a in argv] + subprocess.run( + args, + timeout=CONVERSION_TIMEOUT, + # stdin must be closed: some tools (e.g. ffmpeg) block + # on inherited stdin otherwise. + stdin=subprocess.DEVNULL, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + with open(dst, "rb") as f: return f.read() + except (OSError, subprocess.TimeoutExpired, ValueError): return None + +def _backend_pil(data, dims): + try: + from PIL import Image + from PIL import features + if not features.check("webp"): return None + except ImportError: return None + try: + image = Image.open(io.BytesIO(data)) + image.load() + buf = io.BytesIO() + image.save(buf, format="PNG") + return buf.getvalue() + except Exception: return None + +# Loads the system libwebp via ctypes, caching the result. +def _get_libwebp(): + if _libwebp_cache["probed"]: return _libwebp_cache["lib"] + _libwebp_cache["probed"] = True + lib = None + names = list(_LIBWEBP_NAMES) + found = ctypes.util.find_library("webp") + if found and found not in names: names.insert(0, found) + for name in names: + try: + lib = ctypes.CDLL(name) + break + except OSError: continue + + if lib is not None: + lib.WebPGetInfo.restype = ctypes.c_int + lib.WebPGetInfo.argtypes = [ ctypes.c_char_p, ctypes.c_size_t, + ctypes.POINTER(ctypes.c_int), ctypes.POINTER(ctypes.c_int) ] + + lib.WebPDecodeRGBA.restype = ctypes.c_void_p + lib.WebPDecodeRGBA.argtypes = [ ctypes.c_char_p, ctypes.c_size_t, + ctypes.POINTER(ctypes.c_int), ctypes.POINTER(ctypes.c_int) ] + lib.WebPFree.restype = None + lib.WebPFree.argtypes = [ctypes.c_void_p] + + _libwebp_cache["lib"] = lib + return lib + +def _backend_ctypes(data, dims): + lib = _get_libwebp() + if lib is None: return None + try: + buf = ctypes.create_string_buffer(data, len(data)) + width = ctypes.c_int(0) + height = ctypes.c_int(0) + if not lib.WebPGetInfo(buf, len(data), ctypes.byref(width), ctypes.byref(height)): return None + if width.value <= 0 or height.value <= 0: return None + ptr = lib.WebPDecodeRGBA(buf, len(data), ctypes.byref(width), ctypes.byref(height)) + if not ptr: return None + try: rgba = ctypes.string_at(ptr, width.value * height.value * 4) + finally: lib.WebPFree(ptr) + return _encode_png_rgba(rgba, width.value, height.value) + except Exception: return None + +def _backend_gdkpixbuf(data, dims): + if dims is None: return None + width, height = dims + return _run_cli(["gdk-pixbuf-thumbnailer", "-s", "%dx%d" % (width, height), "{src}", "{dst}"], data) + +def _backend_sips(data, dims): return _run_cli(["sips", "-s", "format", "png", "{src}", "--out", "{dst}"], data) +def _backend_dwebp(data, dims): return _run_cli(["dwebp", "{src}", "-o", "{dst}"], data) +def _backend_magick(data, dims): return _run_cli(["magick", "{src}", "{dst}"], data) +def _backend_convert(data, dims): return _run_cli(["convert", "{src}", "{dst}"], data) +def _backend_gm(data, dims): return _run_cli(["gm", "convert", "{src}", "{dst}"], data) +def _backend_ffmpeg(data, dims): return _run_cli(["ffmpeg", "-y", "-loglevel", "error", "-i", "{src}", "-frames:v", "1", "{dst}"], data) +def _backend_avconv(data, dims): return _run_cli(["avconv", "-y", "-loglevel", "error", "-i", "{src}", "-frames:v", "1", "{dst}"], data) + +_BACKENDS = { "pil": _backend_pil, + "libwebp-ctypes": _backend_ctypes, + "sips": _backend_sips, + "dwebp": _backend_dwebp, + "magick": _backend_magick, + "convert": _backend_convert, + "gm": _backend_gm, + "ffmpeg": _backend_ffmpeg, + "avconv": _backend_avconv, + "gdk-pixbuf": _backend_gdkpixbuf } + +##################### +# Conversion driver # +##################### + +def _webp_dims(data): + from ._imagedata import parse_image_header + try: fmt, width, height = parse_image_header(data) + except ValueError: return None + if fmt != "WebP": return None + return (width, height) + +def _backend_order(): + if _ENV_BACKEND: return [_ENV_BACKEND] if _ENV_BACKEND in BACKENDS else [] + order = list(BACKENDS) + if _winner in order: + order.remove(_winner) + order.insert(0, _winner) + return order + +# Converts WebP data to PNG bytes, or None on failure +def convert_webp_to_png(data): + global _cache_hits + png = _cache_get(data) + if png is not None: + _cache_hits += 1 + return png + return _convert_uncached(data) + +def _convert_uncached(data): + global _convert_count, _winner, _last_backend + dims = _webp_dims(data) + if dims is None: return None + for name in _backend_order(): + try: png = _BACKENDS[name](data, dims) + except Exception: png = None + if png is not None and _png_dims(png) == dims: + _convert_count += 1 + _last_backend = name + if not _ENV_BACKEND: _winner = name + _cache_put(data, png) + return png + + return None + +################################## +# Introspection and test support # +################################## + +# Returns [(name, available), ...] for the built-in chain +def available_backends(): + out = [] + for name in BACKENDS: + if name == "pil": + try: + from PIL import features + out.append((name, bool(features.check("webp")))) + except Exception: out.append((name, False)) + elif name == "libwebp-ctypes": out.append((name, _get_libwebp() is not None)) + elif name == "gdk-pixbuf": out.append((name, shutil.which("gdk-pixbuf-thumbnailer") is not None)) + else: out.append((name, shutil.which(name) is not None)) + + return out + +# A diagnostic string describing the conversion setup +def describe(): + if _ENV_BACKEND: return "backend forced via NOMADNET_IMAGE_BACKEND: %s" % _ENV_BACKEND + present = [name for name, avail in available_backends() if avail] + if not present: return "no conversion backend available (WebP images will not render)" + return "backend(s) available: %s; preferred: %s" % (", ".join(present), _last_backend or _winner or "none yet") + +# Usage statistics for diagnostics +def stats(): + return { "backend": _last_backend, "conversions": _convert_count, + "cache_hits": _cache_hits, "cache_dir": cache_dir() } + +def last_backend(): return _last_backend + +# Forces a specific backend name (testing/debugging); None to clear. +def force_backend(name): + global _ENV_BACKEND + _ENV_BACKEND = name if name in BACKENDS else None + +# Clears dynamic state (winner, counters); used by tests. +def reset(): + global _winner, _last_backend, _convert_count, _cache_hits + _winner = None + _last_backend = None + _convert_count = 0 + _cache_hits = 0 + _libwebp_cache["lib"] = None + _libwebp_cache["probed"] = False diff --git a/nomadnet/ui/textui/images/screen.py b/nomadnet/ui/textui/images/screen.py new file mode 100644 index 0000000..c7a9f59 --- /dev/null +++ b/nomadnet/ui/textui/images/screen.py @@ -0,0 +1,210 @@ +import urwid + +from . import _ctlseqs +from ._imagestore import image_store +from ._termlib import is_kitty_supported +from .widget import ImageWidget + +# A raw display screen subclass with Kitty TGP side-channel support +class ImageScreen(urwid.raw_display.Screen): + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self._ti_images = {} + + ########## + # Output # + ########## + + def write(self, data): + if isinstance(data, bytes): data = data.decode("ascii") + return super().write(data) + + ############# + # Lifecycle # + ############# + + def clear(self): + # Clears all images and forces a full repaint + if is_kitty_supported(): + self.write(_ctlseqs.KITTY_DELETE_ALL) + self._ti_images = {} + image_store.mark_all_untransmitted() + return super().clear() + + def _start(self, *args, **kwargs): + ret = super()._start(*args, **kwargs) + if is_kitty_supported(): + # Clear any images left over from a previous run/instance. + self.write(_ctlseqs.KITTY_DELETE_ALL) + self._ti_images = {} + image_store.mark_all_untransmitted() + return ret + + def _stop(self): + self.purge_images() + try: return super()._stop() + finally: self._ti_images = {} + + def purge_images(self): + # Purges all image data from the terminal buffer. + if not is_kitty_supported(): return + if self._ti_images or image_store.transmitted(): + self.write(_ctlseqs.KITTY_DELETE_ALL) + self._ti_images = {} + image_store.mark_all_untransmitted() + + ########### + # Drawing # + ########### + + def draw_screen(self, size, canvas): + # Draws the canvas, then updates on-screen images within a + # synchronized-update region to prevent flicker. + self.write(_ctlseqs.BEGIN_SYNCED_UPDATE) + try: + super().draw_screen(size, canvas) + self._ti_update(canvas) + finally: + self.write(_ctlseqs.END_SYNCED_UPDATE) + self.flush() + + def _ti_update(self, canvas): + # Places/updates/deletes image placements to match the current + # canvas, and evicts orphaned image data. + if not is_kitty_supported(): return + + entries = self._ti_scan(canvas) + current = {} + for widget, row, col, canvas_cols, trim_top, vis_rows in entries: + current[id(widget)] = (widget, row, col, canvas_cols, trim_top, vis_rows) + + # For widgets that scrolled completely out of view, remove their + # placement only. The image data is kept in the terminal so that + # scrolling back just re-places it. + for wid, info in list(self._ti_images.items()): + if wid in current: continue + self._ti_unplace(info["widget"]) + del self._ti_images[wid] + + # For visible widgets, place/update the visible slice in place. + for wid, (widget, row, col, canvas_cols, trim_top, vis_rows) in current.items(): + info = self._ti_images.setdefault(wid, {"widget": widget, "drawn": None}) + if vis_rows <= 0 or trim_top >= widget._ti_rows: + if info["drawn"] is not None: + self._ti_unplace(widget) + info["drawn"] = None + continue + + key = self._placement_key(widget, row, col, canvas_cols, trim_top, vis_rows) + if key is None: + if info["drawn"] is not None: + self._ti_unplace(widget) + info["drawn"] = None + continue + + if info["drawn"] == key: continue + self._ti_place(widget, key) + info["drawn"] = key + + # Evict image data that no live widget references any more + for key, image_id in image_store.evict_orphans(): + self.write(_ctlseqs.KITTY_DELETE_IMAGE % image_id) + image_store.evict(key) + + ############# + # Internals # + ############# + + def _ti_scan(self, canvas): + # Walks the canvas shards and returns a list of + # (widget, row, col, canvas_cols, trim_top, visible_rows) + # for every visible image canvas. + found = [] + if not isinstance(canvas, urwid.CompositeCanvas): return found + + def process_shard_tails(): + nonlocal col + while col in shard_tails: + *trim, cols, rows, canv = shard_tails[col] + if rows > n_rows: shard_tails[col] = (*trim, cols, rows - n_rows, canv) + else: del shard_tails[col] + col += cols + + shard_tails = {} + row = 0 + for n_rows, cviews in canvas.shards: + col = 0 + for cview in cviews: + process_shard_tails() + *trim, cols, rows, _, canv = cview + widget = self._ti_widget_of(canv) + if widget is not None: found.append((widget, row, col, cols, trim[1], rows)) + if rows > n_rows: shard_tails[col] = (*trim, cols, rows - n_rows, canv) + col += cols + + process_shard_tails() + row += n_rows + + return found + + @staticmethod + def _ti_widget_of(canv): + # Returns the ImageWidget for a leaf canvas, or None + widget_info = getattr(canv, "_widget_info", None) + if not widget_info: return None + widget = widget_info[0] + if isinstance(widget, ImageWidget) and widget._ti_data.ok and not widget._ti_closed: return widget + return None + + def _placement_key(self, widget, row, col, canvas_cols, trim_top, vis_rows): + # Compute placement parameters for the visible slice of + # an image, including the alignment offset. + + cols = widget._ti_cols + rows = widget._ti_rows + if not (cols and rows): return None + place_col = col + if cols < canvas_cols: + align = widget._ti_align + spare = canvas_cols - cols + if align == ">": place_col += spare + elif align == "|": place_col += spare // 2 + + vis_rows = max(1, min(vis_rows, rows)) + y_px = h_px = None + if not (trim_top == 0 and vis_rows == rows): + # Crop to the visible slice of the image, in image pixels + ih = widget._ti_data.height + y_px = min(ih - 1, round(trim_top * ih / rows)) + h_px = max(1, min(ih - y_px, round(vis_rows * ih / rows))) + + return (row, place_col, cols, vis_rows, y_px, h_px) + + def _ti_place(self, widget, key): + # Place (or in-place update) the visible slice of an image + row, place_col, cols, vis_rows, y_px, h_px = key + entry = image_store.entry_for(widget._ti_key) + if entry is None: + # This should not happen, but if the image data was evicted + # while the image was still alive, re-register it so the + # data can be re-transmitted to the terminal buffer. + image_store.register(widget._ti_key, widget._ti_data.data) + entry = image_store.entry_for(widget._ti_key) + image_id = entry["image_id"] + + if not entry["transmitted"]: + payload = widget._ti_data.data + b64_len = (len(payload) * 4 + 2) // 3 + n_chunks = max(1, (b64_len + _ctlseqs.CHUNK_SIZE - 1) // _ctlseqs.CHUNK_SIZE) + self.write(_ctlseqs.kitty_transmission(_ctlseqs.kitty_transmit_control(image_id), payload)) + entry["transmitted"] = True + + self.write(_ctlseqs.CURSOR_POSITION % (row + 1, place_col + 1)) + self.write(_ctlseqs.kitty_placement(image_id, widget._ti_z, cols, vis_rows, widget._ti_z, y=y_px, h=h_px)) + + def _ti_unplace(self, widget): + # Delete a widget's placement, but keep the + # image data in the terminal buffer. + entry = image_store.entry_for(widget._ti_key) + if entry is None: return + self.write(_ctlseqs.KITTY_DELETE_PLACEMENT % (entry["image_id"], widget._ti_z)) diff --git a/nomadnet/ui/textui/images/widget.py b/nomadnet/ui/textui/images/widget.py new file mode 100644 index 0000000..23cdaa0 --- /dev/null +++ b/nomadnet/ui/textui/images/widget.py @@ -0,0 +1,276 @@ +# An urwid widget that occupies the space an image should fill in a layout, +# without embedding any escape sequences in its canvas. The actual terminal +# graphics are transmitted by the ImageScreen class which tracks the widget's +# on-screen position after every draw. +# +# This separation keeps the canvas trivially simple (urwid sees only blank +# rows) and lets the screen handle the hard parts; positioning, deletion when +# scrolled away, terminal resize and redraws - all of which are driven by +# urwid's own shard bookkeeping rather than by fragile escape-sequence +# manipulation inside canvas rows. +# +# When the Kitty graphics protocol is unavailable (or the image cannot be +# loaded), the widget renders a simple text placeholder instead. + +import os +import re +import urwid + +from . import _termlib +from ._imagestore import image_store + +DEFAULT_CELL_SIZE = (1, 2) # Fallback, assume a 1:2 cell aspect ratio +MAX_Z_INDEX = 2 ** 31 - 1 +UI_MARGIN = 8 +_SIZE_SPEC_RE = re.compile(r"(\d+(?:\.\d+)?)%") +_ALIGNS = { "left": "<", "<": "<", + "center": "|", "|": "|", + "right": ">", ">": ">" } + +def _check_align(value): + if value not in _ALIGNS: raise ValueError("invalid alignment %r (expected 'left', 'center' or 'right')" % (value,)) + return _ALIGNS[value] + + +def _check_size_spec(value, name): + if value is None: return None + + if isinstance(value, int): + if value <= 0: raise ValueError("%s must be a positive integer, got %r" % (name, value)) + return value + + if isinstance(value, str): + match = _SIZE_SPEC_RE.fullmatch(value) + if not match or float(match.group(1)) <= 0: + raise ValueError("invalid %s specification %r (expected a positive integer or a percent string like '50%%')" % (name, value)) + return value + + raise TypeError("invalid type for %s (got %s)" % (name, type(value).__name__)) + + +class ImageWidget(urwid.Widget): + # A widget that displays an image via the Kitty graphics protocol. + # The widget is a plain FLOW/BOX widget; its canvas is blank. The + # ImageScreen class performs the actual drawing over the Kitty + # TGP side-channel. + + placeholder_glyph = "" + + _sizing = frozenset((urwid.FLOW, urwid.BOX)) + ignore_focus = True + + # Z-index allocation. Each widget instance gets a unique z-index so that + # the screen can delete exactly one image when it scrolls away or moves. + _ti_next_z = 1 + _ti_free_z = set() + + def __init__(self, path, name=None, width=None, height=None, align="center"): + if path and not isinstance(path, (str, os.PathLike)): raise TypeError("path must be a string or path-like object") + self.path = path + self._notice = None + self._ti_path = None + self._ti_data = None + self._ti_width = _check_size_spec(width, "width") + self._ti_height = _check_size_spec(height, "height") + self._ti_align = _check_align(align) + self._ti_name = name if name is not None else "" + self._ti_z = self._ti_alloc_z() + self.image_id = self._ti_z + + # Display size (cols, rows) from the most recent render. + # Used by the screen to place the image. + self._ti_cols = None + self._ti_rows = None + self._ti_closed = False + self._ti_key = None + super().__init__() + if self.path: self.load() + + def load(self, path=None): + if path: self.path = path + if self.path: + from ._imagedata import ImageData + self._ti_path = os.path.abspath(os.fspath(self.path)) + self._ti_data = ImageData(self._ti_path) + + if self._ti_data.ok: + # Register with the image store (de-duplicates identical data + # and allocates the terminal-side image id on first use). + self._ti_key = self._ti_data.key + image_store.register(self._ti_key, self._ti_data.data) + self._invalidate() + + def notice(self, msg=""): + if not msg: self._notice = None + else: self._notice = str(msg) + self._invalidate() + + def __del__(self): + try: type(self)._ti_free_z.add(self._ti_z) + except Exception: pass + try: self.close() + except Exception: pass + + def close(self): + # Releases the widget's reference to its image data. + # The data is evicted from the terminal by the screen as soon as no + # live widget references it any more. Safe to call multiple times and + # from __del__. + try: + if not self._ti_closed: + self._ti_closed = True + if self._ti_key is not None: image_store.unregister(self._ti_key) + except Exception: pass + + @classmethod + def _ti_alloc_z(cls): + if cls._ti_free_z: return cls._ti_free_z.pop() + z = cls._ti_next_z + if z >= MAX_Z_INDEX: raise RuntimeError("Too many image widgets") + cls._ti_next_z += 1 + return z + + ########## + # Sizing # + ########## + + width = property( + lambda self: self._ti_width, + doc="""Display width specification: None (full width), a positive + int (columns) or a percent string ("NN%") of the available layout + width.""", + ) + + @width.setter + def width(self, value): + self._ti_width = _check_size_spec(value, "width") + self._invalidate() + + height = property( + lambda self: self._ti_height, + doc="""Display height specification: None (derived, aspect + preserving), a positive int (rows) or a percent string ("NN%") of + the visible page height.""", + ) + + @height.setter + def height(self, value): + self._ti_height = _check_size_spec(value, "height") + self._invalidate() + + align = property( + lambda self: self._ti_align, + doc="""Horizontal alignment of the image within its allocated + space: ``"<"`` (left), ``"|"`` (center) or ``">"`` (right).""", + ) + + @align.setter + def align(self, value): + self._ti_align = _check_align(value) + self._invalidate() + + def _short(self): return "img#%x" % (id(self) & 0xFFFFF) + + def _display_size(self, maxcol, maxrow=None): + if not self._ti_data.ok: return (min(maxcol, 1), 1) + iw, ih = self._ti_data.size + cell_w, cell_h = _termlib.get_cell_size() or DEFAULT_CELL_SIZE + w_spec, h_spec = self._ti_width, self._ti_height + + def _pct(value): return float(value[:-1]) / 100.0 + + def _raw_width(): + if w_spec is None: return float(maxcol) + if isinstance(w_spec, int): return float(w_spec) + return maxcol * _pct(w_spec) + + def _raw_height(): + if h_spec is None: return None + if isinstance(h_spec, int): return float(h_spec) + columns, lines = _termlib.get_terminal_size() + page = max(1, lines - UI_MARGIN) + return page * _pct(h_spec) + + def _height_for_width(cols): return cols * cell_w * ih / (cell_h * iw) + def _width_for_height(rows): return rows * cell_h * iw / (cell_w * ih) + + if w_spec is not None and h_spec is not None: + # Stretch mode: Target rectangle, uniformly scaled down only if + # it would overflow the available space. + target_w = max(1.0, _raw_width()) + target_h = max(1.0, _raw_height()) + scale = min(1.0, maxcol / target_w) + if maxrow is not None: scale = min(scale, maxrow / target_h) + cols = max(1, round(target_w * scale)) + rows = max(1, round(target_h * scale)) + elif w_spec is not None: + # Width given: Height derived, aspect preserving. + cols = max(1, min(maxcol, round(_raw_width()))) + rows = max(1, round(_height_for_width(cols))) + if maxrow is not None and rows > maxrow: + rows = maxrow + cols = max(1, min(maxcol, round(_width_for_height(rows)))) + elif h_spec is not None: + # Height given: Width derived, aspect preserving. + rows = max(1, round(_raw_height())) + cols = max(1, min(maxcol, round(_width_for_height(rows)))) + if maxrow is not None and rows > maxrow: + rows = maxrow + cols = max(1, min(maxcol, round(_width_for_height(rows)))) + else: + # Default: Full width, height derived. + cols = maxcol + rows = max(1, round(_height_for_width(cols))) + if maxrow is not None and rows > maxrow: + rows = maxrow + cols = max(1, min(maxcol, round(_width_for_height(rows)))) + + return (cols, rows) + + def rows(self, size, focus=False): + maxcol = size[0] + if not (self._ti_data and self._ti_data.ok and _termlib.is_kitty_supported()): + return len(self._placeholder_lines()) + n = self._display_size(maxcol)[1] + return n + + def render(self, size, focus=False): + if len(size) == 2: + maxcol, maxrow = size + is_box = True + else: + maxcol, maxrow = size[0], None + is_box = False + + if not (self._ti_data and self._ti_data.ok and _termlib.is_kitty_supported()): + return self._placeholder_canvas(maxcol, maxrow, focus) + + cols, rows = self._display_size(maxcol, maxrow) + self._ti_cols, self._ti_rows = cols, rows + # urwid requires canvases to fill the allocated space exactly + # (width == maxcol for FLOW, (maxcol, maxrow) for BOX). The image + # itself is placed by the screen at its resolved size inside the + # canvas, left-aligned; the rest of the canvas stays blank. + if is_box: return ImageCanvas(maxcol, maxrow) + else: return ImageCanvas(maxcol, rows) + + ######################### + # Placeholder rendering # + ######################### + + def _placeholder_lines(self): + return [f"{self.placeholder_glyph}{self._notice or self._ti_name}"] + + def _placeholder_canvas(self, maxcol, maxrow, focus): + align = {"<": "left", "|": "center", ">": "right"}.get(self._ti_align, "center") + text = urwid.Text("\n".join(self._placeholder_lines()), align=align) + if maxrow is not None: return urwid.Filler(text).render((maxcol, maxrow), focus) + else: return text.render((maxcol,), focus) + + +class ImageCanvas(urwid.SolidCanvas): + # A blank canvas marking the space occupied by an image. + # Urwid sets _widget_info on this canvas to the owning + # ImageWidget. ImageScreen finds image canvases through + # that reference during its shard scan. + def __init__(self, cols, rows): super().__init__(" ", cols, rows)