scripts/run_online_daemon.py

#!/usr/bin/env python3
"""Local 1-hour loop: run eval locally, then push results to HF Space."""

from __future__ import annotations

import argparse
import logging
import os
import subprocess
import sys
import time
from pathlib import Path

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
logger = logging.getLogger(__name__)

SPACE_ROOT = Path(__file__).resolve().parents[1]
EVAL_SCRIPT = SPACE_ROOT / "scripts" / "run_space_eval.py"
PUSH_SCRIPT = SPACE_ROOT / "scripts" / "push_results_to_hf.py"
DEFAULT_INTERVAL_SECONDS = int(
    os.getenv(
        "TSFM_BENCH_INTERVAL_SECONDS",
        os.getenv("TSFM_EVAL_INTERVAL_SECONDS", "3600"),
    )
)


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--interval-minutes",
        type=int,
        default=max(1, DEFAULT_INTERVAL_SECONDS // 60),
        help="Minutes between collect + eval + push cycles (default: 60)",
    )
    parser.add_argument(
        "--once",
        action="store_true",
        help="Run a single evaluation (+ push) and exit",
    )
    parser.add_argument(
        "--no-push",
        action="store_true",
        help="Skip pushing results to Hugging Face",
    )
    parser.add_argument(
        "--hf-endpoint",
        default=os.getenv("HF_ENDPOINT", "https://huggingface.co"),
        help="HF API endpoint for upload",
    )
    return parser.parse_args()


def run_eval() -> int:
    cmd = [sys.executable, str(EVAL_SCRIPT)]
    logger.info("Running: %s", " ".join(cmd))
    return subprocess.run(cmd, cwd=str(SPACE_ROOT), check=False).returncode


def push_results(endpoint: str) -> int:
    cmd = [
        sys.executable,
        str(PUSH_SCRIPT),
        "--endpoint",
        endpoint,
    ]
    logger.info("Pushing results to HF Space")
    return subprocess.run(cmd, cwd=str(SPACE_ROOT), check=False).returncode


def run_cycle(args: argparse.Namespace) -> int:
    code = run_eval()
    if code != 0:
        logger.error("Evaluation failed with exit code %s", code)
        return code
    if args.no_push:
        return 0
    push_code = push_results(args.hf_endpoint)
    if push_code != 0:
        logger.error("HF push failed with exit code %s", push_code)
    return push_code


def main() -> None:
    args = parse_args()
    if args.once:
        raise SystemExit(run_cycle(args))

    while True:
        run_cycle(args)
        logger.info("Sleeping %s minutes until next cycle", args.interval_minutes)
        time.sleep(args.interval_minutes * 60)


if __name__ == "__main__":
    main()