""" MCP Tool: run Unified video generation task submission and execution interface. """ import os import time import uuid import threading import contextvars from .common import ( _TASK_DEFINITIONS, _TASKS_DB, _execute_imagegen_pipeline, ) from .error_schema import make_validation_error # ---------------------------------------------------------------- freno PRO # Tope de GPU que ESTE Space puede gastar de la cuota PRO de la cuenta. # 2280 s = 38 min, por debajo de los 40 min diarios, para no tocar creditos. # OJO: el contador es de este Space y vuelve a cero si el Space se reinicia. _TOPE_PRO_S = int(os.environ.get("ESVIAL_TOPE_PRO_S", "2280")) _pro_lock = threading.Lock() _pro_segundos = [0.0] def _pro_gastado() -> int: with _pro_lock: return int(_pro_segundos[0]) def _anota_pro(tarea: dict) -> None: """Suma al contador lo que de verdad gasto la toma.""" res = (tarea or {}).get("result") or {} try: gasto = float(res.get("execution_time_seconds") or 0.0) except (TypeError, ValueError): gasto = 0.0 with _pro_lock: _pro_segundos[0] += gasto def handle_run(params: dict) -> dict: """ Unified MiniMax H3 video generation task execution interface. [SUPPORTED TASK TYPES] - t2va: Text-to-Video. Required: prompt, width, height, duration. - i2va: Image-to-Video. Required: prompt, width, height, duration, first_frame_image. - flf2va: First & Last Frame-to-Video. Required: prompt, width, height, duration, first_frame_image, last_frame_image. - ref2va: Reference-to-Video. Required: prompt, width, height, duration. Optional: ref_image1..9, ref_video1..3, ref_audio1..3, h3_guides. [GLOBAL OPTIONAL PARAMETERS] - seed (int): Random seed (-1 for random seed, >=0 for deterministic reproduction). Default: -1. - zero_gpu_duration (int): GPU execution time quota allocation in seconds on HuggingFace ZeroGPU spaces (default: 120). - cuenta (str): "anonimo" (default) o "pro". "anonimo" -> la toma corre en un hilo nuevo, que NO hereda los contextvars de Gradio, asi que spaces.GPU no ve quien llamo y usa la cuota gratis compartida del Space. "pro" -> se copia el contexto de la llamada al hilo, la identidad viaja, y ZeroGPU cobra a la cuota de la cuenta que llamo. NO hay modo automatico: pasar a "pro" siempre es una decision explicita de quien manda la toma. Ademas hay un freno: cada Space no gasta mas de 38 min en "pro". - loras (list[dict]): List of LoRA configurations (e.g., [{"source": "Hugging Face", "id_or_url": "repo/lora.safetensors", "scale": 1.0}]). - h3_controlnets (list[dict]): List of H3 ControlNet configurations (e.g., [{"video": "https://.../pose.mp4", "strength": 1.0}]). - h3_guides (list[dict]): List of keyframe guide configurations (e.g., [{"image": "https://.../keyframe.png", "time_seconds": 1.5}]). Supported across all H3 tasks (FL2VA & Ref2VA). - control_video (str): Single convenience control video URL/path/base64 for H3 ControlNet. - control_strength (float): Strength for single convenience control video (default: 1.0). [Paste-and-Run json_params Example (t2va)] { "task_type": "t2va", "prompt": "A futuristic city with flying cars at sunset, 4k resolution", "width": 1344, "height": 768, "duration": 5.0, "seed": -1, "zero_gpu_duration": 120, "cuenta": "anonimo" } """ if not isinstance(params, dict): return make_validation_error("Request params must be an object.") valid_tasks = [t["task_type"] for t in _TASK_DEFINITIONS] task_type = params.get("task_type") if not task_type or task_type not in valid_tasks: return make_validation_error( f"Invalid or missing 'task_type'. Must be one of {valid_tasks}.", invalid_fields={"task_type": f"Must be in {valid_tasks}"}, ) task_def = next((t for t in _TASK_DEFINITIONS if t["task_type"] == task_type), None) required_fields = task_def["required_inputs"] if task_def else ["prompt", "width", "height", "duration"] missing = [] for req_field in required_fields: if req_field not in params or params[req_field] is None or params[req_field] == "": missing.append(req_field) if missing: return make_validation_error( f"Missing required parameter(s) for task '{task_type}': {', '.join(missing)}", missing_fields=missing, ) # --- de que bolsa de GPU sale la toma ----------------------------------- cuenta = str(params.get("cuenta", "anonimo")).strip().lower() if cuenta in ("anonymous", "anonimo", "anónimo", ""): cuenta = "anonimo" if cuenta not in ("anonimo", "pro"): return make_validation_error( "Invalid 'cuenta'. Must be 'anonimo' or 'pro'.", invalid_fields={"cuenta": "anonimo | pro"}, ) es_pro = (cuenta == "pro") if es_pro: gastado = _pro_gastado() queda = _TOPE_PRO_S - gastado if queda <= 0: return make_validation_error( "Freno de seguridad: este Space ya gasto %d s de GPU en modo 'pro' " "(tope %d s = %d min). Usa cuenta='anonimo', u otro Space, o espera " "a que se reinicie la cuota diaria." % (gastado, _TOPE_PRO_S, _TOPE_PRO_S // 60), invalid_fields={"cuenta": "tope pro alcanzado en este Space"}, ) try: pedido = int(params.get("zero_gpu_duration", 120) or 120) except (TypeError, ValueError): pedido = 120 if pedido > queda: return make_validation_error( "Freno de seguridad: pides %d s pero en modo 'pro' solo quedan %d s " "en este Space (tope %d s)." % (pedido, queda, _TOPE_PRO_S), invalid_fields={"zero_gpu_duration": "quedan %d s" % queda}, ) task_id = f"v_task_{uuid.uuid4().hex[:10]}" created_at = int(time.time()) _TASKS_DB[task_id] = { "task_id": task_id, "status": "queued", "progress": 0, "created_at": created_at, "modo": cuenta, } async_exec = params.get("async_execution", False) def _correr(): try: _execute_imagegen_pipeline(task_id, params) finally: if es_pro: _anota_pro(_TASKS_DB.get(task_id, {})) if async_exec: if es_pro: ctx = contextvars.copy_context() # la identidad viaja al hilo t = threading.Thread(target=ctx.run, args=(_correr,), daemon=True) else: t = threading.Thread(target=_correr, daemon=True) t.start() return { "status": "queued", "task_id": task_id, "modo": cuenta, "poll_interval_ms": 2000, "message": "Task queued successfully. Poll get_task_status for results.", } else: _correr() return _TASKS_DB[task_id]