Spaces:
Running
Running
| import os | |
| import sys | |
| import threading | |
| from typing import Dict, Any, List | |
| # Add backend directory to path | |
| CURRENT_DIR = os.path.dirname(os.path.abspath(__file__)) | |
| if CURRENT_DIR not in sys.path: | |
| sys.path.append(CURRENT_DIR) | |
| try: | |
| from .sign_bridge_inference_hq import SignBridgeInferenceHQ | |
| except (ImportError, ValueError): | |
| from sign_bridge_inference_hq import SignBridgeInferenceHQ | |
| # HQ Weights are stored in a dedicated directory | |
| MODEL_ROOT_HQ = os.path.join(os.path.dirname(CURRENT_DIR), "weights_hq") | |
| CONFIG_PATH_HQ = os.path.join(os.path.dirname(CURRENT_DIR), "model_configs", "Sign-IDD-HQ.yaml") | |
| class SignModelHQ: | |
| """ | |
| Singleton wrapper for the High Fidelity SignBridgeInference engine. | |
| Uses the uncompressed 1.1GB weights and optimized motion sampling. | |
| """ | |
| _instance = None | |
| _lock = threading.Lock() | |
| def __new__(cls): | |
| with cls._lock: | |
| if cls._instance is None: | |
| cls._instance = super(SignModelHQ, cls).__new__(cls) | |
| cls._instance._initialized = False | |
| return cls._instance | |
| def __init__(self): | |
| if self._initialized: | |
| return | |
| print("Initializing SignBridge HQ Model (High Fidelity Path)...") | |
| self.engine = None | |
| self.is_loaded = False | |
| self._load_error = None | |
| self._initialized = True | |
| # Load in background | |
| threading.Thread(target=self._load_model_async, daemon=True).start() | |
| def _load_model_async(self): | |
| try: | |
| # 1. Ensure weight directory exists | |
| os.makedirs(MODEL_ROOT_HQ, exist_ok=True) | |
| weight_path = os.path.join(MODEL_ROOT_HQ, "best.ckpt") | |
| # 2. Check if weights need to be downloaded (Runtime bypass for 1GB repo limit) | |
| if not os.path.exists(weight_path): | |
| print(f"HQ Weights not found at {weight_path}. Attempting download from Hub...") | |
| from huggingface_hub import hf_hub_download | |
| repo_id = os.environ.get("HF_REPO_ID_HQ", "ExploWebsite/SignBridge-Weights") | |
| token = os.environ.get("HF_TOKEN") # Optional: needed if repo is private | |
| print(f"Downloading HQ Weights from {repo_id}...") | |
| downloaded_file = hf_hub_download( | |
| repo_id=repo_id, | |
| filename="best.ckpt", | |
| local_dir=MODEL_ROOT_HQ, | |
| token=token | |
| ) | |
| print(f"✅ Download complete: {downloaded_file}") | |
| # 3. Initialize the HQ-specific inference engine | |
| self.engine = SignBridgeInferenceHQ(MODEL_ROOT_HQ) | |
| self.is_loaded = True | |
| print("✅ SignBridge HQ Model loaded and ready for high-fidelity inference.") | |
| except Exception as e: | |
| self._load_error = str(e) | |
| print(f"❌ Failed to load SignBridge HQ Model: {e}") | |
| import traceback | |
| traceback.print_exc() | |
| def inference(self, text: str) -> Dict[str, Any]: | |
| if not self.is_loaded: | |
| if self._load_error: | |
| raise RuntimeError(f"HQ Model failed to load: {self._load_error}") | |
| raise RuntimeError("HQ Model is still loading. Please try again in 30 seconds.") | |
| print(f"HQ Inference Request: '{text}'") | |
| try: | |
| # HIGH FIDELITY PARAMS: | |
| # We use 90 steps (matching original training) and potentially different guidance or length heuristics | |
| skeletons = self.engine.translate(text, sampling_steps=90) | |
| import uuid | |
| from video_renderer import render_skeleton_to_video | |
| filename = f"hq_gen_{uuid.uuid4().hex[:8]}.mp4" | |
| output_dir = os.path.join(CURRENT_DIR, "output") | |
| os.makedirs(output_dir, exist_ok=True) | |
| output_path = os.path.join(output_dir, filename) | |
| # Use standard renderer (stable) | |
| render_skeleton_to_video(skeletons, output_path, mode="hq") | |
| # URL resolution (assumes same static mount) | |
| video_url = f"https://explowebsite-sign-idd-inference.hf.space/static/{filename}" | |
| if os.environ.get("LOCAL_DEV"): | |
| video_url = f"http://127.0.0.1:8001/static/{filename}" | |
| return { | |
| "skeletons": None, | |
| "video_url": video_url, | |
| "glosses": self.engine.text_to_glosses(text) | |
| } | |
| except Exception as e: | |
| print(f"HQ Inference error: {e}") | |
| raise e | |
| # Global singleton instance for HQ | |
| sign_model_hq = SignModelHQ() | |