Spaces:
Running
Running
File size: 5,434 Bytes
5063745 8250dbb 99ee5de 5063745 8250dbb 5063745 8250dbb 5063745 62f5d66 99ee5de 62f5d66 99ee5de 62f5d66 5063745 99ee5de 5063745 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 | """Background scheduler: periodically runs live TS-Bench + TSFM.ai evaluation."""
from __future__ import annotations
import json
import logging
import os
import subprocess
import sys
import threading
import time
from datetime import datetime, timezone
from pathlib import Path
from src.benchmark_config import BENCHMARK_INTERVAL_SECONDS, LIVE_EVAL_ENABLED
from src.eval_schedule import compute_next_eval_at_unix, with_next_eval_fields
logger = logging.getLogger(__name__)
SPACE_ROOT = Path(__file__).resolve().parents[1]
EVAL_SCRIPT = SPACE_ROOT / "scripts" / "run_space_eval.py"
DEFAULT_EVAL_INTERVAL = BENCHMARK_INTERVAL_SECONDS
class LiveEvalScheduler:
def __init__(
self,
interval_seconds: int = DEFAULT_EVAL_INTERVAL,
run_on_startup: bool = True,
):
self.interval_seconds = max(300, interval_seconds)
self.run_on_startup = run_on_startup
self._lock = threading.Lock()
self._stop = threading.Event()
self._thread: threading.Thread | None = None
self._next_run_at: float | None = None
@property
def enabled(self) -> bool:
return LIVE_EVAL_ENABLED and bool(os.getenv("TSFM_API_KEY"))
def start(self) -> None:
if self._thread is not None:
return
if not LIVE_EVAL_ENABLED:
logger.info(
"UI-only mode (TSFM_LIVE_EVAL=0): scheduler disabled; "
"leaderboard reads pushed results/."
)
return
if not os.getenv("TSFM_API_KEY"):
logger.warning(
"Live eval disabled: TSFM_API_KEY not set. "
"Leaderboard will only show static CSV results."
)
self._write_idle_state(
"TSFM_API_KEY not set — configure it in HF Space Settings → Secrets."
)
return
self._thread = threading.Thread(target=self._loop, name="live-eval-scheduler", daemon=True)
self._thread.start()
logger.info(
"Live eval scheduler started (interval=%ss, script=%s)",
self.interval_seconds,
EVAL_SCRIPT,
)
def snapshot(self) -> dict:
state_path = SPACE_ROOT / "results" / "eval_state.json"
if state_path.exists():
try:
return json.loads(state_path.read_text())
except json.JSONDecodeError:
pass
return {
"status": "idle" if self.enabled else "disabled",
"message": "Waiting for first evaluation run…" if self.enabled else "No API key",
}
def seconds_until_next_run(self) -> int | None:
if self._next_run_at is None:
return None
return max(0, int(self._next_run_at - time.monotonic()))
def next_eval_at_unix(self) -> int:
state = self.snapshot()
persisted = compute_next_eval_at_unix(state, self.interval_seconds)
remaining = self.seconds_until_next_run()
if remaining is None:
return persisted
in_memory = int(time.time()) + remaining
return min(persisted, in_memory)
def is_running(self) -> bool:
return self.snapshot().get("status") == "running"
def _loop(self) -> None:
if self.run_on_startup:
self._run_once()
while not self._stop.wait(self.interval_seconds):
self._run_once()
def _run_once(self) -> None:
if not self._lock.acquire(blocking=False):
logger.info("Evaluation already in progress, skipping tick")
return
try:
logger.info("Starting scheduled live evaluation")
cmd = [sys.executable, str(EVAL_SCRIPT)]
completed = subprocess.run(cmd, cwd=str(SPACE_ROOT), check=False)
if completed.returncode != 0:
logger.error("Evaluation exited with code %s", completed.returncode)
finally:
self._lock.release()
self._next_run_at = time.monotonic() + self.interval_seconds
self._persist_next_eval_after_run()
def _persist_next_eval_after_run(self) -> None:
state_path = SPACE_ROOT / "results" / "eval_state.json"
if not state_path.exists():
return
try:
state = json.loads(state_path.read_text())
except json.JSONDecodeError:
return
if state.get("status") == "running":
return
anchor = state.get("finished_at") or state.get("updated_at")
if not anchor:
return
enriched = with_next_eval_fields(state, anchor, self.interval_seconds)
state_path.write_text(json.dumps(enriched, indent=4) + "\n")
def _write_idle_state(self, message: str) -> None:
results = SPACE_ROOT / "results"
results.mkdir(parents=True, exist_ok=True)
payload = {
"status": "disabled",
"message": message,
"updated_at": datetime.now(timezone.utc).isoformat(),
}
(results / "eval_state.json").write_text(json.dumps(payload, indent=4) + "\n")
_scheduler: LiveEvalScheduler | None = None
def get_scheduler() -> LiveEvalScheduler:
global _scheduler
if _scheduler is None:
_scheduler = LiveEvalScheduler()
return _scheduler
def start_live_eval_scheduler() -> LiveEvalScheduler:
scheduler = get_scheduler()
scheduler.start()
return scheduler
|