| """ |
| OmniVision AI — Neural Media Studio |
| Punto de entrada principal para Hugging Face Spaces (Gradio SDK + FastAPI). |
| """ |
|
|
| |
| |
| try: |
| import spaces |
| import torch |
| except ImportError: |
| class _SpacesMock: |
| @staticmethod |
| def GPU(fn=None, **kwargs): |
| if fn is None: |
| return lambda f: f |
| return fn |
| spaces = _SpacesMock() |
|
|
| @spaces.GPU |
| def zero_gpu_pipeline(dummy: str = "") -> str: |
| """Acelerador ZeroGPU registrado al inicio de app.py.""" |
| return dummy |
|
|
| import os |
| import sys |
|
|
| |
| try: |
| import huggingface_hub |
| if not hasattr(huggingface_hub, "HfFolder"): |
| class _HfFolderCompat: |
| path_token = "" |
| @staticmethod |
| def get_token(): |
| try: |
| if hasattr(huggingface_hub, "get_token"): |
| tok = huggingface_hub.get_token() |
| if tok: |
| return tok |
| except Exception: |
| pass |
| return os.getenv("HF_TOKEN") |
|
|
| @staticmethod |
| def save_token(token): |
| try: |
| if hasattr(huggingface_hub, "login"): |
| huggingface_hub.login(token=token) |
| except Exception: |
| pass |
|
|
| @staticmethod |
| def delete_token(): |
| try: |
| if hasattr(huggingface_hub, "logout"): |
| huggingface_hub.logout() |
| except Exception: |
| pass |
|
|
| huggingface_hub.HfFolder = _HfFolderCompat |
| except Exception: |
| pass |
|
|
| |
| try: |
| import gradio_client.utils as client_utils |
| _orig_get_type = client_utils.get_type |
| def _safe_get_type(schema): |
| if isinstance(schema, bool): |
| return "boolean" |
| return _orig_get_type(schema) |
| client_utils.get_type = _safe_get_type |
|
|
| _orig_json_schema = client_utils._json_schema_to_python_type |
| def _safe_json_schema_to_python_type(schema, defs=None): |
| if isinstance(schema, bool): |
| return "boolean" |
| return _orig_json_schema(schema, defs) |
| client_utils._json_schema_to_python_type = _safe_json_schema_to_python_type |
| except Exception: |
| pass |
|
|
| |
| try: |
| from starlette.templating import Jinja2Templates |
| _orig_template_response = Jinja2Templates.TemplateResponse |
|
|
| def _safe_template_response(self, *args, **kwargs): |
| if len(args) >= 2 and isinstance(args[0], str) and isinstance(args[1], dict): |
| name = args[0] |
| context = args[1] |
| request = context.get("request") |
| if request is not None: |
| return _orig_template_response(self, request, name, context, *args[2:], **kwargs) |
| elif len(args) == 1 and isinstance(args[0], str) and "context" in kwargs: |
| name = args[0] |
| context = kwargs["context"] |
| request = context.get("request") |
| if request is not None: |
| kwargs["name"] = name |
| kwargs["request"] = request |
| return _orig_template_response(self, **kwargs) |
| return _orig_template_response(self, *args, **kwargs) |
|
|
| Jinja2Templates.TemplateResponse = _safe_template_response |
| except Exception: |
| pass |
|
|
| |
| from config import PORT, logger |
| from client import WorkerEngine |
| from ui import create_ui, setup_fastapi_routes |
|
|
| |
| demo = create_ui(gpu_fn=zero_gpu_pipeline) |
| demo.queue() |
|
|
|
|
| def main(): |
| logger.info("Iniciando motor de worker en segundo plano...") |
| engine = WorkerEngine() |
| engine.start() |
|
|
| logger.info(f"Lanzando servidor Gradio en puerto {PORT}...") |
| fastapi_app, _, _ = demo.launch( |
| server_name="0.0.0.0", |
| server_port=PORT, |
| prevent_thread_lock=True, |
| show_error=True, |
| ) |
|
|
| logger.info("Registrando endpoints REST compatibles (/api/v1/health)...") |
| setup_fastapi_routes(fastapi_app) |
|
|
| logger.info("✅ OmniVision AI activo y listo para conexiones.") |
| demo.block_thread() |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
|
|
