| 1 | import os |
| 2 | import shutil |
| 3 | import json |
| 4 | import logging |
| 5 | import asyncio |
| 6 | import time |
| 7 | from typing import Any, Callable, Optional, Dict, List, Tuple, Literal, Type, TypeVar |
| 8 | from moviepy import VideoFileClip, concatenate_videoclips |
| 9 | from PIL import Image |
| 10 | from agents import * |
| 11 | import yaml |
| 12 | from interfaces import * |
| 13 | from langchain.chat_models import init_chat_model |
| 14 | from tools.render_backend import RenderBackend |
| 15 | from utils.provider_presets import resolve_chat_model_config |
| 16 | |
| 17 | |
| 18 | |
| 19 | |
| 20 | TModel = TypeVar("TModel") |
| 21 | |
| 22 | |
| 23 | def _normalize_model_list(items: Any, model_cls: Type[TModel], field_name: str) -> List[TModel]: |
| 24 | if items is None: |
| 25 | return [] |
| 26 | if not isinstance(items, list): |
| 27 | raise TypeError(f"{field_name} must be a list, got {type(items).__name__}") |
| 28 | normalized: List[TModel] = [] |
| 29 | for idx, item in enumerate(items): |
| 30 | if isinstance(item, model_cls): |
| 31 | normalized.append(item) |
| 32 | elif isinstance(item, dict): |
| 33 | normalized.append(model_cls.model_validate(item)) |
| 34 | else: |
| 35 | raise TypeError(f"{field_name}[{idx}] must be {model_cls.__name__} or dict, got {type(item).__name__}") |
| 36 | return normalized |
| 37 | |
| 38 | |
| 39 | def _group_shots_into_cameras(shot_descriptions: List[ShotDescription]) -> List[Camera]: |
| 40 | cameras_by_idx: Dict[int, Camera] = {} |
| 41 | for shot_description in shot_descriptions: |
| 42 | camera = cameras_by_idx.get(shot_description.cam_idx) |
| 43 | if camera is None: |
| 44 | camera = Camera(idx=shot_description.cam_idx, active_shot_idxs=[]) |
| 45 | cameras_by_idx[shot_description.cam_idx] = camera |
| 46 | camera.active_shot_idxs.append(shot_description.idx) |
| 47 | return list(cameras_by_idx.values()) |
| 48 | |
| 49 | def _collect_priority_shot_idxs(camera_tree: List[Camera]) -> List[int]: |
| 50 | """Shot indices that other cameras depend on.""" |
| 51 | return [camera.parent_shot_idx for camera in camera_tree if camera.parent_shot_idx is not None] |
| 52 | |
| 53 | |
| 54 | def _pipeline_print(quiet: bool, message: str) -> None: |
| 55 | if not quiet: |
| 56 | print(message) |
| 57 | |
| 58 | |
| 59 | def _emit_text_plan_progress(progress, stage: str, message: str, metadata: Dict[str, Any] | None = None) -> None: |
| 60 | if progress is not None: |
| 61 | progress(stage, message, metadata or {}) |
| 62 | |
| 63 | |
| 64 | def _emit_render_progress(progress, stage: str, message: str, metadata: Dict[str, Any] | None = None) -> None: |
| 65 | if progress is not None: |
| 66 | progress(stage, message, metadata or {}) |
| 67 | |
| 68 | |
| 69 | def _scoped_progress(progress, **scope): |
| 70 | if progress is None: |
| 71 | return None |
| 72 | |
| 73 | def emit(stage: str, message: str, metadata: Dict[str, Any] | None = None) -> None: |
| 74 | payload = dict(scope) |
| 75 | payload.update(metadata or {}) |
| 76 | _emit_render_progress(progress, stage, message, payload) |
| 77 | |
| 78 | return emit |
| 79 | |
| 80 | |
| 81 | class Script2VideoPipeline: |
| 82 | |
| 83 | def __init__( |
| 84 | self, |
| 85 | chat_model: str, |
| 86 | image_generator, |
| 87 | video_generator, |
| 88 | working_dir: str, |
| 89 | ): |
| 90 | |
| 91 | self.chat_model = chat_model |
| 92 | self.image_generator = image_generator |
| 93 | self.video_generator = video_generator |
| 94 | |
| 95 | self.character_extractor = CharacterExtractor(chat_model=self.chat_model) |
| 96 | self.character_portraits_generator = CharacterPortraitsGenerator(image_generator=self.image_generator) |
| 97 | self.storyboard_artist = StoryboardArtist(chat_model=self.chat_model) |
| 98 | self.camera_image_generator = CameraImageGenerator(chat_model=self.chat_model, image_generator=self.image_generator, video_generator=self.video_generator) |
| 99 | self.reference_image_selector = ReferenceImageSelector(chat_model=self.chat_model) |
| 100 | |
| 101 | self.working_dir = working_dir |
| 102 | os.makedirs(self.working_dir, exist_ok=True) |
| 103 | self.character_portrait_events = {} |
| 104 | self.shot_desc_events = {} |
| 105 | self.frame_events = {} |
| 106 | |
| 107 | |
| 108 | async def plan_text_artifacts( |
| 109 | self, |
| 110 | script: str, |
| 111 | user_requirement: str, |
| 112 | style: str, |
| 113 | characters: List[CharacterInScene] = None, |
| 114 | progress: Callable[[str, str, Dict[str, Any] | None], None] | None = None, |
| 115 | quiet: bool = False, |
| 116 | ): |
| 117 | """Generate only structured text artifacts required before rendering. |
| 118 | |
| 119 | This helper intentionally stops before character portraits, frame generation, |
| 120 | video generation, and final concatenation so an agent loop can pause for |
| 121 | user review after narrative planning. |
| 122 | """ |
| 123 | self.character_portrait_events = {} |
| 124 | self.shot_desc_events = {} |
| 125 | self.frame_events = {} |
| 126 | |
| 127 | if characters is None: |
| 128 | _emit_text_plan_progress(progress, "extract_characters", "Extracting characters from script") |
| 129 | characters = await self.extract_characters(script=script, quiet=quiet) |
| 130 | else: |
| 131 | characters = _normalize_model_list(characters, CharacterInScene, "characters") |
| 132 | _emit_text_plan_progress(progress, "extract_characters", "Using provided characters", {"provided": True, "count": len(characters)}) |
| 133 | characters_path = os.path.join(self.working_dir, "characters.json") |
| 134 | if not os.path.exists(characters_path): |
| 135 | with open(characters_path, "w", encoding="utf-8") as f: |
| 136 | json.dump([character.model_dump() for character in characters], f, ensure_ascii=False, indent=4) |
| 137 | for character in characters: |
| 138 | self.character_portrait_events[character.idx] = asyncio.Event() |
| 139 | |
| 140 | _emit_text_plan_progress(progress, "design_storyboard", "Designing storyboard") |
| 141 | storyboard = await self.design_storyboard( |
| 142 | script=script, |
| 143 | characters=characters, |
| 144 | user_requirement=user_requirement, |
| 145 | quiet=quiet, |
| 146 | ) |
| 147 | _emit_text_plan_progress(progress, "decompose_shots", "Decomposing shot visual descriptions", {"shot_count": len(storyboard)}) |
| 148 | shot_descriptions = await self.decompose_visual_descriptions( |
| 149 | shot_brief_descriptions=storyboard, |
| 150 | characters=characters, |
| 151 | quiet=quiet, |
| 152 | ) |
| 153 | camera_tree = None |
| 154 | for attempt in range(2): |
| 155 | try: |
| 156 | stage = "construct_camera_tree" if attempt == 0 else "construct_camera_tree_retry" |
| 157 | message = "Constructing camera tree" if attempt == 0 else "Retrying camera tree construction after schema/type failure" |
| 158 | _emit_text_plan_progress(progress, stage, message, {"shot_count": len(shot_descriptions), "attempt": attempt + 1}) |
| 159 | camera_tree = await self.construct_camera_tree( |
| 160 | shot_descriptions=shot_descriptions, |
| 161 | quiet=quiet, |
| 162 | ) |
| 163 | break |
| 164 | except Exception: |
| 165 | camera_tree_path = os.path.join(self.working_dir, "camera_tree.json") |
| 166 | if os.path.exists(camera_tree_path): |
| 167 | os.remove(camera_tree_path) |
| 168 | if attempt == 1: |
| 169 | raise |
| 170 | assert camera_tree is not None |
| 171 | return { |
| 172 | "characters": characters, |
| 173 | "storyboard": storyboard, |
| 174 | "shot_descriptions": shot_descriptions, |
| 175 | "camera_tree": camera_tree, |
| 176 | } |
| 177 | |
| 178 | |
| 179 | @classmethod |
| 180 | def init_from_config(cls, config_path: str): |
| 181 | with open(config_path, "r") as f: |
| 182 | config = yaml.safe_load(f) |
| 183 | |
| 184 | chat_model_args = resolve_chat_model_config(config["chat_model"]["init_args"]) |
| 185 | chat_model = init_chat_model(**chat_model_args) |
| 186 | backend = RenderBackend.from_config(config) |
| 187 | |
| 188 | return cls( |
| 189 | chat_model=chat_model, |
| 190 | image_generator=backend.image_generator, |
| 191 | video_generator=backend.video_generator, |
| 192 | working_dir=config["working_dir"], |
| 193 | ) |
| 194 | |
| 195 | async def __call__( |
| 196 | self, |
| 197 | script: str, |
| 198 | user_requirement: str, |
| 199 | style: str, |
| 200 | characters: List[CharacterInScene] = None, |
| 201 | character_portraits_registry: Optional[Dict[str, Dict[str, Dict[str, str]]]] = None, |
| 202 | quiet: bool = False, |
| 203 | progress: Callable[[str, str, Dict[str, Any] | None], None] | None = None, |
| 204 | ): |
| 205 | _emit_render_progress(progress, "render_start", "Starting script2video render") |
| 206 | if characters is None: |
| 207 | _emit_render_progress(progress, "extract_characters", "Extracting characters before render") |
| 208 | characters = await self.extract_characters(script=script, quiet=quiet) |
| 209 | |
| 210 | # characters_path = os.path.join(self.working_dir, "characters.json") |
| 211 | # if os.path.exists(characters_path): |
| 212 | # with open(characters_path, "r", encoding="utf-8") as f: |
| 213 | # characters = [CharacterInScene.model_validate(c) for c in json.load(f)] |
| 214 | # print(f"🚀 Loaded {len(characters)} characters from existing file.") |
| 215 | # else: |
| 216 | # print(f"🔍 Extracting characters from script...") |
| 217 | # characters = await self.extract_characters(script=script) |
| 218 | # with open(characters_path, "w", encoding="utf-8") as f: |
| 219 | # json.dump([c.model_dump() for c in characters], f, ensure_ascii=False, indent=4) |
| 220 | # print(f"☑️ Extracted {len(characters)} characters from script and saved to {characters_path}.") |
| 221 | else: |
| 222 | characters = _normalize_model_list(characters, CharacterInScene, "characters") |
| 223 | _emit_render_progress(progress, "extract_characters", "Using provided characters for render", {"provided": True, "count": len(characters)}) |
| 224 | for character in characters: |
| 225 | self.character_portrait_events[character.idx] = asyncio.Event() |
| 226 | |
| 227 | if character_portraits_registry is None: |
| 228 | character_portraits_registry_path = os.path.join(self.working_dir, "character_portraits_registry.json") |
| 229 | if os.path.exists(character_portraits_registry_path): |
| 230 | with open(character_portraits_registry_path, "r", encoding="utf-8") as f: |
| 231 | character_portraits_registry = json.load(f) |
| 232 | print(f"🚀 Loaded {len(character_portraits_registry)} character portraits from existing file.") |
| 233 | _emit_render_progress(progress, "character_portraits_loaded", "Loaded existing character portraits", {"count": len(character_portraits_registry)}) |
| 234 | else: |
| 235 | print(f"🔍 Generating character portraits...") |
| 236 | _emit_render_progress(progress, "character_portraits_start", "Generating character portraits", {"character_count": len(characters)}) |
| 237 | character_portraits_registry = await self.generate_character_portraits( |
| 238 | characters=characters, |
| 239 | character_portraits_registry=None, |
| 240 | style=style, |
| 241 | progress=progress, |
| 242 | ) |
| 243 | |
| 244 | with open(character_portraits_registry_path, "w", encoding="utf-8") as f: |
| 245 | json.dump(character_portraits_registry, f, ensure_ascii=False, indent=4) |
| 246 | print(f"☑️ Generated {len(character_portraits_registry)} character portraits and saved to {character_portraits_registry_path}.") |
| 247 | _emit_render_progress(progress, "character_portraits_done", "Character portraits ready", {"count": len(character_portraits_registry)}) |
| 248 | |
| 249 | |
| 250 | |
| 251 | # design shots |
| 252 | _emit_render_progress(progress, "load_storyboard", "Loading or designing storyboard") |
| 253 | storyboard = await self.design_storyboard( |
| 254 | script=script, |
| 255 | characters=characters, |
| 256 | user_requirement=user_requirement, |
| 257 | quiet=quiet, |
| 258 | ) |
| 259 | _emit_render_progress(progress, "storyboard_ready", "Storyboard ready", {"shot_count": len(storyboard)}) |
| 260 | |
| 261 | # decompose visual descriptions of shots |
| 262 | _emit_render_progress(progress, "load_shot_descriptions", "Loading or decomposing shot descriptions", {"shot_count": len(storyboard)}) |
| 263 | shot_descriptions = await self.decompose_visual_descriptions( |
| 264 | shot_brief_descriptions=storyboard, |
| 265 | characters=characters, |
| 266 | quiet=quiet, |
| 267 | ) |
| 268 | _emit_render_progress(progress, "shot_descriptions_ready", "Shot descriptions ready", {"shot_count": len(shot_descriptions)}) |
| 269 | |
| 270 | # construct camera tree |
| 271 | _emit_render_progress(progress, "load_camera_tree", "Loading or constructing camera tree", {"shot_count": len(shot_descriptions)}) |
| 272 | camera_tree = await self.construct_camera_tree( |
| 273 | shot_descriptions=shot_descriptions, |
| 274 | quiet=quiet, |
| 275 | ) |
| 276 | _emit_render_progress(progress, "camera_tree_ready", "Camera tree ready", {"camera_count": len(camera_tree)}) |
| 277 | |
| 278 | priority_shot_idxs = [camera.parent_cam_idx for camera in camera_tree if camera.parent_cam_idx is not None] |
| 279 | _emit_render_progress(progress, "frames_start", "Generating frames for cameras", {"camera_count": len(camera_tree), "shot_count": len(shot_descriptions)}) |
| 280 | tasks = [ |
| 281 | self.generate_frames_for_single_camera( |
| 282 | camera=camera, |
| 283 | shot_descriptions=shot_descriptions, |
| 284 | characters=characters, |
| 285 | character_portraits_registry=character_portraits_registry, |
| 286 | priority_shot_idxs=priority_shot_idxs, |
| 287 | progress=progress, |
| 288 | ) |
| 289 | for camera in camera_tree |
| 290 | ] |
| 291 | |
| 292 | _emit_render_progress(progress, "video_clips_start", "Generating video clips for shots", {"shot_count": len(shot_descriptions)}) |
| 293 | video_tasks = [ |
| 294 | self.generate_video_for_single_shot( |
| 295 | shot_description=shot_description, |
| 296 | progress=progress, |
| 297 | ) |
| 298 | for shot_description in shot_descriptions |
| 299 | ] |
| 300 | tasks.extend(video_tasks) |
| 301 | await asyncio.gather(*tasks) |
| 302 | |
| 303 | final_video_path = os.path.join(self.working_dir, "final_video.mp4") |
| 304 | if os.path.exists(final_video_path): |
| 305 | print(f"🚀 Skipped concatenating videos, already exists.") |
| 306 | _emit_render_progress(progress, "final_video_exists", "Final video already exists", {"path": final_video_path}) |
| 307 | else: |
| 308 | print(f"🎬 Starting concatenating videos...") |
| 309 | _emit_render_progress(progress, "concat_start", "Concatenating video clips", {"shot_count": len(shot_descriptions)}) |
| 310 | video_clips = [ |
| 311 | VideoFileClip(os.path.join(self.working_dir, "shots", f"{shot_description.idx}", "video.mp4")) |
| 312 | for shot_description in shot_descriptions |
| 313 | ] |
| 314 | final_video = concatenate_videoclips(video_clips) |
| 315 | final_video.write_videofile(final_video_path, codec="libx264", preset="medium") |
| 316 | print(f"☑️ Concatenated videos, saved to {final_video_path}.") |
| 317 | _emit_render_progress(progress, "concat_done", "Final video concatenated", {"path": final_video_path}) |
| 318 | |
| 319 | _emit_render_progress(progress, "render_done", "Script2video render complete", {"final_video_path": final_video_path}) |
| 320 | return final_video_path |
| 321 | |
| 322 | |
| 323 | async def generate_frames_for_single_camera( |
| 324 | self, |
| 325 | camera: Camera, |
| 326 | shot_descriptions: List[ShotDescription], |
| 327 | characters: List[CharacterInScene], |
| 328 | character_portraits_registry: Dict[str, Dict[str, Dict[str, str]]], |
| 329 | priority_shot_idxs: List[int], |
| 330 | progress: Callable[[str, str, Dict[str, Any] | None], None] | None = None, |
| 331 | ): |
| 332 | # 1. generate the first_frame of the first shot of the camera |
| 333 | first_shot_idx = camera.active_shot_idxs[0] |
| 334 | first_shot_ff_path = os.path.join(self.working_dir, "shots", f"{first_shot_idx}", "first_frame.png") |
| 335 | _emit_render_progress(progress, "camera_frames_start", f"Generating frames for camera {camera.idx}", {"camera_idx": camera.idx, "active_shot_idxs": camera.active_shot_idxs}) |
| 336 | |
| 337 | if os.path.exists(first_shot_ff_path): |
| 338 | print(f"🚀 Skipped generating first_frame for shot {first_shot_idx}, already exists.") |
| 339 | self.frame_events[first_shot_idx]["first_frame"].set() |
| 340 | _emit_render_progress(progress, "frame_exists", f"First frame for shot {first_shot_idx} already exists", {"camera_idx": camera.idx, "shot_idx": first_shot_idx, "frame_type": "first_frame", "path": first_shot_ff_path}) |
| 341 | |
| 342 | else: |
| 343 | print(f"🖼️ Starting first_frame generation for shot {first_shot_idx}...") |
| 344 | _emit_render_progress(progress, "frame_start", f"Generating first frame for shot {first_shot_idx}", {"camera_idx": camera.idx, "shot_idx": first_shot_idx, "frame_type": "first_frame"}) |
| 345 | available_image_path_and_text_pairs = [] |
| 346 | |
| 347 | for character_idx in shot_descriptions[first_shot_idx].ff_vis_char_idxs: |
| 348 | identifier_in_scene = characters[character_idx].identifier_in_scene |
| 349 | registry_item = character_portraits_registry[identifier_in_scene] |
| 350 | for view, item in registry_item.items(): |
| 351 | available_image_path_and_text_pairs.append((item["path"], item["description"])) |
| 352 | |
| 353 | # generate the first_frame based on the shot_description.ff_desc |
| 354 | if camera.parent_shot_idx is not None: |
| 355 | # generate the first_frame based on the transition video |
| 356 | parent_shot_idx = camera.parent_shot_idx |
| 357 | await self.frame_events[parent_shot_idx]["first_frame"].wait() |
| 358 | parent_shot_ff_path = os.path.join(self.working_dir, "shots", f"{parent_shot_idx}", "first_frame.png") |
| 359 | transition_video_path = os.path.join(self.working_dir, "shots", f"{first_shot_idx}", f"transition_video_from_shot_{parent_shot_idx}.mp4") |
| 360 | |
| 361 | if os.path.exists(transition_video_path): |
| 362 | print(f"🚀 Skipped generating transition video for shot {first_shot_idx} from shot {parent_shot_idx}, already exists.") |
| 363 | _emit_render_progress(progress, "transition_video_exists", f"Transition video for shot {first_shot_idx} already exists", {"camera_idx": camera.idx, "shot_idx": first_shot_idx, "parent_shot_idx": parent_shot_idx, "path": transition_video_path}) |
| 364 | else: |
| 365 | print(f"🖼️ Starting transition video generation for shot {first_shot_idx} from shot {parent_shot_idx}...") |
| 366 | _emit_render_progress(progress, "transition_video_start", f"Generating transition video for shot {first_shot_idx}", {"camera_idx": camera.idx, "shot_idx": first_shot_idx, "parent_shot_idx": parent_shot_idx}) |
| 367 | transition_video_output = await self.camera_image_generator.generate_transition_video( |
| 368 | first_shot_visual_desc=shot_descriptions[parent_shot_idx].visual_desc, |
| 369 | second_shot_visual_desc=shot_descriptions[first_shot_idx].visual_desc, |
| 370 | first_shot_ff_path=parent_shot_ff_path, |
| 371 | progress=_scoped_progress(progress, camera_idx=camera.idx, shot_idx=first_shot_idx, parent_shot_idx=parent_shot_idx, artifact="transition_video"), |
| 372 | ) |
| 373 | transition_video_output.save(transition_video_path) |
| 374 | print(f"☑️ Generated transition video for shot {first_shot_idx} from shot {parent_shot_idx}, saved to {transition_video_path}.") |
| 375 | _emit_render_progress(progress, "transition_video_done", f"Transition video for shot {first_shot_idx} generated", {"camera_idx": camera.idx, "shot_idx": first_shot_idx, "parent_shot_idx": parent_shot_idx, "path": transition_video_path}) |
| 376 | |
| 377 | new_camera_image_path = os.path.join(self.working_dir, "shots", f"{first_shot_idx}", f"new_camera_{camera.idx}.png") |
| 378 | if os.path.exists(new_camera_image_path): |
| 379 | print(f"🚀 Skipped generating new camera image for shot {first_shot_idx}, already exists.") |
| 380 | _emit_render_progress(progress, "new_camera_image_exists", f"New camera image for shot {first_shot_idx} already exists", {"camera_idx": camera.idx, "shot_idx": first_shot_idx, "path": new_camera_image_path}) |
| 381 | else: |
| 382 | print(f"🖼️ Starting new camera image generation for shot {first_shot_idx}...") |
| 383 | _emit_render_progress(progress, "new_camera_image_start", f"Extracting new camera image for shot {first_shot_idx}", {"camera_idx": camera.idx, "shot_idx": first_shot_idx}) |
| 384 | new_camera_image = self.camera_image_generator.get_new_camera_image(transition_video_path) |
| 385 | new_camera_image.save(new_camera_image_path) |
| 386 | print(f"☑️ Generated new camera image for shot {first_shot_idx} (not completed), saved to {new_camera_image_path}.") |
| 387 | _emit_render_progress(progress, "new_camera_image_done", f"New camera image for shot {first_shot_idx} extracted", {"camera_idx": camera.idx, "shot_idx": first_shot_idx, "path": new_camera_image_path}) |
| 388 | |
| 389 | available_image_path_and_text_pairs.append( |
| 390 | ( |
| 391 | new_camera_image_path, |
| 392 | f"The composition and background are correct but some elements may be wrong. The wrong elements should be replaced.\nWrong elements: {camera.missing_info}.\nYou must select this image as the main reference and replace the characters in the image with the provided character portraits. Don't change the background." |
| 393 | ) |
| 394 | ) |
| 395 | |
| 396 | |
| 397 | # 如果子镜头缺少信息，则需要选择参考图像生成 |
| 398 | if camera.parent_shot_idx is None or camera.missing_info is not None: |
| 399 | ff_selector_output_path = os.path.join(self.working_dir, "shots", f"{first_shot_idx}", "first_frame_selector_output.json") |
| 400 | if os.path.exists(ff_selector_output_path): |
| 401 | with open(ff_selector_output_path, 'r', encoding='utf-8') as f: |
| 402 | ff_selector_output = json.load(f) |
| 403 | print(f"🚀 Loaded existing reference image selection and prompt for first_frame of shot {first_shot_idx} from {ff_selector_output_path}.") |
| 404 | _emit_render_progress(progress, "frame_prompt_exists", f"First frame prompt for shot {first_shot_idx} already exists", {"camera_idx": camera.idx, "shot_idx": first_shot_idx, "frame_type": "first_frame", "path": ff_selector_output_path}) |
| 405 | else: |
| 406 | print(f"🔍 Selecting reference images and generating prompt for first_frame of shot {first_shot_idx}...") |
| 407 | _emit_render_progress(progress, "frame_prompt_start", f"Selecting references for first frame of shot {first_shot_idx}", {"camera_idx": camera.idx, "shot_idx": first_shot_idx, "frame_type": "first_frame"}) |
| 408 | ff_selector_output = await self.reference_image_selector.select_reference_images_and_generate_prompt( |
| 409 | available_image_path_and_text_pairs=available_image_path_and_text_pairs, |
| 410 | frame_description=shot_descriptions[first_shot_idx].ff_desc |
| 411 | ) |
| 412 | with open(ff_selector_output_path, 'w', encoding='utf-8') as f: |
| 413 | json.dump(ff_selector_output, f, ensure_ascii=False, indent=4) |
| 414 | |
| 415 | print(f"☑️ Selected reference images and generated prompt for first_frame of shot {first_shot_idx}, saved to {ff_selector_output_path}.") |
| 416 | _emit_render_progress(progress, "frame_prompt_done", f"Selected references for first frame of shot {first_shot_idx}", {"camera_idx": camera.idx, "shot_idx": first_shot_idx, "frame_type": "first_frame", "path": ff_selector_output_path}) |
| 417 | |
| 418 | reference_image_path_and_text_pairs, prompt = ff_selector_output["reference_image_path_and_text_pairs"], ff_selector_output["text_prompt"] |
| 419 | prefix_prompt = "" |
| 420 | for i, (image_path, text) in enumerate(reference_image_path_and_text_pairs): |
| 421 | prefix_prompt += f"Image {i}: {text}\n" |
| 422 | prompt = f"{prefix_prompt}\n{prompt}" |
| 423 | reference_image_paths = [item[0] for item in reference_image_path_and_text_pairs] |
| 424 | ff_image: ImageOutput = await self.image_generator.generate_single_image( |
| 425 | prompt=prompt, |
| 426 | reference_image_paths=reference_image_paths, |
| 427 | size="1600x900", |
| 428 | ) |
| 429 | ff_image.save(first_shot_ff_path) |
| 430 | self.frame_events[first_shot_idx]["first_frame"].set() |
| 431 | print(f"☑️ Generated first_frame for shot {first_shot_idx}, saved to {first_shot_ff_path}.") |
| 432 | _emit_render_progress(progress, "frame_done", f"Generated first frame for shot {first_shot_idx}", {"camera_idx": camera.idx, "shot_idx": first_shot_idx, "frame_type": "first_frame", "path": first_shot_ff_path}) |
| 433 | else: |
| 434 | shutil.copy(new_camera_image_path, first_shot_ff_path) |
| 435 | self.frame_events[first_shot_idx]["first_frame"].set() |
| 436 | print(f"☑️ Generated first_frame for shot {first_shot_idx}, saved to {first_shot_ff_path}.") |
| 437 | _emit_render_progress(progress, "frame_done", f"Generated first frame for shot {first_shot_idx}", {"camera_idx": camera.idx, "shot_idx": first_shot_idx, "frame_type": "first_frame", "path": first_shot_ff_path}) |
| 438 | |
| 439 | |
| 440 | # 2. generate the following frames of the camera |
| 441 | priority_tasks = [] |
| 442 | normal_tasks = [] |
| 443 | |
| 444 | if shot_descriptions[first_shot_idx].variation_type in ["medium", "large"]: |
| 445 | task = self.generate_frame_for_single_shot( |
| 446 | shot_idx=first_shot_idx, |
| 447 | frame_type="last_frame", |
| 448 | first_shot_ff_path_and_text_pair=(first_shot_ff_path, shot_descriptions[first_shot_idx].ff_desc), |
| 449 | frame_desc=shot_descriptions[first_shot_idx].lf_desc, |
| 450 | visible_characters=[characters[idx] for idx in shot_descriptions[first_shot_idx].lf_vis_char_idxs], |
| 451 | character_portraits_registry=character_portraits_registry, |
| 452 | progress=progress, |
| 453 | ) |
| 454 | normal_tasks.append(task) |
| 455 | |
| 456 | for shot_idx in camera.active_shot_idxs[1:]: |
| 457 | first_frame_task = self.generate_frame_for_single_shot( |
| 458 | shot_idx=shot_idx, |
| 459 | frame_type="first_frame", |
| 460 | first_shot_ff_path_and_text_pair=(first_shot_ff_path, shot_descriptions[first_shot_idx].ff_desc), |
| 461 | frame_desc=shot_descriptions[shot_idx].ff_desc, |
| 462 | visible_characters=[characters[idx] for idx in shot_descriptions[shot_idx].ff_vis_char_idxs], |
| 463 | character_portraits_registry=character_portraits_registry, |
| 464 | progress=progress, |
| 465 | ) |
| 466 | if shot_idx in priority_shot_idxs: |
| 467 | priority_tasks.append(first_frame_task) |
| 468 | else: |
| 469 | normal_tasks.append(first_frame_task) |
| 470 | |
| 471 | |
| 472 | if shot_descriptions[shot_idx].variation_type in ["medium", "large"]: |
| 473 | last_frame_task = self.generate_frame_for_single_shot( |
| 474 | shot_idx=shot_idx, |
| 475 | frame_type="last_frame", |
| 476 | first_shot_ff_path_and_text_pair=(first_shot_ff_path, shot_descriptions[first_shot_idx].ff_desc), |
| 477 | frame_desc=shot_descriptions[shot_idx].lf_desc, |
| 478 | visible_characters=[characters[idx] for idx in shot_descriptions[shot_idx].lf_vis_char_idxs], |
| 479 | character_portraits_registry=character_portraits_registry, |
| 480 | progress=progress, |
| 481 | ) |
| 482 | normal_tasks.append(last_frame_task) |
| 483 | |
| 484 | |
| 485 | await asyncio.gather(*priority_tasks) |
| 486 | await asyncio.gather(*normal_tasks) |
| 487 | _emit_render_progress(progress, "camera_frames_done", f"Frames for camera {camera.idx} ready", {"camera_idx": camera.idx, "active_shot_idxs": camera.active_shot_idxs}) |
| 488 | |
| 489 | |
| 490 | |
| 491 | async def generate_video_for_single_shot( |
| 492 | self, |
| 493 | shot_description: ShotDescription, |
| 494 | progress: Callable[[str, str, Dict[str, Any] | None], None] | None = None, |
| 495 | ): |
| 496 | video_path = os.path.join(self.working_dir, "shots", f"{shot_description.idx}", "video.mp4") |
| 497 | if os.path.exists(video_path): |
| 498 | print(f"🚀 Skipped generating video for shot {shot_description.idx}, already exists.") |
| 499 | _emit_render_progress(progress, "video_clip_exists", f"Video clip for shot {shot_description.idx} already exists", {"shot_idx": shot_description.idx, "path": video_path}) |
| 500 | else: |
| 501 | _emit_render_progress(progress, "video_clip_waiting_for_frames", f"Waiting for frames before video clip {shot_description.idx}", {"shot_idx": shot_description.idx}) |
| 502 | await self.frame_events[shot_description.idx]["first_frame"].wait() |
| 503 | if shot_description.variation_type in ["medium", "large"]: |
| 504 | await self.frame_events[shot_description.idx]["last_frame"].wait() |
| 505 | |
| 506 | frame_paths = [] |
| 507 | frame_paths.append(os.path.join(self.working_dir, "shots", f"{shot_description.idx}", "first_frame.png")) |
| 508 | if shot_description.variation_type in ["medium", "large"]: |
| 509 | frame_paths.append(os.path.join(self.working_dir, "shots", f"{shot_description.idx}", "last_frame.png")) |
| 510 | |
| 511 | print(f"🎬 Starting video generation for shot {shot_description.idx}...") |
| 512 | _emit_render_progress(progress, "video_clip_start", f"Generating video clip for shot {shot_description.idx}", {"shot_idx": shot_description.idx, "frame_count": len(frame_paths)}) |
| 513 | video_output = await self.video_generator.generate_single_video( |
| 514 | prompt=shot_description.motion_desc + "\n" + shot_description.audio_desc, |
| 515 | reference_image_paths=frame_paths, |
| 516 | progress=_scoped_progress(progress, shot_idx=shot_description.idx, artifact="video_clip"), |
| 517 | ) |
| 518 | video_output.save(video_path) |
| 519 | print(f"☑️ Generated video for shot {shot_description.idx}, saved to {video_path}.") |
| 520 | _emit_render_progress(progress, "video_clip_done", f"Generated video clip for shot {shot_description.idx}", {"shot_idx": shot_description.idx, "path": video_path}) |
| 521 | |
| 522 | async def generate_frame_for_single_shot( |
| 523 | self, |
| 524 | shot_idx: int, |
| 525 | frame_type: Literal["first_frame", "last_frame"], |
| 526 | first_shot_ff_path_and_text_pair: Tuple[str, str], |
| 527 | frame_desc: str, |
| 528 | visible_characters: List[CharacterInScene], |
| 529 | character_portraits_registry: Dict[str, Dict[str, Dict[str, str]]], |
| 530 | progress: Callable[[str, str, Dict[str, Any] | None], None] | None = None, |
| 531 | ) -> ImageOutput: |
| 532 | |
| 533 | frame_image_path = os.path.join(self.working_dir, "shots", f"{shot_idx}", f"{frame_type}.png") |
| 534 | |
| 535 | if os.path.exists(frame_image_path): |
| 536 | print(f"🚀 Skipped generating {frame_type} for shot {shot_idx}, already exists.") |
| 537 | _emit_render_progress(progress, "frame_exists", f"{frame_type} for shot {shot_idx} already exists", {"shot_idx": shot_idx, "frame_type": frame_type, "path": frame_image_path}) |
| 538 | |
| 539 | else: |
| 540 | print(f"🖼️ Starting {frame_type} generation for shot {shot_idx}...") |
| 541 | _emit_render_progress(progress, "frame_start", f"Generating {frame_type} for shot {shot_idx}", {"shot_idx": shot_idx, "frame_type": frame_type}) |
| 542 | available_image_path_and_text_pairs = [] |
| 543 | for visible_character in visible_characters: |
| 544 | identifier_in_scene = visible_character.identifier_in_scene |
| 545 | registry_item = character_portraits_registry[identifier_in_scene] |
| 546 | for view, item in registry_item.items(): |
| 547 | available_image_path_and_text_pairs.append((item["path"], item["description"])) |
| 548 | |
| 549 | available_image_path_and_text_pairs.append(first_shot_ff_path_and_text_pair) |
| 550 | |
| 551 | selector_output_path = os.path.join(self.working_dir, "shots", f"{shot_idx}", f"{frame_type}_selector_output.json") |
| 552 | if os.path.exists(selector_output_path): |
| 553 | with open(selector_output_path, 'r', encoding='utf-8') as f: |
| 554 | selector_output = json.load(f) |
| 555 | print(f"🚀 Loaded existing reference image selection and prompt for {frame_type} frame of shot {shot_idx} from {selector_output_path}.") |
| 556 | _emit_render_progress(progress, "frame_prompt_exists", f"Prompt for {frame_type} of shot {shot_idx} already exists", {"shot_idx": shot_idx, "frame_type": frame_type, "path": selector_output_path}) |
| 557 | else: |
| 558 | print(f"🔍 Selecting reference images and generating prompt for {frame_type} frame of shot {shot_idx}...") |
| 559 | _emit_render_progress(progress, "frame_prompt_start", f"Selecting references for {frame_type} of shot {shot_idx}", {"shot_idx": shot_idx, "frame_type": frame_type}) |
| 560 | selector_output = await self.reference_image_selector.select_reference_images_and_generate_prompt( |
| 561 | available_image_path_and_text_pairs=available_image_path_and_text_pairs, |
| 562 | frame_description=frame_desc |
| 563 | ) |
| 564 | with open(selector_output_path, 'w', encoding='utf-8') as f: |
| 565 | json.dump(selector_output, f, ensure_ascii=False, indent=4) |
| 566 | print(f"☑️ Selected reference images and generated prompt for {frame_type} frame of shot {shot_idx}, saved to {selector_output_path}.") |
| 567 | _emit_render_progress(progress, "frame_prompt_done", f"Selected references for {frame_type} of shot {shot_idx}", {"shot_idx": shot_idx, "frame_type": frame_type, "path": selector_output_path}) |
| 568 | |
| 569 | reference_image_path_and_text_pairs, prompt = selector_output["reference_image_path_and_text_pairs"], selector_output["text_prompt"] |
| 570 | prefix_prompt = "" |
| 571 | for i, (image_path, text) in enumerate(reference_image_path_and_text_pairs): |
| 572 | prefix_prompt += f"Image {i}: {text}\n" |
| 573 | prompt = f"{prefix_prompt}\n{prompt}" |
| 574 | reference_image_paths = [item[0] for item in reference_image_path_and_text_pairs] |
| 575 | |
| 576 | frame_image: ImageOutput = await self.image_generator.generate_single_image( |
| 577 | prompt=prompt, |
| 578 | reference_image_paths=reference_image_paths, |
| 579 | size="1600x900", |
| 580 | ) |
| 581 | frame_image.save(frame_image_path) |
| 582 | print(f"☑️ Generated {frame_type} frame for shot {shot_idx}, saved to {frame_image_path}.") |
| 583 | _emit_render_progress(progress, "frame_done", f"Generated {frame_type} for shot {shot_idx}", {"shot_idx": shot_idx, "frame_type": frame_type, "path": frame_image_path}) |
| 584 | |
| 585 | |
| 586 | self.frame_events[shot_idx][frame_type].set() |
| 587 | return frame_image_path |
| 588 | |
| 589 | |
| 590 | async def construct_camera_tree( |
| 591 | self, |
| 592 | shot_descriptions: List[ShotDescription], |
| 593 | quiet: bool = False, |
| 594 | ): |
| 595 | camera_tree_path = os.path.join(self.working_dir, "camera_tree.json") |
| 596 | |
| 597 | if os.path.exists(camera_tree_path): |
| 598 | with open(camera_tree_path, "r", encoding="utf-8") as f: |
| 599 | camera_tree = json.load(f) |
| 600 | camera_tree = [Camera.model_validate(camera) for camera in camera_tree] |
| 601 | _pipeline_print(quiet, f"🚀 Loaded {len(camera_tree)} cameras from existing file.") |
| 602 | return camera_tree |
| 603 | |
| 604 | shot_descriptions = _normalize_model_list(shot_descriptions, ShotDescription, "shot_descriptions") |
| 605 | cameras = _group_shots_into_cameras(shot_descriptions) |
| 606 | |
| 607 | camera_tree = await self.camera_image_generator.construct_camera_tree(cameras=cameras, shot_descs=shot_descriptions) |
| 608 | camera_tree = _normalize_model_list(camera_tree, Camera, "camera_tree") |
| 609 | with open(camera_tree_path, "w", encoding="utf-8") as f: |
| 610 | json.dump([camera.model_dump() for camera in camera_tree], f, ensure_ascii=False, indent=4) |
| 611 | _pipeline_print(quiet, f"✅ Constructed camera tree and saved to {camera_tree_path}.") |
| 612 | return camera_tree |
| 613 | |
| 614 | |
| 615 | |
| 616 | |
| 617 | async def extract_characters( |
| 618 | self, |
| 619 | script: str, |
| 620 | quiet: bool = False, |
| 621 | ): |
| 622 | save_path = os.path.join(self.working_dir, "characters.json") |
| 623 | |
| 624 | if os.path.exists(save_path): |
| 625 | with open(save_path, "r", encoding="utf-8") as f: |
| 626 | characters = json.load(f) |
| 627 | characters = [CharacterInScene.model_validate(character) for character in characters] |
| 628 | _pipeline_print(quiet, f"🚀 Loaded {len(characters)} characters from existing file.") |
| 629 | else: |
| 630 | characters = await self.character_extractor.extract_characters(script) |
| 631 | with open(save_path, "w", encoding="utf-8") as f: |
| 632 | json.dump([character.model_dump() for character in characters], f, ensure_ascii=False, indent=4) |
| 633 | _pipeline_print(quiet, f"✅ Extracted {len(characters)} characters from script and saved to {save_path}.") |
| 634 | |
| 635 | for character in characters: |
| 636 | self.character_portrait_events[character.idx] = asyncio.Event() |
| 637 | |
| 638 | return characters |
| 639 | |
| 640 | |
| 641 | async def generate_character_portraits( |
| 642 | self, |
| 643 | characters: List[CharacterInScene], |
| 644 | character_portraits_registry: Optional[Dict[str, Dict[str, Dict[str, str]]]], |
| 645 | style: str, |
| 646 | progress: Callable[[str, str, Dict[str, Any] | None], None] | None = None, |
| 647 | ): |
| 648 | character_portraits_registry_path = os.path.join(self.working_dir, "character_portraits_registry.json") |
| 649 | if character_portraits_registry is None: |
| 650 | if os.path.exists(character_portraits_registry_path): |
| 651 | with open(character_portraits_registry_path, 'r', encoding='utf-8') as f: |
| 652 | character_portraits_registry = json.load(f) |
| 653 | else: |
| 654 | character_portraits_registry = {} |
| 655 | |
| 656 | |
| 657 | tasks = [ |
| 658 | self.generate_portraits_for_single_character(character, style, progress=progress) |
| 659 | for character in characters |
| 660 | if character.identifier_in_scene not in character_portraits_registry |
| 661 | ] |
| 662 | if tasks: |
| 663 | for future in asyncio.as_completed(tasks): |
| 664 | character_portraits_registry.update(await future) |
| 665 | with open(character_portraits_registry_path, 'w', encoding='utf-8') as f: |
| 666 | json.dump(character_portraits_registry, f, ensure_ascii=False, indent=4) |
| 667 | |
| 668 | print(f"✅ Completed character portrait generation for {len(characters)} characters.") |
| 669 | _emit_render_progress(progress, "character_portraits_done", "Completed character portrait generation", {"character_count": len(characters)}) |
| 670 | else: |
| 671 | print("🚀 All characters already have portraits, skipping portrait generation.") |
| 672 | _emit_render_progress(progress, "character_portraits_exist", "All character portraits already exist", {"character_count": len(characters)}) |
| 673 | return character_portraits_registry |
| 674 | |
| 675 | |
| 676 | async def generate_portraits_for_single_character( |
| 677 | self, |
| 678 | character: CharacterInScene, |
| 679 | style: str, |
| 680 | progress: Callable[[str, str, Dict[str, Any] | None], None] | None = None, |
| 681 | ): |
| 682 | character_dir = os.path.join(self.working_dir, "character_portraits", f"{character.idx}_{character.identifier_in_scene}") |
| 683 | os.makedirs(character_dir, exist_ok=True) |
| 684 | _emit_render_progress(progress, "character_portrait_start", f"Generating portraits for {character.identifier_in_scene}", {"character_idx": character.idx, "identifier": character.identifier_in_scene}) |
| 685 | |
| 686 | front_portrait_path = os.path.join(character_dir, "front.png") |
| 687 | if os.path.exists(front_portrait_path): |
| 688 | pass |
| 689 | else: |
| 690 | _emit_render_progress(progress, "character_portrait_front_start", f"Generating front portrait for {character.identifier_in_scene}", {"character_idx": character.idx, "identifier": character.identifier_in_scene}) |
| 691 | front_portrait_output = await self.character_portraits_generator.generate_front_portrait(character, style) |
| 692 | front_portrait_output.save(front_portrait_path) |
| 693 | _emit_render_progress(progress, "character_portrait_front_done", f"Generated front portrait for {character.identifier_in_scene}", {"character_idx": character.idx, "identifier": character.identifier_in_scene, "path": front_portrait_path}) |
| 694 | |
| 695 | |
| 696 | side_portrait_path = os.path.join(character_dir, "side.png") |
| 697 | if os.path.exists(side_portrait_path): |
| 698 | pass |
| 699 | else: |
| 700 | _emit_render_progress(progress, "character_portrait_side_start", f"Generating side portrait for {character.identifier_in_scene}", {"character_idx": character.idx, "identifier": character.identifier_in_scene}) |
| 701 | side_portrait_output = await self.character_portraits_generator.generate_side_portrait(character, front_portrait_path) |
| 702 | side_portrait_output.save(side_portrait_path) |
| 703 | _emit_render_progress(progress, "character_portrait_side_done", f"Generated side portrait for {character.identifier_in_scene}", {"character_idx": character.idx, "identifier": character.identifier_in_scene, "path": side_portrait_path}) |
| 704 | |
| 705 | back_portrait_path = os.path.join(character_dir, "back.png") |
| 706 | if os.path.exists(back_portrait_path): |
| 707 | pass |
| 708 | else: |
| 709 | _emit_render_progress(progress, "character_portrait_back_start", f"Generating back portrait for {character.identifier_in_scene}", {"character_idx": character.idx, "identifier": character.identifier_in_scene}) |
| 710 | back_portrait_output = await self.character_portraits_generator.generate_back_portrait(character, front_portrait_path) |
| 711 | back_portrait_output.save(back_portrait_path) |
| 712 | _emit_render_progress(progress, "character_portrait_back_done", f"Generated back portrait for {character.identifier_in_scene}", {"character_idx": character.idx, "identifier": character.identifier_in_scene, "path": back_portrait_path}) |
| 713 | |
| 714 | self.character_portrait_events[character.idx].set() |
| 715 | |
| 716 | print(f"☑️ Completed character portrait generation for {character.identifier_in_scene}.") |
| 717 | _emit_render_progress(progress, "character_portrait_done", f"Portraits for {character.identifier_in_scene} ready", {"character_idx": character.idx, "identifier": character.identifier_in_scene}) |
| 718 | |
| 719 | return { |
| 720 | character.identifier_in_scene: { |
| 721 | "front": { |
| 722 | "path": front_portrait_path, |
| 723 | "description": f"A front view portrait of {character.identifier_in_scene}.", |
| 724 | }, |
| 725 | "side": { |
| 726 | "path": side_portrait_path, |
| 727 | "description": f"A side view portrait of {character.identifier_in_scene}.", |
| 728 | }, |
| 729 | "back": { |
| 730 | "path": back_portrait_path, |
| 731 | "description": f"A back view portrait of {character.identifier_in_scene}.", |
| 732 | }, |
| 733 | } |
| 734 | } |
| 735 | |
| 736 | |
| 737 | |
| 738 | async def design_storyboard( |
| 739 | self, |
| 740 | script: str, |
| 741 | characters: List[CharacterInScene], |
| 742 | user_requirement: str, |
| 743 | quiet: bool = False, |
| 744 | ): |
| 745 | storyboard_path = os.path.join(self.working_dir, "storyboard.json") |
| 746 | if os.path.exists(storyboard_path): |
| 747 | with open(storyboard_path, 'r', encoding='utf-8') as f: |
| 748 | storyboard = json.load(f) |
| 749 | storyboard = [ShotBriefDescription.model_validate(shot) for shot in storyboard] |
| 750 | _pipeline_print(quiet, f"🚀 Loaded {len(storyboard)} shot brief descriptions from existing file.") |
| 751 | else: |
| 752 | _pipeline_print(quiet, f"🔍 Designing storyboard...") |
| 753 | storyboard = await self.storyboard_artist.design_storyboard( |
| 754 | script=script, |
| 755 | characters=characters, |
| 756 | user_requirement=user_requirement, |
| 757 | retry_timeout=150, |
| 758 | ) |
| 759 | storyboard = _normalize_model_list(storyboard, ShotBriefDescription, "storyboard") |
| 760 | with open(storyboard_path, 'w', encoding='utf-8') as f: |
| 761 | json.dump([shot.model_dump() for shot in storyboard], f, ensure_ascii=False, indent=4) |
| 762 | _pipeline_print(quiet, f"✅ Designed storyboard and saved to {storyboard_path}.") |
| 763 | |
| 764 | for shot_brief_description in storyboard: |
| 765 | self.shot_desc_events[shot_brief_description.idx] = asyncio.Event() |
| 766 | |
| 767 | return storyboard |
| 768 | |
| 769 | |
| 770 | |
| 771 | async def decompose_visual_descriptions( |
| 772 | self, |
| 773 | shot_brief_descriptions: List[ShotBriefDescription], |
| 774 | characters: List[CharacterInScene], |
| 775 | quiet: bool = False, |
| 776 | ): |
| 777 | tasks = [ |
| 778 | self.decompose_visual_description_for_single_shot_brief_description(shot_brief_description, characters, quiet=quiet) |
| 779 | for shot_brief_description in shot_brief_descriptions |
| 780 | ] |
| 781 | |
| 782 | shot_descriptions = await asyncio.gather(*tasks) |
| 783 | return shot_descriptions |
| 784 | |
| 785 | |
| 786 | async def decompose_visual_description_for_single_shot_brief_description( |
| 787 | self, |
| 788 | shot_brief_description: ShotBriefDescription, |
| 789 | characters: List[CharacterInScene], |
| 790 | quiet: bool = False, |
| 791 | ): |
| 792 | shot_description_path = os.path.join(self.working_dir, "shots", f"{shot_brief_description.idx}", "shot_description.json") |
| 793 | os.makedirs(os.path.dirname(shot_description_path), exist_ok=True) |
| 794 | |
| 795 | if os.path.exists(shot_description_path): |
| 796 | with open(shot_description_path, 'r', encoding='utf-8') as f: |
| 797 | shot_description = ShotDescription.model_validate(json.load(f)) |
| 798 | _pipeline_print(quiet, f"🚀 Loaded shot {shot_brief_description.idx} description from existing file.") |
| 799 | else: |
| 800 | shot_description = await self.storyboard_artist.decompose_visual_description( |
| 801 | shot_brief_desc=shot_brief_description, |
| 802 | characters=characters, |
| 803 | retry_timeout=120, |
| 804 | ) |
| 805 | shot_description = _normalize_model_list([shot_description], ShotDescription, "shot_description")[0] |
| 806 | with open(shot_description_path, 'w', encoding='utf-8') as f: |
| 807 | json.dump(shot_description.model_dump(), f, ensure_ascii=False, indent=4) |
| 808 | _pipeline_print(quiet, f"✅ Decomposed visual description for shot {shot_brief_description.idx} and saved to {shot_description_path}.") |
| 809 | |
| 810 | self.shot_desc_events[shot_brief_description.idx].set() |
| 811 | |
| 812 | if shot_description.variation_type in ["medium", "large"]: |
| 813 | self.frame_events[shot_brief_description.idx] = { |
| 814 | "first_frame": asyncio.Event(), |
| 815 | "last_frame": asyncio.Event(), |
| 816 | } |
| 817 | else: |
| 818 | self.frame_events[shot_brief_description.idx] = { |
| 819 | "first_frame": asyncio.Event(), |
| 820 | } |
| 821 | |
| 822 | return shot_description |
| 823 |