Download hdc/os.py from AEUPH/hdcstorage: direct link, hf CLI and curl.
- Browser
- Download file 28.2 kB
-
https://huggingface.co/AEUPH/hdcstorage/resolve/main/hdc/os.py
- Command line
-
hf download hf://AEUPH/hdcstorage/hdc/os.py
-
curl -L -o os.py https://huggingface.co/AEUPH/hdcstorage/resolve/main/hdc/os.py
28.2 kB
| """The webOS: desktop, windows, programs, and the memory map over the HDC machine. | |
| Everything a desktop needs already exists in this project, so this module is the | |
| part that arranges it: | |
| * **memory** — :class:`MemoryMap` reads the same :class:`~hdc.memory.HDCStorage` | |
| the programs run from and buckets cells by what wrote them, plus a bump | |
| allocator so an app can reserve and release cells. | |
| * **filesystem** — :mod:`hdc.fs`, directories by binding, contents exact. | |
| * **programs** — the kernel console, the IDE (assembler + machine + debugger), | |
| the file manager, an editor over the same filesystem, a shader lab that renders | |
| through :mod:`hdc.gfx`, and a system monitor. | |
| * **desktop** — a window manager on an 80×30 character grid, drawn by the | |
| graphics engine: wallpaper shader, window frames, a menu bar and a taskbar. | |
| The compositor is deliberately dull, because it is the honest part: windows are | |
| drawn in z-order into a :class:`~hdc.gfx.Surface`, and an unfocused window is | |
| darkened with ``dim_region`` — an **exact** ``shift`` on the coded cells, not a | |
| multiply on a copy. What the screen shows is the state the machine holds. | |
| Honest limits: character cells, not pixels; no compositing effects beyond dim, | |
| no overlapping-alpha; one desktop, no multi-user session; and the wallpaper shader | |
| is rendered per frame, so a redraw costs tens of milliseconds (see | |
| :func:`hdc.gfx.report`) rather than the sub-millisecond a GPU would take. | |
| """ | |
| from __future__ import annotations | |
| import time | |
| from .fs import FileSystem, seeded | |
| from .gfx import SHADERS, Surface | |
| from .hypervec import DEFAULT_DIM | |
| from .memory import HDCStorage | |
| #: Desktop geometry, in character cells. 80 columns so it maps onto the CRT in | |
| #: the browser without reflow. | |
| COLS = 80 | |
| ROWS = 30 | |
| #: Window presets. ``(width, height)`` at launch; positions cascade. | |
| APPS = { | |
| "terminal": ("Terminal", 46, 14), | |
| "files": ("Files", 40, 13), | |
| "editor": ("Editor", 42, 14), | |
| "ide": ("IDE", 52, 16), | |
| "shaders": ("Shader Lab", 44, 15), | |
| "monitor": ("Monitor", 40, 14), | |
| } | |
| #: Launch order for the menu bar and the function keys. | |
| MENU = ("terminal", "files", "editor", "ide", "shaders", "monitor") | |
| # --------------------------------------------------------------------- memory | |
| class MemoryMap: | |
| """What the cells are being used for, and a bump allocator over them.""" | |
| def __init__(self, storage: HDCStorage): | |
| self.storage = storage | |
| self.reserved: dict[str, list[str]] = {} | |
| def buckets(self) -> dict[str, int]: | |
| """Cells grouped by the prefix of their name — who wrote them.""" | |
| counts: dict[str, int] = {} | |
| for name, _ in self.storage: | |
| key = name.split("/", 1)[0] if "/" in name else "cells" | |
| counts[key] = counts.get(key, 0) + 1 | |
| return dict(sorted(counts.items(), key=lambda item: -item[1])) | |
| def allocate(self, tag: str, count: int) -> list[str]: | |
| """Reserve ``count`` zero cells under ``mem/<tag>/`` and return their names.""" | |
| names: list[str] = [] | |
| for index in range(max(0, count)): | |
| name = "mem/%s/%d" % (tag, index) | |
| self.storage.write(name, 0) | |
| names.append(name) | |
| self.reserved.setdefault(tag, []).extend(names) | |
| return names | |
| def free(self, tag: str) -> int: | |
| names = self.reserved.pop(tag, []) | |
| return sum(1 for name in names if self.storage.remove(name)) | |
| def stat(self) -> dict: | |
| buckets = self.buckets() | |
| return {"cells": len(self.storage), "capacity": self.storage.capacity, | |
| "free": max(0, self.storage.capacity - len(self.storage)), | |
| "buckets": buckets, "reserved": {tag: len(names) | |
| for tag, names in self.reserved.items()}, | |
| "reads": self.storage.reads, "writes": self.storage.writes} | |
| def report(self, width: int = 28) -> str: | |
| stat = self.stat() | |
| lines = ["memory: %d / %d cells (%d free)" | |
| % (stat["cells"], stat["capacity"], stat["free"])] | |
| total = max(1, stat["cells"]) | |
| for name, count in stat["buckets"].items(): | |
| bar = "#" * max(1, round(count * width / total)) | |
| lines.append(" %-8s %5d %s" % (name[:8], count, bar)) | |
| if stat["reserved"]: | |
| lines.append(" reserved: " + ", ".join("%s=%d" % item | |
| for item in stat["reserved"].items())) | |
| lines.append(" reads %d, writes %d" % (stat["reads"], stat["writes"])) | |
| return "\n".join(lines) | |
| # -------------------------------------------------------------------- windows | |
| class Window: | |
| """A rectangle with a title, an app key, and per-window app state.""" | |
| __slots__ = ("id", "app", "title", "x", "y", "w", "h", "state", "scroll") | |
| def __init__(self, ident: int, app: str, title: str, x: int, y: int, w: int, h: int): | |
| self.id = ident | |
| self.app = app | |
| self.title = title | |
| self.x = x | |
| self.y = y | |
| self.w = w | |
| self.h = h | |
| self.state: dict = {} | |
| self.scroll = 0 | |
| def as_dict(self) -> dict: | |
| return {"id": self.id, "app": self.app, "title": self.title, | |
| "x": self.x, "y": self.y, "w": self.w, "h": self.h} | |
| def contains(self, x: int, y: int) -> bool: | |
| return self.x <= x < self.x + self.w and self.y <= y < self.y + self.h | |
| def inner_h(self) -> int: | |
| return max(0, self.h - 2) | |
| def inner_w(self) -> int: | |
| return max(0, self.w - 2) | |
| class App: | |
| """Base class: an app renders lines and handles keys.""" | |
| key = "app" | |
| title = "App" | |
| def render(self, desktop: "Desktop", window: Window) -> list[str]: | |
| return ["(nothing here)"] | |
| def on_key(self, desktop: "Desktop", window: Window, key: str) -> str: | |
| return "" | |
| def on_text(self, desktop: "Desktop", window: Window, text: str) -> str: | |
| return "" | |
| class TerminalApp(App): | |
| """The kernel console, inside a window.""" | |
| key = "terminal" | |
| title = "Terminal" | |
| def render(self, desktop: "Desktop", window: Window) -> list[str]: | |
| window.state.setdefault("buffer", list(desktop.kernel.term.screen()[-window.inner_h:])) | |
| window.state.setdefault("typing", "") | |
| lines = window.state["buffer"][-(window.inner_h - 1):] | |
| lines = [line for line in lines] | |
| lines.append("> " + window.state["typing"]) | |
| return lines | |
| def on_text(self, desktop: "Desktop", window: Window, text: str) -> str: | |
| window.state["typing"] = (window.state.get("typing", "") + text)[-window.inner_w:] | |
| return "" | |
| def on_key(self, desktop: "Desktop", window: Window, key: str) -> str: | |
| typing = window.state.get("typing", "") | |
| if key == "Backspace": | |
| window.state["typing"] = typing[:-1] | |
| return "" | |
| if key != "Enter": | |
| return "" | |
| if not typing.strip(): | |
| window.state["typing"] = "" | |
| return "" | |
| try: | |
| output = desktop.kernel.execute(typing) | |
| except Exception as error: # the console never dies | |
| output = "[error] %s" % error | |
| window.state["buffer"] = (window.state.get("buffer", []) + ["> " + typing] | |
| + output.splitlines())[-200:] | |
| window.state["typing"] = "" | |
| return "kernel: %s" % typing | |
| class FilesApp(App): | |
| """Directory browser over :mod:`hdc.fs`; Enter opens.""" | |
| key = "files" | |
| title = "Files" | |
| def render(self, desktop: "Desktop", window: Window) -> list[str]: | |
| path = window.state.setdefault("cwd", "/home/user") | |
| cursor = window.state.setdefault("cursor", 0) | |
| try: | |
| entries = desktop.fs.ls(path) | |
| except OSError as error: | |
| return ["[%s]" % error, "", "press u to go up"] | |
| window.state["entries"] = entries | |
| if entries: | |
| cursor = min(cursor, len(entries) - 1) | |
| window.state["cursor"] = cursor | |
| head = [path[:window.inner_w]] | |
| head.append("-" * min(window.inner_w, len(path))) | |
| for index, entry in enumerate(entries[:max(0, window.inner_h - 3)]): | |
| marker = ">" if index == cursor else " " | |
| suffix = "/" if entry["kind"] == "dir" else "" | |
| head.append("%s %s%s" % (marker, entry["name"], suffix)) | |
| return head | |
| def on_key(self, desktop: "Desktop", window: Window, key: str) -> str: | |
| entries = window.state.get("entries") or [] | |
| cursor = window.state.get("cursor", 0) | |
| path = window.state.get("cwd", "/home/user") | |
| if key in ("ArrowDown", "j"): | |
| window.state["cursor"] = min(len(entries) - 1, cursor + 1) if entries else 0 | |
| return "" | |
| if key in ("ArrowUp", "k"): | |
| window.state["cursor"] = max(0, cursor - 1) | |
| return "" | |
| if key == "u" or key == "Backspace": | |
| window.state["cwd"] = desktop.fs.parent(path) if path != "/" else "/" | |
| window.state["cursor"] = 0 | |
| return "up" | |
| if key == "Enter" and entries: | |
| entry = entries[min(cursor, len(entries) - 1)] | |
| if entry["kind"] == "dir": | |
| window.state["cwd"] = entry["path"] | |
| window.state["cursor"] = 0 | |
| return entry["path"] | |
| desktop.open_file(entry["path"]) | |
| return "opened " + entry["path"] | |
| return "" | |
| def on_text(self, desktop: "Desktop", window: Window, text: str) -> str: | |
| return "" | |
| class EditorApp(App): | |
| """A viewer that appends: typed text plus Enter writes a line to the file.""" | |
| key = "editor" | |
| title = "Editor" | |
| def render(self, desktop: "Desktop", window: Window) -> list[str]: | |
| path = window.state.get("path") | |
| if not path: | |
| return ["no file open", "", "use the Files window (F2) and press Enter"] | |
| window.state.setdefault("typing", "") | |
| window.state.setdefault("cursor", 0) | |
| try: | |
| lines = desktop.fs.read(path).splitlines() | |
| except OSError as error: | |
| return ["[%s]" % error] | |
| cursor = window.state["cursor"] | |
| visible = window.inner_h - 2 | |
| start = max(0, min(cursor - visible + 1, len(lines) - visible)) if len(lines) > visible else 0 | |
| out = ["%s (%d lines)" % (path[:window.inner_w], len(lines)), ""] | |
| for offset, line in enumerate(lines[start:start + visible]): | |
| number = start + offset + 1 | |
| mark = ">" if number - 1 == cursor else " " | |
| out.append("%s%3d %s" % (mark, number, line[:window.inner_w - 5])) | |
| out.append(": " + window.state["typing"]) | |
| return out | |
| def on_key(self, desktop: "Desktop", window: Window, key: str) -> str: | |
| path = window.state.get("path") | |
| typing = window.state.get("typing", "") | |
| if key == "Backspace": | |
| window.state["typing"] = typing[:-1] | |
| return "" | |
| if not path: | |
| return "" | |
| if key in ("ArrowDown", "j"): | |
| window.state["cursor"] = window.state.get("cursor", 0) + 1 | |
| return "" | |
| if key in ("ArrowUp", "k"): | |
| window.state["cursor"] = max(0, window.state.get("cursor", 0) - 1) | |
| return "" | |
| if key != "Enter": | |
| return "" | |
| if not typing: | |
| window.state["cursor"] = window.state.get("cursor", 0) + 1 | |
| return "" | |
| text = desktop.fs.read(path) | |
| desktop.fs.write(path, (text + "\n" + typing).strip("\n")) | |
| window.state["typing"] = "" | |
| window.state["cursor"] = len(desktop.fs.read(path).splitlines()) - 1 | |
| return "appended to %s" % path | |
| def on_text(self, desktop: "Desktop", window: Window, text: str) -> str: | |
| window.state["typing"] = (window.state.get("typing", "") + text)[-200:] | |
| return "" | |
| class IdeApp(App): | |
| """The native binary-logic IDE: assemble, run, step, inspect.""" | |
| key = "ide" | |
| title = "IDE" | |
| def render(self, desktop: "Desktop", window: Window) -> list[str]: | |
| window.state.setdefault("buffer", [ | |
| "native binary-logic IDE", | |
| "asm <file> | run | step | regs | disasm | bp 2 | bits R1", | |
| "", | |
| ]) | |
| window.state.setdefault("typing", "") | |
| lines = window.state["buffer"][-(window.inner_h - 1):] | |
| return list(lines) + ["> " + window.state["typing"]] | |
| def on_text(self, desktop: "Desktop", window: Window, text: str) -> str: | |
| window.state["typing"] = (window.state.get("typing", "") + text)[-window.inner_w:] | |
| return "" | |
| def on_key(self, desktop: "Desktop", window: Window, key: str) -> str: | |
| typing = window.state.get("typing", "") | |
| if key == "Backspace": | |
| window.state["typing"] = typing[:-1] | |
| return "" | |
| if key != "Enter" or not typing.strip(): | |
| window.state["typing"] = "" | |
| return "" | |
| try: | |
| output = desktop.ide.execute(typing) | |
| except Exception as error: | |
| output = "[error] %s" % error | |
| window.state["buffer"] = (window.state.get("buffer", []) + ["> " + typing] | |
| + (output or "").splitlines())[-200:] | |
| window.state["typing"] = "" | |
| return "ide: %s" % typing | |
| class ShaderApp(App): | |
| """Shader lab: the graphics engine, live inside a window.""" | |
| key = "shaders" | |
| title = "Shader Lab" | |
| def render(self, desktop: "Desktop", window: Window) -> list[str]: | |
| name = window.state.setdefault("shader", "metal") | |
| if name not in SHADERS: | |
| name = "metal" | |
| width, height = window.inner_w, max(1, window.inner_h - 2) | |
| surface = Surface(width, height, d=desktop.d) | |
| surface.apply(name) | |
| if window.state.get("edges"): | |
| surface.edges(mode=window.state.get("edges", "proxy")) | |
| window.state["frame"] = surface | |
| lines = surface.rows() | |
| lines.append("%s | %s | space: next shader, e: edges" | |
| % (name, "dim " + str(window.state.get("dim", 0)))) | |
| return lines | |
| def on_key(self, desktop: "Desktop", window: Window, key: str) -> str: | |
| names = list(SHADERS) | |
| current = window.state.get("shader", names[0]) | |
| if key == " ": | |
| window.state["shader"] = names[(names.index(current) + 1) % len(names)] | |
| return "shader: " + window.state["shader"] | |
| if key == "e": | |
| window.state["edges"] = None if window.state.get("edges") else "proxy" | |
| return "edges: %s" % ("proxy" if window.state.get("edges") else "off") | |
| if key in ("+", "=", "-"): | |
| steps = window.state.get("dim", 0) + (1 if key != "-" else -1) | |
| window.state["dim"] = max(-2, min(3, steps)) | |
| return "dim %+d" % window.state["dim"] | |
| return "" | |
| def on_text(self, desktop: "Desktop", window: Window, text: str) -> str: | |
| return "" | |
| class MonitorApp(App): | |
| """Memory, filesystem, shader timings, and the assistant's readahead.""" | |
| key = "monitor" | |
| title = "Monitor" | |
| def render(self, desktop: "Desktop", window: Window) -> list[str]: | |
| stat = desktop.memory.stat() | |
| files = desktop.fs.stat() | |
| lines = ["memory %d/%d cells, %d free" % (stat["cells"], stat["capacity"], stat["free"])] | |
| for name, count in list(stat["buckets"].items())[:4]: | |
| lines.append(" %-8s %4d cells" % (name[:8], count)) | |
| lines.append("fs %d files, %d dirs, %d bytes" | |
| % (files["files"], files["dirs"], files["bytes"])) | |
| lines.append("kernel %d instr, pc=%d" | |
| % (desktop.kernel.machine.steps, desktop.kernel.machine.pc)) | |
| if desktop.autonomy is not None: | |
| stats = desktop.autonomy.stats() | |
| lines.append("readahead %d/%d correct (%.0f%%), %.2f chance" | |
| % (stats["hits"], stats["graded"], | |
| 100.0 * stats["accuracy"], stats["chance"])) | |
| lines.append("gfx: %s" % ", ".join(SHADERS)) | |
| lines.append("determinism: cells are vectors, not pixels") | |
| return lines | |
| APP_CLASSES = {cls.key: cls for cls in | |
| (TerminalApp, FilesApp, EditorApp, IdeApp, ShaderApp, MonitorApp)} | |
| # -------------------------------------------------------------------- desktop | |
| class Desktop: | |
| """Window manager, launcher, and the frame the browser draws.""" | |
| def __init__(self, d: int = DEFAULT_DIM, kernel=None, fs: FileSystem | None = None, | |
| autonomy=None, cols: int = COLS, rows: int = ROWS): | |
| from .ide import IDE | |
| from .terminal import Kernel | |
| self.d = d | |
| # explicit None checks: an empty Kernel has __len__ 0 and is therefore | |
| # falsy, so ``kernel or Kernel(d)`` would build a second machine | |
| self.kernel = kernel if kernel is not None else Kernel(d=d) | |
| # one store: the filesystem writes into the machine's cells, so the | |
| # memory map and the window contents cannot disagree | |
| self.fs = fs if fs is not None else seeded(d, storage=self.kernel.storage) | |
| self.autonomy = autonomy | |
| self.cols = cols | |
| self.rows = rows | |
| self.memory = MemoryMap(self.kernel.storage) | |
| self.ide = IDE(d=d) | |
| self.windows: list[Window] = [] | |
| self.focus = 0 | |
| self.next_id = 1 | |
| self.status = "F1 terminal F2 files F3 editor F4 ide F5 shaders F6 monitor" | |
| self.wallpaper = "background" | |
| self.frames = 0 | |
| self.last_frame_ms = 0.0 | |
| self.started = time.time() | |
| # --------------------------------------------------------------- windows | |
| def find(self, key: str) -> Window | None: | |
| for window in self.windows: | |
| if window.app == key: | |
| return window | |
| return None | |
| def launch(self, key: str, x: int | None = None, y: int | None = None) -> Window: | |
| existing = self.find(key) | |
| if existing is not None: | |
| self.raise_window(existing) | |
| return existing | |
| if key not in APPS: | |
| raise KeyError("unknown app %r" % (key,)) | |
| title, width, height = APPS[key] | |
| offset = (len(self.windows) * 3) % max(1, self.cols - width - 2) | |
| x = 2 + offset if x is None else x | |
| y = 2 + (len(self.windows) * 2) % max(1, self.rows - height - 4) if y is None else y | |
| window = Window(self.next_id, key, title, x, y, | |
| min(width, self.cols - 1), min(height, self.rows - 3)) | |
| window.state["app"] = APP_CLASSES[key]() | |
| self.next_id += 1 | |
| self.windows.append(window) | |
| self.focus = len(self.windows) - 1 | |
| self.status = "launched " + title | |
| return window | |
| def focused(self) -> Window | None: | |
| if not self.windows: | |
| return None | |
| self.focus = max(0, min(self.focus, len(self.windows) - 1)) | |
| return self.windows[self.focus] | |
| def raise_window(self, window: Window) -> None: | |
| if window in self.windows: | |
| self.windows.remove(window) | |
| self.windows.append(window) | |
| self.focus = len(self.windows) - 1 | |
| def close(self, window: Window | None = None) -> bool: | |
| window = window or self.focused() | |
| if window is None: | |
| return False | |
| self.windows.remove(window) | |
| self.focus = max(0, self.focus - 1) | |
| self.status = "closed " + window.title | |
| return True | |
| def focus_next(self, step: int = 1) -> str: | |
| if not self.windows: | |
| return "" | |
| self.focus = (self.focus + step) % len(self.windows) | |
| window = self.windows[self.focus] | |
| self.status = "focus: " + window.title | |
| return window.title | |
| def move_window(self, dx: int, dy: int) -> str: | |
| window = self.focused() | |
| if window is None: | |
| return "" | |
| window.x = max(0, min(self.cols - window.w, window.x + dx)) | |
| window.y = max(1, min(self.rows - 1 - window.h, window.y + dy)) | |
| self.status = "%s at %d,%d" % (window.title, window.x, window.y) | |
| return self.status | |
| def click(self, x: int, y: int) -> str: | |
| """Topmost window under the cell takes focus. | |
| Sets ``status`` as well as returning it: the browser endpoint reports | |
| ``state()['status']``, so a click that only returned a string looked like | |
| it did nothing at all. | |
| """ | |
| for window in reversed(self.windows): | |
| if window.contains(x, y): | |
| self.raise_window(window) | |
| self.status = "focus: " + window.title | |
| return self.status | |
| return "" | |
| def open_file(self, path: str) -> Window: | |
| window = self.launch("editor") | |
| window.state["path"] = path | |
| window.state["cursor"] = 0 | |
| window.title = "Editor - " + path.rsplit("/", 1)[-1] | |
| return window | |
| # -------------------------------------------------------------- keyboard | |
| #: Function keys, in menu order. | |
| FUNCTION_KEYS = {"F1": "terminal", "F2": "files", "F3": "editor", | |
| "F4": "ide", "F5": "shaders", "F6": "monitor"} | |
| def key(self, name: str) -> dict: | |
| """Dispatch one named key. Returns what happened, for the caller to show.""" | |
| if name in self.FUNCTION_KEYS: | |
| window = self.launch(self.FUNCTION_KEYS[name]) | |
| return {"action": "launch", "window": window.as_dict(), "status": self.status} | |
| if name in ("Tab", "Ctrl+Tab"): | |
| return {"action": "focus", "status": self.focus_next(1 if name == "Tab" else -1)} | |
| if name in ("Ctrl+W", "Escape"): | |
| return {"action": "close", "closed": self.close(), "status": self.status} | |
| if name in ("Ctrl+Left", "Ctrl+Right", "Ctrl+Up", "Ctrl+Down"): | |
| step = {"Left": (-3, 0), "Right": (3, 0), "Up": (0, -1), "Down": (0, 1)}[name.split("+")[1]] | |
| return {"action": "move", "status": self.move_window(*step)} | |
| window = self.focused() | |
| if window is None: | |
| return {"action": "none", "status": "no window: press F1"} | |
| app = window.state.get("app") | |
| try: | |
| status = app.on_key(self, window, name) if app else "" | |
| except Exception as error: # an app may not kill the desktop | |
| status = "[error] %s" % error | |
| if status: | |
| self.status = status | |
| return {"action": "key", "key": name, "status": self.status} | |
| def text(self, value: str) -> dict: | |
| """Typed characters go to the focused window's app.""" | |
| window = self.focused() | |
| if window is None: | |
| return {"action": "none", "status": "no window: press F1"} | |
| app = window.state.get("app") | |
| try: | |
| status = app.on_text(self, window, value) if app else "" | |
| except Exception as error: | |
| status = "[error] %s" % error | |
| if status: | |
| self.status = status | |
| return {"action": "text", "value": value, "status": self.status} | |
| # ----------------------------------------------------------------- frame | |
| def menu_bar(self) -> str: | |
| left = " HDCSTORAGE webOS " | |
| middle = " ".join("F%d %s" % (index + 1, key) for index, key in enumerate(MENU)) | |
| right = " d=%d cells=%d " % (self.d, len(self.kernel.storage)) | |
| width = self.cols | |
| text = left + "| " + middle | |
| pad = max(1, width - len(text) - len(right)) | |
| return (text + " " * pad + right)[:width] | |
| def taskbar(self) -> str: | |
| if not self.windows: | |
| return (" " + self.status)[:self.cols] | |
| titles = " ".join("[%s]" % window.title[:10] for window in self.windows) | |
| return (" " + titles + " | " + self.status)[:self.cols] | |
| def frame(self) -> list[str]: | |
| """Composite the desktop: wallpaper, windows, menu, taskbar.""" | |
| started = time.perf_counter() | |
| surface = Surface(self.cols, self.rows, d=self.d) | |
| surface.apply(self.wallpaper) | |
| surface.dim_region(1) | |
| surface.fill(0, 0, self.cols, 1, 210) | |
| surface.text_at(0, 0, self.menu_bar(), 255) | |
| focused = self.focused() | |
| for window in self.windows: | |
| if window is focused: | |
| continue | |
| self._draw(surface, window, False) | |
| if focused is not None: | |
| self._draw(surface, focused, True) | |
| surface.fill(0, self.rows - 1, self.cols, 1, 200) | |
| surface.text_at(0, self.rows - 1, self.taskbar(), 255) | |
| self.frames += 1 | |
| self.last_frame_ms = (time.perf_counter() - started) * 1000.0 | |
| return surface.rows() | |
| def _draw(self, surface: Surface, window: Window, focused: bool) -> None: | |
| window.x = max(0, min(self.cols - window.w, window.x)) | |
| window.y = max(1, min(self.rows - 2 - window.h, window.y)) | |
| surface.fill(window.x, window.y, window.w, window.h, 4) | |
| surface.frame(window.x, window.y, window.w, window.h, 190 if focused else 70) | |
| title = " %s " % window.title | |
| surface.text_at(window.x + 1, window.y, title[:window.w - 3], 245 if focused else 120) | |
| app = window.state.get("app") | |
| body = app.render(self, window) if app else [] | |
| for index, line in enumerate(body[:window.inner_h]): | |
| surface.text_at(window.x + 1, window.y + 1 + index, | |
| line[:window.inner_w], 220 if focused else 90) | |
| if not focused: | |
| water = 1 | |
| surface.dim_region(water, window.x + 1, window.y + 1, | |
| max(0, window.w - 2), max(0, window.h - 2)) | |
| def start(self) -> "Desktop": | |
| """Open the usual tiled session: terminal, monitor, files. Idempotent.""" | |
| if self.windows: | |
| return self | |
| self.launch("terminal", 1, 1) | |
| self.launch("monitor", 48, 1) | |
| self.launch("files", 1, 15) | |
| self.focus = 0 | |
| self.status = "F1 terminal F2 files F3 editor F4 ide F5 shaders F6 monitor" | |
| return self | |
| # ------------------------------------------------------------------ state | |
| def state(self) -> dict: | |
| stats = self.memory.stat() | |
| return { | |
| "frame": self.frame(), | |
| "cols": self.cols, "rows": self.rows, | |
| "windows": [window.as_dict() for window in self.windows], | |
| "focus": self.focus, | |
| "status": self.status, | |
| "wallpaper": self.wallpaper, | |
| "frames": self.frames, | |
| "frame_ms": round(self.last_frame_ms, 2), | |
| "apps": list(APPS), | |
| "fs": self.fs.stat(), | |
| "memory": stats, | |
| "cwd": self.fs.cwd, | |
| } | |
| def report(self) -> str: | |
| stat = self.memory.stat() | |
| return "\n".join([ | |
| "webOS: %dx%d cells, %d windows, %d frames, %.1f ms/frame" | |
| % (self.cols, self.rows, len(self.windows), self.frames, self.last_frame_ms), | |
| " memory %d/%d cells (%s)" | |
| % (stat["cells"], stat["capacity"], | |
| ", ".join("%s=%d" % item for item in stat["buckets"].items())), | |
| " fs %s" % self.fs.stat(), | |
| " apps: " + ", ".join(APPS), | |
| ]) | |
| def boot(d: int = DEFAULT_DIM) -> Desktop: | |
| """A desktop with the usual windows tiled, as a session would start. | |
| Explicit rectangles rather than a cascade: three windows on an 80x30 grid | |
| look like a desktop when they tile, and like a pile when they cascade. | |
| """ | |
| return Desktop(d=d).start() | |
| def tour(d: int = DEFAULT_DIM) -> str: | |
| """Scripted session, printed — the same actions the browser can send.""" | |
| desktop = boot(d) | |
| desktop.key("F5") | |
| desktop.key("Enter") # files: open the selection | |
| for _ in range(3): | |
| desktop.text("l") | |
| desktop.key("ArrowDown") | |
| desktop.key("F1") | |
| desktop.text("stat") | |
| desktop.key("Enter") | |
| lines = ["=== frame after tour ==="] | |
| lines.extend(desktop.frame()) | |
| lines.append("") | |
| lines.append(desktop.report()) | |
| lines.append(desktop.memory.report()) | |
| return "\n".join(lines) | |
| if __name__ == "__main__": # pragma: no cover - manual inspection | |
| print(tour(2000)) | |