mirror of
https://github.com/markqvist/NomadNet.git
synced 2026-10-05 23:27:24 +00:00
Added image display support
This commit is contained in:
Mark Qvist
+39
-1
@@ -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")
|
||||
|
||||
@@ -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."
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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":
|
||||
|
||||
+15
-12
@@ -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):
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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" )
|
||||
@@ -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 <control-data> ; <base64 payload> 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<message>[^\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 ; <height-px> ; <width-px> t
|
||||
CELL_SIZE_RESPONSE_RE = re.compile(b"\x1b\\[4;(?P<h>\\d+);(?P<w>\\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))
|
||||
@@ -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
|
||||
@@ -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()
|
||||
@@ -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("<I", 30) + b"WEBP" + b"VP8 "
|
||||
+ b"\x00" * 10 + b"\x10\x00\x10\x00" + b"\xff" * 12)
|
||||
check(parse_image_header(junk)[1:] == (16, 16),
|
||||
"junk fixture parses as a 16x16 WebP")
|
||||
fd, webp_path = tempfile.mkstemp(suffix=".webp")
|
||||
with os.fdopen(fd, "wb") as f:
|
||||
f.write(junk)
|
||||
d = ImageData(webp_path)
|
||||
check(not d.ok and "conversion failed" in d.error,
|
||||
"undecodable webp results in the error state")
|
||||
os.unlink(webp_path)
|
||||
finally:
|
||||
if prev_cache_env is not None:
|
||||
os.environ["NOMADNET_IMAGE_CACHE"] = prev_cache_env
|
||||
else:
|
||||
os.environ.pop("NOMADNET_IMAGE_CACHE", None)
|
||||
shutil.rmtree(cache_tmp, ignore_errors=True)
|
||||
_webp.reset()
|
||||
|
||||
|
||||
def test_winsize_parsing():
|
||||
print("== TIOCGWINSZ parsing ==")
|
||||
from nomadnet.ui.textui.images import _termlib
|
||||
# struct order is (rows, cols, xpixel, ypixel)
|
||||
check(_termlib._cell_size_from_winsize([40, 100, 1000, 800]) == (10, 20),
|
||||
"correct winsize -> (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())
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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))
|
||||
@@ -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)
|
||||
Reference in New Issue
Block a user