| """Per-run provenance record: git hash, config, package versions, GPU, timestamps. |
| |
| Spec §1.4: "Log: git hash, config YAML, package versions (pip freeze), GPU model, |
| start/end time." Every script entrypoint should call `start_run` at the top and |
| `finish_run` at the bottom, writing the result into that run's results/ directory. |
| """ |
| from __future__ import annotations |
|
|
| import json |
| import subprocess |
| import sys |
| from datetime import datetime, timezone |
| from pathlib import Path |
| from typing import Any |
|
|
|
|
| def _git_hash() -> str: |
| try: |
| return subprocess.check_output( |
| ["git", "rev-parse", "HEAD"], stderr=subprocess.DEVNULL |
| ).decode().strip() |
| except Exception: |
| return "unknown" |
|
|
|
|
| def _gpu_model() -> str: |
| try: |
| out = subprocess.check_output( |
| ["nvidia-smi", "--query-gpu=name", "--format=csv,noheader"], |
| stderr=subprocess.DEVNULL, |
| ).decode().strip() |
| return out.replace("\n", "; ") |
| except Exception: |
| return "no-gpu" |
|
|
|
|
| def _pip_freeze() -> list[str]: |
| try: |
| return subprocess.check_output( |
| [sys.executable, "-m", "pip", "freeze"], stderr=subprocess.DEVNULL |
| ).decode().splitlines() |
| except Exception: |
| return [] |
|
|
|
|
| def start_run(config: dict[str, Any], seed: int, out_dir: str | Path) -> dict[str, Any]: |
| out_dir = Path(out_dir) |
| out_dir.mkdir(parents=True, exist_ok=True) |
| record = { |
| "git_hash": _git_hash(), |
| "config": config, |
| "seed": seed, |
| "gpu": _gpu_model(), |
| "start_time": datetime.now(timezone.utc).isoformat(), |
| "argv": sys.argv, |
| } |
| (out_dir / "pip_freeze.txt").write_text("\n".join(_pip_freeze())) |
| (out_dir / "run_record.json").write_text(json.dumps(record, indent=2)) |
| return record |
|
|
|
|
| def finish_run(out_dir: str | Path, record: dict[str, Any]) -> None: |
| out_dir = Path(out_dir) |
| record["end_time"] = datetime.now(timezone.utc).isoformat() |
| (out_dir / "run_record.json").write_text(json.dumps(record, indent=2)) |
|
|
