File size: 3,699 Bytes
f4170ee
 
 
 
 
 
 
 
2ab93da
 
284289a
f4170ee
06e6830
f4170ee
 
 
 
 
 
 
 
 
 
 
06e6830
f4170ee
 
 
 
06e6830
f4170ee
 
 
 
 
06e6830
f4170ee
 
 
 
 
 
 
 
 
284289a
f4170ee
 
 
 
 
 
 
284289a
 
06e6830
d9c7ee8
 
 
 
 
 
 
 
 
f4170ee
d9c7ee8
804ef40
284289a
 
 
264d402
284289a
 
f4170ee
284289a
 
 
 
 
 
 
 
 
 
 
f4170ee
 
06e6830
804ef40
284289a
f4170ee
284289a
d9c7ee8
284289a
 
 
 
d9c7ee8
284289a
 
f4170ee
d9c7ee8
284289a
 
 
f4170ee
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
import os
import base64
import json
import zlib
import importlib.machinery
import importlib.util
from pathlib import Path
from typing import Any, Dict
import gradio as gr

# --- CONFIGURATION PROTOCOLS (OMNICFG1) ---
_CONFIG_MARKER = b"OMNICFG1"

def _collect_config_values(namespace: Dict[str, Any]) -> Dict[str, Any]:
    data: Dict[str, Any] = {}
    for key, value in namespace.items():
        if key.isupper() and key != "APP_RUNTIME_CONFIG":
            data[key] = value
    class_obj = namespace.get("APP_RUNTIME_CONFIG")
    if class_obj is not None:
        for key in dir(class_obj):
            if key.isupper() and key not in data:
                data[key] = getattr(class_obj, key)
    return data

def _load_config_from_source(content: str) -> Dict[str, Any]:
    scope: Dict[str, Any] = {"__builtins__": __builtins__, "__name__": "omni_private_config"}
    exec(content, scope, scope)
    return _collect_config_values(scope)

def _write_packed_private_config(path: Path, data: Dict[str, Any]) -> None:
    payload = json.dumps(data, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
    packed = _CONFIG_MARKER + zlib.compress(payload, level=9)
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_bytes(packed)

def _materialize_private_config_from_env() -> None:
    repo_dir = Path(__file__).resolve().parent
    pyc_target = repo_dir / "src" / "config.pyc"
    raw_b64 = (os.getenv("OMNI_PRIVATE_CONFIG_B64") or "").strip()
    raw_py = (os.getenv("OMNI_PRIVATE_CONFIG_PY") or "").strip()
    content = ""
    if raw_b64:
        try:
            content = base64.b64decode(raw_b64.encode("utf-8")).decode("utf-8")
        except Exception:
            return
    elif raw_py:
        content = raw_py
    if content:
        try:
            data = _load_config_from_source(content.rstrip() + "\n")
            _write_packed_private_config(pyc_target, data)
        except Exception:
            pass

def _load_runtime_module():
    repo_dir = Path(__file__).resolve().parent
    try:
        from src import app_lib as module
        return module
    except ImportError:
        return None

# --- INITIALIZATION ---
_materialize_private_config_from_env()
_RUNTIME = _load_runtime_module()

# --- VIDEO INTERFACE BRIDGE ---
# Direct link to the Video Factory space for Iframe embedding
SOURCE_VIDEO_URL = "https://frameai4687-omni-video-factory.hf.space"

def build_video_factory_ui():
    with gr.Blocks(fill_height=True, title="Omni-Video-Factory Bridge") as demo:
        gr.HTML(f"""
            <div style="width: 100%; height: 100vh;">
                <iframe 
                    src="{SOURCE_VIDEO_URL}" 
                    frameborder="0" 
                    width="100%" 
                    height="100%" 
                    style="min-height: 1000px; border-radius: 4px;"
                    allow="accelerometer; autoplay; clipboard-write; encrypted-media; gyroscope; picture-in-picture; web-share" 
                    allowfullscreen>
                </iframe>
            </div>
        """)
    return demo

if __name__ == "__main__":
    print("[startup] Launching Omni-Video-Factory Runtime...")
    
    # Background hooks execution
    if _RUNTIME:
        for hook in ["kickoff_runtime_prepare_background", "kickoff_model_prepare_background"]:
            func = getattr(_RUNTIME, hook, None)
            if callable(func):
                print(f"[startup] {hook}: {func()}")

    # Launching the interface optimized for Colab / External API
    demo = build_video_factory_ui()
    demo.launch(
        share=True,
        server_name="0.0.0.0",
        server_port=int(os.getenv("PORT", "7860")),
        inline=False
    )