diff --git a/.github/workflows/validate.yml b/.github/workflows/validate.yml index 7f0ccb19..bc4d29ec 100644 --- a/.github/workflows/validate.yml +++ b/.github/workflows/validate.yml @@ -213,9 +213,19 @@ jobs: run: python3.9 -m pip install pyyaml "jsonschema[format-nongpl]==4.25.1" # markdown: the site validation test renders a wiki line to prove the - # heading guard's premise, and skipped on every run without it. + # heading guard's premise, and skipped on every run without it. pyte + # reads the installer's full screen back in its end-to-end tests, + # hypothesis fuzzes its key decoder, and vermin holds install.py to 3.8. - name: Install dependencies - run: pip install pyyaml "jsonschema[format-nongpl]==4.26.0" "markdown==3.10.2" + run: pip install pyyaml "jsonschema[format-nongpl]==4.26.0" "markdown==3.10.2" "pyte==0.8.2" "hypothesis==6.170.8" "vermin==1.9.2" + + # The full screen also runs under the 3.8 floor: the end-to-end tests + # start install.py with the interpreter that runs them. + - name: Install the screen tests' dependency for Python 3.8 + run: python3.8 -m pip install "pyte==0.8.2" + + - name: Hold install.py to Python 3.8 + run: vermin --no-tips -t=3.8- --violations install.py # The site is built as deploy-site.yml builds it, from the lockfile. - uses: actions/setup-node@949feb2413d6458794dcd2491c4babbbce0c15c1 # v7.1.0 @@ -257,6 +267,12 @@ jobs: RETROBIOS_OFFLINE: "1" run: python -m unittest discover -s tests -t . -v + - name: Run the full-screen interface tests on Python 3.8 + if: ${{ !cancelled() }} + env: + RETROBIOS_OFFLINE: "1" + run: python3.8 -m unittest tests.test_install_tui -v + # deploy-site.yml runs this after the merge, and a dead link or a stray # heading failed it there with nothing in front of it. Run whatever the # suite returned, so one red step does not hide the other. The release diff --git a/install.ps1 b/install.ps1 index 9edaabee..e8fcc20f 100644 --- a/install.ps1 +++ b/install.ps1 @@ -11,7 +11,7 @@ $ErrorActionPreference = "Stop" $defaultInstallUrl = "https://raw.githubusercontent.com/Abdess/retrobios/main/install.py" - $defaultInstallSha256 = "8cfe5cb80b8df7f8c2736cfe5855a768c3eacad6ebf5bfd7336b3b4f5a4d5d76" + $defaultInstallSha256 = "84ba56c9498a8bf89db4c4af8bd3413871f467771b36aea36158458aa4644ef8" $maximumInstallerBytes = 2MB $installerExit = 0 $installer = if ($PSScriptRoot) { Join-Path $PSScriptRoot "install.py" } else { $null } diff --git a/install.py b/install.py index bf8b2288..a83ada0f 100644 --- a/install.py +++ b/install.py @@ -15,18 +15,23 @@ Usage: from __future__ import annotations import argparse +import codecs import concurrent.futures +import contextlib import datetime import errno +import functools import hashlib import http.client import io import json import os import platform +import queue import re import shlex import shutil +import signal import socket import stat import subprocess @@ -35,13 +40,15 @@ import tarfile import tempfile import threading import time +import traceback +import unicodedata import urllib.error import urllib.parse import urllib.request import xml.etree.ElementTree as ET import zipfile import zlib -from collections.abc import Callable, Collection, Iterator +from collections.abc import Callable, Collection, Iterator, Mapping from pathlib import Path, PurePosixPath from typing import BinaryIO, NamedTuple, TextIO @@ -493,6 +500,8 @@ EVENT_KEYS: dict[str, tuple[str, ...]] = { "file": ("platform", "dest", "systems", "size", "state"), "system": ("platform", "system", "name", "files", "present", "missing", "wrong_hash", "bytes_needed"), + # The files in place being read, at most every half second, and the last. + "checking": ("files_done", "files_total"), "check": ("platform", "name", "bios_dir", "files", "present", "missing", "wrong_hash", "bytes_needed", "data_missing", "not_installable"), "download": ("platform", "files", "bytes"), @@ -3102,7 +3111,9 @@ def _address_lines(url: str) -> list[str]: return lines -def _fetch_json(url: str, limit: int, label: str) -> object: +def _fetch_json( + url: str, limit: int, label: str, attempts: int = MAX_RETRIES, timeout: float = 30, +) -> object: """A JSON document, tried MAX_RETRIES times like every file download. A file list fetched once ended the run on one timeout over flaky Wi-Fi. @@ -3110,26 +3121,26 @@ def _fetch_json(url: str, limit: int, label: str) -> object: nothing. """ _network_allowed(label) - for attempt in range(1, MAX_RETRIES + 1): + for attempt in range(1, attempts + 1): try: - with urllib.request.urlopen(url, timeout=30) as resp: + with urllib.request.urlopen(url, timeout=timeout) as resp: return _read_limited_json(resp, limit, label) except urllib.error.HTTPError as exc: - if exc.code < 500 and exc.code != 429 or attempt == MAX_RETRIES: + if exc.code < 500 and exc.code != 429 or attempt == attempts: raise delay = _retry_delay(attempt, exc.headers.get("Retry-After") if exc.headers else None) problem = f"HTTP {exc.code}" except (urllib.error.URLError, OSError, http.client.HTTPException) as exc: code, problem = network_words(exc) # A wrong clock or no certificate list fails every attempt alike. - if attempt == MAX_RETRIES or code != "network": + if attempt == attempts or code != "network": if isinstance(exc, http.client.HTTPException): raise ConnectionError(f"connection cut: {exc!r}") from exc raise delay = _retry_delay(attempt, None) # Said, or three 30-second attempts read as a frozen program. print(f" The {label} did not arrive ({problem}), trying again " - f"({attempt + 1}/{MAX_RETRIES}) in {delay:.0f} s.", file=sys.stderr) + f"({attempt + 1}/{attempts}) in {delay:.0f} s.", file=sys.stderr) time.sleep(delay) raise AssertionError("unreachable") @@ -3833,13 +3844,20 @@ def check_local( lambda item: (item[0], _classify_local(item[0], item[1])), present ) live = term is not None and term.live and not term.quiet - last = 0.0 + last = evented = 0.0 for done, (entry, verdict) in enumerate(verdicts, 1): if verdict == "up_to_date": up_to_date.append(entry) else: mismatched.append(entry) now = time.monotonic() + # The program driving the installer gets the count the + # terminal gets: a screen with a bare timer over 6 GB on an + # SD card read as hung. + if EVENTS.on and (now - evented >= EVENT_PROGRESS_INTERVAL + or done == len(present)): + evented = now + EVENTS.emit("checking", files_done=done, files_total=len(present)) if live and (now - last >= 0.1 or done == len(present)): last = now print(f"\r read {done}/{len(present)}", end="", file=term.stream, @@ -6908,7 +6926,9 @@ def main() -> None: _INVOCATION.os_type = detect_os() parser = argparse.ArgumentParser( allow_abbrev=False, - description="Download missing BIOS files for retrogaming emulators.", + description="Download missing BIOS files for retrogaming emulators. " + "For questions line by line, as a screen reader reads them, " + "add --plain.", epilog=_help_epilog(), formatter_class=argparse.RawDescriptionHelpFormatter, ) @@ -7089,6 +7109,20 @@ def main() -> None: help="read the file list from FILE (install/.json) instead " "of downloading it", ) + parser.add_argument( + "--tui", + action="store_true", + help="open the full-screen interface, which a keyboard or a controller " + "drives. It opens by itself for an install at a terminal of 40x12 " + "or more", + ) + parser.add_argument( + "--plain", + action="store_true", + help="ask the questions line by line instead of the full-screen " + "interface, as a screen reader reads them (RETROBIOS_PLAIN=1 or " + "ACCESSIBLE=1 do the same)", + ) parser.add_argument( "--version", action="version", @@ -7119,6 +7153,15 @@ def main() -> None: or args.list_systems or args.list_cores or args.list_emulators ): parser.error("--quiet applies to an install, not to --check, --dry-run or a listing") + if args.tui and args.plain: + parser.error("--tui and --plain ask for two different interfaces") + if args.tui: + reason = tui_blocker(args) + if reason: + # The reason alone: fourteen lines of usage hid it. + print(f"{parser.prog}: error: --tui cannot open the full-screen interface " + f"here: {reason}", file=sys.stderr) + sys.exit(2) if args.restore is not None: # The folder says where each file goes: nothing else applies. alone = ("platform", "dest", "target", "every_board", "check", "list_platforms", @@ -7145,6 +7188,8 @@ def main() -> None: if args.no_game_data and (args.list_platforms or args.list_targets): parser.error("--no-game-data narrows a file list, not --list-platforms " "or --list-targets") + if args.tui or not (plain_requested(args) or tui_blocker(args)): + sys.exit(run_tui(args)) if args.events: # The events own stdout. The text a person reads goes to stderr, so a # program never has to tell one from the other. @@ -8051,6 +8096,2575 @@ def _install(args: argparse.Namespace, real_stdout: TextIO) -> None: ) +# Full-screen interface +# +# A bare run at a terminal opens it, and --tui asks for it. It decides and the +# installer acts: every check and install is this file run again as a child, +# with --events, the contract the handheld port reads, and the choices made on +# screen as its options. The line questions stay behind --plain, for screen +# readers and for any terminal this cannot drive. Stdlib only: curses is +# missing from Windows CPython and from some Buildroot images. + +TUI_MIN_COLS = 40 +TUI_MIN_ROWS = 12 +# Text stops growing past this width: on a 200-column window a row and its +# size sat a whole screen apart. +TUI_MAX_WIDTH = 100 +# An arrow key arrives as Esc and two more characters, Esc alone as itself: +# past this wait an Esc with nothing after it is the key. +TUI_ESC_WAIT = 0.04 +TUI_FRAME = 0.1 +_PASTE_START = "\x1b[200~" +_PASTE_END = "\x1b[201~" +# Turned on by the interface, so a pasted path arrives as one piece and its +# newline does not submit it half typed. +_SCREEN_ON = "\x1b[?1049h\x1b[?25l\x1b[?2004h\x1b[2J" +_SCREEN_OFF = "\x1b[?2004l\x1b[0m\x1b[?25h\x1b[?1049l" +_STYLES = {"title": "\x1b[1m", "focus": "\x1b[7m", "dim": "\x1b[2m"} + + +class TuiKey(NamedTuple): + """One key press: a name ("up", "enter"), or "char" or "paste" and the text.""" + + name: str + text: str = "" + + +_CSI_KEYS = {"A": "up", "B": "down", "C": "right", "D": "left", "H": "home", "F": "end", + "Z": "shift-tab"} +_TILDE_KEYS = {"1": "home", "7": "home", "4": "end", "8": "end", "5": "pgup", "6": "pgdn", + "3": "delete", "11": "f1"} +_SS3_KEYS = {"A": "up", "B": "down", "C": "right", "D": "left", "H": "home", "F": "end", + "P": "f1", "M": "enter"} +# msvcrt.getwch gives these after a \x00 or \xe0 prefix. +_WINDOWS_KEYS = {"H": "up", "P": "down", "K": "left", "M": "right", "I": "pgup", + "Q": "pgdn", "G": "home", "O": "end", "S": "delete", ";": "f1"} + + +def _plain_key(char: str) -> TuiKey | None: + """The key one character stands for, None for a control nobody bound.""" + named = {"\r": "enter", "\n": "enter", "\x7f": "backspace", "\x08": "backspace", + "\t": "tab", "\x1b": "esc", "\x03": "ctrl-c", "\x04": "ctrl-d", + "\x15": "ctrl-u"}.get(char) + if named: + return TuiKey(named) + if char < " " or unicodedata.category(char) in ("Cc", "Cs"): + return None + return TuiKey("char", char) + + +def _csi_key(params: str, final: str) -> TuiKey | None: + if final == "~": + if params == "200": + return TuiKey("paste-start") + name = _TILDE_KEYS.get(params.split(";")[0]) + return TuiKey(name) if name else None + name = _CSI_KEYS.get(final) + if name is None: + return None + if params in ("", "1"): + return TuiKey(name) + if params == "1;2" and name in ("up", "down"): + return TuiKey(f"shift-{name}") + return None + + +def _escape(buffer: str) -> tuple[int, TuiKey | None]: + """The sequence at the start of buffer: the characters it takes and its + key (None for one nobody bound), or 0 while more are needed to tell.""" + if len(buffer) < 2: + return 0, None + if buffer[1] == "[": + if buffer[2:3] == "M": + # An X10 mouse report: three bytes of position follow. + return (6, None) if len(buffer) >= 6 else (0, None) + for end in range(2, len(buffer)): + code = ord(buffer[end]) + if 0x40 <= code <= 0x7E: + return end + 1, _csi_key(buffer[2:end], buffer[end]) + if not 0x20 <= code <= 0x3F: + return 1, TuiKey("esc") + return 0, None + if buffer[1] == "O": + if len(buffer) < 3: + return 0, None + name = _SS3_KEYS.get(buffer[2]) + return 3, TuiKey(name) if name else None + return 1, TuiKey("esc") + + +class KeyDecoder: + """Keys from what a terminal sends, however the reads split it. + + Unknown sequences (F13, mouse reports) are dropped, never taken as text. + """ + + def __init__(self) -> None: + self.buffer = "" + self.paste: str | None = None + + def feed(self, text: str) -> list[TuiKey]: + self.buffer += text + return self._drain(final=False) + + def flush(self) -> list[TuiKey]: + """What is left once the wait for the rest of a sequence is over. + + An open paste stays open: over SSH its end arrives later than 40 ms. + """ + return self._drain(final=True) + + def close_paste(self) -> list[TuiKey]: + """The paste so far, its end never seen, without its newlines.""" + text = "".join(char for char in self.paste or "" if char >= " " and char != "\x7f") + self.paste = None + return [TuiKey("paste", text)] if text else [] + + @property + def waiting(self) -> bool: + return bool(self.buffer) or self.paste is not None + + def _drain(self, final: bool) -> list[TuiKey]: + keys: list[TuiKey] = [] + while self.buffer: + if self.paste is not None: + end = self.buffer.find(_PASTE_END) + if end < 0: + # The end marker may be cut between two reads. + keep = next((n for n in range(len(_PASTE_END) - 1, 0, -1) + if self.buffer.endswith(_PASTE_END[:n])), 0) + cut = len(self.buffer) - keep + self.paste += self.buffer[:cut] + self.buffer = self.buffer[cut:] + break + self.paste += self.buffer[:end] + self.buffer = self.buffer[end + len(_PASTE_END):] + keys += self.close_paste() + continue + if self.buffer[0] == "\x1b": + taken, key = _escape(self.buffer) + if not taken: + if final: + if self.buffer == "\x1b": + keys.append(TuiKey("esc")) + self.buffer = "" + break + self.buffer = self.buffer[taken:] + if key is not None and key.name == "paste-start": + self.paste = "" + elif key is not None: + keys.append(key) + continue + key = _plain_key(self.buffer[0]) + self.buffer = self.buffer[1:] + if key is not None: + keys.append(key) + return keys + + +def decode_windows_key(first: str, read_next: Callable[[], str]) -> TuiKey | None: + """A key from msvcrt.getwch: arrows and pages come as a prefix, then a code.""" + if first in ("\x00", "\xe0"): + name = _WINDOWS_KEYS.get(read_next()) + return TuiKey(name) if name else None + return _plain_key(first) + + +def _char_width(char: str) -> int: + if unicodedata.combining(char) or unicodedata.category(char) in ("Mn", "Me", "Cf"): + return 0 + return 2 if unicodedata.east_asian_width(char) in ("W", "F") else 1 + + +def cell_width(text: str) -> int: + """Terminal cells the text takes: a CJK character two, an accent none.""" + return sum(_char_width(char) for char in text) + + +# C0 and C1 controls, DEL and lone surrogates: a file, folder or core name +# holding ESC would drive the terminal (clear it, or set the clipboard with +# OSC 52) instead of being shown. +_CONTROLS = re.compile("[\x00-\x1f\x7f-\x9f\ud800-\udfff]") + + +def screen_text(text: str) -> str: + """Text as a terminal shows it, every control character a "?".""" + return _CONTROLS.sub("?", text) + + +def ascii_text(text: str) -> str: + """The text in ASCII, accents dropped (Edition) and any other character + as one ? per cell it took, so the columns stay where they were.""" + if text.isascii(): + return text + plain = "".join(char for char in unicodedata.normalize("NFKD", text) + if not unicodedata.combining(char)) + return "".join(char if char.isascii() else "?" * _char_width(char) for char in plain) + + +def fit(text: str, width: int) -> str: + """The text cut to width cells, never through a character.""" + text = screen_text(text) + kept: list[str] = [] + used = 0 + for char in text: + size = _char_width(char) + if used + size > width: + break + kept.append(char) + used += size + return "".join(kept) + + +def middle(text: str, width: int) -> str: + """A long path cut in its middle, its start and its last folders kept.""" + if cell_width(text) <= width: + return text + if width <= 3: + return fit(text, width) + room = width - 3 + tail = fit(text[::-1], room // 2)[::-1] + return fit(text, room - cell_width(tail)) + "..." + tail + + +def _pad(text: str, width: int) -> str: + return text + " " * max(0, width - cell_width(text)) + + +def _columns(left: str, right: str, width: int) -> str: + """left, and right against the edge when both fit, else left alone.""" + room = width - cell_width(right) - 2 + if not right or room < 8: + return fit(left, width) + return _pad(fit(left, room), room) + " " + right + + +def _split_word(word: str, room: int) -> list[str]: + """A word longer than room, cut after a slash where it can be: a path + read /home/us and er/bios when cut anywhere.""" + pieces = [] + while cell_width(word) > room: + head = fit(word, max(room, 1)) + cut = head.rfind("/") + 1 + if cut <= len(head) // 3: + cut = len(head) + pieces.append(word[:cut]) + word = word[cut:] + return pieces + [word] if word else pieces + + +def _wrap(text: str, width: int, indent: str = " ") -> list[str]: + """Lines of at most width cells, the ones after the first indented.""" + if cell_width(text) <= width: + return [text] + lead = text[:len(text) - len(text.lstrip(" "))] + lines: list[str] = [] + line = "" + for word in text.split(): + for piece in _split_word(word, width - len(lead + indent)): + prefix = lead + indent if lines else lead + longer = f"{line} {piece}" if line else piece + if not line or cell_width(prefix + longer) <= width: + line = longer + else: + lines.append(prefix + line) + line = piece + lines.append((lead + indent if lines else lead) + line) + return lines + + +class TuiLine(str): + """A screen line and how it is drawn: "title", "focus", "dim" or plain.""" + + style = "" + + +def _styled(text: str, style: str = "") -> TuiLine: + line = TuiLine(screen_text(text)) + line.style = style + return line + + +def _short_path(path: Path | None, home: Path) -> str: + if path is None: + return "" + text = str(path) + home_text = str(home) + if home_text not in ("", "/", ".") and (text == home_text + or text.startswith(home_text + os.sep)): + return "~" + text[len(home_text):] + return text + + +def _folded(text: str) -> str: + """Text compared without case or accents, as a filter reads it.""" + decomposed = unicodedata.normalize("NFKD", text) + return "".join(char for char in decomposed if not unicodedata.combining(char)).casefold() + + +class TuiState: + """Everything the screens show. Nothing here touches the terminal.""" + + def __init__(self, os_type: str = "linux", home: Path | None = None) -> None: + self.os_type = os_type + self.home = home or Path("/home/user") + self.stack: list = ["setup"] + self.focus: dict = {} + self.filters: dict = {} + self.editing: str = "" + self.help: bool = False + self.status: str = "" + self.cols: int = 80 + self.rows: int = 24 + self.now: float = 0.0 + # (platform, folder, label) for each setup found here. + self.setups: list = [] + self.notes: list = [] + self.platform: str = "" + self.dest: Path | None = None + # A --dest given without --platform: the list asks whose files go there. + self.dest_given: Path | None = None + self.folder_default: Path | None = None + self.path_text: str = "" + self.browse_dir: Path | None = None + self.browse_rows: list = [] + self.manifest: dict | None = None + self.manifest_for: str = "" + self.manifest_path: Path | None = None + self.manifest_given: bool = False + self.error: list = [] + self.error_title: str = "" + self.error_code: int = 1 + # Typed over in one key: a field opening on a long path took one + # Backspace per character before anything could be typed. + self.path_selected: bool = False + # The ticks kept across a check run again for new options. + self.keep_ticked: set | None = None + # (files read, files to read) while the check hashes what is in place. + self.reading: tuple | None = None + # The child run in progress: check, confirm, preview or install. + self.run: str = "" + self.running: bool = False + self.run_id: int = 0 + self.started: float = 0.0 + self.checked_flags: list = [] + self.check: dict | None = None + self.files: list = [] + self.systems: list = [] + self.facts: dict = {} + self.dry: dict | None = None + self.dry_files: list = [] + self.dry_summary: dict | None = None + # Choices, each one an option of the command line. + self.ticked: set = set() + self.cli_systems: list = [] + self.regions: list = [] + self.cli_regions: list = [] + self.cores: list = [] + self.no_game_data: bool = False + self.installed_only: bool = False + self.jobs: int = 8 + self.passthrough: list = [] + # The install. + self.download: dict | None = None + self.progress: dict | None = None + # (time, bytes done) of the last progress events: a partial copy + # resumed counts in the first one and never in a speed. + self.samples: list = [] + self.finished: list = [] + self.summary: dict | None = None + self.code: int | None = None + self.stopping: bool = False + self.tail: list = [] + self.log: str = "" + self.exit_code: int | None = None + + @property + def screen(self) -> str: + return self.stack[-1] + + +def _env_on(env: Mapping[str, str], name: str) -> bool: + return env.get(name, "") not in ("", "0") + + +def plain_requested(args: argparse.Namespace, env: Mapping[str, str] | None = None) -> bool: + """Whether the line questions were asked for: --plain, RETROBIOS_PLAIN, ACCESSIBLE.""" + env = os.environ if env is None else env + return bool(args.plain) or _env_on(env, "RETROBIOS_PLAIN") or _env_on(env, "ACCESSIBLE") + + +def _terminal_size(stream: TextIO) -> tuple[int, int]: + try: + size = os.get_terminal_size(stream.fileno()) + except (OSError, ValueError, AttributeError, io.UnsupportedOperation): + return 0, 0 + return size.columns, size.lines + + +def tui_blocker( + args: argparse.Namespace, stdin: TextIO | None = None, stdout: TextIO | None = None, + env: Mapping[str, str] | None = None, +) -> str: + """Why the full-screen interface cannot run here, "" when it can.""" + env = os.environ if env is None else env + stdin = stdin or sys.stdin + stdout = stdout or sys.stdout + for flag, given in ( + ("--check", args.check), ("--dry-run", args.dry_run), + ("--list-platforms", args.list_platforms), ("--list-targets", args.list_targets), + ("--list-systems", args.list_systems), ("--list-cores", args.list_cores), + ("--list-emulators", args.list_emulators), ("--events", args.events), + ("--quiet", args.quiet), ("--no-input", args.no_input), + ("--restore", args.restore is not None), ("--yes", args.yes), + ): + if given: + return f"{flag} runs without questions" + for name, stream in (("input", stdin), ("output", stdout)): + isatty = getattr(stream, "isatty", None) + if not (isatty and isatty()): + return f"the {name} is not a terminal" + if os.name != "nt" and env.get("TERM", "") in ("", "dumb"): + return "TERM is dumb or not set" + # A pseudo-terminal nobody sized reports 0x0: script(1) fed from a pipe + # is one, and the line questions run there as they always have. + cols, rows = _terminal_size(stdout) + if cols < TUI_MIN_COLS or rows < TUI_MIN_ROWS: + return (f"the terminal is {cols}x{rows}, smaller than " + f"{TUI_MIN_COLS}x{TUI_MIN_ROWS}") + if os.name == "nt": + try: + import msvcrt # noqa: F401, PLC0415 + except ImportError: + return "this Python has no msvcrt to read keys" + if not _enable_ansi(): + return "this console does not draw escape sequences (Windows 10 or later does)" + return "" + try: + import termios # noqa: PLC0415 + except ImportError: + return "this Python has no termios to read keys" + try: + termios.tcgetattr(stdin.fileno()) + except (termios.error, OSError, ValueError, AttributeError, + io.UnsupportedOperation) as exc: + return f"single key presses cannot be read here ({exc})" + return "" + + +def tui_flags(state: TuiState) -> list[tuple[str, str | None]]: + """The options that carry the choices made on screen, CLI ones included.""" + flags = list(state.passthrough) + needed = {row["system"] for row in state.systems + if row["missing"] + row["wrong_hash"] > 0} + if state.systems and not needed <= state.ticked: + flags += [("--system", name) for name in sorted(state.ticked)] + else: + flags += [("--system", name) for name in state.cli_systems] + flags += [("--core", name) for name in state.cores] + if state.regions: + flags.append(("--region", ",".join(state.regions))) + if state.no_game_data: + flags.append(("--no-game-data", None)) + if state.installed_only and not any(flag == "--cores-dir" for flag, _v in flags): + flags.append(("--installed-cores", None)) + if state.jobs != 8: + flags.append(("--jobs", str(state.jobs))) + return flags + + +_TUI_MODES = { + "check": ["--check", "--events"], + "confirm": ["--dry-run", "--events"], + "preview": ["--dry-run", "--events"], + "install": ["--events", "--yes"], +} + + +def tui_argv(state: TuiState, run: str, python: str, script: str) -> list[str]: + """The child that does what the screen decided.""" + # --dest=VALUE: a folder typed as -x would otherwise be read as an option. + argv = [python, script, f"--platform={state.platform}", f"--dest={state.dest}"] + if state.manifest_path is not None: + argv.append(f"--manifest={state.manifest_path}") + argv += [flag if value is None else f"{flag}={value}" for flag, value in tui_flags(state)] + return argv + _TUI_MODES[run] + + +def tui_command(state: TuiState) -> str: + """The same choice as one command, in the form the run was started with.""" + hint = Invocation() + hint.os_type = state.os_type + hint.options = ([("--manifest", str(state.manifest_path))] + if state.manifest_given and state.manifest_path else []) + hint.options += tui_flags(state) + return hint.command(state.platform, state.dest) + +class TuiRow(NamedTuple): + """One row a screen lists: what Enter does on it, and what it shows.""" + + action: str + label: str + right: str = "" + value: object = None + style: str = "" + + +# Screens that scroll text instead of listing rows. +_PAGERS = ("preview", "command", "result") +_CHECKLISTS = ("systems", "cores") + + +def _needed(row: dict) -> int: + return row["missing"] + row["wrong_hash"] + + +def _tui_system_label(row: dict, titles: dict | None = None) -> str: + """The name a person reads: the file list's title when it has one, the + platform's own name when it is one (Sony - PlayStation), the id with the + platform's short name otherwise (psx).""" + title = (titles or {}).get(row["system"]) + if isinstance(title, str) and title.strip(): + return title.strip() + name = row.get("name") or row["system"] + if name == row["system"]: + return name + if " " in name or any(char.isupper() for char in name): + return name + return f"{row['system']} ({name})" + + +def _matches(text: str, needle: str) -> bool: + return not needle or _folded(needle) in _folded(text) + + +def _filter_row(state: TuiState) -> TuiRow: + text = state.filters.get(state.screen, "") + if state.editing == state.screen: + return TuiRow("filter", f"Filter: {text}_") + if text: + return TuiRow("filter", f"Filter: {text}") + return TuiRow("filter", "Filter (/ to type)", style="dim") + + +def _system_titles(state: TuiState) -> dict: + titles = (state.manifest or {}).get("system_titles") + return titles if isinstance(titles, dict) else {} + + +def _sorted_systems(state: TuiState) -> list[dict]: + """Systems with something to download first, the ones in place last, each + group in the order of the names a person reads.""" + titles = _system_titles(state) + return sorted(state.systems, key=lambda row: ( + _needed(row) == 0, _tui_system_label(row, titles).casefold())) + + +# Region names as a person writes them, where title case alone is wrong. +_REGION_TITLES = {"asia-ntsc": "Asia (NTSC)", "asia-pal": "Asia (PAL)", + "uk": "United Kingdom", WORLD_REGION: "World"} + + +def region_title(name: str) -> str: + """north-america as North America, asia-ntsc as Asia (NTSC).""" + return _REGION_TITLES.get(name) or " ".join( + word.capitalize() for word in name.split("-")) + + +def tui_choices(state: TuiState) -> dict[str, list[str]]: + """What the file list lets one narrow to, read from the list itself.""" + manifest = state.manifest or {} + return available_choices(manifest.get("files", []), manifest.get("omitted_files", [])) + + +def tui_rows(state: TuiState) -> list[TuiRow]: + """The rows of the current screen, in the order they are shown.""" + screen = state.screen + if screen == "setup": + # The focused one's folder is spelled out below the list. + rows = [TuiRow("setup", label, value=index) + for index, (_plat, _path, label) in enumerate(state.setups)] + return rows + [TuiRow("another", "Another setup or folder")] + if screen == "platforms": + needle = state.filters.get(screen, "") + systems = _systems_reading() + return [_filter_row(state)] + [ + TuiRow("platform", display_name(plat), ", ".join(systems.get(plat, [])), plat) + for plat in AVAILABLE_PLATFORMS + if _matches(" ".join([plat, display_name(plat), *systems.get(plat, [])]), needle) + ] + if screen == "folder": + rows = [TuiRow("default", "Use this folder")] if state.folder_default else [] + return rows + [TuiRow("browse", "Browse for a folder"), TuiRow("type", "Type a path")] + if screen == "browse": + return list(state.browse_rows) + if screen == "create": + label = "Create the whole path" if state.facts.get("risky") else "Create it and continue" + return [TuiRow("create", label), TuiRow("other", "Choose another folder")] + if screen == "cardroot": + return [TuiRow("cardroot", "Use the card root"), TuiRow("other", "Choose another folder")] + if screen == "plan": + return _plan_rows(state) + if screen == "systems": + return _system_rows(state) + if screen == "options": + return _option_rows(state) + if screen == "regions": + ranked = [TuiRow("region", f"[{rank}] {region_title(name)}", value=name) + for rank, name in enumerate(state.regions, 1)] + known = sorted(set(tui_choices(state)["regions"]) | set(state.cli_regions), + key=region_title) + return ranked + [TuiRow("region", f"[ ] {region_title(name)}", value=name) + for name in known if name not in state.regions] + if screen == "cores": + needle = state.filters.get(screen, "") + return [_filter_row(state)] + [ + TuiRow("core", f"[{'x' if name in state.cores else ' '}] {name}", value=name) + for name in tui_choices(state)["cores"] if _matches(name, needle) + ] + if screen == "confirm": + if state.running or state.dry is None: + return [] + return [TuiRow("install", "Install"), TuiRow("back", "Back")] + if screen == "error": + rows = ([TuiRow("retry", "Try again")] + if state.run == "check" and state.error_title else []) + return rows + [TuiRow("back", "Choose another setup or folder" if state.run == "check" + else "Back"), + TuiRow("quit", "Quit")] + if screen == "quit": + return [TuiRow("quit", "Quit"), TuiRow("stay", "Stay")] + return [] + + +def _plan_rows(state: TuiState) -> list[TuiRow]: + check = state.check or {} + if not (_needed(check) if check else 0) and not check.get("data_missing"): + return [TuiRow("close", "Close"), TuiRow("options", "Options")] + size = check.get("bytes_needed", 0) + label = f"Install {format_size(size)}" if size >= CONFIRM_FROM_BYTES else "Install" + rows = [TuiRow("install", label)] + if state.facts.get("writable") is False: + # Every file would fail alike: the way out is another folder. + rows = [TuiRow("folder", "Choose another folder"), TuiRow("install", "Install anyway")] + if records_selection((state.manifest or {}).get("files", [])) and len(state.systems) > 1: + rows.append(TuiRow("systems", "Choose systems")) + return rows + [TuiRow("options", "Options"), TuiRow("quit", "Quit")] + + +def _system_rows(state: TuiState) -> list[TuiRow]: + needle = state.filters.get("systems", "") + titles = _system_titles(state) + rows = [TuiRow("all", "All"), TuiRow("none", "None"), _filter_row(state)] + for row in _sorted_systems(state): + label = _tui_system_label(row, titles) + if not _matches(f"{label} {row['system']}", needle): + continue + box = "[x]" if row["system"] in state.ticked else "[ ]" + need = _needed(row) + right = (f"{_plural(need, 'file')}, {format_size(row['bytes_needed'])}" + if need else "in place") + rows.append(TuiRow("system", f"{box} {label}", right, row["system"], + "" if need else "dim")) + return rows + + +def _cores_dir_given(state: TuiState) -> bool: + return any(flag == "--cores-dir" for flag, _value in state.passthrough) + + +def _yes(value: bool) -> str: + return "yes" if value else "no" + + +def _option_rows(state: TuiState) -> list[TuiRow]: + manifest = state.manifest or {} + choices = tui_choices(state) + rows = [] + if choices["regions"]: + rows.append(TuiRow("regions", "Preferred regions: " + + (", ".join(map(region_title, state.regions)) or "any"), + "--region")) + if manifest.get("categorized"): + rows.append(TuiRow("game_data", f"Leave out game data: {_yes(state.no_game_data)}", + "--no-game-data")) + found = len(state.facts.get("cores", [])) + if _cores_dir_given(state): + rows.append(TuiRow("installed", "Installed cores only: yes, implied by --cores-dir", + "--cores-dir")) + elif state.platform in CORE_PLATFORMS and "system_cores" in manifest and found: + rows.append(TuiRow("installed", f"Installed cores only: " + f"{_yes(state.installed_only)} ({found} found)", + "--installed-cores")) + if choices["cores"]: + chosen = f"{len(state.cores)} chosen" if state.cores else "all" + rows.append(TuiRow("cores", f"Cores: {chosen}", "--core")) + rows += [ + TuiRow("jobs", f"Parallel downloads: {state.jobs}", "--jobs"), + TuiRow("folder", f"BIOS folder: {_short_path(state.dest, state.home)}", "--dest", + style="path"), + TuiRow("preview", "Preview the file list", "--dry-run"), + TuiRow("command", "Show the command line"), + ] + return rows + + +def _push(state: TuiState, screen: str, focus: int = 0) -> list[tuple]: + state.stack.append(screen) + state.focus[screen] = focus + return [] + + +def _replace(state: TuiState, screen: str) -> None: + state.stack[-1] = screen + state.focus[screen] = 0 + + +def _back(state: TuiState) -> list[tuple]: + effects = [("kill",)] if state.running else [] + if state.running: + # What the stopped run still reports belongs to no screen. + state.running = False + state.run_id += 1 + if len(state.stack) == 1: + state.stack.append("quit") + state.focus["quit"] = 0 + return effects + state.stack.pop() + if state.screen == "checking" and not state.running: + # Stay after Esc on a check started from the command line: a + # Checking screen nothing feeds counted seconds forever. + return effects + _start_check(state) + if state.screen == "plan" and _counts_predate_options(state): + # The plan's numbers were read under the options before: they are + # read again, the systems ticked kept. + needed = {row["system"] for row in state.systems if _needed(row)} + state.keep_ticked = None if needed <= state.ticked else set(state.ticked) + return effects + _start_check(state) + return effects + + +def tui_quit(state: TuiState, code: int = 130) -> list[tuple]: + """Leave now: with the install's own code once it ran, else 130.""" + if state.screen == "result": + code = state.code if state.code is not None else 1 + elif state.screen == "error": + code = state.error_code + state.exit_code = code + return [("kill",)] if state.running else [] + + +def tui_interrupt(state: TuiState) -> list[tuple]: + """Ctrl-C: an install stops and says what it did, anything else ends.""" + if state.run == "install" and state.running: + if state.stopping: + return [("kill",)] + state.stopping = True + return [("stop",)] + return tui_quit(state) + + +def _start_run(state: TuiState, run: str) -> list[tuple]: + state.run = run + state.running = True + state.run_id += 1 + state.started = state.now + if run == "check": + # The systems of an earlier check go first: ticks read against them + # dropped the --system the command line gave. + state.check, state.files, state.systems, state.reading = None, [], [], None + state.checked_flags = tui_flags(state) + elif run in ("confirm", "preview"): + state.dry, state.dry_files, state.dry_summary = None, [], None + else: + state.download = state.progress = state.reading = None + state.samples = [] + state.finished, state.summary, state.code, state.stopping = [], None, None, False + return [("run", run)] + + +def _start_check(state: TuiState) -> list[tuple]: + """Check the chosen folder, after fetching the file list once.""" + if "plan" in state.stack: + del state.stack[state.stack.index("plan"):] + while state.stack and state.stack[-1] in ("checking", "error", "create"): + state.stack.pop() + state.stack.append("checking") + state.facts, state.ticked = {}, set() + if state.manifest is not None and state.manifest_for == state.platform: + return _start_run(state, "check") + state.manifest = None + state.run, state.run_id = "check", state.run_id + 1 + state.started = state.now + return [("fetch", state.platform)] + + +def _choose(state: TuiState, plat: str, dest: Path) -> list[tuple]: + state.platform, state.dest = plat, dest + if plat in CARD_PLATFORMS and card_platform(dest) is None: + # The engine refuses a folder inside the card, the list writing + # Bios// below the root: the root is offered instead. + for parent in list(dest.absolute().parents)[:2]: + holder = card_platform(parent) + if holder: + state.facts = {"card_root": parent, "card": holder} + return _push(state, "cardroot") + if not dest.exists(): + nearest = _nearest_folder(dest) + missing = len(dest.absolute().parts) - len(nearest.parts) + state.facts = {"nearest": nearest, "risky": nearest.parent == nearest or missing > 2} + # A path whose start was mistyped would be created from the root: + # the safe answer is the one under A. + return _push(state, "create", 1 if state.facts["risky"] else 0) + return _start_check(state) + + +def _open_folder_screen(state: TuiState, default: Path | None) -> list[tuple]: + state.folder_default = default + return _push(state, "folder") + + +def _mounted_cards() -> list[Path]: + """Mount points of removable media, where a card or a share appears. + + A handheld's card is most often prepared on a PC with a card reader: on + Windows every drive but the system one may be that card. + """ + if os.name == "nt": + system = os.environ.get("SystemDrive", "C:").upper() + return [Path(f"{letter}:\\") for letter in "DEFGHIJKLMNOPQRSTUVWXYZ" + if f"{letter}:" != system and os.path.exists(f"{letter}:\\")] + found: list[Path] = [] + for root in (Path("/media"), Path("/run/media"), Path("/mnt"), Path("/Volumes")): + try: + children = sorted(root.iterdir()) + except OSError: + continue + for child in children: + # WSL's own mounts, and the Windows system drive, hold no card. + if child.name in ("wsl", "wslg") or (root.name == "mnt" and child.name == "c"): + continue + if os.path.ismount(child): + found.append(child) + elif root.name in ("media", "run") or root == Path("/run/media"): + # /media//