File size: 4,699 Bytes
f5cd164
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
e15eab8
f5cd164
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
8e81ebe
f5cd164
 
e15eab8
f5cd164
 
 
 
 
 
 
 
 
 
 
 
 
 
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
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
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()