| """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: |
| 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") |
|
|
