Initial workspace launcher
This commit is contained in:
mpabi
292
workspace.py
Executable file
292
workspace.py
Executable file
@@ -0,0 +1,292 @@
|
||||
#!/usr/bin/env python3
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import shlex
|
||||
import subprocess
|
||||
import sys
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def expand_path(raw_path: str, base_dir: Path) -> Path:
|
||||
expanded = os.path.expandvars(os.path.expanduser(raw_path))
|
||||
path = Path(expanded)
|
||||
if not path.is_absolute():
|
||||
path = base_dir / path
|
||||
return path.resolve()
|
||||
|
||||
|
||||
@dataclass
|
||||
class WorkspaceConfig:
|
||||
config_path: Path
|
||||
workspace_root: Path
|
||||
series_root: Path
|
||||
socket_root: Path
|
||||
tools_root_candidates: list[Path]
|
||||
tools_root: Path
|
||||
defaults: dict[str, str]
|
||||
|
||||
|
||||
def load_config(config_path: Path) -> WorkspaceConfig:
|
||||
raw = json.loads(config_path.read_text(encoding="utf-8"))
|
||||
base_dir = config_path.parent
|
||||
|
||||
workspace_root = expand_path(raw.get("workspace_root", "~/dev/workspace/rv"), base_dir)
|
||||
series_root = expand_path(raw.get("series_root", str(workspace_root / "series")), base_dir)
|
||||
socket_root = expand_path(raw.get("socket_root", str(workspace_root / "sockets")), base_dir)
|
||||
|
||||
candidate_values = raw.get("tools_root_candidates")
|
||||
if not candidate_values:
|
||||
candidate_values = [raw.get("tools_root", "~/dev/workspace/tools/rv32i-hazard3-env")]
|
||||
tools_root_candidates = [expand_path(value, base_dir) for value in candidate_values]
|
||||
|
||||
tools_root = next((path for path in tools_root_candidates if path.exists()), tools_root_candidates[0])
|
||||
defaults = raw.get("defaults", {})
|
||||
|
||||
return WorkspaceConfig(
|
||||
config_path=config_path,
|
||||
workspace_root=workspace_root,
|
||||
series_root=series_root,
|
||||
socket_root=socket_root,
|
||||
tools_root_candidates=tools_root_candidates,
|
||||
tools_root=tools_root,
|
||||
defaults=defaults,
|
||||
)
|
||||
|
||||
|
||||
def series_directories(config: WorkspaceConfig) -> list[Path]:
|
||||
if not config.series_root.is_dir():
|
||||
raise SystemExit(f"Missing series root: {config.series_root}")
|
||||
return sorted(path for path in config.series_root.iterdir() if path.is_dir())
|
||||
|
||||
|
||||
def card_directories(series_path: Path) -> list[Path]:
|
||||
return sorted(path for path in series_path.iterdir() if path.is_dir())
|
||||
|
||||
|
||||
def read_card_title(card_path: Path) -> str:
|
||||
readme_path = card_path / "README.md"
|
||||
if not readme_path.is_file():
|
||||
return ""
|
||||
|
||||
for line in readme_path.read_text(encoding="utf-8").splitlines():
|
||||
stripped = line.strip()
|
||||
if stripped.startswith("#"):
|
||||
return stripped.lstrip("#").strip()
|
||||
return ""
|
||||
|
||||
|
||||
def resolve_selector(
|
||||
config: WorkspaceConfig,
|
||||
series_arg: str | None,
|
||||
card_arg: str | None,
|
||||
) -> tuple[str, str]:
|
||||
default_series = config.defaults.get("series")
|
||||
default_card = config.defaults.get("card")
|
||||
|
||||
if series_arg and "/" in series_arg:
|
||||
if card_arg:
|
||||
raise SystemExit("Pass either '<series> <card>' or '<series/card>', not both forms at once.")
|
||||
series_value, card_value = series_arg.split("/", 1)
|
||||
return series_value, card_value
|
||||
|
||||
if series_arg and not card_arg:
|
||||
if (config.series_root / series_arg).is_dir():
|
||||
if not default_card:
|
||||
raise SystemExit("Missing default card in config.")
|
||||
return series_arg, default_card
|
||||
if default_series and (config.series_root / default_series / series_arg).is_dir():
|
||||
return default_series, series_arg
|
||||
|
||||
series_value = series_arg or default_series
|
||||
card_value = card_arg or default_card
|
||||
if not series_value or not card_value:
|
||||
raise SystemExit("Missing series/card selector and no defaults are configured.")
|
||||
return series_value, card_value
|
||||
|
||||
|
||||
def resolve_card_path(config: WorkspaceConfig, series_name: str, card_name: str) -> Path:
|
||||
card_path = config.series_root / series_name / card_name
|
||||
if not card_path.is_dir():
|
||||
raise SystemExit(f"Missing card path: {card_path}")
|
||||
return card_path
|
||||
|
||||
|
||||
def print_series(config: WorkspaceConfig) -> None:
|
||||
for series_path in series_directories(config):
|
||||
card_count = len(card_directories(series_path))
|
||||
print(f"{series_path.name}\t{card_count}\t{series_path}")
|
||||
|
||||
|
||||
def print_cards(config: WorkspaceConfig, series_arg: str | None) -> None:
|
||||
series_name = series_arg or config.defaults.get("series")
|
||||
if not series_name:
|
||||
raise SystemExit("Missing series name and no default series is configured.")
|
||||
|
||||
series_path = config.series_root / series_name
|
||||
if not series_path.is_dir():
|
||||
raise SystemExit(f"Missing series path: {series_path}")
|
||||
|
||||
for card_path in card_directories(series_path):
|
||||
title = read_card_title(card_path)
|
||||
if title:
|
||||
print(f"{card_path.name}\t{title}\t{card_path}")
|
||||
else:
|
||||
print(f"{card_path.name}\t{card_path}")
|
||||
|
||||
|
||||
def tmux_target_exists(session_name: str) -> bool:
|
||||
result = subprocess.run(
|
||||
["tmux", "has-session", "-t", session_name],
|
||||
stdout=subprocess.DEVNULL,
|
||||
stderr=subprocess.DEVNULL,
|
||||
check=False,
|
||||
)
|
||||
return result.returncode == 0
|
||||
|
||||
|
||||
def build_container_command(
|
||||
config: WorkspaceConfig,
|
||||
selector: str,
|
||||
card_path: Path,
|
||||
instance: str,
|
||||
) -> str:
|
||||
tool_root = config.tools_root
|
||||
rv_script = tool_root / "rv"
|
||||
if not rv_script.is_file():
|
||||
candidates = "\n".join(f" - {path}" for path in config.tools_root_candidates)
|
||||
raise SystemExit(f"Missing rv launcher. Checked:\n{candidates}")
|
||||
|
||||
env_parts = [
|
||||
f"WORKSPACE_ROOT={shlex.quote(str(config.workspace_root))}",
|
||||
f"RV_REPO_PATH={shlex.quote(str(card_path))}",
|
||||
f"RV_CARD={shlex.quote(selector)}",
|
||||
f"RV_INSTANCE={shlex.quote(instance)}",
|
||||
]
|
||||
return " && ".join(
|
||||
[
|
||||
f"cd {shlex.quote(str(tool_root))}",
|
||||
" ".join(env_parts + ["bash", "./rv", "shell"]),
|
||||
]
|
||||
)
|
||||
|
||||
|
||||
def run_tmux_container(config: WorkspaceConfig, args: argparse.Namespace) -> None:
|
||||
series_name, card_name = resolve_selector(config, args.series, args.card)
|
||||
selector = f"{series_name}/{card_name}"
|
||||
card_path = resolve_card_path(config, series_name, card_name)
|
||||
|
||||
session_name = args.session or config.defaults.get("tmux_session", "rv-workspace")
|
||||
window_name = args.window or config.defaults.get("tmux_window", "rv")
|
||||
instance = args.instance or config.defaults.get("instance", "shell")
|
||||
|
||||
if tmux_target_exists(session_name):
|
||||
raise SystemExit(
|
||||
f"tmux session already exists: {session_name}\n"
|
||||
"Use another --session name or remove the existing session first."
|
||||
)
|
||||
|
||||
shell_command = build_container_command(config, selector, card_path, instance)
|
||||
tmux_command = [
|
||||
"tmux",
|
||||
"new-session",
|
||||
"-d",
|
||||
"-s",
|
||||
session_name,
|
||||
"-n",
|
||||
window_name,
|
||||
f"bash -lc {shlex.quote(shell_command)}",
|
||||
]
|
||||
|
||||
if args.dry_run:
|
||||
print("Selector:", selector)
|
||||
print("Card path:", card_path)
|
||||
print("Tools root:", config.tools_root)
|
||||
print("Pane 0 command:")
|
||||
print(shell_command)
|
||||
print("tmux command:")
|
||||
print(shlex.join(tmux_command))
|
||||
return
|
||||
|
||||
subprocess.run(tmux_command, check=True)
|
||||
|
||||
if args.attach:
|
||||
subprocess.run(["tmux", "attach", "-t", session_name], check=True)
|
||||
return
|
||||
|
||||
print(f"Created tmux session: {session_name}")
|
||||
print(f"Window: {window_name}")
|
||||
print("Pane 0: container shell")
|
||||
print(f"Attach with: tmux attach -t {session_name}")
|
||||
|
||||
|
||||
def print_config(config: WorkspaceConfig) -> None:
|
||||
print(f"config_path\t{config.config_path}")
|
||||
print(f"workspace_root\t{config.workspace_root}")
|
||||
print(f"series_root\t{config.series_root}")
|
||||
print(f"socket_root\t{config.socket_root}")
|
||||
print(f"tools_root\t{config.tools_root}")
|
||||
print("tools_root_candidates")
|
||||
for path in config.tools_root_candidates:
|
||||
print(f" {path}")
|
||||
|
||||
|
||||
def build_parser() -> argparse.ArgumentParser:
|
||||
parser = argparse.ArgumentParser(description="Workspace launcher for rv cards and tools.")
|
||||
parser.add_argument(
|
||||
"--config",
|
||||
default=Path(__file__).with_name("workspace.json"),
|
||||
type=Path,
|
||||
help="Path to workspace.json",
|
||||
)
|
||||
|
||||
subparsers = parser.add_subparsers(dest="command", required=True)
|
||||
|
||||
subparsers.add_parser("show-config", help="Print resolved workspace paths.")
|
||||
subparsers.add_parser("list-series", help="List series directories from series_root.")
|
||||
|
||||
cards_parser = subparsers.add_parser("list-cards", help="List cards in a series.")
|
||||
cards_parser.add_argument("series", nargs="?", help="Series id, for example 'inf'.")
|
||||
|
||||
tmux_parser = subparsers.add_parser(
|
||||
"tmux-container",
|
||||
help="Create a tmux session with pane 0 running the container shell for a selected card.",
|
||||
)
|
||||
tmux_parser.add_argument("series", nargs="?", help="Series id or full selector like 'inf/03'.")
|
||||
tmux_parser.add_argument("card", nargs="?", help="Card id, for example '03'.")
|
||||
tmux_parser.add_argument("--session", help="tmux session name.")
|
||||
tmux_parser.add_argument("--window", help="tmux window name.")
|
||||
tmux_parser.add_argument("--instance", help="RV_INSTANCE passed to the tool launcher.")
|
||||
tmux_parser.add_argument("--attach", action="store_true", help="Attach to the session after creation.")
|
||||
tmux_parser.add_argument("--dry-run", action="store_true", help="Print commands instead of running tmux.")
|
||||
|
||||
return parser
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = build_parser()
|
||||
args = parser.parse_args()
|
||||
config = load_config(args.config)
|
||||
|
||||
if args.command == "show-config":
|
||||
print_config(config)
|
||||
return 0
|
||||
if args.command == "list-series":
|
||||
print_series(config)
|
||||
return 0
|
||||
if args.command == "list-cards":
|
||||
print_cards(config, args.series)
|
||||
return 0
|
||||
if args.command == "tmux-container":
|
||||
run_tmux_container(config, args)
|
||||
return 0
|
||||
|
||||
parser.error(f"Unsupported command: {args.command}")
|
||||
return 2
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
Reference in New Issue
Block a user