diff options
| author | Arthur Hanson Rasmusson <arthur@vgpu.io> | 2026-09-29 16:52:18 -0400 |
|---|---|---|
| committer | Arthur Hanson Rasmusson <arthur@vgpu.io> | 2026-09-29 16:52:18 -0400 |
| commit | f049f8613a1e54ba2578e78727f6782a5c45494e (patch) | |
| tree | 20cae54559cd2e2bf15e2e15702808a249e30619 /scripts/browse-project.py | |
| parent | 74260a836396607ded3f7d64494a27802d30abcd (diff) | |
| download | open-iov-f049f8613a1e54ba2578e78727f6782a5c45494e.tar.gz open-iov-f049f8613a1e54ba2578e78727f6782a5c45494e.zip | |
scripts: add browse-project.py - read-only TUI project browser
- docs reader with rendered Markdown and a built-in Mermaid renderer
(flowchart TB/LR with subgraphs, stateDiagram-v2, sequenceDiagram)
- syntax-highlighted code viewer with per-file added-line highlighting
- foldable file tree; search via git grep
- history: browse docs/code at any commit or branch without checkout
- architecture explorer: zoom whole system -> module -> components, with
calls-into and source-file jump
- strictly read-only; non-interactive modes (--render/--diagrams/--tree/--log)
Diffstat (limited to 'scripts/browse-project.py')
| -rwxr-xr-x | scripts/browse-project.py | 2362 |
1 files changed, 2362 insertions, 0 deletions
diff --git a/scripts/browse-project.py b/scripts/browse-project.py new file mode 100755 index 0000000..b757d50 --- /dev/null +++ b/scripts/browse-project.py @@ -0,0 +1,2362 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +browse-project.py -- read-only TUI browser for the Open-IOV tree. + +Browse docs (rendered Markdown + Mermaid diagrams), code (syntax highlighted), +the file tree, the project history (any commit/branch, without checkout) and an +interactive architecture explorer. Reads only: nothing is ever written outside +of this process, no git checkout, no file modification. + +Run from the repository root: + + ./scripts/browse-project.py + +Non-interactive helpers (useful in a pipe / CI / quick look): + + ./scripts/browse-project.py --tree [REV] + ./scripts/browse-project.py --render docs/README.md [--rev REV] [--no-color] + ./scripts/browse-project.py --diagrams docs/03-architecture/09-full-system-picture.md + ./scripts/browse-project.py --log [N] + ./scripts/browse-project.py --version +""" + +from __future__ import annotations + +import argparse +import curses +import shutil +import os +import re +import subprocess +import sys +import textwrap +import unicodedata +from collections import OrderedDict, defaultdict +from dataclasses import dataclass, field + +VERSION = "1.0" +ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + +# -------------------------------------------------------------------------- +# git plumbing (strictly read-only) +# -------------------------------------------------------------------------- + + +class Git: + def __init__(self, root: str): + self.root = root + self._cache: dict = {} + + def run(self, *args: str, check: bool = False) -> str: + try: + p = subprocess.run( + ["git", "-C", self.root, *args], + capture_output=True, text=True, timeout=30, + ) + except (OSError, subprocess.TimeoutExpired) as e: + if check: + raise RuntimeError(f"git {' '.join(args)} failed: {e}") + return "" + if check and p.returncode: + raise RuntimeError(f"git {' '.join(args)}: {p.stderr.strip()}") + return p.stdout + + def head(self) -> str: + return self.run("rev-parse", "--short", "HEAD").strip() or "?" + + def resolve(self, rev: str) -> str: + return self.run("rev-parse", "--short", rev).strip() or rev + + def describe(self, rev: str) -> str: + return self.run("log", "-1", "--format=%h %s", rev).strip() + + def default_branch(self) -> str: + return self.run("rev-parse", "--abbrev-ref", "HEAD").strip() or "HEAD" + + def ls_files(self, rev: str | None) -> list[str]: + key = ("ls", rev) + if key not in self._cache: + if rev: + out = self.run("ls-tree", "-r", "--name-only", rev) + else: + out = self.run("ls-files", "--cached", "--others", "--exclude-standard") + files = [l for l in out.splitlines() if l.strip()] + self._cache[key] = files + return self._cache[key] + + def show(self, path: str, rev: str | None) -> str | None: + key = ("show", rev, path) + if key in self._cache: + return self._cache[key] + text = None + if rev: + text = self.run("show", f"{rev}:{path}") + if not text: + text = None + else: + full = os.path.join(self.root, path) + if os.path.isfile(full): + try: + with open(full, "r", encoding="utf-8", errors="replace") as fh: + text = fh.read() + except OSError: + text = None + self._cache[key] = text + return text + + def log(self, n: int = 400) -> list[str]: + return [l for l in self.run("log", "--oneline", "--decorate", "--no-color", + f"-n{n}").splitlines() if l.strip()] + + def branches(self) -> list[str]: + return [l.strip() for l in self.run( + "branch", "-a", "--no-color", "--format=%(refname:short)" + ).splitlines() if l.strip()] + + def grep(self, pattern: str, rev: str | None, paths: list[str]) -> list[tuple[str, int, str]]: + res: list[tuple[str, int, str]] = [] + args = ["grep", "-n", "-I", "-e", pattern] + if rev: + args.append(rev) + args += ["--", *paths] + out = self.run(*args) + for line in out.splitlines(): + m = re.match(r"^(?:[^:]+:)?([^:]+):(\d+):(.*)$", line) + if m: + res.append((m.group(1), int(m.group(2)), m.group(3))) + return res + + def added_lines(self, path: str, rev: str | None) -> set[int]: + key = ("added", rev, path) + if key in self._cache: + return self._cache[key] + added: set[int] = set() + if rev: + out = self.run("diff", "--no-color", "-U0", f"{rev}~1", rev, "--", path) + else: + out = self.run("diff", "--no-color", "-U0", "HEAD", "--", path) + new_line = 0 + for line in out.splitlines(): + m = re.match(r"^@@ -\d+(?:,\d+)? \+(\d+)(?:,(\d+))? @@", line) + if m: + new_line = int(m.group(1)) + continue + if line.startswith("+") and not line.startswith("+++"): + added.add(new_line) + new_line += 1 + elif line.startswith("-") or line.startswith("\\"): + continue + elif line.startswith(" "): + new_line += 1 + self._cache[key] = added + return added + + def rev_exists(self, rev: str) -> bool: + return bool(self.run("rev-parse", "--verify", "--quiet", f"{rev}^{{commit}}").strip()) + + +# -------------------------------------------------------------------------- +# mermaid renderer (flowchart / stateDiagram-v2 / sequenceDiagram) +# -------------------------------------------------------------------------- + +BOX = { + "single": "┌─┐│└┘", + "double": "╔═╗║╚╝", + "round": "╭─╮│╰╯", +} + + +class Canvas: + def __init__(self, w: int, h: int): + self.w, self.h = w, h + self.g = [[" "] * w for _ in range(h)] + self.obstacles: list[tuple[int, int, int, int]] = [] + self.drawn_boxes: list[tuple[int, int, int, int, list[str], str, str | None]] = [] + + def add_obstacle(self, x0: int, y0: int, x1: int, y1: int): + self.obstacles.append((x0, y0, x1, y1)) + + def blocked(self, x: int, y: int) -> bool: + return any(x0 <= x <= x1 and y0 <= y <= y1 + for (x0, y0, x1, y1) in self.obstacles) + + def col_free(self, x: int, y0: int, y1: int) -> bool: + return all(not self.blocked(x, y) for y in range(y0, y1 + 1)) + + def row_free(self, y: int, x0: int, x1: int) -> bool: + return all(not self.blocked(x, y) for x in range(x0, x1 + 1)) + + def find_free_col(self, sx: int, y0: int, y1: int) -> int | None: + for x in range(sx + 1, self.w - 2): + if self.col_free(x, y0, y1): + return x + for x in range(sx - 1, 1, -1): + if self.col_free(x, y0, y1): + return x + return None + + def find_free_row(self, sy: int, x0: int, x1: int) -> int | None: + lo, hi = min(x0, x1), max(x0, x1) + for y in range(sy + 1, self.h - 2): + if self.row_free(y, lo, hi): + return y + for y in range(sy - 1, 1, -1): + if self.row_free(y, lo, hi): + return y + return None + + def put(self, x: int, y: int, ch: str, merge: bool = True, protect: bool = False): + if 0 <= x < self.w and 0 <= y < self.h: + cur = self.g[y][x] + if protect and self.blocked(x, y) and cur != " ": + return + if protect and self.blocked(x, y): + return + if merge: + ch = merge_char(cur, ch) + self.g[y][x] = ch + + def text(self, x: int, y: int, s: str): + for i, ch in enumerate(s): + self.put(x + i, y, ch, merge=False) + + def hline(self, x0: int, x1: int, y: int, ch: str = "─", protect: bool = False): + for x in range(min(x0, x1), max(x0, x1) + 1): + self.put(x, y, ch, protect=protect) + + def vline(self, x: int, y0: int, y1: int, ch: str = "│", protect: bool = False): + for y in range(min(y0, y1), max(y0, y1) + 1): + self.put(x, y, ch, protect=protect) + + def box(self, x: int, y: int, w: int, h: int, lines: list[str], + style: str = "single", title: str | None = None): + tl, tr, bl, br = {"single": "┌┐└┘", "double": "╔╗╚╝", "round": "╭╮╰╯"}[style] + hz, vt = ("═", "║") if style == "double" else ("─", "│") + self.text(x, y, tl + hz * (w - 2) + tr) + self.text(x, y + h - 1, bl + hz * (w - 2) + br) + for i in range(1, h - 1): + self.put(x, y + i, vt, merge=False) + self.put(x + w - 1, y + i, vt, merge=False) + for i, line in enumerate(lines): + tx = x + max(1, (w - len(line)) // 2) + for k, ch in enumerate(line[: w - 2]): + px = tx + k + if 0 <= px < self.w and self.g[y + 1 + i][px] in (" ",): + self.g[y + 1 + i][px] = ch + if title: + t = f" {title} " + tx = x + max(1, (w - len(t)) // 2) + self.text(tx, y, t) + self.drawn_boxes.append((x, y, w, h, list(lines), style, title)) + + def label(self, x: int, y: int, s: str): + """Write label text only into blank cells (never through boxes/lines).""" + for i, ch in enumerate(s): + px = x + i + if 0 <= px < self.w and 0 <= y < self.h and self.g[y][px] == " ": + self.g[y][px] = ch + + def repair_boxes(self): + for (x, y, w, h, lines, style, title) in self.drawn_boxes: + tl, tr, bl, br = {"single": "┌┐└┘", "double": "╔╗╚╝", "round": "╭╮╰╯"}[style] + hz, vt = ("═", "║") if style == "double" else ("─", "│") + self.text(x, y, tl + hz * (w - 2) + tr) + self.text(x, y + h - 1, bl + hz * (w - 2) + br) + for i in range(1, h - 1): + self.g[y + i][x] = vt + self.g[y + i][x + w - 1] = vt + for i, line in enumerate(lines): + tx = x + max(1, (w - len(line)) // 2) + for k, ch in enumerate(line[: w - 2]): + px = tx + k + if 0 <= px < self.w: + self.g[y + 1 + i][px] = ch + if title: + t = f" {title} " + tx = x + max(1, (w - len(t)) // 2) + self.text(tx, y, t) + + def to_lines(self) -> list[str]: + lines = [] + for row in self.g: + s = "".join(row).rstrip() + lines.append(s) + while lines and not lines[-1]: + lines.pop() + return lines + + +_MERGE = { + ("─", "│"): "┼", ("│", "─"): "┼", + ("─", "┌"): "┬", ("─", "┐"): "┬", ("─", "└"): "┴", ("─", "┘"): "┴", + ("│", "├"): "├", ("│", "┤"): "┤", ("│", "┌"): "├", ("│", "┐"): "┤", + ("│", "└"): "├", ("│", "┘"): "┤", ("│", "┬"): "┼", ("│", "┴"): "┼", + ("┌", "─"): "┬", ("┐", "─"): "┬", ("└", "─"): "┴", ("┘", "─"): "┴", + ("═", "║"): "╬", ("║", "═"): "╬", ("─", "═"): "─", ("═", "─"): "─", + ("│", "║"): "│", ("║", "│"): "│", + ("▶", "─"): "▶", ("─", "▶"): "▶", + ("◀", "─"): "◀", ("─", "◀"): "◀", + ("▼", "│"): "▼", ("│", "▼"): "▼", + ("▲", "│"): "▲", ("│", "▲"): "▲", +} + + +def merge_char(a: str, b: str) -> str: + if a == " ": + return b + if b == " ": + return a + if a == b: + return a + return _MERGE.get((a, b)) or _MERGE.get((b, a)) or b + + +def clean_label(s: str | None) -> str: + if s is None: + return "" + s = s.strip() + if len(s) >= 2 and s[0] == '"' and s[-1] == '"': + s = s[1:-1] + s = s.replace("<br/>", "\n").replace("<br>", "\n").replace("\\n", "\n") + s = s.replace(" ", " ").replace("&", "&").replace("<", "<").replace(">", ">") + return s + + +@dataclass +class GNode: + id: str + label: str = "" + shape: str = "square" + group: int = -1 + seen: int = 0 + x: int = 0 + y: int = 0 + w: int = 0 + h: int = 0 + lines: list[str] = field(default_factory=list) + + @property + def cx(self) -> int: + return self.x + self.w // 2 + + +@dataclass +class GEdge: + a: str + b: str + label: str = "" + dashed: bool = False + back: bool = False + + +_NODE_BRACKETS = [ + ("{{", "}}", "hexagon"), + ("[[", "]]", "square"), + ("[(", ")]", "round"), + ("((", "))", "circle"), + ("[", "]", "square"), + ("(", ")", "round"), + ("{", "}", "rhombus"), + (">", "]", "asym"), +] +_EDGE_RE = re.compile( + r"(?P<a>[A-Za-z0-9_]+(?:\{\{[^}]*\}\}|\[\([^)]*\)\]|\(\([^)]*\)\)|\[[^\]]*\]|\([^)]*\)|\{[^}]*\})?)" + r"\s*(?P<op>-->|---|-\.->|==>|~~~)\s*" + r"(?:\|(?P<lab>[^|]*)\|\s*)?" + r"(?P<b>[A-Za-z0-9_]+(?:\{\{[^}]*\}\}|\[\([^)]*\)\]|\(\([^)]*\)\)|\[[^\]]*\]|\([^)]*\)|\{[^}]*\})?)" +) + + +def parse_node_token(tok: str) -> tuple[str, str, str]: + tok = tok.strip() + m = re.match(r"^([A-Za-z0-9_]+)(.*)$", tok, re.S) + if not m: + return tok, tok, "square" + nid, rest = m.group(1), m.group(2).strip() + if not rest: + return nid, nid, "square" + for open_s, close_s, shape in _NODE_BRACKETS: + if rest.startswith(open_s) and rest.endswith(close_s) and len(rest) >= len(open_s) + len(close_s): + inner = rest[len(open_s):len(rest) - len(close_s)] + return nid, clean_label(inner), shape + return nid, clean_label(rest), "square" + + +def parse_flowchart(src: str) -> tuple[list[GNode], list[GEdge], list[tuple[str, list[str]]]]: + nodes: "OrderedDict[str, GNode]" = OrderedDict() + edges: list[GEdge] = [] + subs: list[tuple[str, list[str]]] = [] + stack: list[tuple[str, list[str]]] = [] + seen = 0 + + def ensure(nid: str, label: str | None = None, shape: str = "square") -> GNode: + nonlocal seen + if nid not in nodes: + seen += 1 + nodes[nid] = GNode(nid, label or nid, shape, -1, seen) + elif label and nodes[nid].label == nodes[nid].id: + nodes[nid].label = label + return nodes[nid] + + lines = src.splitlines() + for raw in lines: + line = raw.strip() + if not line or line.startswith("%%"): + continue + if line.startswith(("classDef ", "class ", "linkStyle ", "style ", "click ")): + continue + if line.startswith("direction "): + continue + m = re.match(r"^subgraph\s+(.*)$", line) + if m: + body = m.group(1).strip() + sm = re.match(r'^([A-Za-z0-9_]+)\s*\[\s*"?(.*?)"?\s*\]$', body) + if sm: + sid, title = sm.group(1), sm.group(2) + else: + sm = re.match(r'^"?(.*?)"?$', body) + title = sm.group(1) if sm else body + sid = re.sub(r"\W+", "_", title) or f"sub{len(subs)}" + stack.append((sid, [])) + continue + if line == "end": + if stack: + sid, members = stack.pop() + subs.append((sid, members)) + if stack: + stack[-1][1].extend(members) + continue + # edges + had_edge = False + pos = 0 + while True: + m = _EDGE_RE.search(line, pos) + if not m: + break + a_tok, b_tok, op, lab = m.group("a"), m.group("b"), m.group("op"), m.group("lab") + aid, alab, ashape = parse_node_token(a_tok) + bid, blab, bshape = parse_node_token(b_tok) + ensure(aid, alab, ashape) + ensure(bid, blab, bshape) + edges.append(GEdge(aid, bid, clean_label(lab), + dashed=op in ("-.->", "~~~"), back=False)) + had_edge = True + nxt = m.start("b") + if nxt <= pos: + break + pos = nxt + if had_edge: + for nid, node in nodes.items(): + if node.group < 0 and stack: + node.group = len(subs) + stack[-1][1].append(nid) + continue + # node definition (possibly "A[label] B[label]" space separated) + found = False + for m in re.finditer( + r"([A-Za-z0-9_]+(?:\{\{[^}]*\}\}|\[\([^)]*\)\]|\(\([^)]*\)\)|\[[^\]]*\]|\([^)]*\)|\{[^}]*\}))", + line, + ): + nid, lab, shape = parse_node_token(m.group(1)) + ensure(nid, lab, shape) + found = True + if found and stack: + for nid, node in nodes.items(): + if node.group < 0: + node.group = len(subs) + stack[-1][1].append(nid) + # group ids -> indices for nodes (use first membership) + idx: dict[str, int] = {} + for i, (sid, members) in enumerate(subs): + for nid in members: + if nid in nodes and nid not in idx: + idx[nid] = i + for nid, node in nodes.items(): + if nid in idx: + node.group = idx[nid] + return list(nodes.values()), edges, subs + + +def assign_layers(nodes: list[GNode], edges: list[GEdge]) -> tuple[str, dict[str, int]]: + ids = [n.id for n in nodes] + adj: dict[str, list[str]] = defaultdict(list) + for e in edges: + adj[e.a].append(e.b) + color: dict[str, int] = {i: 0 for i in ids} + forward: list[tuple[str, str]] = [] + + def dfs(u: str): + color[u] = 1 + for v in adj.get(u, []): + if color.get(v, 0) == 0: + dfs(v) + elif color.get(v, 0) == 1: + # back edge + for e in edges: + if e.a == u and e.b == v: + e.back = True + if color.get(v, 0) != 1: + forward.append((u, v)) + color[u] = 2 + + import sys as _sys + old = _sys.getrecursionlimit() + _sys.setrecursionlimit(max(old, len(ids) * 4 + 100)) + for i in ids: + if color[i] == 0: + dfs(i) + _sys.setrecursionlimit(old) + + layer = {i: 0 for i in ids} + for _ in range(len(ids)): + changed = False + for a, b in forward: + if layer[b] < layer[a] + 1: + layer[b] = layer[a] + 1 + changed = True + if not changed: + break + return layer + + +def _node_size(label: str, shape: str) -> tuple[list[str], int, int]: + lines = label.split("\n") if label else [""] + w = max(len(l) for l in lines) + 6 + w = max(w, 8) + h = len(lines) + 2 + return lines, w, h + + +def _uf_components(nodes: list[GNode], edges: list[GEdge]) -> list[list[GNode]]: + parent: dict[str, str] = {n.id: n.id for n in nodes} + + def find(x: str) -> str: + while parent[x] != x: + parent[x] = parent[parent[x]] + x = parent[x] + return x + + for e in edges: + if e.a in parent and e.b in parent: + ra, rb = find(e.a), find(e.b) + if ra != rb: + parent[ra] = rb + comps: dict[str, list[GNode]] = {} + for n in nodes: + comps.setdefault(find(n.id), []).append(n) + return list(comps.values()) + + +def layout_graph(nodes: list[GNode], edges: list[GEdge], direction: str = "TB", + group_titles: dict[int, str] | None = None) -> Canvas: + """Banded layout: subgraphs (or disconnected components) become bands, laid + out in sequence; within a band, layers follow the DAG.""" + if not nodes: + return Canvas(1, 1) + by_id = {n.id: n for n in nodes} + for n in nodes: + n.lines, n.w, n.h = _node_size(n.label, n.shape) + + grouped: dict[int, list[GNode]] = defaultdict(list) + ungrouped: list[GNode] = [] + for n in nodes: + if n.group >= 0: + grouped[n.group].append(n) + else: + ungrouped.append(n) + + bands: list[dict] = [] + for gi, mem in sorted(grouped.items()): + bands.append({"nodes": mem, "group": gi, "edges": [], + "title": (group_titles or {}).get(gi) or f"group {gi}"}) + if ungrouped: + for comp in _uf_components(ungrouped, edges): + bands.append({"nodes": comp, "group": None, "edges": [], "title": None}) + bands.sort(key=lambda b: min(n.seen for n in b["nodes"])) + + for b in bands: + ids = {n.id for n in b["nodes"]} + b["edges"] = [e for e in edges if e.a in ids and e.b in ids] + bs = b["nodes"] + layer = assign_layers(bs, b["edges"]) + if direction == "TB": + rows: dict[int, list[GNode]] = defaultdict(list) + for n in bs: + rows[layer[n.id]].append(n) + y, maxx = 0, 0 + for k in sorted(rows): + rows[k].sort(key=lambda n: n.seen) + x = 2 + hrow = max(n.h for n in rows[k]) + for n in rows[k]: + n.x, n.y = x, y + x += n.w + 4 + maxx = max(maxx, x) + y += hrow + 3 + b["w"], b["h"] = maxx, y + else: + cols: dict[int, list[GNode]] = defaultdict(list) + for n in bs: + cols[layer[n.id]].append(n) + x, maxy = 0, 0 + for k in sorted(cols): + cols[k].sort(key=lambda n: n.seen) + y = 2 + wcol = max(n.w for n in cols[k]) + for n in cols[k]: + n.x, n.y = x, y + y += n.h + 3 + maxy = max(maxy, y) + x += wcol + 7 + b["w"], b["h"] = x, maxy + + reserve = 10 + if direction == "TB": + total_w = max(b["w"] for b in bands) + 4 + reserve + y = 0 + for b in bands: + dx = max(2, (total_w - reserve - b["w"]) // 2) + for n in b["nodes"]: + n.x += dx + n.y += y + b["top"], b["bottom"] = y, y + b["h"] + y += b["h"] + 2 + total_h = y + 3 + else: + total_h = max(b["h"] for b in bands) + 4 + reserve + x = 0 + for b in bands: + dy = max(2, (total_h - reserve - b["h"]) // 2) + for n in b["nodes"]: + n.y += dy + n.x += x + b["left"], b["right"] = x, x + b["w"] + x += b["w"] + 4 + total_w = x + 6 + cv = Canvas(total_w, total_h) + + for n in nodes: + cv.box(n.x, n.y, n.w, n.h, n.lines, "single") + + band_of: dict[str, int] = {} + for bi, b in enumerate(bands): + for n in b["nodes"]: + band_of[n.id] = bi + + back: list[GEdge] = [] + fwd: list[GEdge] = [] + for e in edges: + if e.back or band_of[e.b] < band_of[e.a]: + back.append(e) + else: + fwd.append(e) + + lane = 0 + for e in back: + a, bb = by_id[e.a], by_id[e.b] + if direction == "TB": + lx = total_w - 2 - lane + lane += 2 + if lx <= a.x + a.w: + continue + cv.hline(a.x + a.w, lx, a.y + a.h // 2, "─", protect=True) + cv.vline(lx, a.y + a.h // 2, bb.y + bb.h // 2, protect=True) + cv.hline(lx, bb.x + bb.w, bb.y + bb.h // 2, "─", protect=True) + cv.put(bb.x + bb.w, bb.y + bb.h // 2, "◀") + if e.label: + cv.label(min(lx + 1, cv.w - len(e.label) - 1), + (a.y + bb.y) // 2, e.label) + else: + ly = total_h - 2 - lane + lane += 1 + if ly <= a.y + a.h: + continue + cv.vline(a.cx, a.y + a.h, ly, "│", protect=True) + cv.hline(a.cx, bb.cx, ly, "─", protect=True) + cv.vline(bb.cx, ly, bb.y + bb.h, "│", protect=True) + cv.put(bb.cx, bb.y + bb.h, "▲") + if e.label: + cv.label(min(a.cx, bb.cx) + 1, ly - 1, e.label) + + detour = 0 + for e in fwd: + a, bb = by_id[e.a], by_id[e.b] + em = "┄" if e.dashed else "─" + if direction == "TB": + sx, sy = a.cx, a.y + a.h + tx, ty = bb.cx, bb.y + if cv.col_free(tx, sy, ty - 1): + my = sy + max(1, (ty - sy) // 2) + cv.vline(sx, sy, my, "│", protect=True) + if sx != tx: + cv.hline(sx, tx, my, em, protect=True) + cv.vline(tx, my, ty - 1, "│", protect=True) + cv.put(tx, ty - 1, "▼") + if e.label: + cv.label(min(tx + 2, cv.w - len(e.label) - 1), sy + 1, e.label) + else: + hold = sy + 1 + dx = cv.find_free_col(sx, hold, ty - 1) + if dx is None or dx == sx: + dx = total_w - 3 - detour + detour += 2 + if not cv.row_free(sy, min(sx, dx), max(sx, dx)): + sy2 = sy + 1 if cv.row_free(sy + 1, min(sx, dx), max(sx, dx)) else sy + else: + sy2 = sy + cv.put(sx, sy, "│") + cv.hline(sx, dx, sy2, em, protect=True) + cv.vline(dx, sy2, ty - 1, "│", protect=True) + cv.hline(dx, tx, ty - 1, em, protect=True) + cv.put(tx, ty - 1, "▼") + if e.label: + cv.label(min(dx, tx) + 2, max(sy2 + 1, (sy2 + ty) // 2), + e.label) + else: + sx, sy = a.x + a.w, a.y + a.h // 2 + ty = bb.y + bb.h // 2 + if (sy == ty or abs(sy - ty) < 2) and cv.row_free(ty, sx, bb.x - 1): + cv.hline(sx, bb.x - 1, ty, em, protect=True) + cv.put(bb.x - 1, ty, "▶") + if e.label: + cv.label(sx + 2, ty - 1, e.label) + else: + mx = sx + max(2, (bb.x - sx) // 2) + dy = cv.find_free_row(sy, sx, mx) + hold = bb.x - 1 + if dy is None or dy == sy: + dy = total_h - 3 - detour + detour += 1 + cv.hline(sx, mx, sy, em, protect=True) + cv.vline(mx, sy, dy, "│", protect=True) + cv.hline(mx, bb.x - 1, dy, em, protect=True) + cv.vline(hold, dy, ty, "│", protect=True) + cv.put(bb.x - 1, ty, "▶") + if e.label: + cv.label(mx + 2, min(sy, ty) + 1, e.label) + + for b in bands: + if b["group"] is None: + continue + xs = [n.x for n in b["nodes"]] + ys = [n.y for n in b["nodes"]] + xe = [n.x + n.w for n in b["nodes"]] + ye = [n.y + n.h for n in b["nodes"]] + x0, y0 = max(0, min(xs) - 3), max(0, min(ys) - 2) + x1, y1 = min(cv.w - 1, max(xe) + 2), min(cv.h - 1, max(ye) + 1) + if x1 - x0 > 4 and y1 - y0 > 2: + cv.box(x0, y0, x1 - x0, y1 - y0, [], "double", + title=b.get("title") or str(b["group"])) + cv.repair_boxes() + return cv + + +def render_flowchart(src: str) -> list[str] | None: + first = "" + direction = "TB" + for l in src.splitlines(): + s = l.strip() + if s and not s.startswith("%%"): + first = s + break + m = re.search(r"\b(TB|TD|LR|RL|BT)\b", first) + if m: + direction = "LR" if m.group(1) in ("LR", "RL") else "TB" + nodes, edges, subs = parse_flowchart(src) + if not nodes: + return None + titles = {i: sid.replace("_", " ") for i, (sid, members) in enumerate(subs)} + cv = layout_graph(nodes, edges, direction, titles) + if direction == "LR": + wide = max((len(l) for l in cv.to_lines()), default=0) + if wide > 118: + cv = layout_graph(nodes, edges, "TB", titles) + return cv.to_lines() + + +def render_state(src: str) -> list[str] | None: + direction = "TB" + nodes: "OrderedDict[str, GNode]" = OrderedDict() + edges: list[GEdge] = [] + notes: list[tuple[str, str]] = [] # (anchor, text) + seen = 0 + lines = src.splitlines() + i = 0 + while i < len(lines): + line = lines[i].strip() + i += 1 + if not line or line.startswith("%%"): + continue + m = re.match(r"^direction\s+(TB|TD|LR|RL|BT)$", line) + if m: + # terminals are vertical; keep state chains readable top-to-bottom + direction = "TB" + continue + m = re.match(r"^note\s+(right|left)\s+of\s+([A-Za-z0-9_\[\]\*]+)\s*$", line) + if m: + anchor, buf = m.group(2), [] + while i < len(lines) and lines[i].strip() != "end note": + buf.append(lines[i].strip()) + i += 1 + i += 1 + notes.append((anchor, " ".join(buf))) + continue + m = re.match(r"^state\s+\"?(.*?)\"?\s+as\s+([A-Za-z0-9_]+)$", line) + if m: + if m.group(2) not in nodes: + seen += 1 + nodes[m.group(2)] = GNode(m.group(2), clean_label(m.group(1)), "round", -1, seen) + continue + m = re.match(r"^state\s+\"?(.*?)\"?$", line) + if m and "-->" not in line: + nid = re.sub(r"\W+", "_", m.group(1)) + if nid not in nodes: + seen += 1 + nodes[nid] = GNode(nid, m.group(1), "round", -1, seen) + continue + m = re.match(r"^(\[\*\]|[A-Za-z0-9_]+)\s*-->\s*(\[\*\]|[A-Za-z0-9_]+)\s*(?::\s*(.*))?$", line) + if m: + a, b, lab = m.group(1), m.group(2), clean_label(m.group(3)) + for nid in (a, b): + if nid not in nodes: + seen += 1 + label = "●" if nid == "[*]" else nid + nodes[nid] = GNode(nid, label, "round", -1, seen) + edges.append(GEdge(a, b, lab)) + continue + if not nodes: + return None + cv = layout_graph(list(nodes.values()), edges, direction) + out = cv.to_lines() + if notes: + out.append("") + for anchor, text in notes: + wrapped = textwrap.wrap(text, 60) or [""] + nb = Canvas(66, len(wrapped) + 4) + nb.box(0, 0, 64, len(wrapped) + 2, wrapped, "round", + title="note") + out += [" " + l for l in nb.to_lines()] + return out + + +def parse_sequence(src: str): + participants: "OrderedDict[str, str]" = OrderedDict() + events = [] + autonum = 0 + for raw in src.splitlines(): + line = raw.strip() + if not line or line.startswith("%%"): + continue + if line.startswith("autonumber"): + autonum = 1 + continue + m = re.match(r"^participant\s+([A-Za-z0-9_]+)(?:\s+as\s+(.*))?$", line) + if m: + participants.setdefault(m.group(1), clean_label(m.group(2) or m.group(1))) + continue + m = re.match(r"^Note\s+(?:over|right of|left of)\s+([A-Za-z0-9_]+)(?:\s*,\s*([A-Za-z0-9_]+))?\s*:\s*(.*)$", + line, re.I) + if m: + events.append(("note", m.group(1), m.group(2) or m.group(1), clean_label(m.group(3)))) + continue + m = re.match(r"^(alt|else|opt|loop|par|critical|break)\b\s*(.*)$", line) + if m: + events.append(("block", m.group(1), "", clean_label(m.group(2)))) + continue + if line == "end": + events.append(("end", "", "", "")) + continue + m = re.match( + r"^([A-Za-z0-9_]+)\s*(-->>|->>|-->|->|--x|-x|--\)|-\))\s*([A-Za-z0-9_]+)\s*:\s*(.*)$", + line) + if m: + a, op, b, text = m.groups() + participants.setdefault(a, a) + participants.setdefault(b, b) + dashed = op.startswith("--") + events.append(("msg", a, b, clean_label(text), dashed)) + continue + return participants, events, autonum + + +def render_sequence(src: str) -> list[str] | None: + participants, events, autonum = parse_sequence(src) + if not participants: + return None + names = list(participants.values()) + widths = [max(12, min(26, len(n) + 4)) for n in names] + xs = [] + x = 2 + for w in widths: + xs.append(x + w // 2) + x += w + 5 + total_w = x + 2 + max_single = 0 + for ev in events: + if ev[0] == "note" and ev[1] == ev[2]: + max_single = max(max_single, min(48, len(ev[3]) + 4)) + total_w = max(total_w, max_single + 8) + + rows: list[list[str]] = [] + + def lifeline_row() -> list[str]: + row = [" "] * total_w + for cx in xs: + row[cx] = "│" + return row + + def put(row, at, s): + for i, ch in enumerate(s): + if 0 <= at + i < total_w: + row[at + i] = ch + + hdr = [" "] * total_w + for i, n in enumerate(names): + w = widths[i] + x0 = xs[i] - w // 2 + b = Canvas(w, 3) + b.box(0, 0, w, 3, [n[: w - 2]], "round") + for l in b.to_lines(): + for k, ch in enumerate(l): + if 0 <= x0 + k < total_w: + hdr[x0 + k] = ch + rows.append(hdr) + rows.append(lifeline_row()) + + num = 0 + for ev in events: + kind = ev[0] + if kind == "msg": + _, a, b, text, dashed = ev + if autonum: + num += 1 + ia, ib = list(participants).index(a), list(participants).index(b) + x0, x1 = xs[ia], xs[ib] + label = (f"{num}. " if autonum else "") + text + if x0 == x1: + lrow = lifeline_row() + put(lrow, min(x0 + 2, total_w - 1), label) + rows.append(lrow) + row = lifeline_row() + for k in range(x0 + 1, min(x0 + 5, total_w)): + row[k] = "╌" if dashed else "─" + if x0 + 5 < total_w: + row[x0 + 5] = "┐" + rows.append(row) + mid = lifeline_row() + if x0 + 5 < total_w: + mid[x0 + 5] = "│" + rows.append(mid) + back = lifeline_row() + for k in range(x0 + 1, min(x0 + 5, total_w)): + back[k] = "╌" if dashed else "─" + if x0 + 5 < total_w: + back[x0 + 5] = "┘" + back[x0] = "◀" + rows.append(back) + else: + lrow = lifeline_row() + lx = min(x0, x1) + 2 if x1 > x0 else max(1, x1 - len(label) - 2) + put(lrow, lx, label) + rows.append(lrow) + row = lifeline_row() + step = 1 if x1 > x0 else -1 + for k in range(x0, x1, step): + row[k] = "╌" if dashed else "─" + if k in xs: + row[k] = "┼" + row[x0] = "│" + row[x1] = "▶" if x1 > x0 else "◀" + rows.append(row) + continue + if kind == "note": + _, a, b, text = ev + ia, ib = list(participants).index(a), list(participants).index(b) + if a == b: + w = min(48, max(28, len(text) + 4)) + x0 = max(0, min(total_w - w, xs[ia] - w // 2)) + else: + x0 = max(0, min(xs[ia], xs[ib]) - 4) + wid = min(total_w - x0, max(20, abs(xs[ib] - xs[ia]) + 10)) + w = wid + wrapped = textwrap.wrap(text, max(8, w - 4)) or [""] + bc = Canvas(w, len(wrapped) + 2) + bc.box(0, 0, w, len(wrapped) + 2, wrapped, "round") + rows.append(lifeline_row()) + for l in bc.to_lines(): + r = lifeline_row() + for k, ch in enumerate(l): + if 0 <= x0 + k < total_w: + r[x0 + k] = ch + rows.append(r) + continue + if kind == "block": + _, word, _, text = ev + label = f"{word} {text}".strip() + row = lifeline_row() + for k in range(total_w): + if row[k] == " ": + row[k] = "┄" + lx = max(2, (total_w - len(label)) // 2) + put(row, lx, f" {label} ") + rows.append(row) + continue + if kind == "end": + row = lifeline_row() + for k in range(total_w): + if row[k] == " ": + row[k] = "┄" + rows.append(row) + continue + return ["".join(r).rstrip() for r in rows] + + +def render_mermaid(src: str) -> list[str] | None: + first = "" + for l in src.splitlines(): + s = l.strip() + if s and not s.startswith("%%"): + first = s + break + try: + if first.startswith("sequenceDiagram"): + return render_sequence(src) + if first.startswith("stateDiagram"): + return render_state(src) + if first.startswith(("flowchart", "graph")): + return render_flowchart(src) + except Exception: + return None + return None + + +# -------------------------------------------------------------------------- +# syntax highlighting +# -------------------------------------------------------------------------- + +C_KEYWORDS = { + "int", "char", "long", "short", "unsigned", "signed", "void", "static", "const", + "struct", "union", "enum", "typedef", "extern", "return", "if", "else", "for", + "while", "do", "switch", "case", "break", "continue", "goto", "sizeof", "inline", + "volatile", "register", "u8", "u16", "u32", "u64", "s8", "s16", "s32", "s64", + "bool", "true", "false", "NULL", +} +PY_KEYWORDS = { + "def", "class", "return", "if", "elif", "else", "for", "while", "import", "from", + "as", "try", "except", "finally", "with", "lambda", "yield", "pass", "break", + "continue", "and", "or", "not", "in", "is", "None", "True", "False", "global", + "nonlocal", "assert", "raise", "del", "async", "await", +} +SH_KEYWORDS = {"if", "then", "else", "fi", "for", "in", "do", "done", "while", "case", + "esac", "function", "local", "export", "return", "echo", "exit", "set"} + + +def highlight(code: str, lang: str) -> list[list[tuple[str, str]]]: + lang = (lang or "").lower() + if lang in ("c", "h", "cpp", "cc", "hpp"): + return _hl_c(code) + if lang in ("py", "python"): + return _hl_py(code) + if lang in ("sh", "bash", "shell", "zsh"): + return _hl_sh(code) + return [[(l, "code")] for l in code.split("\n")] + + +def _hl_c(code: str) -> list[list[tuple[str, str]]]: + out = [] + in_block = False + token_re = re.compile( + r'(?P<block>/\*|\*/)|(?P<line>//.*$)|(?P<str>"(?:[^"\\]|\\.)*"|\'(?:[^\'\\]|\\.)*\')' + r'|(?P<pre>#\s*\w+)|(?P<num>\b\d[\w\.x]*\b)|(?P<id>[A-Za-z_]\w*)') + for line in code.split("\n"): + segs = [] + pos = 0 + if in_block: + end = line.find("*/") + if end < 0: + out.append([(line, "comment")]) + continue + out.append([(line[:end + 2], "comment")]) + pos = end + 2 + in_block = False + rest = _hl_c(line[pos:]) + out[-1] += rest[0] if rest else [] + continue + for m in token_re.finditer(line[pos:]): + s = pos + m.start() + e = pos + m.end() + if s > pos: + segs.append((line[pos:s], "normal")) + if m.group("block"): + if m.group("block") == "/*": + in_block = True + segs.append(("/*", "comment")) + else: + segs.append(("*/", "comment")) + in_block = False + elif m.group("line"): + segs.append((m.group("line"), "comment")) + elif m.group("str"): + segs.append((m.group("str"), "string")) + elif m.group("pre"): + segs.append((m.group("pre"), "preproc")) + elif m.group("num"): + segs.append((m.group("num"), "number")) + elif m.group("id"): + w = m.group("id") + segs.append((w, "keyword" if w in C_KEYWORDS else "normal")) + pos = e + if pos < len(line): + segs.append((line[pos:], "normal")) + out.append(segs or [("", "normal")]) + return out + + +def _hl_py(code: str) -> list[list[tuple[str, str]]]: + out = [] + token_re = re.compile( + r'(?P<str>"""(?:.|\n)*?"""|\'\'\'(?:.|\n)*?\'\'\'|"(?:[^"\\]|\\.)*"|\'(?:[^\'\\]|\\.)*\')' + r'|(?P<line>#.*$)|(?P<num>\b\d[\w\.]*\b)|(?P<id>[A-Za-z_]\w*)', re.S) + for line in code.split("\n"): + segs = [] + pos = 0 + for m in token_re.finditer(line): + s, e = m.start(), m.end() + if s > pos: + segs.append((line[pos:s], "normal")) + if m.group("str"): + segs.append((m.group("str"), "string")) + elif m.group("line"): + segs.append((m.group("line"), "comment")) + elif m.group("num"): + segs.append((m.group("num"), "number")) + elif m.group("id"): + w = m.group("id") + segs.append((w, "keyword" if w in PY_KEYWORDS else "normal")) + pos = e + if pos < len(line): + segs.append((line[pos:], "normal")) + out.append(segs or [("", "normal")]) + return out + + +def _hl_sh(code: str) -> list[list[tuple[str, str]]]: + out = [] + for line in code.split("\n"): + segs = [] + rest = line + m = re.match(r"^(\s*#.*)$", rest) + if m: + out.append([(rest, "comment")]) + continue + m = re.match(r"^(\s*)([\w\.\-/]*)(.*)$", rest) + if m and m.group(2): + segs.append((m.group(1), "normal")) + segs.append((m.group(2), + "keyword" if m.group(2) in SH_KEYWORDS else "normal")) + rest = m.group(3) + for piece in re.split(r'("(?:[^"\\]|\\.)*")', rest): + if piece.startswith('"'): + segs.append((piece, "string")) + else: + segs.append((piece, "normal")) + out.append(segs or [("", "normal")]) + return out + + +# -------------------------------------------------------------------------- +# markdown renderer +# -------------------------------------------------------------------------- + + +def strip_inline(s: str) -> str: + s = re.sub(r"!\[([^\]]*)\]\([^)]*\)", r"[image: \1]", s) + s = re.sub(r"\[([^\]]+)\]\(([^)]+)\)", r"\1", s) + s = re.sub(r"\*\*([^*]+)\*\*", r"\1", s) + s = re.sub(r"__([^_]+)__", r"\1", s) + s = re.sub(r"\*([^*]+)\*", r"\1", s) + s = re.sub(r"_([^_]+)_", r"\1", s) + s = re.sub(r"`([^`]+)`", r"\1", s) + return s + + +def render_markdown(text: str, width: int, mermaid_src: bool = False) -> list[tuple[str, str]]: + """Render Markdown to a list of (line, attribute) pairs for the reader.""" + out: list[tuple[str, str]] = [] + lines = text.split("\n") + i = 0 + while i < len(lines): + line = lines[i] + if line.startswith("```"): + lang = line[3:].strip() + i += 1 + buf = [] + while i < len(lines) and not lines[i].startswith("```"): + buf.append(lines[i]) + i += 1 + i += 1 + if lang == "mermaid": + if mermaid_src: + out.append(("── mermaid source ─────────────────────────", "dim")) + for b in buf: + out.append((b, "code")) + out.append(("───────────────────────────────────────────", "dim")) + else: + first = buf[0].strip().split()[0] if buf[:1] else "diagram" + rendered = render_mermaid("\n".join(buf)) + out.append((f"── mermaid diagram [{first}] ──", "dim")) + if rendered: + for r in rendered: + out.append((r, "mermaid")) + else: + out.append(("(no built-in renderer for this diagram; press m for source)", + "error")) + for b in buf: + out.append((b, "code")) + out.append(("", "dim")) + else: + for segs in highlight("\n".join(buf), lang): + out.append(("".join(t for t, _ in segs), "code")) + continue + i += 1 + if not line.strip(): + out.append(("", "normal")) + continue + m = re.match(r"^(#+)\s*(.*)$", line) + if m: + lvl = min(len(m.group(1)), 3) + title = strip_inline(m.group(2)) + if lvl == 1: + out.append(("", "normal")) + out.append((title, "h1")) + out.append(("═" * min(len(title), width), "h1")) + elif lvl == 2: + out.append((title, "h2")) + under = "─" * min(len(title), width) + if under: + out.append((under, "h2")) + else: + out.append((title, "h3")) + continue + if line.strip() in ("---", "***", "___"): + out.append(("─" * width, "dim")) + continue + if line.lstrip().startswith(">"): + body = strip_inline(re.sub(r"^\s*>\s?", "", line)) + for w in (textwrap.wrap(body, max(10, width - 4)) or [""]): + out.append((" │ " + w, "quote")) + continue + if line.strip().startswith("|") and i < len(lines) and \ + re.match(r"^\s*\|?[\s:|-]+\|", lines[i]): + header = [c.strip() for c in line.strip().strip("|").split("|")] + i += 1 # separator row + rows = [] + while i < len(lines) and lines[i].strip().startswith("|"): + rows.append([c.strip() for c in lines[i].strip().strip("|").split("|")]) + i += 1 + cols = max([len(header)] + [len(r) for r in rows]) + wds = [0] * cols + for r in [header] + rows: + for c in range(len(r)): + wds[c] = max(wds[c], len(strip_inline(r[c]))) + def fmt(cells): + cells = [strip_inline(c).ljust(wds[ci]) for ci, c in enumerate(cells)] + return "│ " + " │ ".join(cells) + " │" + out.append(("┌" + "┬".join("─" * (w + 2) for w in wds) + "┐", "table")) + out.append((fmt(header), "tablehead")) + out.append(("├" + "┼".join("─" * (w + 2) for w in wds) + "┤", "table")) + for r in rows: + out.append((fmt(r), "table")) + out.append(("└" + "┴".join("─" * (w + 2) for w in wds) + "┘", "table")) + continue + m = re.match(r"^(\s*)([-*+]|\d+\.)\s+(.*)$", line) + if m: + indent = len(m.group(1)) + bullet = "•" if not m.group(2)[0].isdigit() else m.group(2) + body = strip_inline(m.group(3)) + wrapped = textwrap.wrap(body, max(10, width - indent - 4)) or [""] + out.append((" " * indent + f"{bullet} " + wrapped[0], "normal")) + for w in wrapped[1:]: + out.append((" " * (indent + len(bullet) + 1) + w, "normal")) + continue + body = strip_inline(line) + for w in (textwrap.wrap(body, width) or [""]): + out.append((w, "normal")) + return out + + +# -------------------------------------------------------------------------- +# architecture explorer data +# -------------------------------------------------------------------------- + +ARCH = { + "id": "root", + "title": "WHOLE SYSTEM", + "status": "[map]", + "desc": "End-state picture: stock guest on the emulated device, one host module, " + "real GSP firmware for execution.", + "sources": [], + "children": ["guest", "hyp", "module", "gpu", "hostos"], + "edges": [("guest", "hyp", "MMIO+DMA+MSI-X"), + ("hyp", "module", "VFIO mdev ops"), + ("module", "gpu", "msgq + push streams")], +} + +COMPONENTS = { + "guest": dict( + title="Guest VM (stock)", + status="[done]", + desc="Unmodified NVIDIA drivers and applications. Sees a TU104 whose GSP " + "answers correctly. No guest-side Open-IOV code, no enlightenment.", + sources=["userspace/cuda/READ-ME-FIRST.md", "userspace/cuda/docs/03-driver-and-gpu-interface.md"], + calls=["CUDA driver API", "RM ioctls", "UVM ioctls", "MMIO/DMA"], + children=[], + ), + "hyp": dict( + title="Hypervisor (KVM+QEMU)", + status="[done]", + desc="VFIO mdev transport for the virtual device, IOMMU/IOMMUFD for per-VM " + "DMA isolation, KVM irqfd/MMIO delivery.", + sources=["docs/03-architecture/06-host-module-and-hybrid-model.md"], + calls=["vfio", "mdev", "iommufd", "kvm irqfd", "eventfd"], + children=[], + ), + "module": dict( + title="open-iov-vfio.ko", + status="[wip]", + desc="The only host component. Virtual device + emulation core + memory " + "model + work path + (planned) execution backend.", + sources=["kernel/main.c", "kernel/gsp_sema_service.c"], + calls=["mdev_register_parent", "vfio_register_emulated_iommu_dev", + "kthreads", "workqueues", "eventfd", "request_firmware"], + children=["vdev", "emucore", "memmodel", "workpath", "backend"], + ), + "vdev": dict( + title="Virtual device", + status="[done]", + desc="PCI config space, BAR0/1/3 shadows, MSI-X table; guest MMIO lands " + "directly in the shadow regions, doorbells observed by polling.", + sources=["kernel/main.c", "kernel/bar_emulation.c", "kernel/bar_init.c", + "kernel/irq.c", "kernel/memory.c"], + calls=["PCI/VFIO region mmap", "mdev", "eventfd -> irqfd"], + children=[], + ), + "emucore": dict( + title="Emulation core", + status="[done]", + desc="GSP boot stages, GSP-RM message queues and RPC, control dispatch, " + "object model, reply fidelity.", + sources=["kernel/gsp_emulation.c", "kernel/gsp_rm_emulation.c", + "kernel/gsp_comm.c", "kernel/gsp_rpc_client.c", + "kernel/dynamic_rpc_dispatcher.c"], + calls=["request_firmware (analysis)", "kthreads", "eventfd"], + children=[], + ), + "memmodel": dict( + title="Memory model", + status="[wip]", + desc="VA to PA discovery from guest structures, pins, VRAM shadow, " + "server-owned page-table population.", + sources=["kernel/gsp_sema_service.c", "kernel/memory.c"], + calls=["vfio_dma_rw", "GMMU formats", "page tables"], + children=[], + ), + "workpath": dict( + title="Work path", + status="[wip]", + desc="Channel pump, GPFIFO entry decode, semaphore/copy/memset execution, " + "completion relay, counter publishing.", + sources=["kernel/gsp_sema_service.c", "tools/test/cuda_drv_test.c"], + calls=["GPFIFO", "USERD/doorbells", "method streams", "tracker writes"], + children=[], + ), + "backend": dict( + title="Execution backend", + status="[planned]", + desc="Module-owned GPU with a direct real-GSP session (host-side RM " + "client): real channels, real page tables, forwarded work.", + sources=["docs/03-architecture/08-physical-gpu-integration.md"], + calls=["real GSP msgq/doorbells", "real MSI-X", "real engines"], + children=[], + ), + "gpu": dict( + title="Physical GPU", + status="[hardware]", + desc="GSP firmware (the real RM server), engines (GR, CE pair, DEC, ENC), " + "GMMU, VRAM, doorbells.", + sources=[], + calls=[], + children=[], + ), + "hostos": dict( + title="Host OS", + status="[vendor]", + desc="Distribution nvidia.ko stays unmodified and unbound while the " + "service owns the GPU; display runs elsewhere.", + sources=[], + calls=["Linux kernel APIs", "DRM/modeset"], + children=[], + ), +} + + +# -------------------------------------------------------------------------- +# curses application +# -------------------------------------------------------------------------- + +PAIRS = { + "normal": 0, "header": 0, "sel": 0, "h1": 0, "h2": 0, "h3": 0, + "code": 0, "comment": 0, "string": 0, "keyword": 0, "number": 0, + "preproc": 0, "dim": 0, "error": 0, "mermaid": 0, "quote": 0, + "table": 0, "tablehead": 0, "accent": 0, "border": 0, "status": 0, + "gutter": 0, "added": 0, "title": 0, +} + + +def init_colors(): + curses.start_color() + curses.use_default_colors() + defs = { + "header": (curses.COLOR_BLACK, curses.COLOR_CYAN), + "sel": (curses.COLOR_BLACK, curses.COLOR_WHITE), + "h1": (curses.COLOR_CYAN, -1), + "h2": (curses.COLOR_BLUE, -1), + "h3": (curses.COLOR_MAGENTA, -1), + "code": (curses.COLOR_YELLOW, -1), + "comment": (curses.COLOR_GREEN, -1), + "string": (curses.COLOR_MAGENTA, -1), + "keyword": (curses.COLOR_CYAN, -1), + "number": (curses.COLOR_YELLOW, -1), + "preproc": (curses.COLOR_RED, -1), + "dim": (curses.COLOR_WHITE, -1), + "error": (curses.COLOR_RED, -1), + "mermaid": (curses.COLOR_GREEN, -1), + "quote": (curses.COLOR_BLUE, -1), + "table": (-1, -1), + "tablehead": (curses.COLOR_CYAN, -1), + "accent": (curses.COLOR_MAGENTA, -1), + "border": (curses.COLOR_BLUE, -1), + "gutter": (curses.COLOR_BLUE, -1), + "added": (curses.COLOR_GREEN, -1), + "title": (curses.COLOR_CYAN, -1), + } + for i, (name, (fg, bg)) in enumerate(defs.items(), start=1): + try: + curses.init_pair(i, fg, bg) + except curses.error: + pass + PAIRS[name] = curses.color_pair(i) + + +def safe_addstr(scr, y, x, s, attr=0): + h, w = scr.getmaxyx() + if y < 0 or y >= h or x >= w: + return + maxlen = w - x - 1 if y == h - 1 else w - x + s = s[:maxlen] + try: + scr.addstr(y, x, s, attr) + except curses.error: + pass + + +class Screen: + def __init__(self, app): + self.app = app + + @property + def top(self) -> int: + return 2 + + @property + def bottom(self) -> int: + return self.app.h - 2 + + def draw(self, scr): + raise NotImplementedError + + def key(self, ch): + raise NotImplementedError + + def hints(self) -> str: + return "" + + def title(self) -> str: + return "" + + +class App: + def __init__(self, git: Git): + self.git = git + self.rev: str | None = None + self.mermaid_src = False + self.stack: list[Screen] = [] + self.running = True + self.status = "" + self._doc_cache: dict = {} + + # ----- plumbing ----- + def push(self, screen: Screen): + self.stack.append(screen) + + def pop(self): + if len(self.stack) > 1: + self.stack.pop() + + def set_rev(self, rev: str | None): + self.rev = rev + self._doc_cache.clear() + + def load_text(self, path: str) -> str | None: + key = (self.rev, path) + if key not in self._doc_cache: + self._doc_cache[key] = self.git.show(path, self.rev) + return self._doc_cache[key] + + def run(self, scr): + init_colors() + curses.curs_set(0) + scr.keypad(True) + scr.timeout(80) + self.scr = scr + self.push(MenuScreen(self)) + while self.running and self.stack: + scr.erase() + self.h, self.w = scr.getmaxyx() + self.draw_topbar(scr) + self.stack[-1].draw(scr) + self.draw_bottombar(scr) + scr.refresh() + ch = scr.getch() + if ch == -1: + continue + if ch == curses.KEY_RESIZE: + continue + try: + self.stack[-1].key(ch) + except Exception as e: # keep the UI alive on unexpected errors + self.status = f"error: {e}" + + def draw_topbar(self, scr): + rev = self.git.resolve(self.rev) if self.rev else f"{self.git.default_branch()} (worktree)" + title = f" open-iov browser v{VERSION} " + info = f" rev: {rev} | mermaid: {'source' if self.mermaid_src else 'rendered'} " + bar = title.ljust(max(0, self.w - len(info) - 1)) + info + safe_addstr(scr, 0, 0, bar[: self.w - 1], PAIRS["header"]) + + def draw_bottombar(self, scr): + hints = self.stack[-1].hints() + extra = f" | {self.status}" if self.status else "" + line = (hints + extra) + safe_addstr(scr, self.h - 1, 0, " " + line[: self.w - 2], PAIRS["header"]) + + def draw_frame(self, scr): + for y in range(self.top + 1, self.h - 1): + safe_addstr(scr, y, 0, " " * (self.w - 1)) + + +class MenuScreen(Screen): + ITEMS = [ + ("Documentation", "browse and read docs/ with rendered Mermaid diagrams"), + ("Code browser", "navigate the source tree, syntax highlighted"), + ("Architecture explorer", "zoom from the whole system into components"), + ("History", "browse docs and code at any commit or branch (no checkout)"), + ("Search", "grep the docs and code across a revision"), + ("Help", "keys, modes and how this tool works"), + ("Quit", "leave the browser"), + ] + + def __init__(self, app): + super().__init__(app) + self.sel = 0 + + def title(self): + return "menu" + + def hints(self): + return "↑/↓ move · Enter open · q quit · h help · v history" + + def draw(self, scr): + logo = [ + " ╭──────────────────────────────────────────────────────────╮", + " │ O P E N - I O V · p r o j e c t b r o w s e r │", + " ╰──────────────────────────────────────────────────────────╯", + ] + y = self.top + 1 + for l in logo: + safe_addstr(scr, y, 2, l, PAIRS["title"]) + y += 1 + y += 1 + rev = self.git.resolve(self.app.rev) if self.app.rev else "worktree (uncommitted changes visible)" + safe_addstr(scr, y, 2, f"revision: {rev}", PAIRS["dim"]) + y += 2 + for i, (name, desc) in enumerate(self.ITEMS): + attr = PAIRS["sel"] if i == self.sel else PAIRS["normal"] + marker = "▸" if i == self.sel else " " + safe_addstr(scr, y, 2, f"{marker} {i+1}. {name}", attr) + safe_addstr(scr, y, 30, desc, PAIRS["dim"] if i != self.sel else attr) + y += 1 + + def key(self, ch): + if ch in (curses.KEY_UP, ord("k")): + self.sel = (self.sel - 1) % len(self.ITEMS) + elif ch in (curses.KEY_DOWN, ord("j")): + self.sel = (self.sel + 1) % len(self.ITEMS) + elif ch in (curses.KEY_ENTER, 10, 13, ord(" ")): + self.activate() + elif ch == ord("q"): + self.app.running = False + elif ch == ord("h"): + self.app.push(HelpScreen(self.app)) + elif ch == ord("v"): + self.app.push(HistoryScreen(self.app)) + elif ch in (ord("1"), ord("2"), ord("3"), ord("4"), ord("5"), ord("6"), ord("7")): + self.sel = ch - ord("1") + self.activate() + + def activate(self): + idx = self.sel + if idx == 0: + self.app.push(TreeScreen(self.app, mode="docs")) + elif idx == 1: + self.app.push(TreeScreen(self.app, mode="code")) + elif idx == 2: + self.app.push(ArchScreen(self.app)) + elif idx == 3: + self.app.push(HistoryScreen(self.app)) + elif idx == 4: + self.app.push(SearchScreen(self.app)) + elif idx == 5: + self.app.push(HelpScreen(self.app)) + else: + self.app.running = False + + +DOC_DIRS = ("docs/",) +CODE_DIRS = ("kernel/", "userspace/", "scripts/", "tools/", "wiki/", "config/", + "drivers/", "include/", "docs/") + + +class TreeScreen(Screen): + """Left: file tree. Right: reader/viewer.""" + + def __init__(self, app, mode): + super().__init__(app) + self.mode = mode + self.files = self._collect() + self.folds: dict[str, bool] = {} + self.sel = 0 + self.offset = 0 + self.focus = "tree" + self.reader = None + self.open_path = None + + def _collect(self): + files = self.app.git.ls_files(self.app.rev) + pref = DOC_DIRS if self.mode == "docs" else CODE_DIRS + if self.mode == "docs": + out = [f for f in files if f.startswith(pref) and f.endswith(".md")] + out = ["README.md"] + out if "README.md" in files else out + else: + out = [f for f in files if f.startswith(pref)] + return sorted(set(out)) + + # tree flattening + def visible(self): + lines = [] + root = {} + for f in self.files: + parts = f.split("/") + node = root + for p in parts[:-1]: + node = node.setdefault(p, {}) + node[parts[-1]] = None + def walk(node, depth, prefix): + for name in sorted(node, key=lambda n: (node[n] is None, n)): + path = prefix + name + if node[name] is None: + lines.append((depth, path)) + else: + folded = self.folds.get(path, False) + lines.append((depth, path + "/", True, folded)) + if not folded: + walk(node[name], depth + 1, path + "/") + walk(root, 0, "") + return lines + + def open_file(self, path): + text = self.app.load_text(path) + if text is None: + self.app.status = f"cannot read {path}" + return + self.open_path = path + if self.mode == "docs": + w = self.app.w - 6 + self.reader = Reader(self.app, path, text, width=w) + else: + self.reader = CodeViewer(self.app, path, text) + self.focus = "reader" + + def title(self): + return "docs" if self.mode == "docs" else "code" + + def hints(self): + if self.focus == "tree": + return ("↑/↓ move · →/Enter open · ← fold · v history · / search · m menu · q menu") + return ("↑/↓ scroll · ←/→ hscroll · PgUp/PgDn · g/G top/bottom · m mermaid source · " + "b back to tree · q menu") + + def draw(self, scr): + if self.reader and self.focus == "reader": + self.reader.draw(scr, top=self.top + 1, left=0, width=self.app.w) + else: + self._draw_tree(scr) + + def _draw_tree(self, scr): + lines = self.visible() + if self.sel >= len(lines): + self.sel = max(0, len(lines) - 1) + avail = self.bottom - self.top - 1 + if self.sel < self.offset: + self.offset = self.sel + if self.sel >= self.offset + avail: + self.offset = self.sel - avail + 1 + for i in range(avail): + idx = self.offset + i + if idx >= len(lines): + break + item = lines[idx] + depth = item[0] + path = item[1] + attr = PAIRS["sel"] if idx == self.sel else PAIRS["normal"] + label = " " * (depth * 2) + if len(item) > 2: + label += ("▸ " if item[3] else "▾ ") + path.rstrip("/").split("/")[-1] + "/" + else: + label += path.split("/")[-1] + safe_addstr(scr, self.top + i, 1, label[:32], attr) + # divider + for y in range(self.top, self.bottom + 1): + safe_addstr(scr, y, 33, "│", PAIRS["gutter"]) + + def key(self, ch): + if self.focus == "reader" and self.reader: + if ch in (ord("b"), 27): + self.focus = "tree" + return + self.reader.key(ch) + return + lines = self.visible() + if ch in (curses.KEY_UP, ord("k")): + self.sel = max(0, self.sel - 1) + elif ch in (curses.KEY_DOWN, ord("j")): + self.sel = min(len(lines) - 1, self.sel + 1) + elif ch in (curses.KEY_ENTER, 10, 13): + item = lines[self.sel] + if len(item) > 2: + self.folds[item[1]] = not self.folds.get(item[1], False) + else: + self.open_file(item[1]) + elif ch in (curses.KEY_RIGHT, ord("l")): + item = lines[self.sel] + if len(item) > 2: + self.folds[item[1]] = False + elif ch in (curses.KEY_LEFT, ord("h")): + item = lines[self.sel] + if len(item) > 2: + self.folds[item[1]] = True + elif ch == ord("q"): + self.app.pop() + elif ch == ord("v"): + self.app.push(HistoryScreen(self.app)) + elif ch == ord("/"): + self.app.push(SearchScreen(self.app)) + elif ch == ord("m"): + self.app.pop() + + +class Reader: + def __init__(self, app, path, text, width): + self.app, self.path, self.text = app, path, text + self.width = max(40, width - 2) + self.offset = 0 + self.xoff = 0 + self.render() + + def render(self): + self.lines = render_markdown(self.text, self.width, + mermaid_src=self.app.mermaid_src) + + def draw(self, scr, top, left, width): + avail = scr.getmaxyx()[0] - top - 2 + if self.offset > max(0, len(self.lines) - avail): + self.offset = max(0, len(self.lines) - avail) + total = max(1, len(self.lines)) + for i in range(avail): + idx = self.offset + i + if idx >= len(self.lines): + break + text, attr = self.lines[idx] + if self.xoff: + text = text[self.xoff:] + safe_addstr(scr, top + i, left, text, PAIRS.get(attr, 0)) + pct = int(100 * min(1.0, (self.offset + avail) / total)) + pos = f" {self.path} [{pct:3d}%] " + safe_addstr(scr, scr.getmaxyx()[0] - 2, left, pos.ljust(width)[:width], + PAIRS["dim"]) + + def key(self, ch): + page = max(1, self.app.h - 4) + if ch in (curses.KEY_DOWN, ord("j")): + self.offset += 1 + elif ch in (curses.KEY_UP, ord("k")): + self.offset = max(0, self.offset - 1) + elif ch == ord(" "): + self.offset += page + elif ch in (curses.KEY_PPAGE,): + self.offset = max(0, self.offset - page) + elif ch in (curses.KEY_NPAGE,): + self.offset += page + elif ch in (ord("g"),): + self.offset = 0 + elif ch in (ord("G"),): + self.offset = max(0, len(self.lines) - 2) + elif ch in (curses.KEY_RIGHT, ord("l")): + self.xoff += 4 + elif ch in (curses.KEY_LEFT, ord("h")): + self.xoff = max(0, self.xoff - 4) + elif ch == ord("m"): + self.app.mermaid_src = not self.app.mermaid_src + self.render() + elif ch == ord("q"): + self.app.pop() + + +class CodeViewer: + def __init__(self, app, path, text): + self.app, self.path, self.text = app, path, text + self.offset = 0 + self.xoff = 0 + self.show_added = True + lang = os.path.splitext(path)[1].lstrip(".") or "txt" + self.segs = highlight(text, lang) + try: + self.added = app.git.added_lines(path, app.rev) if app.rev else set() + except Exception: + self.added = set() + + def draw(self, scr, top, left, width): + avail = scr.getmaxyx()[0] - top - 2 + n = len(self.segs) + if self.offset > max(0, n - avail): + self.offset = max(0, n - avail) + for i in range(avail): + idx = self.offset + i + if idx >= n: + break + gutter = f"{idx+1:5d} " + attr = PAIRS["added"] if idx + 1 in self.added else PAIRS["gutter"] + safe_addstr(scr, top + i, left, gutter, attr) + x = left + len(gutter) + for seg, a in self.segs[idx]: + if x - left >= width: + break + safe_addstr(scr, top + i, x, seg, PAIRS.get(a, 0)) + x += len(seg) + pct = int(100 * min(1.0, (self.offset + avail) / max(1, n))) + pos = f" {self.path} [{pct:3d}%] {'+added' if self.added and self.show_added else ''}" + safe_addstr(scr, scr.getmaxyx()[0] - 2, left, pos.ljust(width)[:width], + PAIRS["dim"]) + + def key(self, ch): + page = max(1, self.app.h - 4) + if ch in (curses.KEY_DOWN, ord("j")): + self.offset += 1 + elif ch in (curses.KEY_UP, ord("k")): + self.offset = max(0, self.offset - 1) + elif ch == ord(" "): + self.offset += page + elif ch in (curses.KEY_PPAGE,): + self.offset = max(0, self.offset - page) + elif ch in (curses.KEY_NPAGE,): + self.offset += page + elif ch == ord("g"): + self.offset = 0 + elif ch == ord("G"): + self.offset = len(self.segs) - 2 + elif ch == ord("a"): + self.show_added = not self.show_added + elif ch in (curses.KEY_RIGHT, ord("l")): + self.xoff += 4 + elif ch in (curses.KEY_LEFT, ord("h")): + self.xoff = max(0, self.xoff - 4) + elif ch == ord("q"): + self.app.pop() + + +class HistoryScreen(Screen): + def __init__(self, app): + super().__init__(app) + self.items = [("HEAD (worktree)", None)] + for b in app.git.branches(): + self.items.append((f"branch {b}", b)) + for l in app.git.log(300): + h = l.split()[0] + self.items.append((l, h)) + self.sel = 0 + self.offset = 0 + + def draw(self, scr): + safe_addstr(scr, self.top, 2, "Select a revision to browse (read-only; no checkout)", + PAIRS["h2"]) + avail = self.bottom - self.top - 2 + if self.sel < self.offset: + self.offset = self.sel + if self.sel >= self.offset + avail: + self.offset = self.sel - avail + 1 + for i in range(avail): + idx = self.offset + i + if idx >= len(self.items): + break + label, rev = self.items[idx] + attr = PAIRS["sel"] if idx == self.sel else PAIRS["normal"] + safe_addstr(scr, self.top + 2 + i, 2, label[: self.app.w - 6], attr) + + def key(self, ch): + if ch in (curses.KEY_UP, ord("k")): + self.sel = max(0, self.sel - 1) + elif ch in (curses.KEY_DOWN, ord("j")): + self.sel = min(len(self.items) - 1, self.sel + 1) + elif ch in (curses.KEY_ENTER, 10, 13): + label, rev = self.items[self.sel] + if rev: + if not self.app.git.rev_exists(rev): + self.app.status = f"cannot resolve {rev}" + return + self.app.set_rev(rev) + d = self.app.git.describe(rev) + self.app.status = f"browsing {d}" + else: + self.app.set_rev(None) + self.app.status = "browsing worktree" + self.app.pop() + elif ch == ord("q"): + self.app.pop() + + def hints(self): + return "↑/↓ choose revision · Enter browse · q back" + + +class SearchScreen(Screen): + def __init__(self, app): + super().__init__(app) + self.query = "" + self.results = [] + self.sel = 0 + self.offset = 0 + self.mode = "docs" + + def draw(self, scr): + safe_addstr(scr, self.top, 2, "search:", PAIRS["h2"]) + safe_addstr(scr, self.top, 10, self.query[: self.app.w - 12], PAIRS["normal"]) + safe_addstr(scr, self.top + 1, 2, "Enter search · Tab scope (docs/code) · Esc back", + PAIRS["dim"]) + avail = self.bottom - self.top - 3 + if self.sel < self.offset: + self.offset = self.sel + if self.sel >= self.offset + avail: + self.offset = self.sel - avail + 1 + for i in range(avail): + idx = self.offset + i + if idx >= len(self.results): + break + path, line, text = self.results[idx] + attr = PAIRS["sel"] if idx == self.sel else PAIRS["normal"] + label = f"{path}:{line}: {text.strip()}" + safe_addstr(scr, self.top + 3 + i, 2, label[: self.app.w - 4], attr) + + def key(self, ch): + if ch in (curses.KEY_ENTER, 10, 13): + self.do_search() + elif ch == 9: # tab + self.mode = "code" if self.mode == "docs" else "docs" + self.app.status = f"scope: {self.mode}" + elif ch in (curses.KEY_UP, ord("k")): + if self.results: + self.sel = max(0, self.sel - 1) + elif self.query: + pass + elif ch in (curses.KEY_DOWN, ord("j")): + if self.results: + self.sel = min(len(self.results) - 1, self.sel + 1) + elif ch in (curses.KEY_BACKSPACE, 127, 8): + self.query = self.query[:-1] + elif ch == 27: + self.app.pop() + elif 32 <= ch < 127: + self.query += chr(ch) + + def do_search(self): + if not self.query: + return + paths = ["docs", "userspace", "kernel", "scripts", "tools", "wiki", "README.md"] + paths = [p for p in paths if os.path.exists(os.path.join(ROOT, p))] + try: + self.results = self.app.git.grep(self.query, self.app.rev, paths) + except Exception as e: + self.app.status = f"search failed: {e}" + self.results = [] + if not self.results: + self.app.status = "no matches" + + def hints(self): + return "type query · Enter search · ↑/↓ results · Tab scope · Esc back" + + +class ArchScreen(Screen): + def __init__(self, app): + super().__init__(app) + self.path = ["root"] + self.sel = 0 + self.info = False + self.render() + + @property + def node(self): + nid = self.path[-1] + if nid == "root": + return ARCH + meta = COMPONENTS.get(nid, {}) + return dict(id=nid, title=meta.get("title", nid), status=meta.get("status", ""), + desc=meta.get("desc", ""), sources=meta.get("sources", []), + children=meta.get("children", []), + edges=[]) + + def node_ids(self): + n = self.node + if n["id"] == "root": + return ["guest", "hyp", "module", "gpu", "hostos"] + return n.get("children", []) + + def edges(self): + if self.node["id"] == "root": + return ARCH["edges"] + if self.node["id"] == "module": + return [("vdev", "emucore", "doorbells"), + ("emucore", "memmodel", "objects"), + ("emucore", "workpath", "pushes"), + ("memmodel", "workpath", "translations"), + ("workpath", "backend", "forward"), + ("backend", "gpu", "real GSP")] + return [] + + def render(self): + n = self.node + ids = self.node_ids() + if n["id"] == "root" or n["id"] == "module": + nodes = [] + for i, cid in enumerate(ids): + meta = COMPONENTS.get(cid, {}) + nodes.append(GNode(cid, f"{meta.get('title', cid)}\n{meta.get('status', '')}", + "round", -1, i + 1)) + edges = [GEdge(a, b, l) for a, b, l in self.edges()] + cv = layout_graph(nodes, edges, "TB") + self.lines = cv.to_lines() + self.positions = {nd.id: (nd.x, nd.y, nd.w, nd.h) for nd in nodes} + else: + self.lines = [] + self.positions = {} + m = COMPONENTS.get(n["id"], {}) + self.detail = m + + def draw(self, scr): + n = self.node + safe_addstr(scr, self.top, 2, f"architecture: {n['title']} {n['status']}", + PAIRS["h1"]) + if n["id"] == "root" or n["id"] == "module": + for i, line in enumerate(self.lines): + if self.top + 2 + i >= self.bottom: + break + safe_addstr(scr, self.top + 1 + i, 2, line, PAIRS["mermaid"]) + y = self.bottom - 2 + safe_addstr(scr, y, 2, "components:", PAIRS["h3"]) + x = 15 + for i, cid in enumerate(self.node_ids()): + meta = COMPONENTS.get(cid, {}) + attr = PAIRS["sel"] if i == self.sel else PAIRS["normal"] + safe_addstr(scr, y, x, f" {meta.get('title', cid)} ", attr) + x += len(meta.get("title", cid)) + 3 + else: + if self.info and self.detail: + d = self.detail + y = self.top + 1 + for line in textwrap.wrap(d.get("desc", ""), self.app.w - 6): + safe_addstr(scr, y, 2, line, PAIRS["normal"]) + y += 1 + y += 1 + if d.get("calls"): + safe_addstr(scr, y, 2, "calls into:", PAIRS["h3"]) + y += 1 + for c in d["calls"]: + safe_addstr(scr, y, 4, "• " + c, PAIRS["normal"]) + y += 1 + y += 1 + if d.get("sources"): + safe_addstr(scr, y, 2, "source files (Enter to open):", PAIRS["h3"]) + y += 1 + for i, s in enumerate(d["sources"]): + attr = PAIRS["sel"] if i == self.sel else PAIRS["normal"] + safe_addstr(scr, y, 4, s, attr) + y += 1 + + def key(self, ch): + n = self.node + if n["id"] in ("root", "module"): + ids = self.node_ids() + if ch in (curses.KEY_LEFT, ord("h")): + self.sel = (self.sel - 1) % len(ids) + elif ch in (curses.KEY_RIGHT, ord("l")): + self.sel = (self.sel + 1) % len(ids) + elif ch in (curses.KEY_ENTER, 10, 13): + cid = ids[self.sel] + meta = COMPONENTS.get(cid, {}) + if meta.get("children"): + self.path.append(cid) + self.sel = 0 + self.info = False + self.render() + else: + self.path.append(cid) + self.info = True + self.sel = 0 + self.render() + elif ch in (ord("b"), 27): + if len(self.path) > 1: + self.path.pop() + self.sel = 0 + self.info = False + self.render() + else: + self.app.pop() + elif ch == ord("q"): + self.app.pop() + else: + d = COMPONENTS.get(n["id"], {}) + srcs = d.get("sources", []) + if ch in (curses.KEY_UP, ord("k")): + self.sel = max(0, self.sel - 1) + elif ch in (curses.KEY_DOWN, ord("j")): + self.sel = min(max(0, len(srcs) - 1), self.sel + 1) + elif ch in (curses.KEY_ENTER, 10, 13) and srcs: + self.app.push(TreeScreen(self.app, mode="code")) + # best effort: open file directly + top = self.app.stack[-1] + top.focus = "reader" + top.files = self.app.git.ls_files(self.app.rev) + top.open_file(srcs[self.sel]) + elif ch in (ord("b"), 27): + self.path.pop() + self.sel = 0 + self.info = False + self.render() + elif ch == ord("q"): + self.app.pop() + + def hints(self): + if self.node["id"] in ("root", "module"): + return "←/→ select · Enter zoom in · b zoom out · q menu" + return "↑/↓ source file · Enter open in code browser · b zoom out · q menu" + + +class HelpScreen(Screen): + def __init__(self, app): + super().__init__(app) + self.offset = 0 + + def draw(self, scr): + help_lines = [ + ("keys", "h2"), + ("", "normal"), + (" Global", "h3"), + (" m back to the main menu", "normal"), + (" q quit / go back", "normal"), + (" v history: browse any commit or branch (no checkout)", "normal"), + (" / search docs and code", "normal"), + ("", "normal"), + (" Docs reader", "h3"), + (" ↑/↓ j/k scroll one line", "normal"), + (" Space/PgDn page down", "normal"), + (" g / G top / bottom", "normal"), + (" h / l scroll horizontally (wide diagrams)", "normal"), + (" m toggle rendered mermaid diagrams vs source", "normal"), + ("", "normal"), + (" Code viewer", "h3"), + (" a toggle highlighting lines added vs the parent commit", "normal"), + (" g / G top / bottom", "normal"), + ("", "normal"), + (" Tree", "h3"), + (" Enter open file / fold directory", "normal"), + (" ←/→ fold / unfold", "normal"), + ("", "normal"), + ("about", "h2"), + ("", "normal"), + (" This browser is read-only: it reads files from the worktree or from", "normal"), + (" git objects (git show / git ls-tree / git grep) and never checks out,", "normal"), + (" writes, stages or commits anything.", "normal"), + ("", "normal"), + (" Mermaid diagrams are rendered by a built-in renderer for flowcharts,", "normal"), + (" state diagrams and sequence diagrams (the types used in this repo).", "normal"), + (" Press m in a reader to see the source of any diagram.", "normal"), + ] + avail = self.bottom - self.top + for i in range(avail): + idx = self.offset + i + if idx >= len(help_lines): + break + text, attr = help_lines[idx] + safe_addstr(scr, self.top + i, 2, text, PAIRS.get(attr, 0)) + + def key(self, ch): + if ch in (curses.KEY_DOWN, ord("j")): + self.offset += 1 + elif ch in (curses.KEY_UP, ord("k")): + self.offset = max(0, self.offset - 1) + elif ch in (ord("q"), 27): + self.app.pop() + + def hints(self): + return "↑/↓ scroll · q back" + + +# -------------------------------------------------------------------------- +# non-interactive helpers +# -------------------------------------------------------------------------- + + +def ansi(text: str, attr: str, enabled: bool) -> str: + if not enabled: + return text + codes = { + "h1": "1;36", "h2": "1;34", "h3": "1;35", "code": "33", + "comment": "32", "string": "35", "keyword": "36", "number": "33", + "preproc": "31", "dim": "90", "error": "31", "mermaid": "32", + "quote": "34", "tablehead": "36", "sel": "7", "title": "1;36", + "added": "32", + }.get(attr) + if not codes: + return text + return f"\x1b[{codes}m{text}\x1b[0m" + + +def cmd_tree(git: Git, rev: str | None): + for f in git.ls_files(rev): + print(f) + + +def cmd_render(git: Git, path: str, rev: str | None, color: bool): + text = git.show(path, rev) + if text is None: + print(f"cannot read {path}", file=sys.stderr) + return 1 + width = shutil.get_terminal_size((100, 24)).columns - 1 + lines = render_markdown(text, max(60, width), mermaid_src=False) + for line, attr in lines: + print(ansi(line, attr, color)) + return 0 + + +def cmd_diagrams(git: Git, path: str, rev: str | None, color: bool): + text = git.show(path, rev) + if text is None: + print(f"cannot read {path}", file=sys.stderr) + return 1 + blocks = re.findall(r"```mermaid\n(.*?)```", text, re.S) + if not blocks: + print("no mermaid blocks") + return 0 + for i, b in enumerate(blocks, 1): + first = b.strip().splitlines()[0].strip().split()[0] + print(ansi(f"── diagram {i} [{first}] ─────────────────────────────", "h2", color)) + rendered = render_mermaid(b) + if rendered: + for l in rendered: + print(ansi(l, "mermaid", color)) + else: + print(ansi("(unrenderable; source follows)", "error", color)) + print(b) + print() + return 0 + + +def cmd_log(git: Git, n: int): + for l in git.log(n): + print(l) + + +# -------------------------------------------------------------------------- +# entry point +# -------------------------------------------------------------------------- + + +def main(argv=None): + if not sys.stdout.isatty(): + try: + import signal + signal.signal(signal.SIGPIPE, signal.SIG_DFL) + except (ImportError, ValueError, AttributeError): + pass + ap = argparse.ArgumentParser(description="Read-only TUI browser for Open-IOV docs/code.") + ap.add_argument("--rev", default=None, help="browse a specific commit/branch") + ap.add_argument("--tree", nargs="?", const="", default=None, + help="print the file tree (optional rev) and exit") + ap.add_argument("--render", metavar="FILE", help="render one Markdown file to stdout") + ap.add_argument("--diagrams", metavar="FILE", help="render mermaid blocks of a file") + ap.add_argument("--log", nargs="?", const="40", default=None, + help="print recent commits (optional count)") + ap.add_argument("--no-color", action="store_true") + ap.add_argument("--version", action="store_true") + args = ap.parse_args(argv) + if not sys.stdout.isatty(): + args.no_color = True + + if args.version: + print(f"browse-project.py {VERSION}") + return 0 + if not os.path.isdir(os.path.join(ROOT, ".git")): + print(f"warning: {ROOT} is not a git repository root", file=sys.stderr) + + git = Git(ROOT) + + if args.tree is not None: + cmd_tree(git, args.rev or (args.tree or None)) + return 0 + if args.render: + return cmd_render(git, args.render, args.rev, not args.no_color) + if args.diagrams: + return cmd_diagrams(git, args.diagrams, args.rev, not args.no_color) + if args.log is not None: + try: + n = int(args.log) + except ValueError: + n = 40 + cmd_log(git, n) + return 0 + + if not sys.stdout.isatty(): + print("interactive mode needs a terminal. Try --render/--tree/--log/--diagrams.", + file=sys.stderr) + return 2 + + app = App(git) + if args.rev: + if not git.rev_exists(args.rev): + print(f"cannot resolve revision: {args.rev}", file=sys.stderr) + return 2 + app.rev = args.rev + curses.wrapper(app.run) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) |
