File size: 2,450 Bytes
5d4272a | 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 | """Provider switch for RADAR scoring backends.
ZeroGPU remains the active backend; Vast.ai is a stub holding config fields
until a Serverless endpoint is provisioned (see backend/vastai/). Frontend
code branches on the same kind strings ("zerogpu" | "vastai").
Import cost: stdlib only. Importing this module must never pull torch,
CUDA, or app state — ZeroGPUBackend resolves app._score_case lazily.
"""
import os
from typing import Protocol
class InferenceBackend(Protocol):
def score_case(self, tmpdir: str, outdir: str) -> tuple[dict, object]:
"""Score one staged case; returns (scores dict, findings DataFrame)."""
...
def get_backend_kind() -> str:
"""Active backend from env; unknown values are validated by callers."""
return (os.getenv("INFERENCE_BACKEND", "zerogpu").strip().lower() or "zerogpu")
class ZeroGPUBackend:
"""Active backend: delegates to app._score_case (the @spaces.GPU step).
score_fn is injected by app.diagnose so this module never imports app
at load time (avoids a circular import and keeps CUDA init in app).
"""
def __init__(self, score_fn=None):
self._score_fn = score_fn
def score_case(self, tmpdir: str, outdir: str) -> tuple[dict, object]:
fn = self._score_fn
if fn is None: # pragma: no cover — production path always injects
from app import _score_case as fn
self._score_fn = fn
return fn(tmpdir, outdir)
class VastAIBackend:
"""Future Vast.ai Serverless backend — not enabled yet.
Holds the config surface the later cutover needs; score_case fails fast
so a misconfigured INFERENCE_BACKEND=vastai can never silently hang.
Real key lives server-side in VAST_API_KEY env, never in frontend code.
"""
NOT_ENABLED = "Vast.ai backend not enabled: set INFERENCE_BACKEND=zerogpu"
def __init__(
self,
endpoint: str = "",
api_key_env: str = "VAST_API_KEY",
):
self.endpoint = endpoint or os.getenv("VAST_ENDPOINT", "")
self.api_key_env = api_key_env
def ensure_available(self) -> None:
"""Fail fast before any staging when this backend is selected."""
import gradio as gr
raise gr.Error(self.NOT_ENABLED)
def score_case(self, tmpdir: str, outdir: str) -> tuple[dict, object]:
self.ensure_available()
raise RuntimeError("unreachable: Vast.ai backend is not enabled")
|