run_analyses.py

"""Run configured project analyses reproducibly on any supported platform."""

from __future__ import annotations

import argparse
import os
import re
import shlex
import subprocess
import sys
import threading
import time
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import Any

try:
    import tomllib
except ModuleNotFoundError as exc:  # pragma: no cover - Python 3.10 and older
    raise SystemExit("error: run_analyses.py requires Python 3.11 or newer") from exc


@dataclass(frozen=True)
class Analysis:
    name: str
    description: str
    enabled: bool
    kind: str
    script: str
    args: tuple[str, ...]
    timeout_seconds: float | None
    working_directory: Path


def positive_seconds(value: str) -> float:
    parsed = float(value)
    if parsed <= 0:
        raise argparse.ArgumentTypeError("must be positive")
    return parsed


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Run analyses declared in a cross-platform TOML manifest."
    )
    parser.add_argument(
        "--config",
        type=Path,
        default=Path(__file__).with_name("analyses.toml"),
        help="TOML manifest (default: analyses.toml beside this script)",
    )
    selection = parser.add_mutually_exclusive_group()
    selection.add_argument(
        "--only",
        nargs="+",
        metavar="NAME",
        help="Run only these analyses, even if disabled in the manifest",
    )
    selection.add_argument(
        "--all",
        action="store_true",
        help="Run every analysis, including disabled analyses",
    )
    parser.add_argument(
        "--skip", nargs="+", default=[], metavar="NAME", help="Skip these analyses"
    )
    parser.add_argument("--list", action="store_true", help="List analyses and exit")
    parser.add_argument(
        "--dry-run", action="store_true", help="Print commands without running them"
    )
    parser.add_argument(
        "--continue-on-error",
        action="store_true",
        help="Continue after an analysis fails or times out",
    )
    parser.add_argument(
        "--show-output",
        action="store_true",
        help="Stream child output in addition to saving it in a log",
    )
    parser.add_argument(
        "--no-logs", action="store_true", help="Do not write per-analysis log files"
    )
    parser.add_argument("--log-dir", type=Path, help="Override the manifest log directory")
    parser.add_argument(
        "--timeout",
        type=positive_seconds,
        help="Override every selected analysis timeout, in seconds",
    )
    parser.add_argument(
        "--rscript", default="Rscript", help="Rscript executable (default: Rscript)"
    )
    return parser.parse_args()


def require_type(value: Any, expected: type, location: str) -> Any:
    if not isinstance(value, expected):
        raise ValueError(f"{location} must be {expected.__name__}")
    return value


def load_manifest(path: Path) -> tuple[dict[str, Any], list[Analysis]]:
    path = path.resolve()
    if not path.is_file():
        raise FileNotFoundError(f"configuration file does not exist: {path}")
    with path.open("rb") as handle:
        document = tomllib.load(handle)
    runner = document.get("runner", {})
    require_type(runner, dict, "[runner]")
    raw_analyses = document.get("analysis", [])
    require_type(raw_analyses, list, "[[analysis]]")
    base_directory = path.parent
    analyses: list[Analysis] = []
    names: set[str] = set()
    for number, raw in enumerate(raw_analyses, start=1):
        location = f"analysis #{number}"
        require_type(raw, dict, location)
        name = require_type(raw.get("name"), str, f"{location}.name")
        if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_.-]*", name):
            raise ValueError(f"{location}.name contains unsupported characters: {name}")
        if name in names:
            raise ValueError(f"duplicate analysis name: {name}")
        names.add(name)
        kind = require_type(raw.get("kind", "python"), str, f"{name}.kind")
        if kind not in {"python", "r", "command"}:
            raise ValueError(f"{name}.kind must be python, r, or command")
        script = require_type(raw.get("script"), str, f"{name}.script")
        arguments = require_type(raw.get("args", []), list, f"{name}.args")
        if any(not isinstance(value, (str, int, float)) for value in arguments):
            raise ValueError(f"{name}.args values must be strings or numbers")
        raw_timeout = raw.get("timeout_seconds")
        timeout = None if raw_timeout in (None, 0) else float(raw_timeout)
        if timeout is not None and timeout <= 0:
            raise ValueError(f"{name}.timeout_seconds must be positive or zero")
        raw_cwd = require_type(
            raw.get("working_directory", "."), str, f"{name}.working_directory"
        )
        cwd = Path(raw_cwd)
        if not cwd.is_absolute():
            cwd = base_directory / cwd
        analyses.append(
            Analysis(
                name=name,
                description=require_type(
                    raw.get("description", ""), str, f"{name}.description"
                ),
                enabled=require_type(raw.get("enabled", False), bool, f"{name}.enabled"),
                kind=kind,
                script=script,
                args=tuple(str(value) for value in arguments),
                timeout_seconds=timeout,
                working_directory=cwd.resolve(),
            )
        )
    if not analyses:
        raise ValueError("the manifest contains no [[analysis]] entries")
    return runner, analyses


def select_analyses(analyses: list[Analysis], args: argparse.Namespace) -> list[Analysis]:
    known = {analysis.name for analysis in analyses}
    requested = set(args.only or []) | set(args.skip)
    unknown = sorted(requested - known)
    if unknown:
        raise ValueError("unknown analysis name(s): " + ", ".join(unknown))
    if args.only:
        selected_names = set(args.only)
    elif args.all:
        selected_names = known
    else:
        selected_names = {analysis.name for analysis in analyses if analysis.enabled}
    selected_names -= set(args.skip)
    return [analysis for analysis in analyses if analysis.name in selected_names]


