#!/usr/bin/env python3 from __future__ import annotations import argparse import json import os import shutil import subprocess import sys import tempfile import textwrap import time from dataclasses import asdict, dataclass from datetime import datetime, timezone from pathlib import Path APP_NAME = "aman" INSTALL_KIND = "portable" SERVICE_NAME = "aman" MANAGED_MARKER = "# managed by aman portable installer" SUPPORTED_PYTHON_TAGS = ("cp310", "cp311", "cp312") DEFAULT_ARCHITECTURE = "x86_64" DEFAULT_SMOKE_CHECK_CODE = textwrap.dedent( """ import gi gi.require_version("Gtk", "3.0") gi.require_version("AppIndicator3", "0.1") from gi.repository import AppIndicator3, Gtk import Xlib import sounddevice """ ).strip() DEFAULT_RUNTIME_DEPENDENCY_HINT = ( "Install the documented GTK, AppIndicator, PyGObject, python-xlib, and " "PortAudio runtime dependencies for your distro, then rerun install.sh." ) class PortableInstallError(RuntimeError): pass @dataclass class InstallPaths: home: Path share_root: Path current_link: Path state_path: Path bin_dir: Path shim_path: Path systemd_dir: Path service_path: Path config_dir: Path cache_dir: Path @classmethod def detect(cls) -> "InstallPaths": home = Path.home() share_root = home / ".local" / "share" / APP_NAME return cls( home=home, share_root=share_root, current_link=share_root / "current", state_path=share_root / "install-state.json", bin_dir=home / ".local" / "bin", shim_path=home / ".local" / "bin" / APP_NAME, systemd_dir=home / ".config" / "systemd" / "user", service_path=home / ".config" / "systemd" / "user" / f"{SERVICE_NAME}.service", config_dir=home / ".config" / APP_NAME, cache_dir=home / ".cache" / APP_NAME, ) def as_serializable(self) -> dict[str, str]: return { "share_root": str(self.share_root), "current_link": str(self.current_link), "state_path": str(self.state_path), "shim_path": str(self.shim_path), "service_path": str(self.service_path), "config_dir": str(self.config_dir), "cache_dir": str(self.cache_dir), } @dataclass class Manifest: app_name: str version: str architecture: str supported_python_tags: list[str] wheelhouse_dirs: list[str] managed_paths: dict[str, str] smoke_check_code: str runtime_dependency_hint: str bundle_format_version: int = 1 @classmethod def default(cls, version: str) -> "Manifest": return cls( app_name=APP_NAME, version=version, architecture=DEFAULT_ARCHITECTURE, supported_python_tags=list(SUPPORTED_PYTHON_TAGS), wheelhouse_dirs=[ "wheelhouse/common", "wheelhouse/cp310", "wheelhouse/cp311", "wheelhouse/cp312", ], managed_paths={ "install_root": "~/.local/share/aman", "current_link": "~/.local/share/aman/current", "shim": "~/.local/bin/aman", "service": "~/.config/systemd/user/aman.service", "state": "~/.local/share/aman/install-state.json", }, smoke_check_code=DEFAULT_SMOKE_CHECK_CODE, runtime_dependency_hint=DEFAULT_RUNTIME_DEPENDENCY_HINT, ) @dataclass class InstallState: app_name: str install_kind: str version: str installed_at: str service_mode: str architecture: str supported_python_tags: list[str] paths: dict[str, str] def _portable_tag() -> str: test_override = os.environ.get("AMAN_PORTABLE_TEST_PYTHON_TAG", "").strip() if test_override: return test_override return f"cp{sys.version_info.major}{sys.version_info.minor}" def _load_manifest(bundle_dir: Path) -> Manifest: manifest_path = bundle_dir / "manifest.json" try: payload = json.loads(manifest_path.read_text(encoding="utf-8")) except FileNotFoundError as exc: raise PortableInstallError(f"missing manifest: {manifest_path}") from exc except json.JSONDecodeError as exc: raise PortableInstallError(f"invalid manifest JSON: {manifest_path}") from exc try: return Manifest(**payload) except TypeError as exc: raise PortableInstallError(f"invalid manifest shape: {manifest_path}") from exc def _load_state(state_path: Path) -> InstallState | None: if not state_path.exists(): return None try: payload = json.loads(state_path.read_text(encoding="utf-8")) except json.JSONDecodeError as exc: raise PortableInstallError(f"invalid install state JSON: {state_path}") from exc try: return InstallState(**payload) except TypeError as exc: raise PortableInstallError(f"invalid install state shape: {state_path}") from exc def _atomic_write(path: Path, content: str, *, mode: int = 0o644) -> None: path.parent.mkdir(parents=True, exist_ok=True) with tempfile.NamedTemporaryFile( "w", encoding="utf-8", dir=path.parent, prefix=f".{path.name}.tmp-", delete=False, ) as handle: handle.write(content) tmp_path = Path(handle.name) os.chmod(tmp_path, mode) os.replace(tmp_path, path) def _atomic_symlink(target: Path, link_path: Path) -> None: link_path.parent.mkdir(parents=True, exist_ok=True) tmp_link = link_path.parent / f".{link_path.name}.tmp-{os.getpid()}" try: if tmp_link.exists() or tmp_link.is_symlink(): tmp_link.unlink() os.symlink(str(target), tmp_link) os.replace(tmp_link, link_path) finally: if tmp_link.exists() or tmp_link.is_symlink(): tmp_link.unlink() def _read_text_if_exists(path: Path) -> str | None: if not path.exists(): return None return path.read_text(encoding="utf-8") def _current_target(current_link: Path) -> Path | None: if current_link.is_symlink(): target = os.readlink(current_link) target_path = Path(target) if not target_path.is_absolute(): target_path = current_link.parent / target_path return target_path if current_link.exists(): return current_link return None def _is_managed_text(content: str | None) -> bool: return bool(content and MANAGED_MARKER in content) def _run( args: list[str], *, check: bool = True, capture_output: bool = False, ) -> subprocess.CompletedProcess[str]: try: return subprocess.run( args, check=check, text=True, capture_output=capture_output, ) except subprocess.CalledProcessError as exc: details = exc.stderr.strip() or exc.stdout.strip() or str(exc) raise PortableInstallError(details) from exc def _run_systemctl(args: list[str], *, check: bool = True) -> subprocess.CompletedProcess[str]: return _run(["systemctl", "--user", *args], check=check, capture_output=True) def _supported_tag_or_raise(manifest: Manifest) -> str: if sys.implementation.name != "cpython": raise PortableInstallError("portable installer requires CPython 3.10, 3.11, or 3.12") tag = _portable_tag() if tag not in manifest.supported_python_tags: version = f"{sys.version_info.major}.{sys.version_info.minor}" raise PortableInstallError( f"unsupported python3 version {version}; supported versions are CPython 3.10, 3.11, and 3.12" ) return tag def _check_preflight(manifest: Manifest, paths: InstallPaths) -> InstallState | None: _supported_tag_or_raise(manifest) if shutil.which("systemctl") is None: raise PortableInstallError("systemctl is required for the supported user service lifecycle") try: import venv as _venv # noqa: F401 except Exception as exc: # pragma: no cover - import failure is environment dependent raise PortableInstallError("python3 venv support is required for the portable installer") from exc state = _load_state(paths.state_path) if state is not None: if state.app_name != APP_NAME or state.install_kind != INSTALL_KIND: raise PortableInstallError(f"unexpected install state in {paths.state_path}") shim_text = _read_text_if_exists(paths.shim_path) if shim_text is not None and (state is None or not _is_managed_text(shim_text)): raise PortableInstallError( f"refusing to overwrite unmanaged shim at {paths.shim_path}; remove it first" ) service_text = _read_text_if_exists(paths.service_path) if service_text is not None and (state is None or not _is_managed_text(service_text)): raise PortableInstallError( f"refusing to overwrite unmanaged service file at {paths.service_path}; remove it first" ) detected_aman = shutil.which(APP_NAME) if detected_aman: expected_paths = {str(paths.shim_path)} current_target = _current_target(paths.current_link) if current_target is not None: expected_paths.add(str(current_target / "venv" / "bin" / APP_NAME)) if detected_aman not in expected_paths: raise PortableInstallError( "detected another Aman install in PATH at " f"{detected_aman}; remove that install before using the portable bundle" ) return state def _require_bundle_file(path: Path, description: str) -> Path: if not path.exists(): raise PortableInstallError(f"missing {description}: {path}") return path def _aman_wheel(common_wheelhouse: Path) -> Path: wheels = sorted(common_wheelhouse.glob(f"{APP_NAME}-*.whl")) if not wheels: raise PortableInstallError(f"no Aman wheel found in {common_wheelhouse}") return wheels[-1] def _render_wrapper(paths: InstallPaths) -> str: exec_path = paths.current_link / "venv" / "bin" / APP_NAME return textwrap.dedent( f"""\ #!/usr/bin/env bash set -euo pipefail {MANAGED_MARKER} exec "{exec_path}" "$@" """ ) def _render_service(template_text: str, paths: InstallPaths) -> str: exec_start = ( f"{paths.current_link / 'venv' / 'bin' / APP_NAME} " f"run --config {paths.home / '.config' / APP_NAME / 'config.json'}" ) return template_text.replace("__EXEC_START__", exec_start) def _write_state(paths: InstallPaths, manifest: Manifest, version_dir: Path) -> None: state = InstallState( app_name=APP_NAME, install_kind=INSTALL_KIND, version=manifest.version, installed_at=datetime.now(timezone.utc).isoformat(), service_mode="systemd-user", architecture=manifest.architecture, supported_python_tags=list(manifest.supported_python_tags), paths={ **paths.as_serializable(), "version_dir": str(version_dir), }, ) _atomic_write(paths.state_path, json.dumps(asdict(state), indent=2, sort_keys=True) + "\n") def _copy_bundle_support_files(bundle_dir: Path, stage_dir: Path) -> None: for name in ("manifest.json", "install.sh", "uninstall.sh", "portable_installer.py"): src = _require_bundle_file(bundle_dir / name, name) dst = stage_dir / name shutil.copy2(src, dst) if dst.suffix in {".sh", ".py"}: os.chmod(dst, 0o755) src_service_dir = _require_bundle_file(bundle_dir / "systemd", "systemd directory") dst_service_dir = stage_dir / "systemd" if dst_service_dir.exists(): shutil.rmtree(dst_service_dir) shutil.copytree(src_service_dir, dst_service_dir) def _run_pip_install(bundle_dir: Path, stage_dir: Path, python_tag: str) -> None: common_dir = _require_bundle_file(bundle_dir / "wheelhouse" / "common", "common wheelhouse") version_dir = _require_bundle_file(bundle_dir / "wheelhouse" / python_tag, f"{python_tag} wheelhouse") aman_wheel = _aman_wheel(common_dir) venv_dir = stage_dir / "venv" _run([sys.executable, "-m", "venv", "--system-site-packages", str(venv_dir)]) _run( [ str(venv_dir / "bin" / "python"), "-m", "pip", "install", "--no-index", "--find-links", str(common_dir), "--find-links", str(version_dir), str(aman_wheel), ] ) def _run_smoke_check(stage_dir: Path, manifest: Manifest) -> None: venv_python = stage_dir / "venv" / "bin" / "python" try: _run([str(venv_python), "-c", manifest.smoke_check_code], capture_output=True) except PortableInstallError as exc: raise PortableInstallError( f"runtime dependency smoke check failed: {exc}\n{manifest.runtime_dependency_hint}" ) from exc def _remove_path(path: Path) -> None: if path.is_symlink() or path.is_file(): path.unlink(missing_ok=True) return if path.is_dir(): shutil.rmtree(path, ignore_errors=True) def _rollback_install( *, paths: InstallPaths, manifest: Manifest, old_state_text: str | None, old_service_text: str | None, old_shim_text: str | None, old_current_target: Path | None, new_version_dir: Path, backup_dir: Path | None, ) -> None: _remove_path(new_version_dir) if backup_dir is not None and backup_dir.exists(): os.replace(backup_dir, new_version_dir) if old_current_target is not None: _atomic_symlink(old_current_target, paths.current_link) else: _remove_path(paths.current_link) if old_shim_text is not None: _atomic_write(paths.shim_path, old_shim_text, mode=0o755) else: _remove_path(paths.shim_path) if old_service_text is not None: _atomic_write(paths.service_path, old_service_text) else: _remove_path(paths.service_path) if old_state_text is not None: _atomic_write(paths.state_path, old_state_text) else: _remove_path(paths.state_path) _run_systemctl(["daemon-reload"], check=False) if old_current_target is not None and old_service_text is not None: _run_systemctl(["enable", "--now", SERVICE_NAME], check=False) def _prune_versions(paths: InstallPaths, keep_version: str) -> None: for entry in paths.share_root.iterdir(): if entry.name in {"current", "install-state.json"}: continue if entry.is_dir() and entry.name != keep_version: shutil.rmtree(entry, ignore_errors=True) def install_bundle(bundle_dir: Path) -> int: manifest = _load_manifest(bundle_dir) paths = InstallPaths.detect() previous_state = _check_preflight(manifest, paths) python_tag = _supported_tag_or_raise(manifest) paths.share_root.mkdir(parents=True, exist_ok=True) stage_dir = paths.share_root / f".staging-{manifest.version}-{os.getpid()}" version_dir = paths.share_root / manifest.version backup_dir: Path | None = None old_state_text = _read_text_if_exists(paths.state_path) old_service_text = _read_text_if_exists(paths.service_path) old_shim_text = _read_text_if_exists(paths.shim_path) old_current_target = _current_target(paths.current_link) service_template_path = _require_bundle_file( bundle_dir / "systemd" / f"{SERVICE_NAME}.service.in", "service template", ) service_template = service_template_path.read_text(encoding="utf-8") cutover_done = False if previous_state is not None: _run_systemctl(["stop", SERVICE_NAME], check=False) _remove_path(stage_dir) stage_dir.mkdir(parents=True, exist_ok=True) try: _run_pip_install(bundle_dir, stage_dir, python_tag) _copy_bundle_support_files(bundle_dir, stage_dir) _run_smoke_check(stage_dir, manifest) if version_dir.exists(): backup_dir = paths.share_root / f".rollback-{manifest.version}-{int(time.time())}" _remove_path(backup_dir) os.replace(version_dir, backup_dir) os.replace(stage_dir, version_dir) _atomic_symlink(version_dir, paths.current_link) _atomic_write(paths.shim_path, _render_wrapper(paths), mode=0o755) _atomic_write(paths.service_path, _render_service(service_template, paths)) _write_state(paths, manifest, version_dir) cutover_done = True _run_systemctl(["daemon-reload"]) _run_systemctl(["enable", "--now", SERVICE_NAME]) except Exception: _remove_path(stage_dir) if cutover_done or backup_dir is not None: _rollback_install( paths=paths, manifest=manifest, old_state_text=old_state_text, old_service_text=old_service_text, old_shim_text=old_shim_text, old_current_target=old_current_target, new_version_dir=version_dir, backup_dir=backup_dir, ) else: _remove_path(stage_dir) raise if backup_dir is not None: _remove_path(backup_dir) _prune_versions(paths, manifest.version) print(f"installed {APP_NAME} {manifest.version} in {version_dir}") return 0 def uninstall_bundle(bundle_dir: Path, *, purge: bool) -> int: _ = bundle_dir paths = InstallPaths.detect() state = _load_state(paths.state_path) if state is None: raise PortableInstallError(f"no portable install state found at {paths.state_path}") if state.app_name != APP_NAME or state.install_kind != INSTALL_KIND: raise PortableInstallError(f"unexpected install state in {paths.state_path}") shim_text = _read_text_if_exists(paths.shim_path) if shim_text is not None and not _is_managed_text(shim_text): raise PortableInstallError(f"refusing to remove unmanaged shim at {paths.shim_path}") service_text = _read_text_if_exists(paths.service_path) if service_text is not None and not _is_managed_text(service_text): raise PortableInstallError(f"refusing to remove unmanaged service at {paths.service_path}") _run_systemctl(["disable", "--now", SERVICE_NAME], check=False) _remove_path(paths.service_path) _run_systemctl(["daemon-reload"], check=False) _remove_path(paths.shim_path) _remove_path(paths.share_root) if purge: _remove_path(paths.config_dir) _remove_path(paths.cache_dir) print(f"uninstalled {APP_NAME} portable bundle") return 0 def write_manifest(version: str, output_path: Path) -> int: manifest = Manifest.default(version) _atomic_write(output_path, json.dumps(asdict(manifest), indent=2, sort_keys=True) + "\n") return 0 def _parse_args(argv: list[str]) -> argparse.Namespace: parser = argparse.ArgumentParser(description="Aman portable bundle helper") subparsers = parser.add_subparsers(dest="command", required=True) install_parser = subparsers.add_parser("install", help="Install or upgrade the portable bundle") install_parser.add_argument("--bundle-dir", default=str(Path.cwd())) uninstall_parser = subparsers.add_parser("uninstall", help="Uninstall the portable bundle") uninstall_parser.add_argument("--bundle-dir", default=str(Path.cwd())) uninstall_parser.add_argument("--purge", action="store_true", help="Remove config and cache too") manifest_parser = subparsers.add_parser("write-manifest", help="Write the portable bundle manifest") manifest_parser.add_argument("--version", required=True) manifest_parser.add_argument("--output", required=True) return parser.parse_args(argv) def main(argv: list[str] | None = None) -> int: args = _parse_args(argv or sys.argv[1:]) try: if args.command == "install": return install_bundle(Path(args.bundle_dir).resolve()) if args.command == "uninstall": return uninstall_bundle(Path(args.bundle_dir).resolve(), purge=args.purge) if args.command == "write-manifest": return write_manifest(args.version, Path(args.output).resolve()) except PortableInstallError as exc: print(str(exc), file=sys.stderr) return 1 return 1 if __name__ == "__main__": raise SystemExit(main())