diff --git a/docs/development/writing-style.md b/docs/development/writing-style.md index a86f7d4..ffd7cb9 100644 --- a/docs/development/writing-style.md +++ b/docs/development/writing-style.md @@ -218,6 +218,7 @@ and in quoted text. | prompt | A dialog or a line that asks for typed text | | | page | All the content of a screen, which can be taller than the terminal. Also a written page | | | frame | The full image that MeshTerm composes for the terminal in one paint: the title bar, the panel, the body, and the footer | | +| video frame | One image of a video, such as the demo video that `scripts/video.py` makes | frame (for this) | | paint | One full update of the terminal | redraw, repaint, refresh (for the terminal) | | animation step | One image of an animation (a spinner) | frame (for this) | | panel | The box with a border around the body of a screen | panel (for the hardware) | diff --git a/scripts/README.md b/scripts/README.md index abb9431..8e912dc 100644 --- a/scripts/README.md +++ b/scripts/README.md @@ -7,6 +7,7 @@ These are tools for the repository. None of them is in the wheel. | `basemap-doctor.py` | Finds why a user's map has no basemap: certificates, DNS, proxy, or colour depth | | `cardputer-zero/` | `build-deb.sh` builds the Debian package for M5Stack's app store. Run it on the Cardputer Zero or on another arm64 Debian 13 system. `store/` holds the listing for the store (`app-builder.json`, which `czdev publish` reads), its icon, and its screenshots. `screenshot.py` saves the display of a Cardputer as a 320×170 PNG over SSH, as the store wants a screenshot. `store-publish.py` sends a package to the store with M5Stack's `czdev`, with two corrections. The release workflow runs it after each release. `launcher-entry.sh` puts a source checkout in the Cardputer Zero's app launcher. `remove` takes it out. `meshterm.png` is the package's icon: the website's logo, at the launcher's 100×100 size. `meshterm-dev.png` is its negative, for the checkout's entry, so that the two entries look different. `type-text.py` types a pasted string (an API key, a Wi-Fi password) on the Cardputer's keyboard from another computer, over SSH | | `screenshots.py` | Makes the screenshots of the website from the demo mesh (`--demo`): the main menu, the chat, the paths of a message, and the map, for the terminal, the Cardputer Zero, and the PicoCalc. They are the same at each run. The result goes to `docs/screenshots//`. It also draws the two wordmarks of the splash into `docs/screenshots/splash/`, which the website uses as its splash | +| `video.py` | Makes the demo video of the website from the demo mesh: a tour of less than two minutes in the terminal, with its poster and its play card. The result goes to `src/` of the `meshterm.net` checkout beside this one | | `picocalc-lyra/` | Setup of the Calculinux device: console font, palette, Wi-Fi. `xiao-radio/` is the kit for the XIAO radio (firmware build, flashing, UART setup) | | `uconsole/` | The uConsole's always-on SPI bridge service. MeshTerm can also control that radio directly, with no script (refer to `docs/devices/uconsole.md`) | @@ -74,3 +75,33 @@ screen, and the narrow mark on a phone. `--only splash` draws only these two ima After a change to a screen or to the art, run the script again, then rebuild the website (refer to the README of the `meshterm.net` repository). + +## video.py + +The website plays a video of MeshTerm under its buttons. This script makes it from the +demo mesh, in the terminal set of `screenshots.py`: the same 80×24 terminal, the same +clock, and the same painter. It types the keys of a tour of less than two minutes: the +chat, the paths of a message, a reply to a message that arrives, a trace, the map, +Repeater admin, and the live feed. A camera copies the screen 30 times each second. + +```bash +python scripts/video.py # into ../meshterm.net/src/ +python scripts/video.py --out some/folder +python scripts/video.py --recording tour.pickle # keep the recording +``` + +It needs ffmpeg with libx264 on the `PATH`, and the needs of `screenshots.py`. A run takes +approximately four minutes. The tour runs on the real clock, then the script paints each +different screen in parallel and encodes the video. With `--recording`, the script keeps +the recording in that file. When the file exists, a later run paints and encodes it again, +and does not run MeshTerm. + +The script writes three files: `demo.mp4`, `demo-poster.png` (the paths of a message, +which the website shows before the video plays), and `demo-play.png` (the poster with a +play mark and the length of the video). The camera stops while the tour waits for +something slow: the tiles of the map, and most of the read of the repeater's settings. +Thus the video jumps over the wait. + +Each run gives the same video to the eye, but not the same file, because the camera copies +the screen on the real clock. After a change to the screens, run the script, then rebuild +the website (refer to the README of the `meshterm.net` repository). diff --git a/scripts/screenshots.py b/scripts/screenshots.py index 7f2b4a5..876ea37 100644 --- a/scripts/screenshots.py +++ b/scripts/screenshots.py @@ -98,35 +98,15 @@ def main(argv: list[str]) -> int: args = parser.parse_args(argv) if args.child: return _child(args.child, args.out) - from meshterm.emulator.font import font_dir - - # The emulator finds its font under the config directory. Each run has a config - # directory of its own, so it gets the font directory of this machine by name. - fonts = os.environ.get("MESHTERM_EMULATOR_FONTS") or str(font_dir()) status = 0 if SPLASH in args.only: for path in _draw_splash(args.out / SPLASH): print(f"{SPLASH}: {path}", flush=True) for name in (name for name in args.only if name in SETS): - home = Path(tempfile.gettempdir()) / "meshterm-screenshots" / name - env = { - **os.environ, - "MESHTERM_HOME": str(home), - "MESHTERM_EMULATOR_FONTS": fonts, - "TZ": TZ, - "MESHTERM_POWERLINE": "full", - "MESHTERM_EMOJI": "1", - # The two wraps of a real terminal (column pins, and the last column of a - # Windows console) are for a terminal that measures glyphs in its own way. - # Here the VT parser measures them as MeshTerm does. - "MESHTERM_COLUMN_SNAP": "0", - "MESHTERM_FULL_WIDTH": "0", - "PYTHONIOENCODING": "utf-8", - } print(f"{name}: capturing...", flush=True) done = subprocess.run( [sys.executable, __file__, "--child", name, "--out", str(args.out)], - env=env, + env=child_env(Path(tempfile.gettempdir()) / "meshterm-screenshots" / name), check=False, ) if done.returncode: @@ -135,6 +115,34 @@ def main(argv: list[str]) -> int: return status +def child_env(home: Path) -> dict[str, str]: + """The environment of a child process that runs MeshTerm for a capture. + + Args: + home: The MeshTerm directory of the run (``MESHTERM_HOME``). The demo writes its + own directory under it again at each start, and keeps its tile cache there. + """ + from meshterm.emulator.font import font_dir + + # The emulator finds its font under the config directory. Each run has a config + # directory of its own, so it gets the font directory of this machine by name. + fonts = os.environ.get("MESHTERM_EMULATOR_FONTS") or str(font_dir()) + return { + **os.environ, + "MESHTERM_HOME": str(home), + "MESHTERM_EMULATOR_FONTS": fonts, + "TZ": TZ, + "MESHTERM_POWERLINE": "full", + "MESHTERM_EMOJI": "1", + # The two wraps of a real terminal (column pins, and the last column of a + # Windows console) are for a terminal that measures glyphs in its own way. + # Here the VT parser measures them as MeshTerm does. + "MESHTERM_COLUMN_SNAP": "0", + "MESHTERM_FULL_WIDTH": "0", + "PYTHONIOENCODING": "utf-8", + } + + # --------------------------------------------------------------------------- the child @@ -195,8 +203,43 @@ class Capture: return [list(row) for row in self.terminal.screen] +#: A tour: it gets the capture and a function that types text into MeshTerm. +Tour = Callable[[Capture, Callable[[str], None]], None] + + def _child(name: str, out: Path) -> int: """Capture one set in this process. The parent set the environment.""" + target = out / name + target.mkdir(parents=True, exist_ok=True) + + def tour(capture: Capture, type_text: Callable[[str], None]) -> None: + def save(screen: str) -> None: + path = target / f"{screen}.png" + image = _draw(name, capture.grid(), capture.terminal.palette) + image.save(path, optimize=True) + print(f"{name}: {path.relative_to(ROOT) if path.is_relative_to(ROOT) else path}") + + _tour(capture, type_text, save) + + return run_demo(name, tour, background=False) + + +def run_demo(name: str, tour: Tour, *, background: bool) -> int: + """Run MeshTerm with ``--demo`` in this process, as the set ``name``, and do ``tour``. + + The tour runs in a thread of its own while MeshTerm runs in the main thread. When the + tour ends, it stops the whole process: exit status 0 if it completed, 1 if it raised + an exception. The parent process set the environment (refer to :func:`child_env`). + + Args: + name: One of :data:`SETS`. + tour: What to do with MeshTerm. + background: Whether the demo device plays the background traffic of the world + (:data:`meshterm.demo.device.PLAY_BACKGROUND`). + + Returns: + 1, which comes only if MeshTerm stopped before the tour ended it. + """ import time_machine traveller = time_machine.travel(NOW, tick=True) @@ -212,7 +255,7 @@ def _child(name: str, out: Path) -> int: from meshterm.emulator.run import _palette, panel_output from meshterm.emulator.vt import Terminal - demo_device.PLAY_BACKGROUND = False + demo_device.PLAY_BACKGROUND = background handheld = DEVICES.get(name) if handheld is not None: cols, rows = handheld.platform.readable_cols, handheld.platform.readable_rows @@ -220,22 +263,14 @@ def _child(name: str, out: Path) -> int: else: cols, rows = TERMINAL_GRID depth, platform = ColorDepth.TRUE_COLOR, "regular" - target = out / name - target.mkdir(parents=True, exist_ok=True) lock = threading.Lock() with create_pipe_input() as pipe: terminal = Terminal(cols, rows, palette=_palette(), reply=pipe.send_text) capture = Capture(terminal, lock) - def save(screen: str) -> None: - path = target / f"{screen}.png" - image = _draw(name, capture.grid(), terminal.palette) - image.save(path, optimize=True) - print(f"{name}: {path.relative_to(ROOT) if path.is_relative_to(ROOT) else path}") - def drive() -> None: try: - _tour(capture, pipe.send_text, save) + tour(capture, pipe.send_text) except BaseException: # noqa: BLE001 - report it, then stop the whole child traceback.print_exc() os._exit(1) @@ -457,7 +492,19 @@ class TerminalPainter: self.pad = self.cw def draw(self, grid): # noqa: ANN001, ANN201 - cells, an Image - """The image of ``grid``.""" + """The image of ``grid``, in a palette of 256 colours. + + A palette of 256 colours, with no dither, looks the same as the full image and + makes the file four times smaller: a terminal has few colours, and the smooth + edges of the text need only a few steps of each. + """ + from PIL import Image + + image = self.render(grid) + return image.quantize(256, method=Image.Quantize.FASTOCTREE, dither=Image.Dither.NONE) + + def render(self, grid): # noqa: ANN001, ANN201 - cells, an Image + """The image of ``grid``, in full RGB colour.""" from PIL import Image, ImageDraw cols, rows = len(grid[0]), len(grid) @@ -488,11 +535,7 @@ class TerminalPainter: [x, y + self.ch - self.ss * 2, x + self.cw - 1, y + self.ch - self.ss], fill=fg, ) - # A palette of 256 colours, with no dither, looks the same as the full image and - # makes the file four times smaller: a terminal has few colours, and the smooth - # edges of the text need only a few steps of each. - image = image.resize((width // 2, height // 2), Image.Resampling.LANCZOS) - return image.quantize(256, method=Image.Quantize.FASTOCTREE, dither=Image.Dither.NONE) + return image.resize((width // 2, height // 2), Image.Resampling.LANCZOS) @staticmethod def _colours(style) -> tuple[tuple[int, int, int], tuple[int, int, int]]: # noqa: ANN001 diff --git a/scripts/video.py b/scripts/video.py new file mode 100644 index 0000000..1ae6601 --- /dev/null +++ b/scripts/video.py @@ -0,0 +1,654 @@ +# SPDX-License-Identifier: Apache-2.0 +"""Make the demo video of the website from the demo mesh. + + python scripts/video.py [--out DIR] [--recording FILE] + +The script runs MeshTerm with ``--demo`` in an 80×24 terminal, as the terminal set of +``scripts/screenshots.py`` does, and types the keys of a tour of less than two minutes: + +1. The main menu. +2. The chat on Public, and the two paths of a message from Beloeil. +3. A message that arrives on Public, and a reply that mentions its sender. +4. A trace to Brossard, across the river, and back. +5. The map. +6. Repeater admin on Mont-Royal: a read of its settings, then an advert. +7. The live feed, and one packet from it. + +A recorder (the camera) copies the screen at each video frame, 30 times each second, while +it rolls. The tour stops the camera while it waits for something slow (the full read of the +settings, the tiles of the map), so the video jumps over the wait. After the tour, the +script paints each different copy in parallel, as the terminal set of the screenshots +paints a screen, and ffmpeg encodes the video. + +The script writes three files into ``DIR``. By default, ``DIR`` is ``src/`` of the website +checkout beside this one (``../meshterm.net/src``), where the build of the website reads +them: + +* ``demo.mp4``: the video, H.264 High in ``yuv420p``, with its ``moov`` box at the start. + Thus a browser can start to play it before all of it arrives. +* ``demo-poster.png``: the video frame that the website shows before the video plays (the + paths of the message). +* ``demo-play.png``: the same image with a play mark and the length of the video. A page + that cannot play the video links to it with this image. + +Each run gives the same video to the eye, but not the same file. The tour types the same +keys at the same pace, and the demo plays the same messages and packets at the same times +after its start. But the camera copies the screen on the real clock, so a change can come +one video frame earlier or later than in another run. + +What it needs, beside the needs of ``scripts/screenshots.py``: ffmpeg, with libx264, on the +``PATH``. +""" + +from __future__ import annotations + +import argparse +import contextlib +import pickle +import random +import shutil +import subprocess +import sys +import tempfile +import threading +import time +from collections.abc import Callable, Iterator +from concurrent.futures import ProcessPoolExecutor +from pathlib import Path + +import screenshots +from screenshots import ROOT, Capture + +#: The video frames in each second. +FPS = 30 + +#: The default folder of the three files: ``src/`` of the website checkout. +SITE_SRC = ROOT.parent / "meshterm.net" / "src" + +#: The colours of the play mark and of the length on the card: the ``brand`` and +#: ``title.brand`` teals of the theme. +BRAND = (0x5E, 0xEA, 0xD4) +BRAND_LIGHT = (0x99, 0xF6, 0xE4) + +#: One copy of the screen: the rows of ``(character, style)`` cells of the VT parser. +Grid = tuple[tuple[tuple[str, object], ...], ...] + + +def main(argv: list[str]) -> int: + """Record the tour in a child process, then paint and encode the video.""" + parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + parser.add_argument("--out", type=Path, default=SITE_SRC) + parser.add_argument( + "--recording", + type=Path, + help="keep the recording in this file. If the file exists, use it and do not run MeshTerm", + ) + parser.add_argument("--child", type=Path, help=argparse.SUPPRESS) + args = parser.parse_args(argv) + if args.child: + _watch_live_start() + return screenshots.run_demo( + "terminal", lambda capture, keys: _record(capture, keys, args.child), background=True + ) + if not args.out.is_dir(): + raise SystemExit(f"video.py: {args.out} is not a folder. Pass --out.") + if shutil.which("ffmpeg") is None: + raise SystemExit("video.py: ffmpeg is not on the PATH") + with tempfile.TemporaryDirectory(prefix="meshterm-video-") as work: + recording = args.recording or Path(work) / "recording.pickle" + if not recording.exists(): + print("recording the tour...", flush=True) + home = Path(tempfile.gettempdir()) / "meshterm-video" + done = subprocess.run( + [sys.executable, __file__, "--child", str(recording)], + env=screenshots.child_env(home), + check=False, + ) + if done.returncode or not recording.exists(): + print(f"the recording failed ({done.returncode})", flush=True) + return 1 + tape = pickle.loads(recording.read_bytes()) + grids, frames = tape["grids"], tape["frames"] + seconds = len(frames) / FPS + print(f"painting {len(grids)} screens for {seconds:.1f} s of video...", flush=True) + paths = _paint(grids, Path(work)) + video = args.out / "demo.mp4" + print("encoding...", flush=True) + _encode(frames, paths, Path(work), video) + poster = grids[frames[min(tape["marks"]["poster"], len(frames) - 1)]] + written = [video, *_stills(poster, seconds, args.out)] + for path in written: + print(f"{path} ({path.stat().st_size // 1024} KB)", flush=True) + return 0 + + +# --------------------------------------------------------------------------- the child + +#: The time on the monotonic clock at which the demo device started its live traffic, +#: or ``None`` before that. The times of the lines of the demo script count from it. +_live_start: float | None = None + + +def _watch_live_start() -> None: + """Keep the time at which the demo device starts its live traffic. + + The device counts the time of each live line from the moment that the app subscribes + to its events, on the clock of the event loop (the monotonic clock). The tour uses this + time to start the camera a moment before a line arrives. + """ + from meshterm.demo.device import DemoDevice + + subscribe = DemoDevice.subscribe_events + + async def subscribe_events(self, on_event): # noqa: ANN001, ANN202 - the same as the original + global _live_start + if _live_start is None: + _live_start = time.monotonic() + return await subscribe(self, on_event) + + DemoDevice.subscribe_events = subscribe_events + + +class Recorder: + """The camera: a copy of the screen at each video frame, while it rolls. + + Each different copy is kept one time (:attr:`grids`), and each video frame is the index + of its copy (:attr:`frames`). When the camera is late, it gives the same copy to each + video frame that it missed. Thus the video keeps the time of the real clock. + """ + + def __init__(self, capture: Capture) -> None: + """Start the thread of the camera. It does not roll yet.""" + self.capture = capture + self.grids: list[Grid] = [] + self.frames: list[int] = [] + self.marks: dict[str, int] = {} + self._index: dict[Grid, int] = {} + self._last: tuple[Grid, int] | None = None + self._rolling = False + self._stopped = False + self._thread = threading.Thread(target=self._run, name="camera", daemon=True) + self._thread.start() + + def roll(self) -> None: + """Start to record.""" + self._rolling = True + + def cut(self) -> None: + """Stop to record. The video continues from here at the next roll.""" + self._rolling = False + + def mark(self, name: str) -> None: + """Give the next video frame a name.""" + self.marks[name] = len(self.frames) + + def stop(self) -> None: + """Stop the camera for good.""" + self._stopped = True + self._thread.join() + + def _run(self) -> None: + period = 1.0 / FPS + due: float | None = None + while not self._stopped: + if not self._rolling: + due = None + time.sleep(period / 2) + continue + now = time.monotonic() + if due is None: + due = now + if now < due: + time.sleep(due - now) + continue + count = int((now - due) / period) + 1 + self.frames.extend([self._copy()] * count) + due += count * period + + def _copy(self) -> int: + """The index of the copy of the screen as it is now.""" + with self.capture.lock: + grid = tuple(tuple(row) for row in self.capture.terminal.screen) + # The cells that did not change are the same objects, so this test is fast. + if self._last is not None and grid == self._last[0]: + return self._last[1] + index = self._index.get(grid) + if index is None: + index = self._index[grid] = len(self.grids) + self.grids.append(grid) + self._last = (grid, index) + return index + + +class Presenter: + """The person at the keyboard: key presses at a human pace, and the camera.""" + + def __init__(self, capture: Capture, type_text: Callable[[str], None], camera: Recorder): + """Keep the capture, the input of MeshTerm, and the camera.""" + self.capture = capture + self.camera = camera + self._type_text = type_text + # A fixed seed: the pace of the typing is the same at each run. + self._rng = random.Random("meshterm-video") + + def key(self, name: str = "", text: str = "", *, ctrl: bool = False, pause: float = 0.35): + """Press one keyboard key, then wait for ``pause`` seconds. + + After Esc, the wait is at least 0.7 s. prompt_toolkit waits a moment to see if an + Esc starts a longer sequence, and a key that comes in that time joins the Esc. + """ + from meshterm.emulator.keys import Key, encode + + self._type_text(encode(Key(name=name, text=text, ctrl=ctrl))) + time.sleep(max(pause, 0.7) if name == "escape" else pause) + + def type(self, text: str) -> None: + """Type ``text``, one character at a time, at the pace of a person.""" + for char in text: + self._type_text(char) + time.sleep(self._rng.uniform(0.06, 0.14)) + + def hold(self, seconds: float) -> None: + """Wait, while the camera records what is on the screen.""" + time.sleep(seconds) + + def wait_for(self, text: str, timeout: float = 30.0) -> None: + """Wait until ``text`` is on the screen.""" + self.capture.wait_for(text, timeout) + + def wait_until(self, moment: float) -> None: + """Wait until ``moment`` on the monotonic clock.""" + time.sleep(max(0.0, moment - time.monotonic())) + + def shows(self, text: str) -> bool: + """Whether ``text`` is on the screen now.""" + with self.capture.lock: + return text in self.capture.terminal.text() + + def gone(self, text: str, timeout: float = 120.0) -> None: + """Wait until ``text`` is not on the screen. + + Raises: + TimeoutError: The text stayed. + """ + start = time.monotonic() + while time.monotonic() - start < timeout: + if not self.shows(text): + return + time.sleep(0.1) + raise TimeoutError(f"{text!r} stayed on the screen") + + def highlight(self, label: str, key: str, tries: int = 12) -> None: + """Press ``key`` until the highlight is on the row of ``label``. + + Raises: + TimeoutError: The highlight did not come to the row. + """ + for _ in range(tries): + if self._highlighted(label): + return + self.key(key, pause=0.3) + self.capture.settle(quiet=0.3, timeout=5.0) + if not self._highlighted(label): + raise TimeoutError(f"the highlight did not come to {label!r}") + + def _highlighted(self, label: str) -> bool: + with self.capture.lock: + lines = self.capture.terminal.text().splitlines() + return any("❯" in line and label in line for line in lines) + + @contextlib.contextmanager + def off_camera(self) -> Iterator[None]: + """Stop the camera for the body of a ``with`` block, then roll it again.""" + self.camera.cut() + try: + yield + finally: + self.camera.roll() + + +def _record(capture: Capture, type_text: Callable[[str], None], path: Path) -> None: + """Record the tour, and write the recording to ``path``.""" + camera = Recorder(capture) + _tour(Presenter(capture, type_text, camera)) + camera.cut() + camera.stop() + tape = {"grids": camera.grids, "frames": camera.frames, "marks": camera.marks} + path.write_bytes(pickle.dumps(tape, protocol=pickle.HIGHEST_PROTOCOL)) + print(f"recorded {len(camera.frames) / FPS:.1f} s, {len(camera.grids)} screens", flush=True) + + +def _tour(p: Presenter) -> None: + """Type the keys of the tour, and roll the camera on what the video shows.""" + from meshterm.core import preferences + from meshterm.demo import script + + capture = p.capture + capture.wait_for("Chat", timeout=60) + capture.settle(quiet=0.5) + + print("the main menu", flush=True) + p.camera.roll() + p.hold(4.0) + + print("the chat, and the paths of a message", flush=True) + p.type("chat") + p.hold(0.5) + p.key("enter") + p.wait_for("pick a conversation") + p.hold(2.0) + p.type("pub") + p.hold(0.4) + p.key("enter") + p.wait_for(screenshots.PATHS_MESSAGE[2]) + p.hold(2.5) + p.key("up", pause=0.6) + p.key("up", pause=1.2) + # A key press can be lost while the transcript draws. The camera does not show the + # tries: it shows the dialog of the right message as it opens. + with p.off_camera(): + _open_paths(p) + p.hold(2.5) + # The second path. The first one is often a straight line. + p.key("down", pause=0.6) + p.camera.mark("poster") + p.hold(3.4) + p.key("escape") + p.key("escape") # the first Esc closes the dialog, the second unpicks the message + + print("a message arrives, and a reply", flush=True) + offset, _, line = next(item for item in script.LIVE_CHANNEL if item[1] == "Public") + with p.off_camera(): + p.wait_until(_start() + offset - 1.5) + p.wait_for(line.text, timeout=20) + p.hold(2.5) + p.key("up", pause=0.8) # select the new message + p.key("enter", pause=0.8) # its mention goes into the compose line + p.type(" count me in! I'll bring coffee") + p.hold(0.7) + p.key("enter", pause=0.0) + p.wait_for("0/130") # the compose line is empty: the message went + p.hold(4.0) + + print("a trace", flush=True) + _menu(p, "trace t") + p.wait_for("pick a target") + p.hold(1.2) + p.type("brossard") + p.hold(0.4) + p.key("enter") + p.wait_for("Trace — Brossard") + p.hold(2.5) + p.key("enter", pause=0.0) + p.wait_for("Per-hop medians", timeout=30) + p.hold(3.0) + p.key("pagedown", pause=0.0) + p.hold(4.0) + + print("the map", flush=True) + _menu(p, "map", enter=False) + with p.off_camera(): + p.key("enter", pause=0.0) + capture.settle(quiet=3.0, timeout=120.0) + p.hold(3.5) + p.key("pageup", pause=0.0) + capture.settle(quiet=1.0, timeout=30.0) + p.hold(2.5) + for _ in range(3): + p.key("right", pause=0.5) + capture.settle(quiet=1.0, timeout=30.0) + p.hold(2.5) + + print("a repeater administered over the air", flush=True) + _menu(p, "repeater a") + p.wait_for("node to manage") + p.hold(1.2) + p.type("mont-r") + p.hold(0.4) + p.key("enter") + p.wait_for("Repeater admin — Mont-Royal") + p.hold(2.0) + p.key("end", pause=0.6) + p.highlight("Read settings", "up") + p.hold(0.8) + p.key("enter", pause=0.0) + p.wait_for("Reading — Mont-Royal") + p.hold(2.5) + # The read is one round trip for each setting, with the transmit cooldown between two + # reads: more than three minutes in all. The camera shows the first reads at the real + # pace. Then, off camera, a short cooldown makes the run shorter, and the video + # does not show it. + with p.off_camera(): + cooldown = preferences.current().trace_cooldown_s + preferences.current().set("trace_cooldown_s", 0.2) + try: + p.gone("Reading — Mont-Royal") + finally: + preferences.current().set("trace_cooldown_s", cooldown) + capture.settle(quiet=0.5) + if p.shows("No reply from"): + p.key("enter") # the list of the settings that did not answer + capture.settle(quiet=0.5) + p.key("home", pause=0.0) + capture.settle(quiet=0.5) + p.hold(3.0) + p.key("pagedown") + p.hold(2.5) + p.key("end", pause=0.6) + p.highlight("Send advert", "up") + p.hold(0.8) + p.key("enter", pause=0.0) + p.wait_for("announce itself") + p.hold(1.5) + p.key("enter", pause=0.0) + p.wait_for("Advert sent", timeout=30) + p.hold(2.0) + p.key("enter", pause=0.8) + + print("the live feed, and one packet", flush=True) + _menu(p, "live f") + p.wait_for("Live feed") + p.hold(5.0) + for _ in range(3): + p.key("down", pause=0.5) + p.hold(0.5) + p.key("enter", pause=0.0) + p.hold(5.0) + p.key("escape") + p.hold(0.5) + # The end: the main menu, as at the start, with no filter. + _back_to_menu(p) + p.hold(3.0) + + +def _start() -> float: + """The start of the live traffic of the demo device (refer to :func:`_watch_live_start`).""" + if _live_start is None: + raise RuntimeError("the demo device did not start its live traffic") + return _live_start + + +def _back_to_menu(p: Presenter) -> None: + """Go back to the main menu (^W), and clear the filter that it can still hold. + + The Esc waits until the menu is on the screen. If it comes while ^W still unwinds the + screens, a screen that closes gets it, and the filter stays. + """ + p.key(text="w", ctrl=True, pause=0.0) + p.wait_for("What would you like to do?") + p.hold(0.6) + p.key("escape") + + +def _menu(p: Presenter, name: str, *, enter: bool = True) -> None: + """Go back to the main menu, and type ``name`` to filter it (then Enter).""" + _back_to_menu(p) + p.type(name) + p.hold(0.4) + if enter: + p.key("enter") + + +def _open_paths(p: Presenter) -> None: + """Open the paths dialog of the message of :data:`screenshots.PATHS_MESSAGE`. + + The message is selected before the call. If the dialog shows another message, close it + and select the next older message. + + Raises: + TimeoutError: No dialog showed the message. + """ + snippet = screenshots.PATHS_MESSAGE[2] + for _ in range(6): + p.key(text="p", ctrl=True, pause=0.0) + p.capture.settle(quiet=1.0) + quote = p.capture.quote() + if snippet in quote: + return + if quote: + p.key("escape") + p.capture.settle(quiet=0.6) + p.key("up") + p.capture.settle(quiet=0.6) + raise TimeoutError(f"no paths dialog showed {snippet!r}") + + +# --------------------------------------------------------------------------- the video + +#: The painter of each worker process (refer to :func:`_paint`). +_painter = None + + +def _start_painter() -> None: + global _painter + _painter = screenshots.TerminalPainter() + + +def _paint_one(job: tuple[int, Grid, str]) -> str: + index, grid, folder = job + path = Path(folder) / f"{index:05d}.png" + _painter.render(grid).save(path, compress_level=1) + return str(path) + + +def _paint(grids: list[Grid], folder: Path) -> list[Path]: + """Paint each copy of the screen into a PNG file in ``folder``, in parallel. + + Returns: + The path of the image of each copy, in the order of ``grids``. + """ + jobs = ((index, grid, str(folder)) for index, grid in enumerate(grids)) + with ProcessPoolExecutor(initializer=_start_painter) as pool: + return [Path(path) for path in pool.map(_paint_one, jobs, chunksize=8)] + + +def _encode(frames: list[int], paths: list[Path], work: Path, target: Path) -> None: + """Encode the video frames into ``target`` with ffmpeg. + + The video frames go to ffmpeg as a list of images, each with the time that it stays on + the screen (the concat demuxer). Thus each different image goes to ffmpeg one time, and + the ``fps`` filter makes the video frames again. The RGB to YUV conversion uses the + BT.709 matrix, which is the matrix that a browser uses for an HD video. The default + of ffmpeg is the BT.601 matrix, which would change the colours a little. + + The settings come from a comparison on one recording. A keyframe costs approximately + 120 KB (300 KB on the map), and the keyframes were half of the file. Thus there is one + keyframe each 30 s, and one at each change of scene. CRF 22 looks the same as CRF 20, + and the file is 3.4 MB instead of 3.8 MB. + """ + runs: list[list[int]] = [] + for index in frames: + if runs and runs[-1][0] == index: + runs[-1][1] += 1 + else: + runs.append([index, 1]) + lines = ["ffconcat version 1.0"] + for index, count in runs: + lines += [f"file '{paths[index].as_posix()}'", f"duration {count / FPS:.6f}"] + # The concat demuxer uses the duration of the last image only if the image comes again. + # The image that comes again adds its own time, and -t cuts it. + lines.append(f"file '{paths[runs[-1][0]].as_posix()}'") + listing = work / "frames.ffconcat" + listing.write_text("\n".join(lines) + "\n", encoding="utf-8") + colours = "color_primaries=bt709:color_trc=iec61966-2-1:colorspace=bt709:range=tv" + command = [ + "ffmpeg", "-y", "-hide_banner", "-loglevel", "error", + "-f", "concat", "-safe", "0", "-i", str(listing), + "-vf", f"fps={FPS},scale=out_color_matrix=bt709:out_range=tv,format=yuv420p," + f"setparams={colours}", + "-t", f"{len(frames) / FPS:.3f}", + "-c:v", "libx264", "-preset", "veryslow", "-tune", "animation", "-crf", "22", + "-profile:v", "high", "-g", str(FPS * 30), + "-movflags", "+faststart", "-an", str(target), + ] # fmt: skip + subprocess.run(command, check=True) + + +def _stills(grid: Grid, seconds: float, folder: Path) -> list[Path]: + """Write the poster and the card of the video into ``folder``. + + Returns: + The paths of the two images. + """ + from PIL import Image + + painter = screenshots.TerminalPainter() + image = painter.render(grid) + poster = folder / "demo-poster.png" + card = folder / "demo-play.png" + quantize = {"method": Image.Quantize.FASTOCTREE, "dither": Image.Dither.NONE} + image.quantize(256, **quantize).save(poster, optimize=True) + _card(image, seconds).quantize(256, **quantize).save(card, optimize=True) + return [poster, card] + + +def _card(image, seconds: float): # noqa: ANN001, ANN202 - an Image, an Image + """The poster with a play mark and the length of the video. + + The image is a little darker, so that the mark is clear on it. The length is on it, + because a person who decides to click must know that the video is short. + """ + from PIL import Image, ImageDraw, ImageEnhance, ImageFont + + width, height = image.size + ss = 4 # draw at four times the size, then scale down, for smooth edges + layer = Image.new("RGBA", (width * ss, height * ss), (0, 0, 0, 0)) + draw = ImageDraw.Draw(layer) + cx, cy = width * ss / 2, height * ss / 2 + radius = width * ss * 0.055 + ring = radius * 0.07 + draw.ellipse( + [cx - radius, cy - radius, cx + radius, cy + radius], + fill=(0, 0, 0, 150), + outline=(*BRAND, 255), + width=round(ring), + ) + draw.polygon( + [ + (cx - radius * 0.3, cy - radius * 0.42), + (cx + radius * 0.46, cy), + (cx - radius * 0.3, cy + radius * 0.42), + ], + fill=(*BRAND, 255), + ) + minutes, rest = divmod(round(seconds), 60) + label = f"{minutes}:{rest:02d}" + font = ImageFont.truetype(screenshots.TEXT_FONTS[0], round(width * ss * 0.024)) + left, top, right, bottom = draw.textbbox((0, 0), label, font=font) + pad = font.size * 0.45 + margin = width * ss * 0.012 + box = [ + width * ss - margin - (right - left) - 2 * pad, + height * ss - margin - (bottom - top) - 2 * pad, + width * ss - margin, + height * ss - margin, + ] + draw.rounded_rectangle(box, radius=font.size * 0.3, fill=(12, 12, 12, 225)) + draw.text((box[0] + pad - left, box[1] + pad - top), label, font=font, fill=(*BRAND_LIGHT, 255)) + layer = layer.resize((width, height), Image.Resampling.LANCZOS) + base = ImageEnhance.Brightness(image.convert("RGB")).enhance(0.75).convert("RGBA") + return Image.alpha_composite(base, layer).convert("RGB") + + +if __name__ == "__main__": + sys.exit(main(sys.argv[1:]))