Spaces:
Sleeping
Sleeping
File size: 2,593 Bytes
51992f4 36f68cc 51992f4 36f68cc 51992f4 36f68cc 51992f4 36f68cc a6f63e9 36f68cc a6f63e9 51992f4 36f68cc | 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 | """Backend client for Modal inference with explicit local sample mode."""
from __future__ import annotations
import base64
import os
from dataclasses import dataclass
from io import BytesIO
from typing import Any
import requests
from PIL import Image
from snap2sim.schema import EXAMPLE_ANALYSIS, select_render_mode, validate_analysis
@dataclass(frozen=True)
class Settings:
backend: str = os.getenv("INFERENCE_BACKEND", "modal")
analyze_url: str = os.getenv("MODAL_ANALYZE_URL", "")
generate_url: str = os.getenv("MODAL_GENERATE_URL", "")
api_token: str = os.getenv("SNAP2SIM_API_TOKEN", "")
timeout_seconds: int = int(os.getenv("INFERENCE_TIMEOUT_SECONDS", "300"))
def encode_image(image: Image.Image) -> str:
buffer = BytesIO()
image.convert("RGB").save(buffer, format="JPEG", quality=92)
return base64.b64encode(buffer.getvalue()).decode("ascii")
class InferenceClient:
def __init__(self, settings: Settings | None = None) -> None:
self.settings = settings or Settings()
def analyze_image(self, image: Image.Image | None) -> dict[str, Any]:
if self.settings.backend == "modal":
if image is None:
raise ValueError("Upload an image before analysis.")
return self._post_json(
self.settings.analyze_url,
{"image_base64": encode_image(image)},
)
return validate_analysis(dict(EXAMPLE_ANALYSIS))
def generate_scene(self, analysis: dict[str, Any], threshold: Any = None) -> dict[str, Any]:
valid_analysis = validate_analysis(analysis)
render_mode = select_render_mode(valid_analysis, threshold)
renderer = "three" if render_mode == "three" else "photo"
return {"renderer": renderer, "render_mode": render_mode, "analysis": valid_analysis}
def _post_json(self, url: str, payload: dict[str, Any]) -> dict[str, Any]:
if not url:
raise RuntimeError("Modal backend selected but endpoint URL is not configured.")
if not self.settings.api_token:
raise RuntimeError("Modal backend selected but SNAP2SIM_API_TOKEN is not configured.")
response = requests.post(
url,
json=payload,
headers={"Authorization": f"Bearer {self.settings.api_token}"},
timeout=self.settings.timeout_seconds,
)
response.raise_for_status()
data = response.json()
if not isinstance(data, dict):
raise RuntimeError("Inference backend returned a non-object JSON payload.")
return data
|