Spaces:
Running
Running
File size: 1,883 Bytes
99ee5de | 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 | """Persisted schedule for the next live evaluation cycle."""
from __future__ import annotations
from datetime import datetime, timedelta, timezone
from src.benchmark_config import BENCHMARK_INTERVAL_SECONDS
def parse_iso_to_unix(value: str) -> int:
normalized = value.replace("Z", "+00:00")
return int(datetime.fromisoformat(normalized).timestamp())
def compute_next_eval_at_iso(finished_at: str, interval_seconds: int | None = None) -> str:
interval = interval_seconds or BENCHMARK_INTERVAL_SECONDS
finished_dt = datetime.fromisoformat(finished_at.replace("Z", "+00:00"))
if finished_dt.tzinfo is None:
finished_dt = finished_dt.replace(tzinfo=timezone.utc)
return (finished_dt + timedelta(seconds=interval)).isoformat()
def compute_next_eval_at_unix(
state: dict,
interval_seconds: int | None = None,
) -> int:
"""Absolute unix time for the next eval; survives process restarts and page reloads."""
interval = interval_seconds or BENCHMARK_INTERVAL_SECONDS
if state.get("status") == "running":
return int(datetime.now(timezone.utc).timestamp())
if raw_unix := state.get("next_eval_at_unix"):
return int(raw_unix)
if next_at := state.get("next_eval_at"):
return parse_iso_to_unix(next_at)
anchor = state.get("finished_at") or state.get("updated_at")
if anchor:
return parse_iso_to_unix(anchor) + interval
return int(datetime.now(timezone.utc).timestamp()) + interval
def with_next_eval_fields(
payload: dict,
finished_at: str,
interval_seconds: int | None = None,
) -> dict:
interval = interval_seconds or BENCHMARK_INTERVAL_SECONDS
next_at = compute_next_eval_at_iso(finished_at, interval)
enriched = dict(payload)
enriched["next_eval_at"] = next_at
enriched["next_eval_at_unix"] = parse_iso_to_unix(next_at)
return enriched
|