Keep ZeroGPU lease active during pipeline; add per-run GPU env overrides

app.py

@@ -37,6 +37,73 @@ def _err_payload(message: str) -> Dict[str, Any]:
     return {"status": "error", "message": message}
 
 
+ZERO_GPU_DURATION_SEC = int(os.getenv("ZERO_GPU_DURATION_SEC", "7200"))
+ZERO_GPU_POLL_SEC = float(os.getenv("ZERO_GPU_POLL_SEC", "2.0"))
+
+
+def _start_pipeline_job(
+    variant: str,
+    input_mode: str,
+    video_file_path: Optional[str],
+    video_url: Optional[str],
+    out_dir: Optional[str],
+    python_bin: Optional[str],
+    deepgram_model: str,
+    deepgram_language: Optional[str],
+    deepgram_request_timeout_sec: float,
+    deepgram_connect_timeout_sec: float,
+    deepgram_retries: int,
+    deepgram_retry_backoff_sec: float,
+    force_deepgram: bool,
+    force_keyframes: bool,
+    pre_roll_sec: float,
+    gemini_model: str,
+    similarity_threshold: float,
+    temperature: float,
+    log_heartbeat_sec: float,
+    env_overrides: Optional[Dict[str, str]] = None,
+) -> Tuple[str, Dict[str, Any], str, str]:
+    chosen_video_file = None
+    chosen_video_url = None
+    mode = (input_mode or "").strip().lower()
+
+    if mode == "upload file":
+        chosen_video_file = _clean_optional(video_file_path)
+        if not chosen_video_file:
+            raise ValueError("Select a video file for Upload File mode.")
+    elif mode == "video url":
+        chosen_video_url = _clean_optional(video_url)
+        if not chosen_video_url:
+            raise ValueError("Provide video_url for Video URL mode.")
+    else:
+        raise ValueError("Invalid input mode.")
+
+    result = start_run(
+        variant=variant,
+        video_file_path=chosen_video_file,
+        video_url=chosen_video_url,
+        out_dir=_clean_optional(out_dir),
+        python_bin=_clean_optional(python_bin),
+        deepgram_model=deepgram_model,
+        deepgram_language=_clean_optional(deepgram_language),
+        deepgram_request_timeout_sec=float(deepgram_request_timeout_sec),
+        deepgram_connect_timeout_sec=float(deepgram_connect_timeout_sec),
+        deepgram_retries=int(deepgram_retries),
+        deepgram_retry_backoff_sec=float(deepgram_retry_backoff_sec),
+        force_deepgram=bool(force_deepgram),
+        force_keyframes=bool(force_keyframes),
+        pre_roll_sec=float(pre_roll_sec),
+        gemini_model=gemini_model,
+        similarity_threshold=float(similarity_threshold),
+        temperature=float(temperature),
+        log_heartbeat_sec=float(log_heartbeat_sec),
+        env_overrides=env_overrides or {},
+    )
+    run_id = str(result["run_id"])
+    logs = get_logs(run_id, tail_lines=120)
+    return run_id, result, logs, run_id
+
+
 def start_pipeline(
     variant: str,
     input_mode: str,
@@ -59,50 +126,33 @@ def start_pipeline(
     log_heartbeat_sec: float,
 ) -> Tuple[str, Dict[str, Any], str, str]:
     try:
-        chosen_video_file = None
-        chosen_video_url = None
-        mode = (input_mode or "").strip().lower()
-
-        if mode == "upload file":
-            chosen_video_file = _clean_optional(video_file_path)
-            if not chosen_video_file:
-                raise ValueError("Select a video file for Upload File mode.")
-        elif mode == "video url":
-            chosen_video_url = _clean_optional(video_url)
-            if not chosen_video_url:
-                raise ValueError("Provide video_url for Video URL mode.")
-        else:
-            raise ValueError("Invalid input mode.")
-
-        result = start_run(
-            variant=variant,
-            video_file_path=chosen_video_file,
-            video_url=chosen_video_url,
-            out_dir=_clean_optional(out_dir),
-            python_bin=_clean_optional(python_bin),
-            deepgram_model=deepgram_model,
-            deepgram_language=_clean_optional(deepgram_language),
-            deepgram_request_timeout_sec=float(deepgram_request_timeout_sec),
-            deepgram_connect_timeout_sec=float(deepgram_connect_timeout_sec),
-            deepgram_retries=int(deepgram_retries),
-            deepgram_retry_backoff_sec=float(deepgram_retry_backoff_sec),
-            force_deepgram=bool(force_deepgram),
-            force_keyframes=bool(force_keyframes),
-            pre_roll_sec=float(pre_roll_sec),
-            gemini_model=gemini_model,
-            similarity_threshold=float(similarity_threshold),
-            temperature=float(temperature),
-            log_heartbeat_sec=float(log_heartbeat_sec),
+        return _start_pipeline_job(
+            variant,
+            input_mode,
+            video_file_path,
+            video_url,
+            out_dir,
+            python_bin,
+            deepgram_model,
+            deepgram_language,
+            deepgram_request_timeout_sec,
+            deepgram_connect_timeout_sec,
+            deepgram_retries,
+            deepgram_retry_backoff_sec,
+            force_deepgram,
+            force_keyframes,
+            pre_roll_sec,
+            gemini_model,
+            similarity_threshold,
+            temperature,
+            log_heartbeat_sec,
         )
-        run_id = str(result["run_id"])
-        logs = get_logs(run_id, tail_lines=120)
-        return run_id, result, logs, run_id
     except Exception as e:
         msg = f"{type(e).__name__}: {e}"
         return "", _err_payload(msg), msg, ""
 
 
-@spaces.GPU
+@spaces.GPU(duration=ZERO_GPU_DURATION_SEC)
 def start_pipeline_gpu(
     variant: str,
     input_mode: str,
@@ -124,27 +174,68 @@ def start_pipeline_gpu(
     temperature: float,
     log_heartbeat_sec: float,
 ) -> Tuple[str, Dict[str, Any], str, str]:
-    return start_pipeline(
-        variant,
-        input_mode,
-        video_file_path,
-        video_url,
-        out_dir,
-        python_bin,
-        deepgram_model,
-        deepgram_language,
-        deepgram_request_timeout_sec,
-        deepgram_connect_timeout_sec,
-        deepgram_retries,
-        deepgram_retry_backoff_sec,
-        force_deepgram,
-        force_keyframes,
-        pre_roll_sec,
-        gemini_model,
-        similarity_threshold,
-        temperature,
-        log_heartbeat_sec,
-    )
+    try:
+        run_id, start_result, _, _ = _start_pipeline_job(
+            variant,
+            input_mode,
+            video_file_path,
+            video_url,
+            out_dir,
+            python_bin,
+            deepgram_model,
+            deepgram_language,
+            deepgram_request_timeout_sec,
+            deepgram_connect_timeout_sec,
+            deepgram_retries,
+            deepgram_retry_backoff_sec,
+            force_deepgram,
+            force_keyframes,
+            pre_roll_sec,
+            gemini_model,
+            similarity_threshold,
+            temperature,
+            log_heartbeat_sec,
+            env_overrides={
+                "OCR_MODE": "gpu",
+                "OCR_BACKEND_GPU": "easyocr",
+                "YOLO_DEVICE": "0",
+                "CUDA_VISIBLE_DEVICES": "0",
+            },
+        )
+
+        started = time.time()
+        deadline = started + float(ZERO_GPU_DURATION_SEC)
+        sleep_sec = max(1.0, float(ZERO_GPU_POLL_SEC))
+        final_status: Dict[str, Any] = {}
+        while True:
+            final_status = get_status(run_id)
+            state = str(final_status.get("status", "running")).lower()
+            if state in {"succeeded", "failed"}:
+                break
+            if time.time() >= deadline:
+                logs = get_logs(run_id, tail_lines=500)
+                return run_id, {
+                    "status": "running",
+                    "run_id": run_id,
+                    "message": (
+                        f"Run is still active after {ZERO_GPU_DURATION_SEC}s. "
+                        "Continue monitoring in Track Run."
+                    ),
+                    "start_response": start_result,
+                    "latest_status": final_status,
+                }, logs, run_id
+            time.sleep(sleep_sec)
+
+        logs = get_logs(run_id, tail_lines=600)
+        return run_id, {
+            "status": str(final_status.get("status", "unknown")),
+            "run_id": run_id,
+            "start_response": start_result,
+            "final_status": final_status,
+        }, logs, run_id
+    except Exception as e:
+        msg = f"{type(e).__name__}: {e}"
+        return "", _err_payload(msg), msg, ""
 
 
 def refresh_status_logs(run_id: str, tail_lines: int) -> Tuple[Dict[str, Any], str]:
@@ -343,4 +434,4 @@ if __name__ == "__main__":
     if "ssr_mode" in inspect.signature(gr.Blocks.launch).parameters:
         launch_kwargs["ssr_mode"] = False
 
-    demo.queue(default_concurrency_limit=2).launch(**launch_kwargs)
+    demo.queue(default_concurrency_limit=1).launch(**launch_kwargs)

pipelines/smart_keyframes_and_classify.py

@@ -360,7 +360,11 @@ def _resolve_ocr_backend_for_mode(mode: str) -> Tuple[str, bool]:
     mode = _choose_ocr_mode(mode)
     gpu_available = _has_cuda()
 
-    if mode == "gpu":
+    # If GPU mode is requested but CUDA is not visible in this process,
+    # prefer CPU-first backends to avoid expensive GPU-oriented model downloads.
+    if mode == "gpu" and not gpu_available:
+        candidates = [OCR_BACKEND_CPU, "rapidocr", OCR_BACKEND_GPU, "easyocr", "paddle"]
+    elif mode == "gpu":
         candidates = [OCR_BACKEND_GPU, "easyocr", "paddle", "rapidocr"]
     else:
         candidates = [OCR_BACKEND_CPU, "rapidocr", "easyocr", "paddle"]

run_manager.py

@@ -412,6 +412,7 @@ def start_run(
     similarity_threshold: float,
     temperature: float,
     log_heartbeat_sec: float = 10.0,
+    env_overrides: Optional[Dict[str, str]] = None,
 ) -> Dict[str, Any]:
     script_name = {
         "full": "run_pipeline_all.py",
@@ -487,6 +488,12 @@ def start_run(
     child_env.setdefault("OCR_MODE", "cpu")
     child_env.setdefault("OCR_BACKEND_CPU", "rapidocr")
     child_env.setdefault("OCR_BACKEND_GPU", "easyocr")
+    child_env.setdefault("YOLO_DEVICE", "cpu")
+    for k, v in (env_overrides or {}).items():
+        key = str(k or "").strip()
+        if not key or v is None:
+            continue
+        child_env[key] = str(v)
 
     log_fh = open(logs_path, "a", encoding="utf-8", buffering=1)
     log_fh.write(
@@ -497,7 +504,8 @@ def start_run(
         f"[runner] python_unbuffered=1\n"
         f"[runner] ocr_mode={child_env.get('OCR_MODE')} "
         f"ocr_backend_cpu={child_env.get('OCR_BACKEND_CPU')} "
-        f"ocr_backend_gpu={child_env.get('OCR_BACKEND_GPU')}\n\n"
+        f"ocr_backend_gpu={child_env.get('OCR_BACKEND_GPU')} "
+        f"yolo_device={child_env.get('YOLO_DEVICE')}\n\n"
     )
     log_fh.flush()
 
@@ -524,6 +532,7 @@ def start_run(
         "out_dir": str(out_path),
         "logs_path": str(logs_path),
         "heartbeat_interval_sec": float(log_heartbeat_sec),
+        "env_overrides": {k: str(v) for k, v in (env_overrides or {}).items() if v is not None},
         "output_files": _build_output_files(out_path, variant),
     }
     _write_json(_meta_path(run_id), meta)