237 lines
8.4 KiB
Python
Executable File
237 lines
8.4 KiB
Python
Executable File
#!/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())
|