hdcstorage / hdc /os.py
AEUPH's picture
HDCSTORAGE: hyperdimensional computer, webOS and agentic MCP layer (stdlib-only Python)
ad474c9 verified
Raw History Blame Contribute Delete
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
@property
def inner_h(self) -> int:
return max(0, self.h - 2)
@property
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))