| 1 | # Copyright (C) 2025 AIDC-AI |
| 2 | # |
| 3 | # Licensed under the Apache License, Version 2.0 (the "License"); |
| 4 | # you may not use this file except in compliance with the License. |
| 5 | # You may obtain a copy of the License at |
| 6 | # http://www.apache.org/licenses/LICENSE-2.0 |
| 7 | # Unless required by applicable law or agreed to in writing, software |
| 8 | # distributed under the License is distributed on an "AS IS" BASIS, |
| 9 | # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 10 | # See the License for the specific language governing permissions and |
| 11 | # limitations under the License. |
| 12 | |
| 13 | """ |
| 14 | Frame processor - Process single frame through complete pipeline |
| 15 | |
| 16 | Orchestrates: TTS → Image Generation → Frame Composition → Video Segment |
| 17 | |
| 18 | Key Feature: |
| 19 | - TTS-driven video duration: Audio duration from TTS is passed to video generation workflows |
| 20 | to ensure perfect sync between audio and video (no padding, no trimming needed) |
| 21 | """ |
| 22 | |
| 23 | from typing import Callable, Optional |
| 24 | |
| 25 | import httpx |
| 26 | from loguru import logger |
| 27 | |
| 28 | from pixelle_video.models.progress import ProgressEvent |
| 29 | from pixelle_video.models.storyboard import Storyboard, StoryboardFrame, StoryboardConfig |
| 30 | |
| 31 | |
| 32 | class FrameProcessor: |
| 33 | """Frame processor""" |
| 34 | |
| 35 | def __init__(self, pixelle_video_core): |
| 36 | """ |
| 37 | Initialize |
| 38 | |
| 39 | Args: |
| 40 | pixelle_video_core: PixelleVideoCore instance |
| 41 | """ |
| 42 | self.core = pixelle_video_core |
| 43 | |
| 44 | async def __call__( |
| 45 | self, |
| 46 | frame: StoryboardFrame, |
| 47 | storyboard: 'Storyboard', |
| 48 | config: StoryboardConfig, |
| 49 | total_frames: int = 1, |
| 50 | progress_callback: Optional[Callable[[ProgressEvent], None]] = None |
| 51 | ) -> StoryboardFrame: |
| 52 | """ |
| 53 | Process single frame through complete pipeline |
| 54 | |
| 55 | Steps: |
| 56 | 1. Generate audio (TTS) |
| 57 | 2. Generate image (ComfyKit) |
| 58 | 3. Compose frame (add subtitle) |
| 59 | 4. Create video segment (image + audio) |
| 60 | |
| 61 | Args: |
| 62 | frame: Storyboard frame to process |
| 63 | storyboard: Storyboard instance |
| 64 | config: Storyboard configuration |
| 65 | total_frames: Total number of frames in storyboard |
| 66 | progress_callback: Optional callback for progress updates (receives ProgressEvent) |
| 67 | |
| 68 | Returns: |
| 69 | Processed frame with all paths filled |
| 70 | """ |
| 71 | logger.info(f"Processing frame {frame.index}...") |
| 72 | |
| 73 | frame_num = frame.index + 1 |
| 74 | |
| 75 | # Determine if this frame needs image generation |
| 76 | # If image_path or video_path is already set (e.g. asset-based pipeline), we consider it "has existing media" but skip generation |
| 77 | has_existing_media = frame.image_path is not None or frame.video_path is not None |
| 78 | needs_generation = frame.image_prompt is not None |
| 79 | |
| 80 | try: |
| 81 | # Step 1: Generate audio (TTS) |
| 82 | if not frame.audio_path: |
| 83 | if progress_callback: |
| 84 | progress_callback(ProgressEvent( |
| 85 | event_type="frame_step", |
| 86 | progress=0.0, |
| 87 | frame_current=frame_num, |
| 88 | frame_total=total_frames, |
| 89 | step=1, |
| 90 | action="audio" |
| 91 | )) |
| 92 | await self._step_generate_audio(frame, config) |
| 93 | else: |
| 94 | logger.debug(f" 1/4: Using existing audio: {frame.audio_path}") |
| 95 | |
| 96 | # Step 2: Generate media (image or video, conditional) |
| 97 | if needs_generation: |
| 98 | if progress_callback: |
| 99 | progress_callback(ProgressEvent( |
| 100 | event_type="frame_step", |
| 101 | progress=0.25, |
| 102 | frame_current=frame_num, |
| 103 | frame_total=total_frames, |
| 104 | step=2, |
| 105 | action="media" |
| 106 | )) |
| 107 | await self._step_generate_media(frame, config) |
| 108 | elif has_existing_media: |
| 109 | # Log appropriate message based on media type |
| 110 | if frame.video_path: |
| 111 | logger.debug(f" 2/4: Using existing video: {frame.video_path}") |
| 112 | else: |
| 113 | logger.debug(f" 2/4: Using existing image: {frame.image_path}") |
| 114 | else: |
| 115 | frame.image_path = None |
| 116 | frame.media_type = None |
| 117 | logger.debug(f" 2/4: Skipped media generation (not required by template)") |
| 118 | |
| 119 | # Step 3: Compose frame (add subtitle) |
| 120 | if progress_callback: |
| 121 | progress_callback(ProgressEvent( |
| 122 | event_type="frame_step", |
| 123 | progress=0.50 if (needs_generation or has_existing_media) else 0.33, |
| 124 | frame_current=frame_num, |
| 125 | frame_total=total_frames, |
| 126 | step=3, |
| 127 | action="compose" |
| 128 | )) |
| 129 | await self._step_compose_frame(frame, storyboard, config) |
| 130 | |
| 131 | # Step 4: Create video segment |
| 132 | if progress_callback: |
| 133 | progress_callback(ProgressEvent( |
| 134 | event_type="frame_step", |
| 135 | progress=0.75 if (needs_generation or has_existing_media) else 0.67, |
| 136 | frame_current=frame_num, |
| 137 | frame_total=total_frames, |
| 138 | step=4, |
| 139 | action="video" |
| 140 | )) |
| 141 | |
| 142 | await self._step_create_video_segment(frame, config) |
| 143 | |
| 144 | logger.info(f"✅ Frame {frame.index} completed") |
| 145 | return frame |
| 146 | |
| 147 | except Exception as e: |
| 148 | logger.error(f"❌ Failed to process frame {frame.index}: {e}") |
| 149 | raise |
| 150 | |
| 151 | async def _step_generate_audio( |
| 152 | self, |
| 153 | frame: StoryboardFrame, |
| 154 | config: StoryboardConfig |
| 155 | ): |
| 156 | """Step 1: Generate audio using TTS""" |
| 157 | logger.debug(f" 1/4: Generating audio for frame {frame.index}...") |
| 158 | |
| 159 | # Generate output path using task_id |
| 160 | from pixelle_video.utils.os_util import get_task_frame_path |
| 161 | output_path = get_task_frame_path(config.task_id, frame.index, "audio") |
| 162 | |
| 163 | # Build TTS params based on inference mode |
| 164 | tts_params = { |
| 165 | "text": frame.narration, |
| 166 | "inference_mode": config.tts_inference_mode, |
| 167 | "output_path": output_path, |
| 168 | "index": frame.index + 1, # 1-based index for workflow |
| 169 | } |
| 170 | |
| 171 | if config.tts_inference_mode == "local": |
| 172 | # Local mode: pass voice and speed |
| 173 | if config.voice_id: |
| 174 | tts_params["voice"] = config.voice_id |
| 175 | if config.tts_speed is not None: |
| 176 | tts_params["speed"] = config.tts_speed |
| 177 | else: # comfyui |
| 178 | # ComfyUI mode: pass workflow, voice, speed, and ref_audio |
| 179 | if config.tts_workflow: |
| 180 | tts_params["workflow"] = config.tts_workflow |
| 181 | if config.voice_id: |
| 182 | tts_params["voice"] = config.voice_id |
| 183 | if config.tts_speed is not None: |
| 184 | tts_params["speed"] = config.tts_speed |
| 185 | if config.ref_audio: |
| 186 | tts_params["ref_audio"] = config.ref_audio |
| 187 | |
| 188 | audio_path = await self.core.tts(**tts_params) |
| 189 | |
| 190 | frame.audio_path = audio_path |
| 191 | |
| 192 | # Get audio duration |
| 193 | frame.duration = await self._get_audio_duration(audio_path) |
| 194 | |
| 195 | logger.debug(f" ✓ Audio generated: {audio_path} ({frame.duration:.2f}s)") |
| 196 | |
| 197 | async def _step_generate_media( |
| 198 | self, |
| 199 | frame: StoryboardFrame, |
| 200 | config: StoryboardConfig |
| 201 | ): |
| 202 | """Step 2: Generate media (image or video) using ComfyKit""" |
| 203 | logger.debug(f" 2/4: Generating media for frame {frame.index}...") |
| 204 | |
| 205 | # Determine media type based on workflow/template. |
| 206 | # video_ prefix in workflow name indicates ComfyUI video generation; |
| 207 | # video_* templates can also use direct API video workflows. |
| 208 | workflow_name = config.media_workflow or "" |
| 209 | from pixelle_video.utils.template_util import get_template_type |
| 210 | template_type = get_template_type(config.frame_template or "") |
| 211 | is_video_workflow = "video_" in workflow_name.lower() or template_type == "video" |
| 212 | media_type = "video" if is_video_workflow else "image" |
| 213 | |
| 214 | logger.debug(f" → Media type: {media_type} (workflow: {workflow_name})") |
| 215 | |
| 216 | # Build media generation parameters |
| 217 | from pixelle_video.utils.os_util import get_task_frame_path |
| 218 | output_path = get_task_frame_path(config.task_id, frame.index, media_type) |
| 219 | api_video_params = dict(config.api_video_params or {}) if media_type == "video" else {} |
| 220 | if media_type == "video" and workflow_name.startswith("api/"): |
| 221 | await self._prepare_api_video_inputs(frame, config, api_video_params) |
| 222 | |
| 223 | media_params = { |
| 224 | "prompt": frame.image_prompt, |
| 225 | "workflow": config.media_workflow, # Pass workflow from config (None = use default) |
| 226 | "media_type": media_type, |
| 227 | "width": config.media_width, |
| 228 | "height": config.media_height, |
| 229 | "output_path": output_path, |
| 230 | "image_path": frame.image_path, |
| 231 | "index": frame.index + 1, # 1-based index for workflow |
| 232 | } |
| 233 | media_params.update(api_video_params) |
| 234 | |
| 235 | # For video workflows: pass audio duration as target video duration |
| 236 | # This ensures video length matches audio length from the source |
| 237 | if is_video_workflow and frame.duration: |
| 238 | media_params["duration"] = frame.duration |
| 239 | logger.info(f" → Generating video with target duration: {frame.duration:.2f}s (from TTS audio)") |
| 240 | |
| 241 | # Call Media generation |
| 242 | media_result = await self.core.media(**media_params) |
| 243 | |
| 244 | # Store media type |
| 245 | frame.media_type = media_result.media_type |
| 246 | |
| 247 | if media_result.is_image: |
| 248 | # Download image to local (pass task_id) |
| 249 | local_path = await self._download_media( |
| 250 | media_result.url, |
| 251 | frame.index, |
| 252 | config.task_id, |
| 253 | media_type="image" |
| 254 | ) |
| 255 | frame.image_path = local_path |
| 256 | logger.debug(f" ✓ Image generated: {local_path}") |
| 257 | |
| 258 | elif media_result.is_video: |
| 259 | # Download video to local (pass task_id) |
| 260 | local_path = await self._download_media( |
| 261 | media_result.url, |
| 262 | frame.index, |
| 263 | config.task_id, |
| 264 | media_type="video" |
| 265 | ) |
| 266 | frame.video_path = local_path |
| 267 | |
| 268 | # Update duration from video if available |
| 269 | if media_result.duration: |
| 270 | frame.duration = media_result.duration |
| 271 | logger.debug(f" ✓ Video generated: {local_path} (duration: {frame.duration:.2f}s)") |
| 272 | else: |
| 273 | # Get video duration from file |
| 274 | frame.duration = await self._get_video_duration(local_path) |
| 275 | logger.debug(f" ✓ Video generated: {local_path} (duration: {frame.duration:.2f}s)") |
| 276 | |
| 277 | else: |
| 278 | raise ValueError(f"Unknown media type: {media_result.media_type}") |
| 279 | |
| 280 | async def _prepare_api_video_inputs( |
| 281 | self, |
| 282 | frame: StoryboardFrame, |
| 283 | config: StoryboardConfig, |
| 284 | api_video_params: dict, |
| 285 | ) -> None: |
| 286 | """Prepare provider-specific inputs for API video models.""" |
| 287 | from pixelle_video.utils.os_util import get_task_frame_path |
| 288 | |
| 289 | if api_video_params.pop("use_narration_audio_as_driving_audio", False): |
| 290 | api_video_params["audio_path"] = frame.audio_path |
| 291 | |
| 292 | if frame.image_path or api_video_params.get("first_clip_path") or api_video_params.get("first_video_path"): |
| 293 | return |
| 294 | |
| 295 | first_frame_workflow = api_video_params.pop("first_frame_workflow", None) |
| 296 | if not first_frame_workflow: |
| 297 | return |
| 298 | |
| 299 | first_frame_path = get_task_frame_path(config.task_id, frame.index, "image") |
| 300 | logger.info(f" → Generating API video first frame via {first_frame_workflow}") |
| 301 | image_result = await self.core.media( |
| 302 | prompt=frame.image_prompt, |
| 303 | workflow=first_frame_workflow, |
| 304 | media_type="image", |
| 305 | width=config.media_width, |
| 306 | height=config.media_height, |
| 307 | output_path=first_frame_path, |
| 308 | index=frame.index + 1, |
| 309 | ) |
| 310 | frame.image_path = await self._download_media( |
| 311 | image_result.url, |
| 312 | frame.index, |
| 313 | config.task_id, |
| 314 | media_type="image", |
| 315 | ) |
| 316 | |
| 317 | async def _step_compose_frame( |
| 318 | self, |
| 319 | frame: StoryboardFrame, |
| 320 | storyboard: 'Storyboard', |
| 321 | config: StoryboardConfig |
| 322 | ): |
| 323 | """Step 3: Compose frame with subtitle using HTML template""" |
| 324 | logger.debug(f" 3/4: Composing frame {frame.index}...") |
| 325 | |
| 326 | # Generate output path using task_id |
| 327 | from pixelle_video.utils.os_util import get_task_frame_path |
| 328 | output_path = get_task_frame_path(config.task_id, frame.index, "composed") |
| 329 | |
| 330 | # For video type: render HTML as transparent overlay image |
| 331 | # For image type: render HTML with image background |
| 332 | # In both cases, we need the composed image |
| 333 | composed_path = await self._compose_frame_html(frame, storyboard, config, output_path) |
| 334 | |
| 335 | frame.composed_image_path = composed_path |
| 336 | |
| 337 | logger.debug(f" ✓ Frame composed: {composed_path}") |
| 338 | |
| 339 | async def _compose_frame_html( |
| 340 | self, |
| 341 | frame: StoryboardFrame, |
| 342 | storyboard: 'Storyboard', |
| 343 | config: StoryboardConfig, |
| 344 | output_path: str |
| 345 | ) -> str: |
| 346 | """Compose frame using HTML template""" |
| 347 | from pixelle_video.services.frame_html import HTMLFrameGenerator |
| 348 | from pixelle_video.utils.template_util import resolve_template_path |
| 349 | |
| 350 | # Resolve template path (handles various input formats) |
| 351 | template_path = resolve_template_path(config.frame_template) |
| 352 | |
| 353 | # Get content metadata from storyboard |
| 354 | content_metadata = storyboard.content_metadata if storyboard else None |
| 355 | |
| 356 | # Build ext data |
| 357 | ext = { |
| 358 | "index": frame.index + 1, |
| 359 | } |
| 360 | |
| 361 | # Add custom template parameters |
| 362 | if config.template_params: |
| 363 | ext.update(config.template_params) |
| 364 | |
| 365 | # Generate frame using HTML (size is auto-parsed from template path) |
| 366 | generator = HTMLFrameGenerator(template_path) |
| 367 | |
| 368 | # Use video_path for video media, image_path for images |
| 369 | media_path = frame.video_path if frame.media_type == "video" else frame.image_path |
| 370 | logger.debug(f"Generating frame with media: '{media_path}' (type: {frame.media_type})") |
| 371 | |
| 372 | composed_path = await generator.generate_frame( |
| 373 | title=storyboard.title, |
| 374 | text=frame.narration, |
| 375 | image=media_path, # HTMLFrameGenerator handles both image and video paths |
| 376 | ext=ext, |
| 377 | output_path=output_path |
| 378 | ) |
| 379 | |
| 380 | return composed_path |
| 381 | |
| 382 | async def _step_create_video_segment( |
| 383 | self, |
| 384 | frame: StoryboardFrame, |
| 385 | config: StoryboardConfig |
| 386 | ): |
| 387 | """Step 4: Create video segment from media + audio""" |
| 388 | logger.debug(f" 4/4: Creating video segment for frame {frame.index}...") |
| 389 | |
| 390 | # Generate output path using task_id |
| 391 | from pixelle_video.utils.os_util import get_task_frame_path |
| 392 | output_path = get_task_frame_path(config.task_id, frame.index, "segment") |
| 393 | |
| 394 | from pixelle_video.services.video import VideoService |
| 395 | video_service = VideoService() |
| 396 | |
| 397 | # Branch based on media type |
| 398 | if frame.media_type == "video": |
| 399 | # Video workflow: overlay HTML template on video, then add audio |
| 400 | logger.debug(f" → Using video-based composition with HTML overlay") |
| 401 | |
| 402 | # Step 1: Overlay transparent HTML image on video |
| 403 | # The composed_image_path contains the rendered HTML with transparent background |
| 404 | temp_video_with_overlay = get_task_frame_path(config.task_id, frame.index, "video") + "_overlay.mp4" |
| 405 | |
| 406 | video_service.overlay_image_on_video( |
| 407 | video=frame.video_path, |
| 408 | overlay_image=frame.composed_image_path, |
| 409 | output=temp_video_with_overlay, |
| 410 | scale_mode="contain" # Scale video to fit template size (contain mode) |
| 411 | ) |
| 412 | |
| 413 | # Step 2: Add narration audio to the overlaid video |
| 414 | # Note: The video might have audio (replaced) or be silent (audio added) |
| 415 | segment_path = video_service.merge_audio_video( |
| 416 | video=temp_video_with_overlay, |
| 417 | audio=frame.audio_path, |
| 418 | output=output_path, |
| 419 | replace_audio=True, # Replace video audio with narration |
| 420 | audio_volume=1.0 |
| 421 | ) |
| 422 | |
| 423 | # Clean up temp file |
| 424 | import os |
| 425 | if os.path.exists(temp_video_with_overlay): |
| 426 | os.unlink(temp_video_with_overlay) |
| 427 | |
| 428 | elif frame.media_type == "image" or frame.media_type is None: |
| 429 | # Image workflow: Use composed image directly |
| 430 | # The asset_default.html template includes the image in the composition |
| 431 | logger.debug(f" → Using image-based composition") |
| 432 | |
| 433 | segment_path = video_service.create_video_from_image( |
| 434 | image=frame.composed_image_path, |
| 435 | audio=frame.audio_path, |
| 436 | output=output_path, |
| 437 | fps=config.video_fps |
| 438 | ) |
| 439 | |
| 440 | else: |
| 441 | raise ValueError(f"Unknown media type: {frame.media_type}") |
| 442 | |
| 443 | frame.video_segment_path = segment_path |
| 444 | |
| 445 | logger.debug(f" ✓ Video segment created: {segment_path}") |
| 446 | |
| 447 | async def _get_audio_duration(self, audio_path: str) -> float: |
| 448 | """Get audio duration in seconds""" |
| 449 | try: |
| 450 | # Try using ffmpeg-python |
| 451 | import ffmpeg |
| 452 | probe = ffmpeg.probe(audio_path) |
| 453 | duration = float(probe['format']['duration']) |
| 454 | return duration |
| 455 | except Exception as e: |
| 456 | logger.warning(f"Failed to get audio duration: {e}, using estimate") |
| 457 | # Fallback: estimate based on file size (very rough) |
| 458 | import os |
| 459 | file_size = os.path.getsize(audio_path) |
| 460 | # Assume ~16kbps for MP3, so 2KB per second |
| 461 | estimated_duration = file_size / 2000 |
| 462 | return max(1.0, estimated_duration) # At least 1 second |
| 463 | |
| 464 | async def _download_media( |
| 465 | self, |
| 466 | url: str, |
| 467 | frame_index: int, |
| 468 | task_id: str, |
| 469 | media_type: str |
| 470 | ) -> str: |
| 471 | """Download media (image or video) from URL to local file""" |
| 472 | import os |
| 473 | from pixelle_video.utils.os_util import get_task_frame_path |
| 474 | output_path = get_task_frame_path(task_id, frame_index, media_type) |
| 475 | |
| 476 | if url.startswith("file://"): |
| 477 | local_path = url[7:] |
| 478 | if not os.path.exists(local_path): |
| 479 | raise FileNotFoundError(f"Generated media file not found: {local_path}") |
| 480 | return local_path |
| 481 | |
| 482 | if os.path.exists(url): |
| 483 | return url |
| 484 | |
| 485 | timeout = httpx.Timeout(connect=10.0, read=60, write=60, pool=60) |
| 486 | async with httpx.AsyncClient(timeout=timeout) as client: |
| 487 | response = await client.get(url) |
| 488 | response.raise_for_status() |
| 489 | |
| 490 | with open(output_path, 'wb') as f: |
| 491 | f.write(response.content) |
| 492 | |
| 493 | return output_path |
| 494 | |
| 495 | async def _get_video_duration(self, video_path: str) -> float: |
| 496 | """Get video duration in seconds""" |
| 497 | try: |
| 498 | import ffmpeg |
| 499 | probe = ffmpeg.probe(video_path) |
| 500 | duration = float(probe['format']['duration']) |
| 501 | return duration |
| 502 | except Exception as e: |
| 503 | logger.warning(f"Failed to get video duration: {e}, using audio duration") |
| 504 | # Fallback: use audio duration if available |
| 505 | return 1.0 # Default to 1 second if unable to determine |
| 506 |