Generalize Pi sandbox launcher
This commit is contained in:
Executable
+236
@@ -0,0 +1,236 @@
|
||||
#!/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())
|
||||
Reference in New Issue
Block a user