From 69f9dbd9b107aa71e337f49882ebaa04ac5c699f Mon Sep 17 00:00:00 2001 From: pleb Date: Wed, 1 Jul 2026 11:44:03 -0700 Subject: [PATCH] Add profile-aware plugin TUI --- .gitignore | 2 + README.md | 46 +++-- plugin-helper.toml.example | 11 ++ pyproject.toml | 4 +- src/plugin_helper/cli.py | 295 +++++++++++--------------------- src/plugin_helper/config.py | 113 +++++++++++- src/plugin_helper/operations.py | 47 +++++ src/plugin_helper/reports.py | 78 +++++++++ src/plugin_helper/tui.py | 275 +++++++++++++++++++++++++++++ tests/test_plugin_helper.py | 270 ++++++++++++++++++++++++----- 10 files changed, 876 insertions(+), 265 deletions(-) create mode 100644 plugin-helper.toml.example create mode 100644 src/plugin_helper/operations.py create mode 100644 src/plugin_helper/reports.py create mode 100644 src/plugin_helper/tui.py diff --git a/.gitignore b/.gitignore index 1c26636..bf74cf6 100644 --- a/.gitignore +++ b/.gitignore @@ -1,5 +1,6 @@ /.state/ /.state-*/ +/plugin-helper.local.toml /.pytest_cache/ /build/ /dist/ @@ -7,3 +8,4 @@ /src/*.egg-info/ /__pycache__/ *.pyc +.venv diff --git a/README.md b/README.md index 2580b58..aab56ce 100644 --- a/README.md +++ b/README.md @@ -26,10 +26,17 @@ multiple explicit roots, separate them with `:`. The helper is intended to manage both the local Linux BSManager install and the mounted Windows install. Lockfiles and registry entries are shared by Beat Saber version, but install state is target-specific. When the same instance name -exists under both roots, such as `1.44.1`, use an explicit `--instances-root` -and a separate state directory for each target. +exists under both roots, such as `1.44.1`, give each install profile its own +state directory. -Suggested repo-local convention: +Copy the example profile config and adjust paths if needed: + +```sh +cp plugin-helper.toml.example plugin-helper.local.toml +``` + +`plugin-helper.local.toml` is ignored by git. The default example uses this +repo-local convention: ```text .state/ local Linux BSManager state @@ -40,35 +47,38 @@ Examples: ```sh PYTHONPATH=src python -m plugin_helper \ - --instances-root /home/pleb/.local/share/BSManager/BSInstances \ - --state-dir .state \ + --profile linux \ installed --instance 1.44.1 PYTHONPATH=src python -m plugin_helper \ - --instances-root /home/pleb/Windows/Users/pleb/BSManager/BSInstances \ - --state-dir .state-windows \ + --profile windows \ installed --instance 1.44.1 ``` -Do not reuse the same state directory for both targets when their instance names -match. The current state layout is keyed by instance name, so sharing one state -directory would mix bootstrap records, generated plans, backups, and installed -file records for different game trees. +Explicit `--instances-root` and `--state-dir` still work and override profile +values. Do not reuse the same state directory for both targets when their +instance names match. The current state layout is keyed by instance name, so +sharing one state directory would mix bootstrap records, generated plans, +backups, and installed file records for different game trees. ## Commands -For normal use, run the menu from the repo root. Use repo-local state so the -menu sees the same plans, downloads, and install records used by the helper -workflow: +For normal use, run the Textual menu from the repo root: ```sh -PYTHONPATH=src python -m plugin_helper --state-dir .state menu +PYTHONPATH=src python -m plugin_helper menu ``` +The menu reads `plugin-helper.local.toml` when present, shows each discovered +Beat Saber install with its resolved state directory, and lets you toggle +managed plugins with arrow keys and Space. In the plugin table, use `d` to +disable all currently enabled managed plugins and `e` to enable all currently +disabled managed plugins. + The individual subcommands are mostly for automation and debugging. If you use -them, pass `--state-dir .state` unless you intentionally want the default live -state outside this repo or are intentionally targeting the Windows install with -`.state-windows`. +them, prefer `--profile linux` or `--profile windows`. Pass `--state-dir` +directly only when you intentionally want to override profile state or use the +default live state outside this repo. Install assets are currently expected to already exist locally, usually under: diff --git a/plugin-helper.toml.example b/plugin-helper.toml.example new file mode 100644 index 0000000..c150215 --- /dev/null +++ b/plugin-helper.toml.example @@ -0,0 +1,11 @@ +[[profiles]] +id = "linux" +label = "Local Linux BSManager" +instances_root = "~/.local/share/BSManager/BSInstances" +state_dir = ".state" + +[[profiles]] +id = "windows" +label = "Mounted Windows BSManager" +instances_root = "~/Windows/Users/pleb/BSManager/BSInstances" +state_dir = ".state-windows" diff --git a/pyproject.toml b/pyproject.toml index 93708df..f0f8a46 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -10,7 +10,9 @@ readme = "README.md" requires-python = ">=3.11" license = "MIT" authors = [{ name = "plugin-helper contributors" }] -dependencies = [] +dependencies = [ + "textual>=8.2,<9", +] [project.scripts] plugin-helper = "plugin_helper.cli:main" diff --git a/src/plugin_helper/cli.py b/src/plugin_helper/cli.py index ea8bd3a..fedbfde 100644 --- a/src/plugin_helper/cli.py +++ b/src/plugin_helper/cli.py @@ -4,9 +4,9 @@ import argparse import json import sys from pathlib import Path -from typing import Any, Callable +from typing import Any -from .config import instances_roots, repo_root, state_root +from .config import Profile, repo_root, resolve_runtime_config from .bootstrap import run_bootstrap from .bsipa import check_bsipa_health from .checker import check_lock @@ -14,8 +14,9 @@ from .github import fetch_releases from .installer import apply_plan, disable_plugin, uninstall_plugin from .instances import get_instance, list_instances from .models import load_lockfile, load_registry -from .models import Lockfile, Registry +from .operations import enable_disabled_plugin from .planner import create_plan +from .reports import installed_plugins_report, print_installed_plugins from .scanner import scan_instance from .state import load_installed_state from .updates import check_updates @@ -26,79 +27,6 @@ def _json(data: Any) -> None: print(json.dumps(data, indent=2, sort_keys=True)) -def installed_plugins_report( - *, - installed_state: dict[str, Any], - registry: Registry, - lockfile: Lockfile, -) -> dict[str, Any]: - locked_by_id = {plugin.id: plugin for plugin in lockfile.plugins} - plugins: list[dict[str, Any]] = [] - state_plugins = [ - (plugin_id, plugin_state, "enabled") - for plugin_id, plugin_state in installed_state.get("plugins", {}).items() - ] - state_plugins.extend( - (plugin_id, plugin_state, "disabled") - for plugin_id, plugin_state in installed_state.get("disabledPlugins", {}).items() - ) - for plugin_id, plugin_state, status in sorted(state_plugins): - registry_plugin = registry.get(plugin_id) - locked = locked_by_id.get(plugin_id) - files = plugin_state.get("files", []) - plugins.append( - { - "id": plugin_id, - "name": registry_plugin.name if registry_plugin else plugin_id, - "version": locked.tag if locked and locked.tag else "(not locked)", - "asset": locked.asset if locked and locked.asset else "(unknown)", - "repo": (locked.repo if locked and locked.repo else None) - or (registry_plugin.repo if registry_plugin else None) - or "(unknown)", - "installedAt": plugin_state.get("installedAt", "(unknown)"), - "disabledAt": plugin_state.get("disabledAt"), - "status": status, - "fileCount": len(files), - "files": files, - } - ) - return { - "instance": installed_state.get("instance", lockfile.instance), - "beatSaberVersion": installed_state.get("beatSaberVersion", lockfile.beat_saber_version), - "plugins": plugins, - } - - -def print_installed_plugins(report: dict[str, Any]) -> None: - plugins = report["plugins"] - print(f"{report['instance']} managed plugins ({len(plugins)})") - if not plugins: - print("No plugins have been installed by plugin-helper yet.") - return - - headers = ("Plugin", "Status", "Version", "Asset", "Files", "Installed") - rows = [ - ( - f"{plugin['name']} ({plugin['id']})", - plugin["status"], - plugin["version"], - plugin["asset"], - str(plugin["fileCount"]), - plugin["disabledAt"] or plugin["installedAt"], - ) - for plugin in plugins - ] - widths = [ - max(len(headers[index]), *(len(row[index]) for row in rows)) - for index in range(len(headers)) - ] - header = " ".join(label.ljust(widths[index]) for index, label in enumerate(headers)) - print(header) - print(" ".join("-" * width for width in widths)) - for row in rows: - print(" ".join(value.ljust(widths[index]) for index, value in enumerate(row))) - - def print_updates(report: dict[str, Any]) -> None: plugins = report["plugins"] summary = report["summary"] @@ -135,6 +63,8 @@ def _add_common(parser: argparse.ArgumentParser, *, suppress_default: bool = Fal default = argparse.SUPPRESS if suppress_default else None parser.add_argument("--instances-root", default=default, help="BSManager instances root") parser.add_argument("--state-dir", default=default, help="plugin-helper state directory") + parser.add_argument("--profile", default=default, help="Configured profile id from plugin-helper.local.toml") + parser.add_argument("--config", default=default, help="Path to plugin-helper TOML config") def build_parser() -> argparse.ArgumentParser: @@ -295,120 +225,101 @@ def _common_parent() -> argparse.ArgumentParser: return parent -def _ask_choice( +def _menu_profiles( *, - title: str, - choices: list[tuple[str, str] | tuple[str, str, str]], - input_func: Callable[[str], str] | None = None, -) -> str | None: - ask = input_func or input - print() - print(title) - for index, choice in enumerate(choices, start=1): - label = choice[1] - print(f" {index}. {label}") - if len(choice) > 2: - print(f" {choice[2]}") - print(" q. Quit") - - while True: - answer = ask("> ").strip().lower() - if answer in {"q", "quit", "exit"}: - return None - if answer.isdigit(): - index = int(answer) - if 1 <= index <= len(choices): - return choices[index - 1][0] - print("Choose a listed number, or q to quit.") - - -def _run_menu(inst_roots: list[Path], st_root: Path, input_func: Callable[[str], str] | None = None) -> int: - ask = input_func or input - instances = list_instances(inst_roots) - if not instances: - print(f"No Beat Saber instances found under {', '.join(str(root) for root in inst_roots)}") - return 1 - - instances_by_choice = {str(index): item for index, item in enumerate(instances, start=1)} - instance_choices = [(str(index), f"{item.name} {item.path}") for index, item in enumerate(instances, start=1)] - action_choices = [ - ("installed", "Show managed installs", "Lists plugins recorded in plugin-helper state with locked versions and files."), - ("updates", "Check locked plugin updates", "Looks at GitHub releases for newer assets matching locked plugins."), - ("scan", "Scan installed files", "Counts files currently present in Plugins/, Libs/, and IPA/Pending/."), - ("check", "Check lockfile and assets", "Validates registry entries, lockfile data, local assets, and SHA-256 values."), - ("bootstrap", "Bootstrap BSIPA", "Fetches the locked BSIPA archive, installs it, runs IPA.exe -n, and records bootstrap files."), - ("bootstrap-check", "Check BSIPA bootstrap", "Verifies recorded bootstrap state and the latest BSIPA log evidence."), - ("plan", "Create install plan", "Writes a dry-run JSON plan for locked plugin files before anything is applied."), - ("apply", "Apply a plan by path", "Installs exactly the file changes from a previously generated plan JSON."), - ("disable", "Disable managed plugin", "Removes one recorded plugin from the game while keeping helper state and assets."), - ("enable", "Enable managed plugin", "Reinstalls one disabled locked plugin from local assets."), - ("backup-userdata", "Back up UserData", "Copies UserData and AppData into the adjacent backups repo."), - ("restore-userdata", "Restore UserData", "Restores UserData and AppData from the backups repo into an instance."), - ("change", "Choose another version", "Returns to the Beat Saber version picker."), + profiles: tuple[Profile, ...], + selected_profile: Profile | None, + instances_roots: list[Path], + state_root: Path, + use_config_profiles: bool, +) -> list[Profile]: + if selected_profile: + return [ + Profile( + id=selected_profile.id if len(instances_roots) == 1 else f"{selected_profile.id}-{index}", + label=selected_profile.label, + instances_root=root, + state_dir=state_root, + ) + for index, root in enumerate(instances_roots, start=1) + ] + if use_config_profiles and profiles: + return list(profiles) + return [ + Profile( + id=f"root-{index}", + label=str(root), + instances_root=root, + state_dir=state_root, + ) + for index, root in enumerate(instances_roots, start=1) ] - selected_instance_key = _ask_choice( - title="Choose Beat Saber version", - choices=instance_choices, - input_func=ask, + +def _run_menu( + *, + runtime: Any, + explicit_instances_root: bool, + explicit_state_dir: bool, +) -> int: + try: + from .tui import InstallationChoice, PluginHelperTui + except ImportError as exc: + print(f"Textual is required for the menu TUI: {exc}") + print("Install project dependencies, for example: python -m pip install -e .") + return 1 + + profiles = _menu_profiles( + profiles=runtime.profiles, + selected_profile=runtime.selected_profile, + instances_roots=runtime.instances_roots, + state_root=runtime.state_root, + use_config_profiles=runtime.config_loaded and not explicit_instances_root and not explicit_state_dir, ) - if selected_instance_key is None: - return 0 - selected_instance = instances_by_choice[selected_instance_key] - - while True: - selected_action = _ask_choice( - title=f"Choose action for {selected_instance.name}", - choices=action_choices, - input_func=ask, - ) - if selected_action is None: - return 0 - if selected_action == "change": - selected_instance_key = _ask_choice( - title="Choose Beat Saber version", - choices=instance_choices, - input_func=ask, + choices: list[InstallationChoice] = [] + for profile in profiles: + for instance in list_instances(profile.instances_root): + choices.append( + InstallationChoice( + profile_id=profile.id, + profile_label=profile.label, + instance_name=instance.name, + instance_path=instance.path, + state_root=profile.state_dir, + ) ) - if selected_instance_key is None: - return 0 - selected_instance = instances_by_choice[selected_instance_key] - continue + if not choices: + searched = ", ".join(str(profile.instances_root) for profile in profiles) + print(f"No Beat Saber instances found under {searched}") + return 1 - command = [ - "--instances-root", - str(selected_instance.path.parent), - "--state-dir", - str(st_root), - selected_action, - ] - if selected_action == "apply": - plan_path = ask("Plan path> ").strip() - if not plan_path: - print("No plan path entered.") - continue - command.append(plan_path) - elif selected_action in {"disable", "enable"}: - plugin_id = ask("Plugin id> ").strip() - if not plugin_id: - print("No plugin id entered.") - continue - command.extend(["--instance", selected_instance.name, plugin_id]) - else: - command.extend(["--instance", selected_instance.name]) - - print() - status = run(command) - print(f"Command exited with status {status}") + setup_hint = None + if not runtime.config_loaded and not runtime.selected_profile and not explicit_instances_root and not explicit_state_dir: + setup_hint = ( + f"No {runtime.config_path.name} found; using default roots and default state. " + f"Copy plugin-helper.toml.example to {runtime.config_path.name} to map each install to a state dir." + ) + app = PluginHelperTui(choices=choices, repo_root=repo_root(), setup_hint=setup_hint) + result = app.run() + return int(result or 0) def run(argv: list[str] | None = None) -> int: parser = build_parser() args = parser.parse_args(argv) - inst_roots = instances_roots(getattr(args, "instances_root", None)) - st_root = state_root(getattr(args, "state_dir", None)) try: + explicit_instances_root = getattr(args, "instances_root", None) is not None + explicit_state_dir = getattr(args, "state_dir", None) is not None + runtime = resolve_runtime_config( + instances_root_value=getattr(args, "instances_root", None), + state_dir_value=getattr(args, "state_dir", None), + profile_id=getattr(args, "profile", None), + config_path=getattr(args, "config", None), + ) + inst_roots = runtime.instances_roots + st_root = runtime.state_root + if args.command == "instances": found = list_instances(inst_roots) if not found: @@ -427,7 +338,11 @@ def run(argv: list[str] | None = None) -> int: return 0 if args.command == "menu": - return _run_menu(inst_roots, st_root) + return _run_menu( + runtime=runtime, + explicit_instances_root=explicit_instances_root, + explicit_state_dir=explicit_state_dir, + ) if args.command == "scan": instance = get_instance(inst_roots, args.instance) @@ -617,30 +532,16 @@ def run(argv: list[str] | None = None) -> int: if args.command == "enable": instance = get_instance(inst_roots, args.instance) - installed_state = load_installed_state(st_root, args.instance) - if args.plugin not in installed_state.get("disabledPlugins", {}): - raise KeyError(f"plugin is not recorded as disabled: {args.plugin}") - root = repo_root() - registry_path = (root / args.registry).resolve() if not Path(args.registry).is_absolute() else Path(args.registry) - lock_path = Path(args.lockfile) if args.lockfile else root / "locks" / f"{args.instance}.lock.toml" - if not lock_path.is_absolute(): - lock_path = (root / lock_path).resolve() - lockfile = load_lockfile(lock_path) - if not any(plugin.id == args.plugin for plugin in lockfile.plugins): - raise KeyError(f"plugin is disabled but not locked for this instance: {args.plugin}") - plan, path = create_plan( + result = enable_disabled_plugin( instance=args.instance, instance_path=instance.path, - beat_saber_version=lockfile.beat_saber_version, - registry=load_registry(registry_path), - lockfile=lockfile, state_root=st_root, - repo_root=root, - selected={args.plugin}, + plugin_id=args.plugin, + registry=args.registry, + lockfile=args.lockfile, ) - result = apply_plan(plan, st_root) print(f"Enabled: {args.plugin}") - print(f"Plan: {path}") + print(f"Plan: {result['planPath']}") print(f"Applied: {len(result['applied'])}") print(f"State: {result['statePath']}") return 0 diff --git a/src/plugin_helper/config.py b/src/plugin_helper/config.py index 9b6f5e2..0c5cd2a 100644 --- a/src/plugin_helper/config.py +++ b/src/plugin_helper/config.py @@ -1,13 +1,39 @@ from __future__ import annotations import os +import tomllib +from dataclasses import dataclass from pathlib import Path +from typing import Any WINDOWS_INSTANCES_ROOT = Path("/home/pleb/Windows/Users/pleb/BSManager/BSInstances") LOCAL_INSTANCES_ROOT = Path.home() / ".local/share/BSManager/BSInstances" DEFAULT_INSTANCES_ROOTS = (WINDOWS_INSTANCES_ROOT, LOCAL_INSTANCES_ROOT) DEFAULT_INSTANCES_ROOT = WINDOWS_INSTANCES_ROOT +LOCAL_CONFIG_NAME = "plugin-helper.local.toml" + + +@dataclass(frozen=True) +class Profile: + id: str + label: str + instances_root: Path + state_dir: Path + + +@dataclass(frozen=True) +class RuntimeConfig: + instances_roots: list[Path] + state_root: Path + profiles: tuple[Profile, ...] + selected_profile: Profile | None + config_path: Path + config_loaded: bool + + +def repo_root() -> Path: + return Path(__file__).resolve().parents[2] def instances_root(value: str | None = None) -> Path: @@ -23,11 +49,92 @@ def instances_roots(value: str | None = None) -> list[Path]: def state_root(value: str | None = None) -> Path: if value: - return Path(value).expanduser() + return _resolve_path(value, repo_root()) xdg_state = os.environ.get("XDG_STATE_HOME") base = Path(xdg_state).expanduser() if xdg_state else Path.home() / ".local" / "state" return base / "plugin-helper" -def repo_root() -> Path: - return Path(__file__).resolve().parents[2] +def default_config_path(root: Path | None = None) -> Path: + return (root or repo_root()) / LOCAL_CONFIG_NAME + + +def _resolve_path(value: str | Path, base: Path) -> Path: + path = Path(value).expanduser() + if path.is_absolute(): + return path + return (base / path).resolve() + + +def load_profiles(config_path: str | Path | None = None, *, root: Path | None = None) -> tuple[tuple[Profile, ...], Path, bool]: + repo = root or repo_root() + path = _resolve_path(config_path, repo) if config_path else default_config_path(repo) + if not path.exists(): + if config_path: + raise FileNotFoundError(f"plugin-helper config not found: {path}") + return (), path, False + + with path.open("rb") as handle: + data: dict[str, Any] = tomllib.load(handle) + + profiles: list[Profile] = [] + seen: set[str] = set() + base = path.parent + for item in data.get("profiles", []): + profile_id = item["id"] + if profile_id in seen: + raise ValueError(f"{path}: duplicate profile id: {profile_id}") + seen.add(profile_id) + profiles.append( + Profile( + id=profile_id, + label=item.get("label", profile_id), + instances_root=_resolve_path(item["instances_root"], base), + state_dir=_resolve_path(item["state_dir"], base), + ) + ) + return tuple(profiles), path, True + + +def profile_by_id(profiles: tuple[Profile, ...], profile_id: str) -> Profile: + for profile in profiles: + if profile.id == profile_id: + return profile + available = ", ".join(profile.id for profile in profiles) or "(none)" + raise KeyError(f"unknown profile: {profile_id}; available profiles: {available}") + + +def resolve_runtime_config( + *, + instances_root_value: str | None = None, + state_dir_value: str | None = None, + profile_id: str | None = None, + config_path: str | Path | None = None, + root: Path | None = None, +) -> RuntimeConfig: + repo = root or repo_root() + profiles, loaded_path, loaded = load_profiles(config_path, root=repo) + selected = profile_by_id(profiles, profile_id) if profile_id else None + + if instances_root_value: + resolved_instances = instances_roots(instances_root_value) + elif selected: + resolved_instances = [selected.instances_root] + else: + resolved_instances = instances_roots(None) + + if state_dir_value: + resolved_state = _resolve_path(state_dir_value, repo) + elif selected: + resolved_state = selected.state_dir + else: + resolved_state = state_root(None) + + return RuntimeConfig( + instances_roots=resolved_instances, + state_root=resolved_state, + profiles=profiles, + selected_profile=selected, + config_path=loaded_path, + config_loaded=loaded, + ) diff --git a/src/plugin_helper/operations.py b/src/plugin_helper/operations.py new file mode 100644 index 0000000..501f140 --- /dev/null +++ b/src/plugin_helper/operations.py @@ -0,0 +1,47 @@ +from __future__ import annotations + +from pathlib import Path +from typing import Any + +from .config import repo_root +from .installer import apply_plan +from .models import load_lockfile, load_registry +from .planner import create_plan +from .state import load_installed_state + + +def enable_disabled_plugin( + *, + instance: str, + instance_path: Path, + state_root: Path, + plugin_id: str, + registry: str = "registry/plugins.toml", + lockfile: str | None = None, + repo: Path | None = None, +) -> dict[str, Any]: + installed_state = load_installed_state(state_root, instance) + if plugin_id not in installed_state.get("disabledPlugins", {}): + raise KeyError(f"plugin is not recorded as disabled: {plugin_id}") + + root = repo or repo_root() + registry_path = (root / registry).resolve() if not Path(registry).is_absolute() else Path(registry) + lock_path = Path(lockfile) if lockfile else root / "locks" / f"{instance}.lock.toml" + if not lock_path.is_absolute(): + lock_path = (root / lock_path).resolve() + loaded_lockfile = load_lockfile(lock_path) + if not any(plugin.id == plugin_id for plugin in loaded_lockfile.plugins): + raise KeyError(f"plugin is disabled but not locked for this instance: {plugin_id}") + + plan, path = create_plan( + instance=instance, + instance_path=instance_path, + beat_saber_version=loaded_lockfile.beat_saber_version, + registry=load_registry(registry_path), + lockfile=loaded_lockfile, + state_root=state_root, + repo_root=root, + selected={plugin_id}, + ) + result = apply_plan(plan, state_root) + return {"planPath": str(path), **result} diff --git a/src/plugin_helper/reports.py b/src/plugin_helper/reports.py new file mode 100644 index 0000000..218e7fb --- /dev/null +++ b/src/plugin_helper/reports.py @@ -0,0 +1,78 @@ +from __future__ import annotations + +from typing import Any + +from .models import Lockfile, Registry + + +def installed_plugins_report( + *, + installed_state: dict[str, Any], + registry: Registry, + lockfile: Lockfile, +) -> dict[str, Any]: + locked_by_id = {plugin.id: plugin for plugin in lockfile.plugins} + plugins: list[dict[str, Any]] = [] + state_plugins = [ + (plugin_id, plugin_state, "enabled") + for plugin_id, plugin_state in installed_state.get("plugins", {}).items() + ] + state_plugins.extend( + (plugin_id, plugin_state, "disabled") + for plugin_id, plugin_state in installed_state.get("disabledPlugins", {}).items() + ) + for plugin_id, plugin_state, status in sorted(state_plugins): + registry_plugin = registry.get(plugin_id) + locked = locked_by_id.get(plugin_id) + files = plugin_state.get("files", []) + plugins.append( + { + "id": plugin_id, + "name": registry_plugin.name if registry_plugin else plugin_id, + "version": locked.tag if locked and locked.tag else "(not locked)", + "asset": locked.asset if locked and locked.asset else "(unknown)", + "repo": (locked.repo if locked and locked.repo else None) + or (registry_plugin.repo if registry_plugin else None) + or "(unknown)", + "installedAt": plugin_state.get("installedAt", "(unknown)"), + "disabledAt": plugin_state.get("disabledAt"), + "status": status, + "fileCount": len(files), + "files": files, + } + ) + return { + "instance": installed_state.get("instance", lockfile.instance), + "beatSaberVersion": installed_state.get("beatSaberVersion", lockfile.beat_saber_version), + "plugins": plugins, + } + + +def print_installed_plugins(report: dict[str, Any]) -> None: + plugins = report["plugins"] + print(f"{report['instance']} managed plugins ({len(plugins)})") + if not plugins: + print("No plugins have been installed by plugin-helper yet.") + return + + headers = ("Plugin", "Status", "Version", "Asset", "Files", "Installed") + rows = [ + ( + f"{plugin['name']} ({plugin['id']})", + plugin["status"], + plugin["version"], + plugin["asset"], + str(plugin["fileCount"]), + plugin["disabledAt"] or plugin["installedAt"], + ) + for plugin in plugins + ] + widths = [ + max(len(headers[index]), *(len(row[index]) for row in rows)) + for index in range(len(headers)) + ] + header = " ".join(label.ljust(widths[index]) for index, label in enumerate(headers)) + print(header) + print(" ".join("-" * width for width in widths)) + for row in rows: + print(" ".join(value.ljust(widths[index]) for index, value in enumerate(row))) diff --git a/src/plugin_helper/tui.py b/src/plugin_helper/tui.py new file mode 100644 index 0000000..0343dd9 --- /dev/null +++ b/src/plugin_helper/tui.py @@ -0,0 +1,275 @@ +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +from rich.text import Text +from textual.app import App, ComposeResult +from textual.binding import Binding +from textual.widgets import DataTable, Footer, Header, Static + +from .installer import disable_plugin +from .models import load_lockfile, load_registry +from .operations import enable_disabled_plugin +from .reports import installed_plugins_report +from .state import load_installed_state + + +@dataclass(frozen=True) +class InstallationChoice: + profile_id: str + profile_label: str + instance_name: str + instance_path: Path + state_root: Path + + +class PluginHelperTui(App[int]): + CSS = """ + #title { + padding: 0 1; + text-style: bold; + } + + #status { + padding: 0 1; + color: $text-muted; + } + """ + BINDINGS = [ + Binding("enter", "select", "Select", priority=True), + Binding("space", "toggle_plugin", "Toggle", priority=True), + Binding("d", "disable_all_plugins", "Disable all", priority=True), + Binding("e", "enable_all_plugins", "Enable all", priority=True), + Binding("r", "refresh", "Refresh"), + Binding("b", "back", "Back"), + Binding("q", "quit", "Quit"), + Binding("ctrl+q", "quit", "Quit", show=False, priority=True), + ] + + def __init__( + self, + *, + choices: list[InstallationChoice], + repo_root: Path, + setup_hint: str | None = None, + ) -> None: + super().__init__() + self.choices = choices + self.repo_root = repo_root + self.setup_hint = setup_hint + self.mode = "installations" + self.selected_installation: InstallationChoice | None = None + self.plugin_rows: list[dict[str, Any]] = [] + self.status_message = "" + + def compose(self) -> ComposeResult: + yield Header(show_clock=False) + yield Static("", id="title") + yield DataTable(id="table") + yield Static("", id="status") + yield Footer() + + def on_mount(self) -> None: + table = self.query_one(DataTable) + table.cursor_type = "row" + self._show_installations() + + def action_select(self) -> None: + if self.mode != "installations": + return + index = self._cursor_index(len(self.choices)) + if index is None: + return + self.selected_installation = self.choices[index] + self._show_plugins() + + def action_back(self) -> None: + if self.mode == "plugins": + self._show_installations() + + def action_refresh(self) -> None: + if self.mode == "plugins": + self._show_plugins() + else: + self._show_installations() + + def action_toggle_plugin(self) -> None: + if self.mode != "plugins" or self.selected_installation is None: + return + index = self._cursor_index(len(self.plugin_rows)) + if index is None: + return + plugin = self.plugin_rows[index] + plugin_id = plugin["id"] + target = self.selected_installation + try: + if plugin["status"] == "enabled": + result = disable_plugin( + target.instance_name, + target.instance_path, + target.state_root, + plugin_id, + force=False, + ) + if not result["stateUpdated"]: + self._set_status(f"Could not disable {plugin_id}: {self._format_skipped(result['skipped'])}") + return + self._set_status(f"Disabled {plugin_id}; removed {len(result['removed'])} files.") + elif plugin["status"] == "disabled": + result = enable_disabled_plugin( + instance=target.instance_name, + instance_path=target.instance_path, + state_root=target.state_root, + plugin_id=plugin_id, + repo=self.repo_root, + ) + self._set_status(f"Enabled {plugin_id}; applied {len(result['applied'])} files.") + else: + self._set_status(f"Cannot toggle {plugin_id}: unknown status {plugin['status']}.") + return + except Exception as exc: + self._set_status(f"Could not toggle {plugin_id}: {exc}") + return + self._show_plugins(preserve_status=True) + + def action_disable_all_plugins(self) -> None: + if self.mode != "plugins" or self.selected_installation is None: + return + target = self.selected_installation + enabled = [plugin for plugin in self.plugin_rows if plugin["status"] == "enabled"] + changed = 0 + errors: list[str] = [] + for plugin in enabled: + plugin_id = plugin["id"] + try: + result = disable_plugin( + target.instance_name, + target.instance_path, + target.state_root, + plugin_id, + force=False, + ) + if result["stateUpdated"]: + changed += 1 + else: + errors.append(f"{plugin_id}: {self._format_skipped(result['skipped'])}") + except Exception as exc: + errors.append(f"{plugin_id}: {exc}") + self._set_bulk_status("Disabled", changed, errors) + self._show_plugins(preserve_status=True) + + def action_enable_all_plugins(self) -> None: + if self.mode != "plugins" or self.selected_installation is None: + return + target = self.selected_installation + disabled = [plugin for plugin in self.plugin_rows if plugin["status"] == "disabled"] + changed = 0 + errors: list[str] = [] + for plugin in disabled: + plugin_id = plugin["id"] + try: + enable_disabled_plugin( + instance=target.instance_name, + instance_path=target.instance_path, + state_root=target.state_root, + plugin_id=plugin_id, + repo=self.repo_root, + ) + changed += 1 + except Exception as exc: + errors.append(f"{plugin_id}: {exc}") + self._set_bulk_status("Enabled", changed, errors) + self._show_plugins(preserve_status=True) + + def _show_installations(self) -> None: + self.mode = "installations" + self.plugin_rows = [] + self._set_title("Choose Beat Saber Installation") + table = self.query_one(DataTable) + table.clear(columns=True) + table.add_columns("Profile", "Version", "Instance Path", "State Dir") + for choice in self.choices: + table.add_row( + choice.profile_label, + choice.instance_name, + str(choice.instance_path), + str(choice.state_root), + ) + if self.setup_hint: + self._set_status(self.setup_hint) + else: + self._set_status("Enter selects an installation. q quits.") + + def _show_plugins(self, *, preserve_status: bool = False) -> None: + if self.selected_installation is None: + self._show_installations() + return + target = self.selected_installation + self.mode = "plugins" + self._set_title(f"{target.profile_label} / {target.instance_name}") + table = self.query_one(DataTable) + table.clear(columns=True) + table.add_columns("Status", "Name", "ID", "Version", "Files", "Asset") + + try: + lockfile = load_lockfile(self.repo_root / "locks" / f"{target.instance_name}.lock.toml") + report = installed_plugins_report( + installed_state=load_installed_state(target.state_root, target.instance_name), + registry=load_registry(self.repo_root / "registry" / "plugins.toml"), + lockfile=lockfile, + ) + self.plugin_rows = report["plugins"] + except Exception as exc: + self.plugin_rows = [] + if not preserve_status: + self._set_status(f"Could not load plugins: {exc}") + return + + for plugin in self.plugin_rows: + table.add_row( + self._status_marker(plugin["status"]), + plugin["name"], + plugin["id"], + plugin["version"], + str(plugin["fileCount"]), + plugin["asset"], + ) + if not preserve_status: + if self.plugin_rows: + self._set_status("Space toggles selected. d disables all. e enables all. b returns to installations.") + else: + self._set_status("No managed plugins recorded for this installation.") + + def _cursor_index(self, row_count: int) -> int | None: + table = self.query_one(DataTable) + row = table.cursor_coordinate.row + if 0 <= row < row_count: + return row + return None + + def _set_title(self, message: str) -> None: + self.query_one("#title", Static).update(message) + + def _set_status(self, message: str) -> None: + self.status_message = message + self.query_one("#status", Static).update(message) + + def _set_bulk_status(self, verb: str, changed: int, errors: list[str]) -> None: + if errors: + preview = "; ".join(errors[:3]) + suffix = f"; {len(errors) - 3} more" if len(errors) > 3 else "" + self._set_status(f"{verb} {changed} plugins; {len(errors)} failed: {preview}{suffix}") + else: + self._set_status(f"{verb} {changed} plugins.") + + @staticmethod + def _status_marker(status: str) -> Text: + return Text("[x]" if status == "enabled" else "[ ]", no_wrap=True) + + @staticmethod + def _format_skipped(skipped: list[dict[str, str]]) -> str: + if not skipped: + return "no files changed" + return "; ".join(f"{item['path']} {item['reason']}" for item in skipped) diff --git a/tests/test_plugin_helper.py b/tests/test_plugin_helper.py index 5811590..612f9cb 100644 --- a/tests/test_plugin_helper.py +++ b/tests/test_plugin_helper.py @@ -4,22 +4,27 @@ import json import tempfile import unittest import os -from io import StringIO from pathlib import Path from unittest.mock import patch from zipfile import ZipFile +from rich.text import Text +from textual.coordinate import Coordinate +from textual.widgets import DataTable + from plugin_helper.bootstrap import _run_ipa from plugin_helper.beatmods import by_version_id, normalize_mods from plugin_helper.checker import check_lock from plugin_helper.cli import installed_plugins_report, run +from plugin_helper.config import load_profiles, profile_by_id, resolve_runtime_config from plugin_helper.fsutil import sha256_file from plugin_helper.installer import apply_plan, disable_plugin, uninstall_plugin from plugin_helper.instances import get_instance, list_instances from plugin_helper.models import Lockfile, LockedPlugin, Registry, RegistryPlugin from plugin_helper.planner import create_plan from plugin_helper.scanner import scan_bootstrap_files, scan_instance -from plugin_helper.state import downloads_dir, load_installed_state, plugin_downloads_dir, save_bootstrap_state +from plugin_helper.state import downloads_dir, load_installed_state, plugin_downloads_dir, save_bootstrap_state, save_installed_state +from plugin_helper.tui import InstallationChoice, PluginHelperTui from plugin_helper.updates import check_updates from plugin_helper.userdata import ( backup_userdata, @@ -119,58 +124,61 @@ class PluginHelperTests(unittest.TestCase): with self.assertRaisesRegex(ValueError, "ambiguous"): get_instance([windows, local], "1.44.1") - def test_menu_selects_instance_and_action(self) -> None: + def test_profile_config_loads_and_resolves_paths(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) - instance = root / "instances" / "1.40.8" - state = root / "state" - (instance / "Beat Saber_Data").mkdir(parents=True) - (instance / "Plugins").mkdir() + config = root / "plugin-helper.local.toml" + config.write_text( + """ +[[profiles]] +id = "linux" +label = "Linux" +instances_root = "~/BSInstances" +state_dir = ".state" - answers = iter(["1", "3", "q"]) - output = StringIO() - with patch("builtins.input", side_effect=lambda _: next(answers)), patch("sys.stdout", output): - status = run( - [ - "--instances-root", - str(root / "instances"), - "--state-dir", - str(state), - "menu", - ] - ) +[[profiles]] +id = "windows" +label = "Windows" +instances_root = "mounted/BSInstances" +state_dir = ".state-windows" +""".lstrip(), + encoding="utf-8", + ) - self.assertEqual(status, 0) - self.assertIn("Counts files currently present", output.getvalue()) + profiles, loaded_path, loaded = load_profiles(config, root=root) - def test_menu_routes_duplicate_instance_names_by_selected_root(self) -> None: + self.assertTrue(loaded) + self.assertEqual(loaded_path, config) + self.assertEqual(profile_by_id(profiles, "linux").instances_root, Path("~/BSInstances").expanduser()) + self.assertEqual(profile_by_id(profiles, "windows").instances_root, root / "mounted" / "BSInstances") + self.assertEqual(profile_by_id(profiles, "windows").state_dir, root / ".state-windows") + + def test_profile_runtime_explicit_overrides(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) - first_root = root / "a-root" - second_root = root / "z-root" - first = first_root / "1.44.1" - second = second_root / "1.44.1" - state = root / "state" - (first / "Beat Saber_Data").mkdir(parents=True) - (second / "Beat Saber_Data").mkdir(parents=True) - (second / "Plugins").mkdir() - (second / "Plugins" / "Example.dll").write_bytes(b"dll") + config = root / "plugin-helper.local.toml" + config.write_text( + """ +[[profiles]] +id = "linux" +label = "Linux" +instances_root = "profile-root" +state_dir = ".state" +""".lstrip(), + encoding="utf-8", + ) - answers = iter(["2", "3", "q"]) - output = StringIO() - with patch("builtins.input", side_effect=lambda _: next(answers)), patch("sys.stdout", output): - status = run( - [ - "--instances-root", - os.pathsep.join([str(first_root), str(second_root)]), - "--state-dir", - str(state), - "menu", - ] - ) + runtime = resolve_runtime_config( + instances_root_value=str(root / "explicit-root"), + state_dir_value="explicit-state", + profile_id="linux", + config_path=config, + root=root, + ) - self.assertEqual(status, 0) - self.assertIn("1.44.1: 1 files", output.getvalue()) + self.assertEqual(runtime.instances_roots, [root / "explicit-root"]) + self.assertEqual(runtime.state_root, root / "explicit-state") + self.assertEqual(runtime.selected_profile.id, "linux") def test_run_ipa_timeout_returns_control(self) -> None: with tempfile.TemporaryDirectory() as tmp: @@ -343,7 +351,7 @@ sha256 = "{sha256_file(asset)}" save_installed_state(state, "1.40.8", disabled) - with patch("plugin_helper.cli.repo_root", return_value=work): + with patch("plugin_helper.operations.repo_root", return_value=work): status = run( [ "--instances-root", @@ -971,5 +979,175 @@ sha256 = "{sha256_file(asset)}" self.assertEqual(result["plugins"][0]["latestAssetSha256"], "new") +def _make_tui_fixture(root: Path, *, disabled: bool = False, hash_mismatch: bool = False) -> tuple[PluginHelperTui, Path, Path]: + repo = root / "repo" + instance_root = root / "instances" + instance = instance_root / "1.40.8" + state = root / "state" + (repo / "registry").mkdir(parents=True) + (repo / "locks").mkdir() + (instance / "Beat Saber_Data").mkdir(parents=True) + (instance / "Plugins").mkdir() + + asset = plugin_downloads_dir(state, "1.40.8", "example") / "Example.dll" + asset.write_bytes(b"managed dll") + (repo / "registry" / "plugins.toml").write_text( + """ +[[plugins]] +id = "example" +name = "Example" +repo = "owner/example" +asset_patterns = ["*.dll"] +install_strategy = "dll-to-plugins" +""".lstrip(), + encoding="utf-8", + ) + (repo / "locks" / "1.40.8.lock.toml").write_text( + f""" +beat_saber_version = "1.40.8" +instance = "1.40.8" + +[[plugins]] +id = "example" +repo = "owner/example" +tag = "v1.0.0" +asset = "Example.dll" +sha256 = "{sha256_file(asset)}" +""".lstrip(), + encoding="utf-8", + ) + + target = instance / "Plugins" / "Example.dll" + if disabled: + plugins = {} + disabled_plugins = { + "example": { + "installedAt": "2026-06-14T17:18:40Z", + "disabledAt": "2026-06-14T17:20:00Z", + "files": [{"path": "Plugins/Example.dll", "sha256": sha256_file(asset), "size": asset.stat().st_size}], + } + } + else: + target.write_bytes(b"changed dll" if hash_mismatch else b"managed dll") + plugins = { + "example": { + "installedAt": "2026-06-14T17:18:40Z", + "files": [{"path": "Plugins/Example.dll", "sha256": sha256_file(asset), "size": asset.stat().st_size}], + } + } + disabled_plugins = {} + save_installed_state( + state, + "1.40.8", + {"instance": "1.40.8", "plugins": plugins, "disabledPlugins": disabled_plugins}, + ) + choice = InstallationChoice( + profile_id="test", + profile_label="Test Profile", + instance_name="1.40.8", + instance_path=instance, + state_root=state, + ) + return PluginHelperTui(choices=[choice], repo_root=repo), instance, state + + +class PluginHelperTuiTests(unittest.IsolatedAsyncioTestCase): + async def test_installation_picker_shows_duplicate_instances_with_state_dirs(self) -> None: + choices = [ + InstallationChoice( + profile_id="linux", + profile_label="Linux", + instance_name="1.44.1", + instance_path=Path("/tmp/linux/1.44.1"), + state_root=Path("/tmp/state-linux"), + ), + InstallationChoice( + profile_id="windows", + profile_label="Windows", + instance_name="1.44.1", + instance_path=Path("/tmp/windows/1.44.1"), + state_root=Path("/tmp/state-windows"), + ), + ] + app = PluginHelperTui(choices=choices, repo_root=Path("/tmp/repo")) + + async with app.run_test(): + table = app.query_one(DataTable) + self.assertEqual(table.row_count, 2) + self.assertEqual(app.mode, "installations") + + async def test_space_disables_enabled_plugin_without_id_prompt(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + app, instance, state = _make_tui_fixture(Path(tmp)) + + async with app.run_test() as pilot: + await pilot.press("enter") + self.assertEqual(app.plugin_rows[0]["status"], "enabled") + table = app.query_one(DataTable) + self.assertEqual(table.get_cell_at(Coordinate(0, 0)), Text("[x]", no_wrap=True)) + await pilot.press("space") + + self.assertFalse((instance / "Plugins" / "Example.dll").exists()) + updated = load_installed_state(state, "1.40.8") + self.assertNotIn("example", updated["plugins"]) + self.assertIn("example", updated["disabledPlugins"]) + + async def test_space_enables_disabled_plugin_from_asset(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + app, instance, state = _make_tui_fixture(Path(tmp), disabled=True) + + async with app.run_test() as pilot: + await pilot.press("enter") + self.assertEqual(app.plugin_rows[0]["status"], "disabled") + await pilot.press("space") + + self.assertEqual((instance / "Plugins" / "Example.dll").read_bytes(), b"managed dll") + updated = load_installed_state(state, "1.40.8") + self.assertIn("example", updated["plugins"]) + self.assertNotIn("example", updated["disabledPlugins"]) + + async def test_disable_all_disables_enabled_plugins(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + app, instance, state = _make_tui_fixture(Path(tmp)) + + async with app.run_test() as pilot: + await pilot.press("enter") + await pilot.press("d") + + self.assertFalse((instance / "Plugins" / "Example.dll").exists()) + updated = load_installed_state(state, "1.40.8") + self.assertNotIn("example", updated["plugins"]) + self.assertIn("example", updated["disabledPlugins"]) + self.assertIn("Disabled 1 plugins", app.status_message) + + async def test_enable_all_enables_disabled_plugins(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + app, instance, state = _make_tui_fixture(Path(tmp), disabled=True) + + async with app.run_test() as pilot: + await pilot.press("enter") + await pilot.press("e") + + self.assertEqual((instance / "Plugins" / "Example.dll").read_bytes(), b"managed dll") + updated = load_installed_state(state, "1.40.8") + self.assertIn("example", updated["plugins"]) + self.assertNotIn("example", updated["disabledPlugins"]) + self.assertIn("Enabled 1 plugins", app.status_message) + + async def test_space_reports_hash_mismatch_without_state_update(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + app, instance, state = _make_tui_fixture(Path(tmp), hash_mismatch=True) + + async with app.run_test() as pilot: + await pilot.press("enter") + await pilot.press("space") + + self.assertEqual((instance / "Plugins" / "Example.dll").read_bytes(), b"changed dll") + updated = load_installed_state(state, "1.40.8") + self.assertIn("example", updated["plugins"]) + self.assertNotIn("example", updated.get("disabledPlugins", {})) + self.assertIn("hash mismatch", app.status_message) + + if __name__ == "__main__": unittest.main()