Image to Video

1import os2import sys3import json4import traceback5import random67from gais import Gais89INPUT_DIR = "/data/input"10OUTPUT_DIR = "/data/output"111213def main():14    try:15        input_json = sys.stdin.read()16        execution_input = json.loads(input_json)17        inputs = execution_input.get("inputs", {})1819        prompt = inputs.get("prompt", "")20        negative_prompt = inputs.get("negative_prompt", "")21        aspect_ratio = float(inputs.get("aspect_ratio", 1.7778) or 1.7778)22        megapixel = float(inputs.get("megapixel", 1.0) or 1.0)23        resolution = (inputs.get("resolution") or "").strip()24        num_frames = int(inputs.get("num_frames", 125) or 125)25        # fps unset → let the active video-creation backend pick its default26        # (Wan 2.2 = 16, LTX 2.3 = 24). Only pass when the user explicitly set it.27        fps_in = inputs.get("fps")28        fps = int(fps_in) if fps_in not in (None, "") else None2930        seed_mode = inputs.get("seed_mode", "random")31        seed_in = inputs.get("seed", -1)32        seed_input = int(seed_in) if seed_in not in (None, "") else -133        if seed_mode == "fixed" and seed_input >= 0:34            seed_value = seed_input35        else:36            seed_value = random.randint(0, 2**31 - 1)3738        if not prompt:39            raise ValueError("Prompt is required")4041        image_name = inputs.get("image", "")42        if not image_name:43            raise ValueError("An image is required")44        image_path = os.path.join(INPUT_DIR, image_name)45        if not os.path.exists(image_path):46            raise FileNotFoundError(f"Input image not found: {image_path}")4748        def _opt_input(name):49            v = (inputs.get(name) or "").strip()50            if not v:51                return None52            p = os.path.join(INPUT_DIR, v)53            if not os.path.exists(p):54                raise FileNotFoundError(f"Input file for {name!r} not found: {p}")55            return p5657        last_frame_path = _opt_input("last_frame")58        audio_path = _opt_input("audio")59        motion_video_path = _opt_input("motion_video")6061        # Width/height: explicit `resolution` override wins; otherwise62        # derive from aspect_ratio + megapixel, rounded to a multiple of 863        # (most video codecs / vendor backends prefer 8-divisible dims).64        if resolution:65            try:66                w_str, h_str = resolution.lower().split("x")67                width = int(w_str)68                height = int(h_str)69            except Exception as e:70                raise ValueError(71                    f"Invalid resolution {resolution!r}; expected 'WxH'"72                ) from e73        else:74            import math75            target_pixels = max(0.25, min(8.0, megapixel)) * 1_000_000.076            h_raw = math.sqrt(target_pixels / max(0.1, aspect_ratio))77            height = max(8, int(round(h_raw / 8) * 8))78            width = max(8, int(round((height * aspect_ratio) / 8) * 8))79            print(80                f"[derived] aspect_ratio={aspect_ratio} megapixel={megapixel} "81                f"-> {width}x{height} (backend may snap further)",82                file=sys.stderr,83            )8485        if fps is not None and fps <= 0:86            raise ValueError(f"fps must be > 0, got {fps}")87        # When the user provided fps, derive duration_s locally so the service88        # gets a precise duration. When fps is unset, leave duration_s out too89        # — the backend will compute it from num_frames at its own default fps.90        duration_s = (num_frames / fps) if fps else None9192        os.makedirs(OUTPUT_DIR, exist_ok=True)9394        fps_label = f"{fps}" if fps else "backend-default"95        dur_label = f"{duration_s:.2f}s" if duration_s else "backend-derived"96        extras = ", ".join(97            n for n, p in (98                ("last_frame", last_frame_path),99                ("audio", audio_path),100                ("motion_video", motion_video_path),101            ) if p102        ) or "none"103        print(104            f"Requesting i2v via gais-video-creation slot: "105            f"{width}x{height} @ {fps_label}fps, num_frames={num_frames} "106            f"({dur_label}), seed={seed_value}, optional inputs: {extras}",107            file=sys.stderr,108        )109110        # Profile-routed: backend (Wan22 / LTX / Helios / Luma / future) is111        # picked by HARDWARE_PROFILE via PROFILE_OVERRIDES in _registry.py.112        result = Gais.video.create_i2v(113            first_frame=image_path,114            prompt=prompt,115            negative_prompt=negative_prompt,116            width=width,117            height=height,118            duration_s=duration_s,119            fps=fps,120            seed=seed_value,121            num_frames=num_frames,122            last_frame=last_frame_path,123            audio=audio_path,124            motion_video=motion_video_path,125        )126127        out_filename = "generated_video.mp4"128        out_path = os.path.join(OUTPUT_DIR, out_filename)129        with open(out_path, "wb") as f:130            f.write(result.content)131132        inference_time = result.metadata.get("inference_time_ms", "unknown")133        provider = result.metadata.get("provider_name", "local")134        print(135            f"video generated via {provider}: time={inference_time}ms, seed={seed_value}",136            file=sys.stderr,137        )138139        # Probe output dimensions and emit aspect_ratio + resolution so140        # downstream nodes can chain.141        try:142            import av143            with av.open(out_path) as _c:144                _s = next(s for s in _c.streams if s.type == "video")145                _w = int(_s.codec_context.width)146                _h = int(_s.codec_context.height)147        except Exception:148            _w, _h = 0, 0149150        print(json.dumps({151            "video": out_filename,152            "aspect_ratio": round(_w / _h, 4) if _h else 0.0,153            "resolution": f"{_w}x{_h}",154        }, indent=2))155156    except Exception as e:157        err = {158            "error": str(e),159            "errorType": type(e).__name__,160        }161        # Backend/provider failures (policy violations, quota, remote 4xx)162        # already carry a human-readable message — a Python traceback is163        # pure noise in the execution-error panel. Keep it for genuine164        # template bugs only.165        try:166            from gais._errors import GaisError167            is_backend_err = isinstance(e, GaisError)168        except Exception:169            is_backend_err = False170        if not is_backend_err:171            err["traceback"] = traceback.format_exc()172        print(json.dumps(err), file=sys.stderr)173        sys.exit(1)174175176if __name__ == "__main__":177    main()