File size: 12,082 Bytes
ad474c9
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
"""The boot chain: POST screen, shader splash, AUTOEXEC, desktop.

Three phases, each rendered by the graphics engine rather than assembled as
strings, and each doing real work rather than miming it:

1. **POST.**  The 16-bit BIOS in :mod:`hdc.x86` runs for real (IVT installed,
   boot sector read from LBA 0, `int 10h` teletype) and its text buffer is painted
   onto a phosphor surface with a scanline pass.  The lines on screen are the
   lines the BIOS produced.
2. **Splash.**  A shader frame: gradient, fractal noise, a `max_blend` of a second
   noise field, scanline and vignette — regenerated per frame with a shifted seed,
   so the animation *is* the shader pipeline running, not a GIF.  A title and a
   progress bar are drawn into the coded cells with exact levels.
3. **AUTOEXEC.**  A boot script whose every line is executed against the live
   desktop: `GFX.INIT` measures the shader pass, `MOUNT` changes the filesystem's
   working directory, `FS.SEARCH` runs the real hypervector search, `START`
   launches the real windows, `AUTONOMY` turns the readahead loop on.

``BootSequence.run()`` records the frames it drew, so a client can replay the
boot without re-running it, and the desktop it hands over is the state that
actually resulted from those commands.
"""

from __future__ import annotations

import time

from .fs import tokenize  # noqa: F401  (kept for symmetry with fs.search's contract)
from .gfx import Surface
from .hypervec import DEFAULT_DIM

#: Lines executed in order; the handler returns the text shown beside each.
AUTOEXEC: tuple[str, ...] = (
    "SET DIM     %d",
    "GFX.INIT    render shader passes",
    "MOUNT       /home/user",
    "FS.SEARCH   shader notes about edges",
    "START       terminal",
    "START       monitor",
    "START       files",
    "AUTONOMY    enable readahead",
)

#: Splash animation frames and how long each is held.
SPLASH_FRAMES = 6
SPLASH_MS = 110


