98 lines
4 KiB
Python
98 lines
4 KiB
Python
"""Command-line entry point for live and baseline evaluation."""
|
|
|
|
import argparse
|
|
import asyncio
|
|
import os
|
|
from pathlib import Path
|
|
|
|
from baseline import record_baselines, write_comparison
|
|
from live import LiveLimits, ScenarioResult, run_live_scenarios
|
|
from scenario import SCENARIO_PATH, load_scenarios
|
|
|
|
EVAL_ROOT = Path(__file__).parent
|
|
|
|
|
|
def main() -> int:
|
|
"""Run the selected evaluation arms and return a process status."""
|
|
arguments = _parse_arguments()
|
|
scenarios = load_scenarios(arguments.scenarios)
|
|
exit_code = 0
|
|
if arguments.base_url is None:
|
|
raise SystemExit("--base-url is required")
|
|
limits = LiveLimits(
|
|
minimum_tracks=arguments.min_tracks,
|
|
maximum_tracks=arguments.max_tracks,
|
|
minimum_artists=arguments.min_artists,
|
|
first_track_budget_ms=arguments.first_track_budget_ms,
|
|
total_budget_ms=arguments.total_budget_ms,
|
|
)
|
|
results = asyncio.run(
|
|
run_live_scenarios(scenarios, arguments.base_url, arguments.report_dir, limits)
|
|
)
|
|
_print_summary(results)
|
|
if not all(result.passed for result in results):
|
|
exit_code = 1
|
|
|
|
if arguments.baseline:
|
|
client_id = os.environ.get("SPOTIFY_CLIENT_ID")
|
|
client_secret = os.environ.get("SPOTIFY_CLIENT_SECRET")
|
|
if not client_id or not client_secret:
|
|
print("Spotify client credentials are absent; skipping baseline.")
|
|
else:
|
|
baseline_root = arguments.snapshot_dir / "baseline"
|
|
asyncio.run(record_baselines(scenarios, client_id, client_secret, baseline_root))
|
|
comparison_path = write_comparison(
|
|
scenarios,
|
|
arguments.snapshot_dir / "baseline",
|
|
arguments.report_dir,
|
|
arguments.comparison_top_n,
|
|
)
|
|
print(f"Comparison: {comparison_path}")
|
|
return exit_code
|
|
|
|
|
|
def _parse_arguments() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--base-url")
|
|
parser.add_argument("--baseline", action="store_true")
|
|
parser.add_argument("--scenarios", type=Path, default=SCENARIO_PATH)
|
|
parser.add_argument("--snapshot-dir", type=Path, default=EVAL_ROOT / "snapshots")
|
|
parser.add_argument("--report-dir", type=Path, default=EVAL_ROOT / "reports")
|
|
parser.add_argument("--min-tracks", type=int, default=8)
|
|
parser.add_argument("--max-tracks", type=int, default=15)
|
|
parser.add_argument("--min-artists", type=int, default=5)
|
|
parser.add_argument("--first-track-budget-ms", type=int, default=20_000)
|
|
parser.add_argument("--total-budget-ms", type=int, default=30_000)
|
|
parser.add_argument("--comparison-top-n", type=int, default=10)
|
|
arguments = parser.parse_args()
|
|
if arguments.min_tracks < 0 or arguments.max_tracks < arguments.min_tracks:
|
|
parser.error("track bounds are invalid")
|
|
if min(arguments.min_artists, arguments.first_track_budget_ms, arguments.total_budget_ms) < 0:
|
|
parser.error("artist and latency limits must be non-negative")
|
|
if arguments.comparison_top_n < 1:
|
|
parser.error("comparison top N must be positive")
|
|
return arguments
|
|
|
|
|
|
def _print_summary(results: tuple[ScenarioResult, ...]) -> None:
|
|
print("| Scenario | Status | Tracks | Artists | First track | Total |")
|
|
print("| --- | --- | ---: | ---: | ---: | ---: |")
|
|
for result in results:
|
|
artists = {str(artist).casefold() for event in result.tracks for artist in _artists(event)}
|
|
first = f"{result.first_track_ms} ms" if result.first_track_ms is not None else "n/a"
|
|
total = f"{result.total_ms} ms" if result.total_ms is not None else "n/a"
|
|
status = "PASS" if result.passed else "FAIL"
|
|
print(
|
|
f"| {result.scenario.key} | {status} | {len(result.tracks)} | "
|
|
f"{len(artists)} | {first} | {total} |"
|
|
)
|
|
|
|
|
|
def _artists(event: dict[str, object]) -> list[object]:
|
|
track = event.get("track")
|
|
artists = track.get("artists") if isinstance(track, dict) else None
|
|
return artists if isinstance(artists, list) else []
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|