{ "cells": [ { "cell_type": "markdown", "id": "license-header", "metadata": {}, "source": [ "" ] }, { "cell_type": "markdown", "id": "d88fe9a8", "metadata": {}, "source": [ "# Cosmos3 Generator Audiovisual with SGLang\n", "\n", "This notebook calls already-running SGLang Cosmos3 servers with direct `curl` requests from Python.\n", "\n", "The examples are split into Cosmos3-Nano and Cosmos3-Super sections. Each section is self-contained, so you can run just one. Each section targets the matching model endpoint.\n" ] }, { "cell_type": "markdown", "id": "49df4e61", "metadata": {}, "source": [ "## 1. Prerequisites\n", "\n", "Use a running SGLang server and set endpoint environment variables before the setup cell if you are not using the local default. Text-to-image uses `/v1/images/generations`; video modes use `/v1/videos`.\n", "\n", "```bash\n", "export COSMOS3_SGLANG_BASE_URL=http://localhost:30000\n", "export COSMOS3_SGLANG_NANO_BASE_URL=http://localhost:30000\n", "export COSMOS3_SGLANG_SUPER_BASE_URL=http://localhost:30000\n", "```\n" ] }, { "cell_type": "markdown", "id": "26776c50", "metadata": {}, "source": [ "## 2. Start the Server\n", "\n", "Run the SGLang server before running the request cells. Use the Docker image for every modality on this page. Mount any directory that contains local media or action files you want the server to read.\n", "\n", "### Docker Image: Cosmos3-Nano\n", "\n", "```bash\n", "docker run --runtime nvidia --gpus all \\\n", " -v ~/.cache/huggingface:/root/.cache/huggingface \\\n", " -v \"$(pwd):/workspace\" \\\n", " -p 30000:30000 \\\n", " --ipc=host \\\n", " lmsysorg/sglang:dev \\\n", " sglang serve \\\n", " --model-path nvidia/Cosmos3-Nano \\\n", " --host 0.0.0.0\n", "```\n", "\n", "### Docker Image: Cosmos3-Super\n", "\n", "`Cosmos3-Super` is the larger 64B model, so it usually needs more GPU memory than `Cosmos3-Nano`. `--tp-size` splits model weights across multiple GPUs and reduces per-GPU memory use. Use `--performance-mode memory` preset to preserve memory.\n", "\n", "For example, on four GPUs:\n", "\n", "```bash\n", "docker run --runtime nvidia --gpus all \\\n", " -v ~/.cache/huggingface:/root/.cache/huggingface \\\n", " -v \"$(pwd):/workspace\" \\\n", " -p 30000:30000 \\\n", " --ipc=host \\\n", " lmsysorg/sglang:dev \\\n", " sglang serve \\\n", " --model-path nvidia/Cosmos3-Super \\\n", " --host 0.0.0.0 \\\n", " --num-gpus 4 \\\n", " --performance-mode memory\n", "```\n", "\n", "### CFG Parallel\n", "\n", "Use `--cfg-parallel-size 2` to run the positive and negative CFG branches in parallel on two GPUs:\n", "\n", "```bash\n", "sglang serve \\\n", " --model-path nvidia/Cosmos3-Nano \\\n", " --host 0.0.0.0 \\\n", " --num-gpus 2 \\\n", " --enable-cfg-parallel \\\n", " --cfg-parallel-size 2\n", "```\n", "\n", "For Cosmos3, set CFG strength with the request-level `guidance_scale` field. Do not use `true_cfg_scale` for CFG Parallel with these Cosmos3 examples.\n" ] }, { "cell_type": "markdown", "id": "4412f2f9", "metadata": {}, "source": [ "## 3. Configure Paths and Endpoints\n", "\n", "This setup cell only configures repo/output paths and SGLang endpoint settings.\n" ] }, { "cell_type": "code", "execution_count": null, "id": "23f04a90", "metadata": {}, "outputs": [], "source": [ "from pathlib import Path\n", "import os\n", "\n", "\n", "def find_repo_root(start: Path) -> Path:\n", " for path in [start, *start.parents]:\n", " if (path / \"README.md\").exists() and (path / \"cookbooks\").exists():\n", " return path\n", " return start\n", "\n", "\n", "COSMOS_ROOT = find_repo_root(Path.cwd().resolve())\n", "COSMOS3_AUDIOVISUAL_ROOT = COSMOS_ROOT / \"cookbooks\" / \"cosmos3\" / \"generator\" / \"audiovisual\"\n", "COSMOS3_AUDIOVISUAL_OUTPUT_ROOT = Path(\n", " os.environ.get(\"COSMOS3_AUDIOVISUAL_OUTPUT_ROOT\", COSMOS3_AUDIOVISUAL_ROOT / \"outputs\" / \"notebooks\")\n", ").resolve()\n", "DEFAULT_SGLANG_BASE_URL = os.environ.get(\"COSMOS3_SGLANG_BASE_URL\", \"http://localhost:30000\")\n", "SGLANG_ENDPOINTS = {\n", " \"Cosmos3-Nano\": os.environ.get(\"COSMOS3_SGLANG_NANO_BASE_URL\", DEFAULT_SGLANG_BASE_URL),\n", " \"Cosmos3-Super\": os.environ.get(\"COSMOS3_SGLANG_SUPER_BASE_URL\", DEFAULT_SGLANG_BASE_URL),\n", "}\n", "\n", "os.environ[\"COSMOS3_AUDIOVISUAL_OUTPUT_ROOT\"] = str(COSMOS3_AUDIOVISUAL_OUTPUT_ROOT)\n", "os.environ.setdefault(\"COSMOS3_SGLANG_API_KEY\", \"\")\n", "\n", "print(\"COSMOS_ROOT:\", COSMOS_ROOT)\n", "print(\"COSMOS3_AUDIOVISUAL_OUTPUT_ROOT:\", COSMOS3_AUDIOVISUAL_OUTPUT_ROOT)\n", "for model, endpoint in SGLANG_ENDPOINTS.items():\n", " print(f\"{model} endpoint: {endpoint}\")\n" ] }, { "cell_type": "markdown", "id": "73369e7f", "metadata": {}, "source": [ "## 4. Verify Endpoint Configuration\n" ] }, { "cell_type": "code", "execution_count": null, "id": "1c50e183", "metadata": {}, "outputs": [], "source": [ "from urllib.parse import urlparse\n", "\n", "for model, base_url in SGLANG_ENDPOINTS.items():\n", " api_root = base_url.rstrip(\"/\")\n", " if not api_root.endswith(\"/v1\"):\n", " api_root = f\"{api_root}/v1\"\n", " parsed = urlparse(api_root)\n", " print(model)\n", " print(\" api root:\", api_root)\n", " print(\" images generations:\", f\"{api_root}/images/generations\")\n", " print(\" videos async:\", f\"{api_root}/videos\")\n", " print(\" scheme:\", parsed.scheme)\n", " print(\" host:\", parsed.netloc)\n" ] }, { "cell_type": "markdown", "id": "c1d161a6", "metadata": {}, "source": [ "## 5. Preview Available Inputs\n" ] }, { "cell_type": "code", "execution_count": null, "id": "973ea472", "metadata": {}, "outputs": [], "source": [ "from pathlib import Path\n", "import json\n", "from IPython.display import Image, display\n", "\n", "assets_dir = COSMOS3_AUDIOVISUAL_ROOT / \"assets\"\n", "for prompt_dir in sorted((assets_dir / \"prompts\").iterdir()):\n", " if not prompt_dir.is_dir():\n", " continue\n", " print(f\"{prompt_dir.relative_to(assets_dir)}:\")\n", " for prompt_path in sorted(prompt_dir.glob(\"*.json\")):\n", " data = json.loads(prompt_path.read_text())\n", " caption = (\n", " data.get(\"temporal_caption\")\n", " or data.get(\"comprehensive_t2i_caption\")\n", " or data.get(\"extra\", {}).get(\"prompt\", \"\")\n", " )\n", " print(f\" {prompt_path.name}: {caption[:180]}{'...' if len(caption) > 180 else ''}\")\n", " print()\n", "\n", "for image_dir in sorted((assets_dir / \"images\").iterdir()):\n", " if not image_dir.is_dir():\n", " continue\n", " print(f\"{image_dir.relative_to(assets_dir)}:\")\n", " for image_path in sorted(image_dir.iterdir()):\n", " if image_path.suffix.lower() in {\".jpg\", \".jpeg\", \".png\", \".webp\", \".bmp\"}:\n", " print(f\" {image_path.name}\")\n", " display(Image(filename=str(image_path), width=420))\n" ] }, { "cell_type": "markdown", "id": "cfa34351", "metadata": {}, "source": [ "## 6. Define Asset Sets, Payload Helpers, Request Helpers, and Viewer Helpers\n" ] }, { "cell_type": "code", "execution_count": null, "id": "af5dd1c8", "metadata": {}, "outputs": [], "source": [ "import json\n", "import os\n", "from pathlib import Path\n", "from IPython.display import Image, display, Video\n", "\n", "IMAGE_EXTENSIONS = {\".jpg\", \".jpeg\", \".png\", \".webp\", \".bmp\"}\n", "\n", "FIXED_SAMPLING = {\n", " \"num_steps\": 35,\n", " \"guidance\": 6.0,\n", " \"shift\": 10.0,\n", " \"fps\": 24,\n", " \"num_frames\": 189,\n", " \"resolution\": \"720\",\n", " \"aspect_ratio\": \"16,9\",\n", " \"seed\": 0,\n", "}\n", "\n", "# All asset paths are repo-relative under cookbooks/cosmos3/generator/audiovisual.\n", "# Model and sound choices live in this manifest; folders are organized only by modality.\n", "ASSET_SETS = {\n", " \"t2i\": {\n", " \"model\": \"Cosmos3-Nano\",\n", " \"mode\": \"text2image\",\n", " \"prompt\": \"assets/prompts/text2image/robot_draping.json\",\n", " \"enable_sound\": False,\n", " },\n", " \"t2i_super\": {\n", " \"model\": \"Cosmos3-Super\",\n", " \"mode\": \"text2image\",\n", " \"prompt\": \"assets/prompts/text2image/robot_draping.json\",\n", " \"enable_sound\": False,\n", " },\n", " \"t2v_nano_noaudio\": {\n", " \"model\": \"Cosmos3-Nano\",\n", " \"mode\": \"text2video\",\n", " \"prompt\": \"assets/prompts/text2video/robot_kitchen.json\",\n", " \"enable_sound\": False,\n", " },\n", " \"t2vs\": {\n", " \"model\": \"Cosmos3-Nano\",\n", " \"mode\": \"text2video\",\n", " \"prompt\": \"assets/prompts/text2video/robot_pouring_water_audio.json\",\n", " \"enable_sound\": True,\n", " },\n", " \"i2v_nano_noaudio\": {\n", " \"model\": \"Cosmos3-Nano\",\n", " \"mode\": \"image2video\",\n", " \"prompt\": \"assets/prompts/image2video/car_driving.json\",\n", " \"image\": \"assets/images/image2video/car_driving.jpg\",\n", " \"enable_sound\": False,\n", " },\n", " \"i2vs\": {\n", " \"model\": \"Cosmos3-Nano\",\n", " \"mode\": \"image2video\",\n", " \"prompt\": \"assets/prompts/image2video/coastal_road_audio.json\",\n", " \"image\": \"assets/images/image2video/coastal_road_audio.jpg\",\n", " \"enable_sound\": True,\n", " },\n", " \"v2vs\": {\n", " \"model\": \"Cosmos3-Nano\",\n", " \"mode\": \"video2video\",\n", " \"prompt\": \"assets/prompts/image2video/car_driving.json\",\n", " \"negative_prompt\": \"assets/negative_prompts/image2video/neg_prompt.json\",\n", " \"video\": \"assets/videos/car_driving_plain.mp4\",\n", " \"enable_sound\": True,\n", " },\n", " \"t2v_super_noaudio\": {\n", " \"model\": \"Cosmos3-Super\",\n", " \"mode\": \"text2video\",\n", " \"prompt\": \"assets/prompts/text2video/robot_kitchen.json\",\n", " \"enable_sound\": False,\n", " },\n", " \"i2v_super_noaudio\": {\n", " \"model\": \"Cosmos3-Super\",\n", " \"mode\": \"image2video\",\n", " \"prompt\": \"assets/prompts/image2video/car_driving.json\",\n", " \"image\": \"assets/images/image2video/car_driving.jpg\",\n", " \"enable_sound\": False,\n", " },\n", "}\n", "\n", "\n", "def asset_path(relative_path: str) -> Path:\n", " path = COSMOS3_AUDIOVISUAL_ROOT / relative_path\n", " if not path.exists():\n", " raise FileNotFoundError(path)\n", " return path.resolve()\n", "\n", "\n", "def compact_json_file(path: Path) -> str:\n", " return json.dumps(json.loads(path.read_text()), ensure_ascii=True, separators=(\",\", \":\"))\n", "\n", "\n", "def normalize_negative_prompt(value) -> str:\n", " if value is None:\n", " return \"\"\n", " if isinstance(value, str):\n", " return value\n", " if isinstance(value, dict):\n", " parts = []\n", " for v in value.values():\n", " if isinstance(v, str):\n", " parts.append(v)\n", " elif isinstance(v, list):\n", " parts.extend(str(x) for x in v)\n", " return \"\\n\".join(parts)\n", " return str(value)\n", "\n", "\n", "def payload_dimensions(payload: dict) -> tuple[int, int]:\n", " if payload.get(\"resolution\") == \"720\" and payload.get(\"aspect_ratio\") == \"16,9\":\n", " return 720, 1280\n", " if payload.get(\"resolution\") == \"256\" and payload.get(\"aspect_ratio\") == \"16,9\":\n", " return 192, 320\n", " raise ValueError(f\"Unsupported payload resolution/aspect ratio: {payload.get('resolution')} {payload.get('aspect_ratio')}\")\n", "\n", "\n", "def resolve_payload_path(payload_path: Path, value: str) -> Path:\n", " path = Path(value)\n", " if path.is_absolute():\n", " return path\n", " return (payload_path.parent / path).resolve()\n", "\n", "\n", "def create_payload(use_case: str, *, backend: str) -> tuple[Path, Path, str]:\n", " spec = ASSET_SETS[use_case]\n", " payload_dir = Path(os.environ[\"COSMOS3_AUDIOVISUAL_OUTPUT_ROOT\"]) / backend / \"payloads\" / use_case\n", " output_dir = Path(os.environ[\"COSMOS3_AUDIOVISUAL_OUTPUT_ROOT\"]) / backend / use_case\n", " payload_dir.mkdir(parents=True, exist_ok=True)\n", " output_dir.mkdir(parents=True, exist_ok=True)\n", "\n", " prompt_path = asset_path(spec[\"prompt\"])\n", " negative_prompt = \"\"\n", " if spec[\"mode\"] != \"text2image\":\n", " negative_prompt_path = asset_path(f\"assets/negative_prompts/{spec['mode']}/neg_prompt.json\") if not spec.get(\"negative_prompt\") else spec[\"negative_prompt\"]\n", " negative_prompt = normalize_negative_prompt(negative_prompt_path)\n", " payload_path = payload_dir / f\"{use_case}.json\"\n", " payload = {\n", " \"model_mode\": spec[\"mode\"],\n", " \"name\": use_case,\n", " \"prompt\": compact_json_file(prompt_path),\n", " \"negative_prompt\": negative_prompt,\n", " \"enable_sound\": spec[\"enable_sound\"],\n", " **FIXED_SAMPLING,\n", " }\n", " if spec[\"mode\"] in {\"image2video\"}:\n", " image_path = asset_path(spec[\"image\"])\n", " payload[\"vision_path\"] = os.path.relpath(image_path, payload_path.parent)\n", " \n", " if spec[\"mode\"] in {\"video2video\"}:\n", " video_path = asset_path(spec[\"video\"])\n", " payload[\"vision_path\"] = os.path.relpath(video_path, payload_path.parent)\n", "\n", " payload_path.write_text(json.dumps(payload, indent=2) + \"\\n\")\n", "\n", " os.environ[f\"COSMOS3_{backend.upper()}_{use_case.upper()}_INPUT\"] = str(payload_path)\n", " os.environ[f\"COSMOS3_{backend.upper()}_{use_case.upper()}_OUTPUT\"] = str(output_dir)\n", "\n", " print(f\"model: {spec['model']}\")\n", " print(f\"payload: {payload_path}\")\n", " print(f\"output: {output_dir}\")\n", " print(f\"prompt: {prompt_path.relative_to(COSMOS_ROOT)}\")\n", " if \"vision_path\" in payload:\n", " if payload[\"model_mode\"] == \"image2video\":\n", " image_display_path = resolve_payload_path(payload_path, payload[\"vision_path\"])\n", " print(f\"image: {image_display_path.relative_to(COSMOS_ROOT)}\")\n", " display(Image(filename=str(image_display_path), width=420))\n", " elif payload[\"model_mode\"] == \"video2video\":\n", " video_display_path = resolve_payload_path(payload_path, payload[\"vision_path\"])\n", " print(f\"video: {video_display_path.relative_to(COSMOS_ROOT)}\")\n", " display(Video(filename=str(video_display_path), width=420))\n", " print(json.dumps({k: payload[k] for k in [\"model_mode\", \"name\", \"enable_sound\", \"num_steps\", \"guidance\", \"shift\", \"fps\", \"num_frames\", \"resolution\", \"aspect_ratio\", \"seed\"]}, indent=2))\n", " return payload_path, output_dir, spec[\"model\"]\n", "\n", "\n", "import base64\n", "import html\n", "import json\n", "import os\n", "import subprocess\n", "import time\n", "from pathlib import Path\n", "from IPython.display import HTML, display\n", "\n", "\n", "def api_root_url(base_url: str) -> str:\n", " normalized = base_url.rstrip(\"/\")\n", " if not normalized.endswith(\"/v1\"):\n", " normalized = f\"{normalized}/v1\"\n", " return normalized\n", "\n", "\n", "def video_api_url(base_url: str) -> str:\n", " return f\"{api_root_url(base_url)}/videos\"\n", "\n", "\n", "def image_api_url(base_url: str) -> str:\n", " return f\"{api_root_url(base_url)}/images/generations\"\n", "\n", "def build_sglang_video_form(payload: dict) -> dict[str, str]:\n", " height, width = payload_dimensions(payload)\n", " extra_params = {\n", " \"use_resolution_template\": False,\n", " \"use_duration_template\": False,\n", " \"guardrails\": True,\n", " }\n", " form = {\n", " \"prompt\": payload[\"prompt\"],\n", " \"negative_prompt\": payload[\"negative_prompt\"],\n", " \"size\": f\"{width}x{height}\",\n", " \"num_frames\": str(payload[\"num_frames\"]),\n", " \"fps\": str(payload[\"fps\"]),\n", " \"num_inference_steps\": str(payload[\"num_steps\"]),\n", " \"guidance_scale\": str(payload[\"guidance\"]),\n", " \"flow_shift\": str(payload[\"shift\"]),\n", " \"seed\": str(payload[\"seed\"]),\n", " \"extra_params\": json.dumps(extra_params, separators=(\",\", \":\")),\n", " }\n", " if payload[\"enable_sound\"]:\n", " form[\"generate_sound\"] = \"true\"\n", " form[\"sound_duration\"] = f\"{payload['num_frames'] / payload['fps']:.3f}\"\n", " return form\n", "\n", "\n", "def build_sglang_image_body(payload: dict) -> dict:\n", " height, width = payload_dimensions(payload)\n", " return {\n", " \"prompt\": payload[\"prompt\"],\n", " \"negative_prompt\": payload.get(\"negative_prompt\", \"\"),\n", " \"size\": f\"{width}x{height}\",\n", " \"n\": 1,\n", " \"num_inference_steps\": payload[\"num_steps\"],\n", " \"guidance_scale\": payload[\"guidance\"],\n", " \"flow_shift\": payload[\"shift\"],\n", " \"seed\": payload[\"seed\"],\n", " \"response_format\": \"b64_json\",\n", " \"extra_params\": {\n", " \"use_resolution_template\": False,\n", " \"guardrails\": True,\n", " },\n", " }\n", "\n", "\n", "def post_video(*, payload_path: Path, payload: dict, output_path: Path, model: str) -> None:\n", " url = video_api_url(SGLANG_ENDPOINTS[model])\n", " api_key = os.environ.get(\"COSMOS3_SGLANG_API_KEY\") or None\n", " tmp_path = Path(f\"{output_path}.tmp\")\n", " error_path = Path(f\"{output_path}.error.txt\")\n", " if tmp_path.exists():\n", " tmp_path.unlink()\n", " if error_path.exists():\n", " error_path.unlink()\n", "\n", " cmd = [\n", " \"curl\",\n", " \"-sS\",\n", " \"--fail-with-body\",\n", " \"-X\",\n", " \"POST\",\n", " url,\n", " \"-H\",\n", " \"Accept: video/mp4\",\n", " ]\n", " if api_key is not None:\n", " cmd += [\"-H\", f\"Authorization: Bearer {api_key}\"]\n", "\n", " for key, value in build_sglang_video_form(payload).items():\n", " cmd += [\"--form-string\", f\"{key}={value}\"]\n", "\n", " if payload[\"model_mode\"] == \"image2video\":\n", " image_path = resolve_payload_path(payload_path, payload[\"vision_path\"])\n", " cmd += [\"-F\", f\"input_reference=@{image_path}\"]\n", "\n", " if payload[\"model_mode\"] == \"video2video\":\n", " video_path = resolve_payload_path(payload_path, payload[\"vision_path\"])\n", " cmd += [\"-F\", f\"video_reference=@{video_path};type=video/mp4\"]\n", "\n", " result = subprocess.run(cmd, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)\n", " if result.returncode != 0:\n", " error_path.write_text((result.stdout or \"\") + (result.stderr or \"\"))\n", " raise RuntimeError(f\"SGLang request failed with exit code {result.returncode}; see {error_path}\")\n", "\n", " # poll output video\n", " initial = json.loads(result.stdout)\n", " video_id = initial[\"id\"]\n", "\n", " while True:\n", " poll = subprocess.run(\n", " [\"curl\", \"-sS\", \"--fail-with-body\", f\"{url}/{video_id}\"],\n", " text=True,\n", " stdout=subprocess.PIPE,\n", " stderr=subprocess.PIPE,\n", " )\n", " if poll.returncode != 0:\n", " error_path.write_text((poll.stdout or \"\") + (poll.stderr or \"\"))\n", " raise RuntimeError(f\"SGLang video poll failed; see {error_path}\")\n", "\n", " final = json.loads(poll.stdout)\n", " if final.get(\"status\") == \"completed\":\n", " break\n", " if final.get(\"status\") in {\"failed\", \"cancelled\"}:\n", " error_path.write_text(json.dumps(final, indent=2))\n", " raise RuntimeError(f\"SGLang video failed; see {error_path}\")\n", " time.sleep(5)\n", "\n", " download = subprocess.run(\n", " [\"curl\", \"-sS\", \"--fail-with-body\", f\"{url}/{video_id}/content\", \"-o\", str(tmp_path)],\n", " text=True,\n", " stdout=subprocess.PIPE,\n", " stderr=subprocess.PIPE,\n", " )\n", " if download.returncode != 0:\n", " error_path.write_text(json.dumps(final, indent=2) + \"\\n\" + (download.stderr or \"\"))\n", " raise RuntimeError(f\"SGLang video download failed; see {error_path}\")\n", "\n", " tmp_path.replace(output_path)\n", "\n", "\n", "def post_image(*, payload: dict, output_path: Path, model: str) -> None:\n", " url = image_api_url(SGLANG_ENDPOINTS[model])\n", " api_key = os.environ.get(\"COSMOS3_SGLANG_API_KEY\") or None\n", " tmp_path = Path(f\"{output_path}.tmp\")\n", " error_path = Path(f\"{output_path}.error.txt\")\n", " if tmp_path.exists():\n", " tmp_path.unlink()\n", " if error_path.exists():\n", " error_path.unlink()\n", "\n", " cmd = [\n", " \"curl\",\n", " \"-sS\",\n", " \"--fail-with-body\",\n", " \"-X\",\n", " \"POST\",\n", " url,\n", " \"-H\",\n", " \"Content-Type: application/json\",\n", " ]\n", " if api_key is not None:\n", " cmd += [\"-H\", f\"Authorization: Bearer {api_key}\"]\n", " cmd += [\"-d\", json.dumps(build_sglang_image_body(payload), separators=(\",\", \":\"))]\n", "\n", " result = subprocess.run(cmd, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)\n", " if result.returncode != 0:\n", " error_path.write_text((result.stdout or \"\") + (result.stderr or \"\"))\n", " raise RuntimeError(f\"SGLang image request failed with exit code {result.returncode}; see {error_path}\")\n", " try:\n", " response = json.loads(result.stdout)\n", " b64_json = response[\"data\"][0][\"b64_json\"]\n", " tmp_path.write_bytes(base64.b64decode(b64_json))\n", " except Exception as exc:\n", " error_path.write_text((result.stdout or \"\") + (result.stderr or \"\"))\n", " raise RuntimeError(f\"Could not decode SGLang image response; see {error_path}\") from exc\n", " tmp_path.replace(output_path)\n", "\n", "\n", "def run_sglang_payload(payload_path: Path, output_dir: str | Path, *, model: str) -> Path:\n", " payload_path = Path(payload_path)\n", " output_dir = Path(output_dir)\n", " output_dir.mkdir(parents=True, exist_ok=True)\n", " payload = json.loads(payload_path.read_text())\n", " output_ext = \".png\" if payload[\"model_mode\"] == \"text2image\" else \".mp4\"\n", " output_path = output_dir / f\"{payload['name']}{output_ext}\"\n", " endpoint = image_api_url(SGLANG_ENDPOINTS[model]) if payload[\"model_mode\"] == \"text2image\" else video_api_url(SGLANG_ENDPOINTS[model])\n", " print(\"endpoint:\", endpoint)\n", " print(\"payload:\", payload_path)\n", " print(\"output:\", output_path)\n", " if payload[\"model_mode\"] == \"image2video\":\n", " print(\"input image:\", resolve_payload_path(payload_path, payload[\"vision_path\"]))\n", " t0 = time.time()\n", " if payload[\"model_mode\"] == \"text2image\":\n", " post_image(payload=payload, output_path=output_path, model=model)\n", " else:\n", " post_video(payload_path=payload_path, payload=payload, output_path=output_path, model=model)\n", " print(f\"wrote {output_path} in {time.time() - t0:.1f}s\")\n", " return output_path\n", "\n", "\n", "def display_video(path: Path, *, width: int = 720) -> None:\n", " data = base64.b64encode(path.read_bytes()).decode(\"ascii\")\n", " label = html.escape(str(path))\n", " markup = f\"\"\"\n", "\n", "
{label}
\n", "\"\"\"\n", " display(HTML(markup))\n", "\n", "\n", "def view_run(output_dir: str | Path) -> None:\n", " output_dir = Path(output_dir)\n", " videos = [\n", " path\n", " for path in sorted(output_dir.rglob(\"*.mp4\"))\n", " if not path.name.endswith((\"_preview.mp4\", \"_browser.mp4\"))\n", " ]\n", " images = sorted(output_dir.rglob(\"*.png\"))\n", " if not videos and not images:\n", " print(f\"No generated media found under {output_dir}\")\n", " return\n", " for src in videos:\n", " print(f\"source: {src} ({src.stat().st_size // 1024} KB)\")\n", " display_video(src)\n", " for src in images:\n", " print(f\"source: {src} ({src.stat().st_size // 1024} KB)\")\n", " display(Image(filename=str(src), width=720))\n" ] }, { "cell_type": "markdown", "id": "9ffaab0e", "metadata": {}, "source": [ "Run each use case top-to-bottom: create the JSON payload, run inference, then view the generated media. The examples are grouped by model size below. The Cosmos3-Nano and Cosmos3-Super sections are independent, so you can run just one.\n" ] }, { "cell_type": "markdown", "id": "e7d284db", "metadata": {}, "source": [ "# Cosmos3-Nano Examples\n", "\n", "Use cases for the `Cosmos3-Nano` model. This section is self-contained; you can run it without the Cosmos3-Super section below.\n" ] }, { "cell_type": "markdown", "id": "27bf6dce", "metadata": {}, "source": [ "## Nano: Text to Image\n", "\n", "Nano text-to-image generation using a structured JSON prompt.\n", "\n", "### Create Payload\n" ] }, { "cell_type": "code", "execution_count": null, "id": "383f6351", "metadata": {}, "outputs": [], "source": [ "t2i_payload, t2i_output, t2i_model = create_payload(\"t2i\", backend=\"sglang\")\n" ] }, { "cell_type": "markdown", "id": "d3c59500", "metadata": {}, "source": [ "### Run\n" ] }, { "cell_type": "code", "execution_count": null, "id": "97b4b312", "metadata": {}, "outputs": [], "source": [ "run_sglang_payload(t2i_payload, t2i_output, model=\"Cosmos3-Nano\")\n" ] }, { "cell_type": "markdown", "id": "c027b956", "metadata": {}, "source": [ "### View Results\n" ] }, { "cell_type": "code", "execution_count": null, "id": "8fe6e3ed", "metadata": {}, "outputs": [], "source": [ "view_run(t2i_output)\n" ] }, { "cell_type": "markdown", "id": "0e7d3092", "metadata": {}, "source": [ "## Nano: Text to Video Without Audio\n", "\n", "Nano text-to-video generation with audio disabled.\n", "\n", "### Create Payload\n" ] }, { "cell_type": "code", "execution_count": null, "id": "b2e85424", "metadata": {}, "outputs": [], "source": [ "t2v_nano_noaudio_payload, t2v_nano_noaudio_output, t2v_nano_noaudio_model = create_payload(\"t2v_nano_noaudio\", backend=\"sglang\")\n" ] }, { "cell_type": "markdown", "id": "e4fa0cdd", "metadata": {}, "source": [ "### Run\n" ] }, { "cell_type": "code", "execution_count": null, "id": "9db2a1a5", "metadata": {}, "outputs": [], "source": [ "run_sglang_payload(t2v_nano_noaudio_payload, t2v_nano_noaudio_output, model=\"Cosmos3-Nano\")\n" ] }, { "cell_type": "markdown", "id": "1bc8a331", "metadata": {}, "source": [ "### View Results\n" ] }, { "cell_type": "code", "execution_count": null, "id": "87917755", "metadata": {}, "outputs": [], "source": [ "view_run(t2v_nano_noaudio_output)\n" ] }, { "cell_type": "markdown", "id": "816e9b09", "metadata": {}, "source": [ "## Nano: Text to Video with Audio\n", "\n", "Nano text-to-video generation with generated audio.\n", "\n", "### Create Payload\n" ] }, { "cell_type": "code", "execution_count": null, "id": "e8450feb", "metadata": {}, "outputs": [], "source": [ "t2vs_payload, t2vs_output, t2vs_model = create_payload(\"t2vs\", backend=\"sglang\")\n" ] }, { "cell_type": "markdown", "id": "b6bd6ec8", "metadata": {}, "source": [ "### Run\n" ] }, { "cell_type": "code", "execution_count": null, "id": "685d18d6", "metadata": {}, "outputs": [], "source": [ "run_sglang_payload(t2vs_payload, t2vs_output, model=\"Cosmos3-Nano\")\n" ] }, { "cell_type": "markdown", "id": "93743159", "metadata": {}, "source": [ "### View Results\n" ] }, { "cell_type": "code", "execution_count": null, "id": "a4bc049a", "metadata": {}, "outputs": [], "source": [ "view_run(t2vs_output)\n" ] }, { "cell_type": "markdown", "id": "1cf05248", "metadata": {}, "source": [ "## Nano: Image to Video Without Audio\n", "\n", "Nano image-to-video generation using its paired image asset, with audio disabled.\n", "\n", "### Create Payload\n" ] }, { "cell_type": "code", "execution_count": null, "id": "91e28e16", "metadata": {}, "outputs": [], "source": [ "i2v_nano_noaudio_payload, i2v_nano_noaudio_output, i2v_nano_noaudio_model = create_payload(\"i2v_nano_noaudio\", backend=\"sglang\")\n" ] }, { "cell_type": "markdown", "id": "ef42f84d", "metadata": {}, "source": [ "### Run\n" ] }, { "cell_type": "code", "execution_count": null, "id": "4ec848e8", "metadata": {}, "outputs": [], "source": [ "run_sglang_payload(i2v_nano_noaudio_payload, i2v_nano_noaudio_output, model=\"Cosmos3-Nano\")\n" ] }, { "cell_type": "markdown", "id": "dea78891", "metadata": {}, "source": [ "### View Results\n" ] }, { "cell_type": "code", "execution_count": null, "id": "ae7a16ed", "metadata": {}, "outputs": [], "source": [ "view_run(i2v_nano_noaudio_output)\n" ] }, { "cell_type": "markdown", "id": "1860e911", "metadata": {}, "source": [ "## Nano: Image to Video with Audio\n", "\n", "Nano image-to-video generation using its paired image asset and generated audio.\n", "\n", "### Create Payload\n" ] }, { "cell_type": "code", "execution_count": null, "id": "37c40d8e", "metadata": {}, "outputs": [], "source": [ "i2vs_payload, i2vs_output, i2vs_model = create_payload(\"i2vs\", backend=\"sglang\")\n" ] }, { "cell_type": "markdown", "id": "637e5e87", "metadata": {}, "source": [ "### Run\n" ] }, { "cell_type": "code", "execution_count": null, "id": "7c899d2a", "metadata": {}, "outputs": [], "source": [ "run_sglang_payload(i2vs_payload, i2vs_output, model=\"Cosmos3-Nano\")\n" ] }, { "cell_type": "markdown", "id": "62994915", "metadata": {}, "source": [ "### View Results\n" ] }, { "cell_type": "code", "execution_count": null, "id": "58a28e11", "metadata": {}, "outputs": [], "source": [ "view_run(i2vs_output)\n" ] }, { "cell_type": "markdown", "id": "39d8b5a5", "metadata": {}, "source": [ "## Nano: Video to Video with Audio\n", "\n", "Nano video-to-video generation with sound.\n", "\n", "### Create Payload\n" ] }, { "cell_type": "code", "execution_count": null, "id": "6e86736f", "metadata": {}, "outputs": [], "source": [ "v2vs_payload, v2vs_output, v2vs_model = create_payload(\"v2vs\", backend=\"sglang\")" ] }, { "cell_type": "markdown", "id": "37f4d18d", "metadata": {}, "source": [ "### Run\n" ] }, { "cell_type": "code", "execution_count": null, "id": "b4a4c308", "metadata": {}, "outputs": [], "source": [ "run_sglang_payload(v2vs_payload, v2vs_output, model=\"Cosmos3-Nano\")\n" ] }, { "cell_type": "markdown", "id": "bfd6854b", "metadata": {}, "source": [ "### View Results\n" ] }, { "cell_type": "code", "execution_count": null, "id": "b84f2449", "metadata": {}, "outputs": [], "source": [ "view_run(v2vs_output)\n" ] }, { "cell_type": "markdown", "id": "fe62931e", "metadata": {}, "source": [ "# Cosmos3-Super Examples\n", "\n", "The same use cases for the larger `Cosmos3-Super` model. This section is self-contained; you can run it without the Cosmos3-Nano section above.\n" ] }, { "cell_type": "markdown", "id": "466c6bac", "metadata": {}, "source": [ "## Super: Text to Image\n", "\n", "Super text-to-image generation using the same structured JSON prompt.\n", "\n", "### Create Payload\n" ] }, { "cell_type": "code", "execution_count": null, "id": "575653c1", "metadata": {}, "outputs": [], "source": [ "t2i_super_payload, t2i_super_output, t2i_super_model = create_payload(\"t2i_super\", backend=\"sglang\")\n" ] }, { "cell_type": "markdown", "id": "42776c8b", "metadata": {}, "source": [ "### Run\n" ] }, { "cell_type": "code", "execution_count": null, "id": "824783c8", "metadata": {}, "outputs": [], "source": [ "run_sglang_payload(t2i_super_payload, t2i_super_output, model=\"Cosmos3-Super\")\n" ] }, { "cell_type": "markdown", "id": "1ebf0b15", "metadata": {}, "source": [ "### View Results\n" ] }, { "cell_type": "code", "execution_count": null, "id": "b0fba2bb", "metadata": {}, "outputs": [], "source": [ "view_run(t2i_super_output)\n" ] }, { "cell_type": "markdown", "id": "f608eaa7", "metadata": {}, "source": [ "## Super: Text to Video Without Audio\n", "\n", "Super text-to-video generation with audio disabled.\n", "\n", "### Create Payload\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "t2v_super_noaudio_payload, t2v_super_noaudio_output, t2v_super_noaudio_model = create_payload(\"t2v_super_noaudio\", backend=\"sglang\")\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Run\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "run_sglang_payload(t2v_super_noaudio_payload, t2v_super_noaudio_output, model=\"Cosmos3-Super\")\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### View Results\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "view_run(t2v_super_noaudio_output)\n" ] }, { "cell_type": "markdown", "id": "367f4161", "metadata": {}, "source": [ "## Super: Image to Video Without Audio\n", "\n", "Super image-to-video generation using its paired image asset, with audio disabled.\n", "\n", "### Create Payload\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "i2v_super_noaudio_payload, i2v_super_noaudio_output, i2v_super_noaudio_model = create_payload(\"i2v_super_noaudio\", backend=\"sglang\")\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Run\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "run_sglang_payload(i2v_super_noaudio_payload, i2v_super_noaudio_output, model=\"Cosmos3-Super\")\n" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### View Results\n" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "view_run(i2v_super_noaudio_output)\n" ] } ], "metadata": { "kernelspec": { "display_name": ".venv", "language": "python", "name": "python3" }, "language_info": { "codemirror_mode": { "name": "ipython", "version": 3 }, "file_extension": ".py", "mimetype": "text/x-python", "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", "version": "3.12.3" } }, "nbformat": 4, "nbformat_minor": 5 }