class BootSequence:
    """Run and record the boot: BIOS, shader splash, AUTOEXEC."""

    def __init__(self, desktop, d: int | None = None):
        self.desktop = desktop
        self.d = d or desktop.d
        self.frames: list[dict] = []
        self.log: list[dict] = []
        self.started = 0.0
        self.elapsed_ms = 0.0

    # ---------------------------------------------------------------- phases
    def _post_lines(self) -> list[str]:
        from .x86 import Boot

        boot = Boot()
        boot.post()
        result = boot.boot()
        lines = [line for line in result["screen"] if line.strip()]
        self.log.append({"phase": "post", "lines": len(lines),
                         "stage": result["stage"],
                         "instructions": result["instructions"]})
        return lines

    def _post_frame(self, lines: list[str]) -> dict:
        surface = Surface(self.desktop.cols, self.desktop.rows, d=self.d)
        surface.clear(4)
        surface.scanline(1)
        for index, line in enumerate(lines[: self.desktop.rows - 2]):
            surface.text_at(1, 1 + index, line, 200)
        surface.text_at(1, self.desktop.rows - 2,
                        "Press F1-F6 after boot   |   HDCSTORAGE BIOS", 120)
        return {"phase": "post", "lines": surface.rows(), "delay": 260,
                "status": "BIOS POST complete: %d lines" % len(lines)}

    def _splash_base(self) -> Surface:
        """The expensive part of the splash, rendered once.

        ``max_blend`` resolves each cell with a similarity comparator (up to 64
        probes), which measured ~350 ms for a full frame — 2.1 s across six
        frames.  The base is computed once here and every animation frame copies
        it and applies exact shifts, so the motion is still HDC arithmetic
        (``shift``) while the boot stays under a second.
        """
        cols, rows = self.desktop.cols, self.desktop.rows
        surface = Surface(cols, rows, d=self.d)
        surface.gradient(top=6, bottom=34, vertical=True)
        surface.noise(scale=9, octaves=3, seed=21, amplitude=120)
        surface.max_blend(scale=5, seed=24, amplitude=70)
        surface.scanline(1)
        surface.vignette(2)
        self.base_ms = 0.0
        return surface

    def _splash_frames(self) -> list[dict]:
        out: list[dict] = []
        cols, rows = self.desktop.cols, self.desktop.rows
        title = "H D C S T O R A G E"
        subtitle = "hyperdimensional computer  |  everything is a vector"
        box_x, box_w = cols // 2 - 19, 39
        top = max(2, rows // 2 - 4)
        bar_width = 33
        base = self._splash_base()
        for index in range(SPLASH_FRAMES):
            progress = (index + 1) / float(SPLASH_FRAMES)
            surface = base.copy()
            # a bright band sweeping down the screen: two exact shifts, and the
            # reason the splash looks like it is doing something
            band = 2 + (index * (rows - 8)) // max(1, SPLASH_FRAMES - 1)
            surface.brighten_region(1, 0, band, cols, 2)
            surface.fill(box_x, top - 1, box_w, 9, 12)
            surface.frame(box_x, top - 1, box_w, 9, 190)
            surface.text_at((cols - len(title)) // 2, top, title, 250)
            surface.text_at((cols - len(subtitle)) // 2, top + 2, subtitle, 170)
            filled = int(bar_width * progress)
            surface.fill(box_x + 4, top + 4, max(0, filled), 1, 245)
            surface.text_at(box_x + 3, top + 4, "[", 180)
            surface.text_at(box_x + 4 + bar_width, top + 4, "]", 180)
            percent = "%3d%%" % int(progress * 100)
            surface.text_at((cols - 12) // 2, top + 6, percent + "  gfx engine", 220)
            out.append({"phase": "splash", "lines": surface.rows(),
                        "delay": SPLASH_MS,
                        "status": "splash %s  shader noise+max_blend, sweep row %d"
                                  % (percent, band)})
        return out

    # -------------------------------------------------------------- autoexec
    def _autoexec_frames(self) -> list[dict]:
        out: list[dict] = []
        shown: list[str] = ["AUTOEXEC.BAT - HDCSTORAGE boot script", ""]
        for index, template in enumerate(AUTOEXEC):
            command = template % self.d if "%d" in template else template
            result = self._run_autoexec(command)
            shown.append("%-12s %s" % (command.split(None, 1)[0], result))
            self.log.append({"phase": "autoexec", "command": command, "result": result})
            if index % 2 == 1 or index == len(AUTOEXEC) - 1:
                out.append(self._text_frame(shown, "autoexec: " + result))
        return out

    def _run_autoexec(self, command: str) -> str:
        """Execute one boot line against the live desktop; return what happened."""
        words = command.split()
        verb = words[0].upper() if words else ""
        desktop = self.desktop
        try:
            if verb == "SET":
                return "d=%d" % self.d if len(words) < 2 else " ".join(words[1:])
            if verb == "GFX.INIT":
                from .gfx import SHADERS

                surface = Surface(desktop.cols, desktop.rows, d=self.d).apply("background")
                check = surface.verify()
                return "%d shaders, %d cells, verify %s" % (
                    len(SHADERS), desktop.cols * desktop.rows,
                    "ok" if check["ok"] else "FAILED")
            if verb == "MOUNT":
                target = words[1] if len(words) > 1 else "/"
                desktop.fs.chdir(target)
                stat = desktop.fs.stat()
                return "%s  %d files, %d dirs" % (target, stat["files"], stat["dirs"])
            if verb == "FS.SEARCH":
                query = " ".join(words[1:]) or "readme"
                hits = desktop.fs.search(query, limit=1)
                if not hits:
                    return "no match for %r" % query
                return "%s  (%+.3f)" % (hits[0][0], hits[0][1])
            if verb == "START":
                key = words[1] if len(words) > 1 else "terminal"
                # tiled, not cascaded: a booted desktop should look arranged
                spots = {"terminal": (1, 1), "monitor": (48, 1), "files": (1, 15)}
                spot = spots.get(key)
                window = (desktop.launch(key, *spot) if spot else desktop.launch(key))
                return "%s at %d,%d" % (window.title, window.x, window.y)
            if verb == "AUTONOMY":
                if desktop.autonomy is None:
                    return "no assistant attached"
                desktop.autonomy.enabled = True
                desktop.autonomy.tick()
                return "readahead on, %d intents" % len(desktop.autonomy.intents)
        except Exception as error:                  # a boot line must not abort boot
            return "[error] %s" % error
        return "unknown boot command %r" % command

    def _text_frame(self, lines: list[str], status: str) -> dict:
        surface = Surface(self.desktop.cols, self.desktop.rows, d=self.d)
        surface.clear(5)
        surface.scanline(1)
        for index, line in enumerate(lines[: self.desktop.rows - 2]):
            surface.text_at(2, 1 + index, line, 210)
        surface.text_at(2, self.desktop.rows - 2, "AUTOEXEC running ...", 140)
        return {"phase": "autoexec", "lines": surface.rows(), "delay": 220,
                "status": status}

    # ------------------------------------------------------------------- run
    def run(self) -> dict:
        """Play every phase against the live desktop and record the frames."""
        started = time.perf_counter()
        self.frames = []
        self.log = []
        self.frames.append(self._post_frame(self._post_lines()))
        self.frames.extend(self._splash_frames())
        self.frames.extend(self._autoexec_frames())
        self.frames.append({"phase": "desktop", "lines": self.desktop.frame(),
                            "delay": 420, "status": "desktop ready"})
        self.elapsed_ms = (time.perf_counter() - started) * 1000.0
        self.started = time.time()
        return self.result()

    def result(self) -> dict:
        phases: list[str] = []
        for frame in self.frames:
            if not phases or phases[-1] != frame["phase"]:
                phases.append(frame["phase"])
        return {"frames": self.frames, "phases": phases, "log": self.log,
                "elapsed_ms": round(self.elapsed_ms, 1),
                "frame_count": len(self.frames),
                "total_ms": sum(int(frame["delay"]) for frame in self.frames)}

    def report(self) -> str:
        result = self.result()
        lines = ["boot: %d frames in %.0f ms of rendering, %d ms of playback"
                 % (result["frame_count"], result["elapsed_ms"], result["total_ms"]),
                 "  phases: " + " -> ".join(result["phases"])]
        for entry in self.log:
            if entry["phase"] == "autoexec":
                lines.append("    %-10s %s" % (entry["command"].split()[0], entry["result"]))
            elif entry["phase"] == "post":
                lines.append("    POST       %d lines, stage %s, %d instructions"
                             % (entry["lines"], entry["stage"], entry["instructions"]))
        return "\n".join(lines)


def render(frames: list[dict], marks: bool = True) -> str:
    """Printable form: frames separated by a phase banner (for the CLI)."""
    out: list[str] = []
    for index, frame in enumerate(frames):
        if marks:
            out.append("--- frame %d/%d  [%s]  %s ---"
                       % (index + 1, len(frames), frame["phase"], frame["status"]))
        out.extend(frame["lines"])
    return "\n".join(out)


def demo(d: int = DEFAULT_DIM) -> str:
    """Boot a throwaway desktop and print what the sequence produced."""
    from .os import Desktop

    desktop = Desktop(d=d)
    sequence = BootSequence(desktop)
    sequence.run()
    return "\n".join([render(sequence.frames), "", sequence.report()])


if __name__ == "__main__":  # pragma: no cover - manual inspection
    print(demo(2000))