def analysis_command(analysis: Analysis, rscript: str) -> list[str]:
    if analysis.kind == "python":
        prefix = [sys.executable]
    elif analysis.kind == "r":
        prefix = [rscript]
    else:
        prefix = []
    return [*prefix, analysis.script, *analysis.args]


def display_command(command: list[str]) -> str:
    if os.name == "nt":
        return subprocess.list2cmdline(command)
    return shlex.join(command)


def safe_log_name(name: str) -> str:
    return re.sub(r"[^A-Za-z0-9_.-]+", "_", name)


def run_analysis(
    analysis: Analysis,
    command: list[str],
    log_path: Path | None,
    show_output: bool,
    timeout: float | None,
) -> tuple[str, int | None, float]:
    started = time.perf_counter()
    log_handle = log_path.open("w", encoding="utf-8", newline="") if log_path else None
    header = (
        f"Analysis: {analysis.name}\n"
        f"Working directory: {analysis.working_directory}\n"
        f"Command: {display_command(command)}\n\n"
    )
    if log_handle:
        log_handle.write(header)
        log_handle.flush()
    process: subprocess.Popen[str] | None = None
    status = "failed"
    return_code: int | None = None
    try:
        process = subprocess.Popen(
            command,
            cwd=analysis.working_directory,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            text=True,
            encoding="utf-8",
            errors="replace",
            bufsize=1,
        )

        def copy_output() -> None:
            assert process is not None and process.stdout is not None
            for line in process.stdout:
                if log_handle:
                    log_handle.write(line)
                    log_handle.flush()
                if show_output:
                    print(line, end="", flush=True)

        reader = threading.Thread(target=copy_output, daemon=True)
        reader.start()
        try:
            return_code = process.wait(timeout=timeout)
            reader.join()
            status = "passed" if return_code == 0 else "failed"
        except subprocess.TimeoutExpired:
            status = "timed out"
            process.terminate()
            try:
                process.wait(timeout=5)
            except subprocess.TimeoutExpired:
                process.kill()
                process.wait()
            reader.join(timeout=5)
            return_code = process.returncode
    except FileNotFoundError as exc:
        message = f"error: {exc}\n"
        if log_handle:
            log_handle.write(message)
        if show_output:
            print(message, end="", file=sys.stderr)
    except KeyboardInterrupt:
        if process is not None and process.poll() is None:
            process.terminate()
        raise
    finally:
        if log_handle:
            log_handle.close()
    return status, return_code, time.perf_counter() - started


def main() -> int:
    args = parse_args()
    runner, analyses = load_manifest(args.config)
    if args.list:
        print("enabled  kind     name                         description")
        for analysis in analyses:
            print(
                f"{'yes' if analysis.enabled else 'no ':7}  "
                f"{analysis.kind:7}  {analysis.name:27}  {analysis.description}"
            )
        return 0
    selected = select_analyses(analyses, args)
    if not selected:
        print("No analyses selected. Enable jobs in the manifest or use --only/--all.")
        return 0

    config_directory = args.config.resolve().parent
    raw_log_dir = args.log_dir or Path(runner.get("log_directory", "logs"))
    log_directory = raw_log_dir if raw_log_dir.is_absolute() else config_directory / raw_log_dir
    show_output = args.show_output or bool(runner.get("show_output", False))
    continue_on_error = args.continue_on_error or bool(
        runner.get("continue_on_error", False)
    )
    timestamp = datetime.now().astimezone().strftime("%Y%m%d_%H%M%S")
    if not args.no_logs and not args.dry_run:
        log_directory.mkdir(parents=True, exist_ok=True)

    print(f"Selected {len(selected)} analysis job(s).")
    results: list[tuple[str, str, int | None, float, Path | None]] = []
    overall_start = time.perf_counter()
    for position, analysis in enumerate(selected, start=1):
        command = analysis_command(analysis, args.rscript)
        timeout = args.timeout if args.timeout is not None else analysis.timeout_seconds
        log_path = None
        if not args.no_logs and not args.dry_run:
            log_path = log_directory / f"{timestamp}_{safe_log_name(analysis.name)}.log"
        print(f"\n[{position}/{len(selected)}] {analysis.name}")
        print(f"  {display_command(command)}")
        if args.dry_run:
            results.append((analysis.name, "dry run", None, 0.0, None))
            continue
        status, return_code, elapsed = run_analysis(
            analysis, command, log_path, show_output, timeout
        )
        results.append((analysis.name, status, return_code, elapsed, log_path))
        detail = f"exit {return_code}" if return_code is not None else "not started"
        print(f"  {status}: {detail}; {elapsed:.2f} seconds")
        if log_path:
            print(f"  log: {log_path}")
        if status != "passed" and not continue_on_error:
            print("Stopping after failure; use --continue-on-error to run remaining jobs.")
            break

    print("\nRun summary:")
    for name, status, return_code, elapsed, _ in results:
        code = "" if return_code is None else f" (exit {return_code})"
        print(f"  {name}: {status}{code}; {elapsed:.2f} seconds")
    print(f"Overall elapsed: {time.perf_counter() - overall_start:.2f} seconds")
    return 1 if any(status in {"failed", "timed out"} for _, status, _, _, _ in results) else 0


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except (FileNotFoundError, OSError, ValueError, tomllib.TOMLDecodeError) as exc:
        print(f"error: {exc}", file=sys.stderr)
        raise SystemExit(1)