Spaces:
Sleeping
Sleeping
| from __future__ import annotations | |
| import json | |
| from pathlib import Path | |
| from typing import Any | |
| RUNS_DIR = Path(__file__).resolve().parents[2] / "runs" | |
| def run_dir(run_id: str) -> Path: | |
| path = RUNS_DIR / run_id | |
| path.mkdir(parents=True, exist_ok=True) | |
| return path | |
| def append_stage_trace(run_id: str, trace: dict[str, Any]) -> None: | |
| path = run_dir(run_id) / "stage_traces.jsonl" | |
| with path.open("a", encoding="utf-8") as handle: | |
| handle.write(json.dumps(trace, ensure_ascii=False) + "\n") | |
| def write_run_export(run_id: str, export: dict[str, Any]) -> None: | |
| path = run_dir(run_id) / "run_export.json" | |
| path.write_text(json.dumps(export, indent=2, ensure_ascii=False), encoding="utf-8") | |
| def read_stage_traces_jsonl(run_id: str) -> str: | |
| path = run_dir(run_id) / "stage_traces.jsonl" | |
| return path.read_text(encoding="utf-8") if path.exists() else "" | |