133 lines
3.8 KiB
Python
133 lines
3.8 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import sys
|
|
import traceback
|
|
from collections.abc import Callable, Sequence
|
|
from enum import IntEnum
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from .workspace import (
|
|
EvaluationWorkspace,
|
|
WorkspaceNotFoundError,
|
|
WorkspaceValidationError,
|
|
)
|
|
|
|
|
|
class ExitCode(IntEnum):
|
|
SUCCESS = 0
|
|
FAILURE = 1
|
|
INVALID_ARGUMENTS = 2
|
|
INVALID_WORKSPACE = 3
|
|
PARTIAL = 4
|
|
INTERRUPTED = 130
|
|
|
|
|
|
class CliError(Exception):
|
|
def __init__(self, message: str, exit_code: ExitCode = ExitCode.FAILURE) -> None:
|
|
self.exit_code = exit_code
|
|
super().__init__(message)
|
|
|
|
|
|
def standard_parser(description: str) -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(description=description)
|
|
parser.add_argument(
|
|
"--verbose",
|
|
action="store_true",
|
|
help="Show a traceback when an unexpected error occurs",
|
|
)
|
|
return parser
|
|
|
|
|
|
def evaluation_parser(description: str) -> argparse.ArgumentParser:
|
|
parser = standard_parser(description)
|
|
parser.add_argument(
|
|
"evaluation",
|
|
type=Path,
|
|
help="Evaluation directory",
|
|
)
|
|
return parser
|
|
|
|
|
|
def target_parser(description: str) -> argparse.ArgumentParser:
|
|
parser = standard_parser(description)
|
|
parser.add_argument(
|
|
"target",
|
|
type=Path,
|
|
help="Evaluation directory or nested file to process",
|
|
)
|
|
return parser
|
|
|
|
|
|
def evaluation_workspace(
|
|
path: str | Path,
|
|
*,
|
|
repository: str | Path | None = None,
|
|
) -> EvaluationWorkspace:
|
|
root = Path(path).expanduser().resolve()
|
|
if not root.exists():
|
|
raise CliError(
|
|
f"Evaluation directory does not exist: {root}",
|
|
ExitCode.INVALID_WORKSPACE,
|
|
)
|
|
if not root.is_dir():
|
|
raise CliError(
|
|
f"Evaluation path is not a directory: {root}",
|
|
ExitCode.INVALID_WORKSPACE,
|
|
)
|
|
return EvaluationWorkspace(root, Path(repository) if repository is not None else None)
|
|
|
|
|
|
def workspace_from_args(
|
|
args: argparse.Namespace,
|
|
*,
|
|
repository: str | Path | None = None,
|
|
) -> EvaluationWorkspace:
|
|
return evaluation_workspace(args.evaluation, repository=repository)
|
|
|
|
|
|
def workspace_from_target(
|
|
args: argparse.Namespace,
|
|
*,
|
|
repository: str | Path | None = None,
|
|
) -> tuple[EvaluationWorkspace, Path]:
|
|
target = Path(args.target).expanduser().resolve()
|
|
if not target.exists():
|
|
raise CliError(f"Target does not exist: {target}", ExitCode.INVALID_WORKSPACE)
|
|
repository_path = Path(repository) if repository is not None else None
|
|
if target.is_dir() and EvaluationWorkspace.looks_like_evaluation(target):
|
|
return EvaluationWorkspace(target, repository_path), target
|
|
workspace = EvaluationWorkspace.discover(target, repository=repository_path)
|
|
return workspace, target
|
|
|
|
|
|
def execute(
|
|
parser: argparse.ArgumentParser,
|
|
argv: Sequence[str] | None,
|
|
handler: Callable[[argparse.Namespace], int | ExitCode | None],
|
|
) -> int:
|
|
args = parser.parse_args(argv)
|
|
try:
|
|
result = handler(args)
|
|
except CliError as exc:
|
|
print(f"Error: {exc}", file=sys.stderr)
|
|
return int(exc.exit_code)
|
|
except (WorkspaceNotFoundError, WorkspaceValidationError) as exc:
|
|
print(f"Error: {exc}", file=sys.stderr)
|
|
return int(ExitCode.INVALID_WORKSPACE)
|
|
except KeyboardInterrupt:
|
|
print("Interrupted by user.", file=sys.stderr)
|
|
return int(ExitCode.INTERRUPTED)
|
|
except Exception as exc: # noqa: BLE001 - executable boundary
|
|
if getattr(args, "verbose", False):
|
|
traceback.print_exc()
|
|
else:
|
|
print(f"Error: {exc}", file=sys.stderr)
|
|
return int(ExitCode.FAILURE)
|
|
return int(ExitCode.SUCCESS if result is None else result)
|
|
|
|
|
|
def stderr(message: Any) -> None:
|
|
print(message, file=sys.stderr)
|