#!/usr/bin/env python3 """Manage directory-to-profile mappings for pi-sbx.""" from __future__ import annotations import argparse import json import os import subprocess import tempfile from pathlib import Path from typing import Any DEFAULT_CONFIG_PATH = Path("~/.config/pi-sbx/config.json") RUNTIME_FIELDS = ("nodeExecutable", "piExecutable", "piAgentDirectory") def config_path() -> Path: override = os.environ.get("PI_SBX_CONFIG") if override: selected_path = Path(override).expanduser() if not selected_path.is_absolute(): raise ValueError("PI_SBX_CONFIG must be an absolute path or start with ~") return selected_path xdg_config_home = os.environ.get("XDG_CONFIG_HOME") if xdg_config_home: selected_directory = Path(xdg_config_home).expanduser() if selected_directory.is_absolute(): return selected_directory / "pi-sbx" / "config.json" return DEFAULT_CONFIG_PATH.expanduser() def validate_runtime(runtime: Any, field_name: str, *, require_all: bool) -> None: if not isinstance(runtime, dict): raise ValueError(f"{field_name} must be a JSON object") unknown_fields = set(runtime) - set(RUNTIME_FIELDS) if unknown_fields: unknown = ", ".join(sorted(unknown_fields)) raise ValueError(f"{field_name} contains unknown fields: {unknown}") if require_all: missing_fields = set(RUNTIME_FIELDS) - set(runtime) if missing_fields: missing = ", ".join(sorted(missing_fields)) raise ValueError(f"{field_name} is missing fields: {missing}") for key, value in runtime.items(): if not isinstance(value, str) or not value: raise ValueError(f"{field_name}.{key} must be a non-empty path") if not Path(value).expanduser().is_absolute(): raise ValueError(f"{field_name}.{key} must be an absolute path or start with ~") def validate_config(data: Any) -> dict[str, Any]: if not isinstance(data, dict): raise ValueError("configuration must be a JSON object") default_profile = data.get("defaultProfile") if not isinstance(default_profile, str) or not default_profile: raise ValueError("defaultProfile must be a non-empty profile name") directories = data.get("directories") if not isinstance(directories, dict): raise ValueError("directories must be a JSON object") for directory, profile in directories.items(): if not isinstance(directory, str): raise ValueError("directories keys must be canonical absolute paths") configured_path = Path(directory).expanduser() if not configured_path.is_absolute() or str(configured_path.resolve()) != directory: raise ValueError("directories keys must be canonical absolute physical paths") if not isinstance(profile, str) or not profile: raise ValueError("directories values must be non-empty profile names") validate_runtime(data.get("defaultRuntime"), "defaultRuntime", require_all=True) profile_runtimes = data.get("profileRuntimes", {}) if not isinstance(profile_runtimes, dict): raise ValueError("profileRuntimes must be a JSON object") for profile, runtime in profile_runtimes.items(): if not isinstance(profile, str) or not profile: raise ValueError("profileRuntimes keys must be non-empty profile names") validate_runtime(runtime, f"profileRuntimes.{profile}", require_all=False) return data def load_config(path: Path | None = None) -> dict[str, Any]: selected_path = path or config_path() try: contents = selected_path.read_text(encoding="utf-8") data = json.loads(contents) except (OSError, json.JSONDecodeError) as error: raise ValueError(f"cannot load configuration from {selected_path}: {error}") from error return validate_config(data) def save_config(data: dict[str, Any], path: Path | None = None) -> None: validate_config(data) selected_path = path or config_path() selected_path.parent.mkdir(parents=True, exist_ok=True) descriptor, temporary_name = tempfile.mkstemp(prefix=".config.", dir=selected_path.parent) temporary = Path(temporary_name) try: with os.fdopen(descriptor, "w", encoding="utf-8") as handle: json.dump(data, handle, indent=2, sort_keys=True) handle.write("\n") temporary.chmod(0o600) os.replace(temporary, selected_path) finally: temporary.unlink(missing_ok=True) def resolve_profile(data: dict[str, Any], path: Path) -> tuple[str, str, Path | None]: launch_path = path.expanduser().resolve() matches: list[tuple[Path, str]] = [] for directory_value, profile in data["directories"].items(): directory = Path(directory_value).expanduser().resolve() if launch_path == directory or directory in launch_path.parents: matches.append((directory, profile)) if matches: directory, profile = max(matches, key=lambda match: len(match[0].parts)) return profile, "directory", directory return data["defaultProfile"], "default", None def runtime_for_profile(data: dict[str, Any], profile: str) -> dict[str, str]: runtime = dict(data["defaultRuntime"]) runtime.update(data.get("profileRuntimes", {}).get(profile, {})) return {key: str(Path(runtime[key]).expanduser()) for key in RUNTIME_FIELDS} def resolved_configuration(data: dict[str, Any], path: Path) -> dict[str, Any]: launch_path = path.expanduser().resolve() profile, source, directory = resolve_profile(data, launch_path) return { "launchPath": str(launch_path), "matchedDirectory": str(directory) if directory else None, "profile": profile, "runtime": runtime_for_profile(data, profile), "source": source, } def profile_exists(profile: str) -> bool: result = subprocess.run( ["nono", "profile", "show", profile], check=False, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) return result.returncode == 0 def create_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description=__doc__) subparsers = parser.add_subparsers(dest="command", required=True) resolve_parser = subparsers.add_parser("resolve") resolve_parser.add_argument("path", nargs="?", default=os.getcwd()) resolve_parser.add_argument( "--format", choices=("profile", "launcher", "json"), default="profile", ) show_parser = subparsers.add_parser("show") show_parser.add_argument("path", nargs="?", default=os.getcwd()) set_parser = subparsers.add_parser("set") set_parser.add_argument("profile") set_parser.add_argument("path", nargs="?", default=os.getcwd()) unset_parser = subparsers.add_parser("unset") unset_parser.add_argument("path", nargs="?", default=os.getcwd()) subparsers.add_parser("list") return parser def main() -> int: parser = create_parser() args = parser.parse_args() try: data = load_config() except ValueError as error: parser.error(str(error)) if args.command == "list": print(json.dumps(data, indent=2, sort_keys=True)) return 0 path = Path(args.path).expanduser().resolve() if args.command in ("resolve", "show"): resolved = resolved_configuration(data, path) if args.command == "show" or args.format == "json": print(json.dumps(resolved, indent=2, sort_keys=True)) elif args.format == "launcher": runtime = resolved["runtime"] print(resolved["profile"]) print(runtime["nodeExecutable"]) print(runtime["piExecutable"]) print(runtime["piAgentDirectory"]) else: print(resolved["profile"]) return 0 if args.command == "set": if not profile_exists(args.profile): parser.error(f"nono profile does not exist: {args.profile}") data["directories"][str(path)] = args.profile save_config(data) print(f"{path} -> {args.profile}") return 0 if args.command == "unset": removed = data["directories"].pop(str(path), None) save_config(data) if removed is None: print(f"no mapping for {path}") else: print(f"removed {path} -> {removed}") return 0 raise AssertionError(f"unhandled command: {args.command}") if __name__ == "__main__": raise SystemExit(main())