monitor progress

scripts/run_hf_remote_job_uv.py

@@ -24,7 +24,10 @@ import shutil
 import subprocess
 import sys
 import tempfile
+import threading
+import time
 import traceback
+from typing import Callable
 
 
 def _log(msg: str, logs: io.StringIO) -> None:
@@ -32,6 +35,117 @@ def _log(msg: str, logs: io.StringIO) -> None:
     logs.write(msg + "\n")
 
 
+def _progress_interval_s() -> float:
+    """Heartbeat cadence (seconds) for the remote progress reporter."""
+    raw = os.environ.get("FFASR_REMOTE_PROGRESS_INTERVAL_S", "30").strip()
+    try:
+        v = float(raw) if raw else 30.0
+    except ValueError:
+        v = 30.0
+    return max(2.0, min(v, 600.0))
+
+
+def _format_eta(seconds: float | None) -> str:
+    if seconds is None or seconds == float("inf"):
+        return "?"
+    if seconds < 60.0:
+        return f"{seconds:.0f}s"
+    if seconds < 3600.0:
+        return f"{seconds/60.0:.1f}m"
+    return f"{seconds/3600.0:.2f}h"
+
+
+def _make_progress_reporter() -> tuple[Callable[[int, int, str], None], Callable[[], None]]:
+    """
+    Return ``(progress_cb, stop)``.
+
+    ``progress_cb(done, total, condition)`` is the callback passed to ``run_evaluation``.
+    A daemon thread emits a heartbeat line periodically (``FFASR_REMOTE_PROGRESS_INTERVAL_S``,
+    default 30s) so the Hub Jobs log shows liveness even while the model is loading
+    (no per-sample callback fires during that phase).
+
+    Each emitted line is prefixed with ``[ffasr-progress]`` so it's easy to grep in the
+    streaming Hub log.
+    """
+    started = time.monotonic()
+    state = {
+        "done": 0,
+        "total": 0,
+        "condition": "",
+        "last_emit": 0.0,
+        "saw_sample": False,
+        "finished": False,
+    }
+    stop_event = threading.Event()
+    lock = threading.Lock()
+    interval = _progress_interval_s()
+
+    def _format(now: float) -> str:
+        elapsed = now - started
+        done, total = state["done"], state["total"]
+        cond = state["condition"] or "-"
+        if state["saw_sample"] and total > 0:
+            pct = 100.0 * done / total
+            rate = done / elapsed if elapsed > 0 else 0.0
+            eta = (total - done) / rate if rate > 0 else None
+            return (
+                f"[ffasr-progress] {done}/{total} ({pct:.1f}%) "
+                f"condition={cond} elapsed={elapsed:.0f}s "
+                f"rate={rate:.2f} samples/s ETA={_format_eta(eta)}"
+            )
+        if total > 0:
+            return (
+                f"[ffasr-progress] loading model / first sample pending "
+                f"(total={total} samples) elapsed={elapsed:.0f}s"
+            )
+        return f"[ffasr-progress] starting up… elapsed={elapsed:.0f}s"
+
+    def _emit_locked(now: float) -> None:
+        state["last_emit"] = now
+        print(_format(now), flush=True)
+
+    def cb(done: int, total: int, condition: str) -> None:
+        now = time.monotonic()
+        with lock:
+            state["done"] = int(done)
+            state["total"] = int(total)
+            state["condition"] = str(condition or "")
+            if done > 0:
+                state["saw_sample"] = True
+            # Always emit on boundaries (start, condition change, finish); otherwise
+            # throttle to ~5s so the log isn't flooded for fast models.
+            since_last = now - state["last_emit"]
+            at_boundary = (
+                done == 0
+                or (total > 0 and done == total)
+                or state["last_emit"] == 0.0
+            )
+            if at_boundary or since_last >= 5.0:
+                _emit_locked(now)
+
+    def heartbeat() -> None:
+        # Emit at least one line per `interval` seconds even when no sample callback
+        # fires (long model loads, slow first transcription, hung backend).
+        while not stop_event.wait(interval):
+            now = time.monotonic()
+            with lock:
+                if state["finished"]:
+                    return
+                # Only emit if we haven't already emitted recently from the callback.
+                if now - state["last_emit"] >= interval - 1.0:
+                    _emit_locked(now)
+
+    t = threading.Thread(target=heartbeat, name="ffasr-progress-heartbeat", daemon=True)
+    t.start()
+
+    def stop() -> None:
+        with lock:
+            state["finished"] = True
+        stop_event.set()
+
+    return cb, stop
+
+
 def _ensure_git() -> None:
     if shutil.which("git"):
         return
@@ -128,13 +242,23 @@ def main() -> int:
         condition_keys = None
         if cond_raw:
             condition_keys = json.loads(cond_raw)
-        result = run_evaluation(
-            model_id,
-            family_id=family_id,
-            progress_cb=None,
-            custom_script=custom_script,
-            condition_keys=condition_keys,
+        _log(
+            f"[ffasr-progress] beginning evaluation: model={model_id!r} "
+            f"family={family_id!r} conditions={condition_keys or 'default-set'}",
+            logs,
         )
+        progress_cb, stop_progress = _make_progress_reporter()
+        try:
+            result = run_evaluation(
+                model_id,
+                family_id=family_id,
+                progress_cb=progress_cb,
+                custom_script=custom_script,
+                condition_keys=condition_keys,
+            )
+        finally:
+            stop_progress()
+        _log("[ffasr-progress] evaluation finished, uploading artifact…", logs)
         artifact = build_artifact(
             model_id=model_id,
             family_id=family_id,