Coverage for apps/gen_video_chunked.py: 83%

1""" 

2Generate a video in chunks of a maximum length. 

3It uses Hunyuan FramePack to generate a long sketch video at low resolution, 

4then splits the audio into subvideos based on silences, and for each subvideo, 

5it generates video+audio at medium resolution using Fantasy Talking. 

6 

7Pipeline: 

81. Split audio into silence-aligned subvideos 

92. Generate a long low-res sketch video (Hunyuan FramePack) 

103. Stream sketch frames and schedule subvideo generation 

114. Generate video+audio chunks (Fantasy Talking) 

125. Concatenate results 

13""" 

14import sys 

15import os 

16import asyncio 

17import aiofiles 

18import logging 

19 

20from typing import List 

21from typing import Dict 

22from typing import Optional 

23from typing import Union 

24from typing import Tuple 

25from typing import cast 

26import math 

27 

28from PIL import Image 

29 

30# Local relative imports 

31sys.path.append("..") # noqa: E402 

32 

33from lmm_generator import LMMGenerator 

34 

35from client import ServiceRequest 

36 

37from video import get_num_video_frames_from_duration 

38from video import MAX_FT_DURATION_SECS 

39from video import FANTASYTALKING_FPS 

40from video import HUNYUANFRAMEPACK_FPS 

41from video import HUNYUANFRAMEPACK_VAE_T 

42from video import VAE_T 

43 

44from tts_utils import get_audio_chunks_by_silences 

45 

46from console_utils import bytes_to_human 

47 

48from file_utils import read_file_base64 

49from file_utils import save_base64_as_binary 

50 

51from media_utils import get_video_file_info 

52from media_utils import get_audio_duration 

53from media_utils import get_video_with_text 

54from media_utils import get_video_frames_at_fps 

55from media_utils import get_font_size 

56from media_utils import split_text_lines 

57from media_utils import add_text_to_frame 

58from media_utils import chunk_audio_base64 

59from media_utils import concatenate_videos 

60 

61MAX_IMG_LINE_CHARS = 50 

62 

63 

64class SubVideoInfo: 

65 """Subvideo information.""" 

66 

67 def __init__( 

68 self, 

69 start_seconds: float, 

70 end_seconds: float, 

71 ) -> None: 

72 self.start_seconds = start_seconds 

73 self.end_seconds = end_seconds 

74 

75 def get_seconds(self) -> float: 

76 """Get subvideo duration in seconds.""" 

77 return self.end_seconds - self.start_seconds 

78 

79 def get_start_frame( 

80 self, 

81 fps: float, 

82 ) -> int: 

83 """Get start frame.""" 

84 return int(math.ceil(self.start_seconds * fps)) 

85 

86 def get_end_frame( 

87 self, 

88 fps: float, 

89 ) -> int: 

90 """Get end frame.""" 

91 return int(math.ceil(self.end_seconds * fps)) 

92 

93 def get_frames(self, fps: float) -> Tuple[int, int]: 

94 """Get start and end frames.""" 

95 start_frames = self.get_start_frame(fps) 

96 end_frames = self.get_end_frame(fps) 

97 return start_frames, end_frames 

98 

99 def __str__(self) -> str: 

100 return f"SubVideoInfo({self.start_seconds:.3f}-{self.end_seconds:.3f}s)" 

101 

102 

103class GenVideoChunked: 

104 """ 

105 Generate a video in chunks of a maximum length. 

106 """ 

107 

108 def __init__( 

109 self, 

110 video_id: int, 

111 gen: LMMGenerator, 

112 job_path: str, 

113 logger: logging.Logger, 

114 ) -> None: 

115 """Initialize the video generator.""" 

116 self.video_id = video_id 

117 self.gen = gen 

118 self.job_id = gen.job_id 

119 self.job_path = job_path 

120 self.logger = logger 

121 

122 async def _prepare_audio( 

123 self, 

124 audio_path: str, 

125 max_duration: int = MAX_FT_DURATION_SECS, 

126 ) -> Tuple[str, float, List[SubVideoInfo]]: 

127 """Prepare audio: load, get duration, split into subvideos. """ 

128 if not os.path.exists(audio_path): 

129 raise FileNotFoundError(f"Audio file not found: {audio_path}") 

130 

131 audio_b64 = await read_file_base64(audio_path) 

132 duration = get_audio_duration(audio_b64) 

133 

134 if duration < max_duration: 

135 raise ValueError(f"Audio too short for chunked generation ({duration}/{max_duration}).") 

136 

137 audio_splits = get_audio_chunks_by_silences( 

138 audio_path, 

139 max_duration, 

140 chunk_alignment_seconds=1.0 / FANTASYTALKING_FPS) 

141 

142 if len(audio_splits) > math.ceil(duration / max_duration): 

143 self.logger.warning( 

144 f"[{self.video_id}] Too many subvideos ({len(audio_splits)}) for audio with " 

145 f"{duration:.3f} seconds and max {max_duration:.3f}.") 

146 

147 subvideos = [ 

148 SubVideoInfo(start_secs, end_secs) 

149 for start_secs, end_secs in audio_splits 

150 ] 

151 

152 for subvideo_id, subvideo in enumerate(subvideos): 

153 self.logger.info( 

154 f"[{self.video_id}.{subvideo_id}] " 

155 f"{subvideo.start_seconds:.3f}-{subvideo.end_seconds:.3f} " 

156 f"({subvideo.get_seconds():.3f}s)") 

157 if subvideo.get_seconds() > max_duration: 

158 self.logger.error( 

159 f"[{self.video_id}.{subvideo_id}] Subvideo too long: " 

160 f"{subvideo.get_seconds():.3f}s > {max_duration:.3f}s.") 

161 elif subvideo.get_seconds() < 0.5: 

162 self.logger.warning( 

163 f"[{self.video_id}.{subvideo_id}] Subvideo too short: " 

164 f"{subvideo.get_seconds():.3f}s < 0.5s.") 

165 

166 return audio_b64, duration, subvideos 

167 

168 async def gen_video_chunked( 

169 self, 

170 audio_path: str, 

171 image: Image.Image, 

172 prompt: str, 

173 neg_prompt: str, 

174 width: int, 

175 height: int, 

176 num_steps: int, 

177 subvideo_duration_max: int = MAX_FT_DURATION_SECS, 

178 upscaling: bool = False, 

179 debug: bool = False, 

180 deadline: Optional[float] = None, 

181 ) -> bytes: 

182 """ 

183 Generate a video with audio and video starting with image at high resolution. 

184 1. Split audio into subvideos based on silences. 

185 2. Generate sketch at low resolution with Hunyuan FramePack. 

186 3. Generate video+audio at medium resolution with Fantasy Talking. 

187 Returns video synced with the audio in binary. 

188 """ 

189 audio_base64, audio_duration, subvideos = await self._prepare_audio( 

190 audio_path, 

191 subvideo_duration_max 

192 ) 

193 

194 sketch_request, sketch_num_frames = await self._gen_sketch( 

195 duration=audio_duration, 

196 image=image, 

197 prompt=prompt, 

198 neg_prompt=neg_prompt, 

199 width=width, 

200 height=height, 

201 num_steps=num_steps, 

202 deadline=deadline, 

203 ) 

204 

205 subvideo_tasks = await self._schedule_subvideos( 

206 sketch_request=sketch_request, 

207 sketch_num_frames=sketch_num_frames, 

208 subvideos=subvideos, 

209 audio_base64=audio_base64, 

210 width=width, 

211 height=height, 

212 num_steps=num_steps, 

213 prompt=prompt, 

214 neg_prompt=neg_prompt, 

215 upscaling=upscaling, 

216 debug=debug, 

217 deadline=deadline, 

218 ) 

219 

220 await self._save_sketch(sketch_request) 

221 

222 subvideo_binaries = await self._collect_subvideos( 

223 subvideo_tasks, 

224 subvideos, 

225 width, 

226 height, 

227 ) 

228 

229 if upscaling: 

230 subvideo_binaries = await self._upscale_subvideos(subvideo_binaries) 

231 

232 video_binary = await self._concatenate_subvideos(subvideo_binaries) 

233 

234 return video_binary 

235 

236 async def _gen_sketch( 

237 self, 

238 duration: float, 

239 image: Image.Image, 

240 prompt: str, 

241 neg_prompt: str, 

242 width: int, 

243 height: int, 

244 num_steps: int, 

245 deadline: Optional[float] = None, 

246 ) -> Tuple[ServiceRequest, int]: 

247 """Start long sketch video (no audio) generation in the background.""" 

248 sketch_num_frames = get_num_video_frames_from_duration( 

249 duration, 

250 HUNYUANFRAMEPACK_FPS, 

251 HUNYUANFRAMEPACK_VAE_T) 

252 

253 sketch_request = await self.gen.gen_video( 

254 image, 

255 prompt, 

256 neg_prompt, 

257 width=width, 

258 height=height, 

259 num_frames=sketch_num_frames, 

260 steps=num_steps // 2, # TODO Less steps for sketch video 

261 task_id=f"{self.video_id:03d}_sketch", 

262 wait_request=False, 

263 deadline=deadline, 

264 ) 

265 

266 self.logger.info( 

267 f"[{self.video_id}] Generating long sketch video with {sketch_num_frames} frames, " 

268 f"{duration:.3f} seconds, " 

269 f"{HUNYUANFRAMEPACK_FPS} FPS, and " 

270 f"{width}x{height} pixels.") 

271 

272 return sketch_request, sketch_num_frames 

273 

274 async def _save_sketch( 

275 self, 

276 sketch_request: ServiceRequest, 

277 ) -> str: 

278 """Save full sketch video (no audio).""" 

279 content_type, video_binary = await sketch_request.future 

280 if not video_binary: 

281 raise ValueError("Cannot generate sketch video.") 

282 if content_type != "video/mp4": 

283 raise ValueError(f"Invalid content type for video: {content_type}.") 

284 

285 self._log_video_info(f"[{self.video_id}] Video sketch", video_binary) 

286 video_path = f"{self.job_path}/{self.video_id:03d}_chunks_sketch.mp4" 

287 async with aiofiles.open(video_path, "wb") as file: 

288 await file.write(video_binary) 

289 return video_path 

290 

291 async def _schedule_subvideos( 

292 self, 

293 sketch_request: ServiceRequest, 

294 sketch_num_frames: int, 

295 subvideos: List[SubVideoInfo], 

296 audio_base64: str, 

297 width: int, 

298 height: int, 

299 num_steps: int, 

300 prompt: str, 

301 neg_prompt: str, 

302 upscaling: bool, 

303 debug: bool, 

304 deadline: Optional[float], 

305 ) -> List[Optional[asyncio.Task]]: 

306 """Schedule subvideo generation while sketch video is being generated.""" 

307 tasks: List[Optional[asyncio.Task]] = [None] * len(subvideos) 

308 

309 sketch_frames: List[Image.Image] = [] 

310 base_url = sketch_request.get_base_request_url() 

311 subvideo_id = 0 

312 

313 # Get frames while the sketch request is running 

314 video_gen_request_done = False 

315 while len(sketch_frames) < sketch_num_frames and not video_gen_request_done: 

316 if sketch_request.done(): 

317 video_gen_request_done = True 

318 elif not sketch_request.is_running() or not sketch_request.url: 

319 RETRY_SLEEP_SECONDS = 1.0 

320 await asyncio.sleep(RETRY_SLEEP_SECONDS) # Wait for it to be running before checking again 

321 continue 

322 

323 # Get intermediate frames (while long sketch video generation is running) 

324 async for frame in self.gen.gen_intermediate_video_frames( 

325 base_url, 

326 task_id=f"{self.video_id:03d}_sketch", 

327 video_gen_request=sketch_request, 

328 ): 

329 sketch_frames.append(frame) 

330 

331 # Schedule as many subvideos as possible with available frames 

332 while self._can_schedule_subvideo(subvideos, subvideo_id, sketch_frames): 

333 # Enough frames for the subvideo, generate video+audio 

334 subvideo = subvideos[subvideo_id] 

335 task = await self._gen_subvideo( 

336 subvideo_id=subvideo_id, 

337 subvideo_info=subvideo, 

338 width=width, 

339 height=height, 

340 num_steps=num_steps, 

341 video_frames=sketch_frames, 

342 audio_base64=audio_base64, 

343 video_prompt=prompt, 

344 video_neg_prompt=neg_prompt, 

345 upscaling=upscaling, 

346 debug=debug, 

347 deadline=deadline, 

348 ) 

349 if task is None: 

350 self.logger.error(f"[{self.video_id}.{subvideo_id}] Cannot generate subvideo task.") 

351 else: 

352 tasks[subvideo_id] = task 

353 subvideo_id += 1 

354 

355 if len(sketch_frames) >= sketch_num_frames: 

356 break # All frames received 

357 

358 # Schedule the remaining subvideos if any 

359 while subvideo_id < len(subvideos): 

360 subvideo = subvideos[subvideo_id] 

361 task = await self._gen_subvideo( 

362 subvideo_id=subvideo_id, 

363 subvideo_info=subvideo, 

364 width=width, 

365 height=height, 

366 num_steps=num_steps, 

367 video_frames=sketch_frames, 

368 audio_base64=audio_base64, 

369 video_prompt=prompt, 

370 video_neg_prompt=neg_prompt, 

371 upscaling=upscaling, 

372 debug=debug, 

373 deadline=deadline, 

374 ) 

375 if task is None: 

376 self.logger.error(f"[{self.video_id}.{subvideo_id}] Cannot generate subvideo task.") 

377 else: 

378 tasks[subvideo_id] = task 

379 subvideo_id += 1 

380 

381 # Final checks 

382 self.logger.info( 

383 f"[{self.video_id}] Got {len(sketch_frames)}/{sketch_num_frames} streamed video frames.") 

384 

385 if len(sketch_frames) < sketch_num_frames: 

386 self.logger.warning( 

387 f"[{self.video_id}] Not enough frames ({len(sketch_frames)} < {sketch_num_frames}).") 

388 

389 return tasks 

390 

391 def _can_schedule_subvideo( 

392 self, 

393 subvideos: List[SubVideoInfo], 

394 idx: int, 

395 frames: List[Image.Image], 

396 ) -> bool: 

397 """Check if we can schedule the next subvideo given available frames.""" 

398 if idx >= len(subvideos): 

399 return False 

400 subvideo = subvideos[idx] 

401 needed = subvideo.get_end_frame(HUNYUANFRAMEPACK_FPS) + VAE_T 

402 return len(frames) >= needed 

403 

404 async def _collect_subvideos( 

405 self, 

406 subvideo_tasks: List[Optional[asyncio.Task]], 

407 subvideos: List[SubVideoInfo], 

408 width: int, 

409 height: int, 

410 ) -> List[bytes]: 

411 """Collecting video+audio subvideos.""" 

412 self.logger.info(f"[{self.video_id}] Generating {len(subvideo_tasks)} video+audio subvideos...") 

413 

414 subvideo_binaries: List[bytes] = [b""] * len(subvideos) 

415 

416 subvideo_ids = list(range(len(subvideo_tasks))) 

417 gather_tasks: List[asyncio.Task] = [] 

418 subvideo_to_task: Dict[int, int] = {} 

419 for subvideo_id, subvideo_task in enumerate(subvideo_tasks): 

420 if subvideo_task is not None: 

421 gather_task_id = len(gather_tasks) 

422 subvideo_to_task[subvideo_id] = gather_task_id 

423 gather_tasks.append(subvideo_task) 

424 

425 gather_results = await asyncio.gather( 

426 *gather_tasks, 

427 return_exceptions=True 

428 ) 

429 self.logger.info(f"[{self.video_id}] Generated {len(gather_tasks)} video+audio subvideos.") 

430 

431 if not gather_results: 

432 raise ValueError(f"No subvideos generated for video {self.video_id}.") 

433 

434 for subvideo_id in subvideo_ids: 

435 subvideo = subvideos[subvideo_id] 

436 duration_seconds = subvideo.get_seconds() 

437 task_id = subvideo_to_task.get(subvideo_id) 

438 

439 subvideo_binary = None 

440 if task_id is not None: 

441 subvideo_binary = gather_results[task_id] 

442 

443 if isinstance(subvideo_binary, bytes): 

444 self._log_video_info(f"[{self.video_id}.{subvideo_id}] Generated video", subvideo_binary) 

445 else: 

446 # Error case -> replace with static error video 

447 # TODO add audio 

448 err_msg = "No video generated" 

449 if subvideo_binary is not None: 

450 err_msg = str(subvideo_binary) # This was an exception 

451 self.logger.error( 

452 f"[{self.video_id}.{subvideo_id}] {err_msg}. " 

453 f"Adding error video with {duration_seconds:.3f} seconds and {width}x{height} pixels.") 

454 subvideo_binary = await self._gen_error_subvideo( 

455 subvideo_id=subvideo_id, 

456 width=width, 

457 height=height, 

458 duration_seconds=duration_seconds, 

459 fps=FANTASYTALKING_FPS, 

460 err_msg=err_msg) 

461 

462 subvideo_binaries[subvideo_id] = subvideo_binary 

463 

464 video_path = f"{self.job_path}/{self.video_id:03d}_{subvideo_id:03d}_chunks.mp4" 

465 async with aiofiles.open(video_path, "wb") as file: 

466 await file.write(subvideo_binary) 

467 return subvideo_binaries 

468 

469 async def _gen_error_subvideo( 

470 self, 

471 subvideo_id: int, 

472 width: int, 

473 height: int, 

474 duration_seconds: float, 

475 fps: float, 

476 err_msg: str, 

477 ) -> bytes: 

478 frame_text = "Cannot generate video\n" 

479 frame_text += f"Subvideo {self.video_id:03d}.{subvideo_id:03d}\n" 

480 frame_text += "\n".join(split_text_lines(err_msg, MAX_IMG_LINE_CHARS)) 

481 font_size = get_font_size(width, height) 

482 subvideo_binary = await get_video_with_text( 

483 width=width, 

484 height=height, 

485 text=frame_text, 

486 font_size=font_size, 

487 fps=fps, 

488 duration_seconds=duration_seconds) 

489 return subvideo_binary 

490 

491 async def _upscale_subvideos( 

492 self, 

493 subvideo_binaries: List[bytes], 

494 ) -> List[bytes]: 

495 """Upscale subvideos.""" 

496 # TODO 

497 # await self.gen.gen_video_upscale() 

498 return subvideo_binaries 

499 

500 async def _concatenate_subvideos( 

501 self, 

502 subvideo_binaries: List[bytes], 

503 ) -> bytes: 

504 """Concatenate subvideos into final video.""" 

505 self.logger.info(f"[{self.video_id}] Concatenating {len(subvideo_binaries)} subvideos...") 

506 video_binary = await concatenate_videos( 

507 subvideo_binaries, 

508 fast_copy=False) # move to True once we fix the durations 

509 if not video_binary: 

510 raise ValueError(f"Cannot concatenate subvideos for video {self.video_id}") 

511 self._log_video_info( 

512 f"[{self.video_id}] Concatenated {len(subvideo_binaries)} videos into video", 

513 video_binary) 

514 return video_binary 

515 

516 async def _gen_subvideo( 

517 self, 

518 subvideo_id: int, 

519 subvideo_info: SubVideoInfo, 

520 width: int, 

521 height: int, 

522 num_steps: int, 

523 video_frames: List[Image.Image], # video_hy_video_frames 

524 audio_base64: str, 

525 video_prompt: str = "", 

526 video_neg_prompt: str = "", 

527 upscaling: bool = False, 

528 debug: bool = False, 

529 deadline: Optional[float] = None, 

530 ) -> Optional[asyncio.Task]: 

531 """ 

532 Generate video+audio chunk with Fantasy Talking from sketch frames (Hunyuan FramePack) and audio_base64. 

533 """ 

534 start_frame = subvideo_info.get_start_frame(HUNYUANFRAMEPACK_FPS) 

535 end_frame = subvideo_info.get_end_frame(HUNYUANFRAMEPACK_FPS) 

536 

537 if start_frame > len(video_frames): 

538 self.logger.error( 

539 f"[{self.video_id}.{subvideo_id}] Not enough sketch frames" 

540 f"[{start_frame}..{end_frame} ] > {len(video_frames)}.") 

541 return None 

542 

543 # Add some frames to account for the 1+4n alignment 

544 end_frame += VAE_T 

545 if end_frame > len(video_frames): 

546 self.logger.warning( 

547 f"[{self.video_id}.{subvideo_id}] Sketch frames truncated " 

548 f"{end_frame} > {len(video_frames)}.") 

549 end_frame = len(video_frames) # Don't go beyond available frames 

550 

551 # TODO align audio lengths 

552 # TODO check if this is just silence 

553 

554 # Audio 

555 subaudio_b64 = chunk_audio_base64( 

556 audio_base64, 

557 subvideo_info.start_seconds, 

558 subvideo_info.end_seconds) 

559 subvideo_duration = subvideo_info.end_seconds - subvideo_info.start_seconds 

560 subvideo_audio_num_frames = get_num_video_frames_from_duration(subvideo_duration) 

561 

562 subvideo_audio_path = f"{self.job_path}/{self.video_id:03d}_{subvideo_id:03d}.wav" 

563 await save_base64_as_binary( 

564 subvideo_audio_path, 

565 subaudio_b64) 

566 

567 # Video 

568 # Hunyuan FramePack -> Fantasy Talking intermediate frames 

569 hy_frames = video_frames[start_frame:end_frame] 

570 ft_frames = get_video_frames_at_fps( 

571 hy_frames, 

572 src_fps=HUNYUANFRAMEPACK_FPS, 

573 dst_fps=FANTASYTALKING_FPS) 

574 

575 # Adjust video frames portion to match audio if needed 

576 len_video = len(ft_frames) 

577 if len_video == 0: 

578 self.logger.error(f"[{self.video_id}.{subvideo_id}] No video frames for subvideo.") 

579 return None 

580 

581 if len_video < subvideo_audio_num_frames: 

582 msg = f"[{self.video_id}.{subvideo_id}] Video < Audio ({len_video}<{subvideo_audio_num_frames}). Extend." 

583 self.logger.warning(msg) 

584 ft_frames += [ft_frames[-1]] * (subvideo_audio_num_frames - len_video) 

585 elif len_video > subvideo_audio_num_frames: 

586 msg = f"[{self.video_id}.{subvideo_id}] Video > Audio ({len_video}>{subvideo_audio_num_frames}). Trim." 

587 if len_video > subvideo_audio_num_frames + VAE_T: 

588 self.logger.warning(msg) 

589 else: 

590 self.logger.debug(msg) # If it is only a few frames, it is just VAE 1+4n rounding 

591 ft_frames = ft_frames[:subvideo_audio_num_frames] 

592 if debug: 

593 ft_frames = self._add_debug(subvideo_id, ft_frames) 

594 

595 # TODO 

596 # width, height = self.width, self.height 

597 if upscaling: 

598 # width, height = RESOLUTIONS[self.aspect_ratio]["medium"] 

599 width = width // 2 

600 height = height // 2 

601 # TODO run the upscaling after Fantasy Talking 

602 

603 # num_steps = self.get_num_steps() 

604 task = asyncio.create_task( 

605 self.gen.gen_video_audio_from_video( 

606 ft_frames, 

607 subaudio_b64, 

608 prompt=video_prompt, 

609 neg_prompt=video_neg_prompt, 

610 width=width, 

611 height=height, 

612 steps=num_steps, 

613 task_id=f"{self.video_id:03d}_{subvideo_id:03d}", 

614 deadline=deadline, 

615 )) 

616 

617 self.logger.info( 

618 f"[{self.video_id}.{subvideo_id}] Generating video+audio using " 

619 f"{len(hy_frames)}@{HUNYUANFRAMEPACK_FPS}FPS->" 

620 f"{len(ft_frames)}@{FANTASYTALKING_FPS}FPS frames...") 

621 

622 return task 

623 

624 def _add_debug( 

625 self, 

626 subvideo_id: int, 

627 ft_frames: List[Image.Image], 

628 ) -> List[Image.Image]: 

629 """ 

630 Add debug text to frames. 

631 We are adding this pre Fantasy Talking which could make it worse. 

632 """ 

633 # Id 

634 frame_text = f"{self.video_id:03d}.{subvideo_id:03d}" 

635 ft_frames = [ 

636 cast(Image.Image, add_text_to_frame(frame, text=frame_text, position="top-left")) 

637 for frame in ft_frames 

638 ] 

639 

640 # Size and number of frames 

641 width, height = ft_frames[0].size 

642 frame_text = f"{width}x{height} {len(ft_frames)} frames" 

643 ft_frames = [ 

644 cast(Image.Image, add_text_to_frame(frame, text=frame_text, position="top-right")) 

645 for frame in ft_frames 

646 ] 

647 return ft_frames 

648 

649 def _log_video_info( 

650 self, 

651 prefix: str, 

652 video_content: Union[bytes, str], 

653 ) -> None: 

654 """Log video information.""" 

655 video_file_info = get_video_file_info(video_content) 

656 video_num_bytes = video_file_info["overall"]["num_bytes"] 

657 

658 video_info = video_file_info["video"] 

659 video_fps = video_info["fps"] 

660 video_duration = video_info["duration_seconds"] 

661 video_num_frames = video_info["num_frames"] 

662 width, height = video_info["width"], video_info["height"] 

663 

664 self.logger.info( 

665 f"{prefix} with " 

666 f"{video_duration:.3f} seconds, " 

667 f"{video_num_frames} frames, " 

668 f"{video_fps} FPS, " 

669 f"{bytes_to_human(video_num_bytes)}, and " 

670 f"{width}x{height} pixels.") 

671 

672 

673# Backup code before the refactor 

674""" 

675async def gen_scene_chunks( 

676 self, 

677 scene_id: int, 

678 audio_path: str, 

679 image: Image.Image, 

680 video_prompt: str = VIDEO_PROMPT, 

681 video_neg_prompt: str = VIDEO_NEG_PROMPT, 

682 sub_scene_duration_max: int = MAX_FT_DURATION_SECS, 

683) -> bytes: 

684 " "" 

685 Generate a scene with audio and video starting with image at high resolution. 

686 1. Split audio into sub-scenes based on silences. 

687 2. Generate sketch at low resolution with Hunyuan FramePack. 

688 3. Generate video+audio at medium resolution with Fantasy Talking. 

689 Returns video synced with the audio in binary. 

690 " "" 

691 

692 # Split into sub-scenes based on silences (aligned with Fantasy Talking frames) 

693 audio_base64 = await read_file_base64(audio_path) 

694 audio_duration = get_audio_duration(audio_base64) 

695 audio_splits = get_audio_chunks_by_silences( 

696 audio_path, 

697 sub_scene_duration_max, 

698 chunk_alignment_seconds=1.0 / FANTASYTALKING_FPS) 

699 

700 if len(audio_splits) > audio_duration / sub_scene_duration_max + 1: 

701 self.logger.warning( 

702 f"[{scene_id}] Too many sub-scenes ({len(audio_splits)}) for audio with " 

703 f"{audio_duration:.3f} seconds and max {sub_scene_duration_max:.3f}.") 

704 

705 # Calculate #frames for each sub-scene 

706 scene_info = SceneInfo() 

707 for sub_scene_id, (start_secs, end_secs) in enumerate(audio_splits): 

708 sub_scene = SubSceneInfo(start_secs, end_secs) 

709 scene_info.append(sub_scene) 

710 num_audio_frames = sub_scene.get_num_audio_frames() 

711 duration_secs = end_secs - start_secs 

712 self.logger.info( 

713 f"[{scene_id}.{sub_scene_id}] Sub-scene: {start_secs:6.3f}-{end_secs:6.3f} ({duration_secs:6.3f}). " 

714 f"{num_audio_frames:4d} frames. FP:" 

715 f"{sub_scene.get_start_frame(HUNYUANFRAMEPACK_FPS):4d}-" 

716 f"{sub_scene.get_end_frame(HUNYUANFRAMEPACK_FPS):4d} frames.") 

717 if duration_secs > sub_scene_duration_max: 

718 self.logger.error( 

719 f"[{scene_id}.{sub_scene_id}] Sub-scene too long: {start_secs:.3f}-{end_secs:.3f} " 

720 f"({duration_secs:.3f} seconds) > {sub_scene_duration_max:.3f}.") 

721 elif duration_secs < 0.5: 

722 self.logger.warning( 

723 f"[{scene_id}.{sub_scene_id}] Sub-scene too short: {start_secs:.3f}-{end_secs:.3f} " 

724 f"({duration_secs:.3f} seconds).") 

725 

726 # Start long sketch video (no audio) generation in the background 

727 width, height = RESOLUTIONS[self.aspect_ratio]["low"] 

728 video_num_frames = get_num_video_frames_from_duration( 

729 audio_duration, 

730 HUNYUANFRAMEPACK_FPS, 

731 HUNYUANFRAMEPACK_VAE_T) 

732 

733 num_steps = self.get_num_steps() 

734 video_gen_request = await self.gen.gen_video( 

735 image, 

736 video_prompt, 

737 video_neg_prompt, 

738 width=width, 

739 height=height, 

740 num_frames=video_num_frames, 

741 steps=num_steps // 2, # Less steps for sketch video 

742 task_id=f"{scene_id:03d}_sketch", 

743 deadline=self.get_scene_deadline(scene_id), 

744 wait_request=False, 

745 ) 

746 self.logger.info( 

747 f"[{scene_id}] Generating long sketch video with {video_num_frames} frames, " 

748 f"{audio_duration:.3f} seconds, " 

749 f"{HUNYUANFRAMEPACK_FPS} FPS, and " 

750 f"{width}x{height} pixels.") 

751 

752 # Get frames while the request is running 

753 scene_hy_video_frames: List[Image.Image] = [] 

754 sub_scene_tasks: Dict[int, asyncio.Task] = {} 

755 video_gen_request_done = False 

756 while len(scene_hy_video_frames) < video_num_frames and not video_gen_request_done: 

757 if video_gen_request.done(): 

758 video_gen_request_done = True 

759 elif not video_gen_request.is_running() or not video_gen_request.url: 

760 RETRY_SLEEP_SECONDS = 1.0 

761 await asyncio.sleep(RETRY_SLEEP_SECONDS) # Wait for it to be running before checking again 

762 continue 

763 

764 # Get intermediate frames (while long sketch video generation is running) 

765 sub_scene_id = 0 

766 base_url = video_gen_request.get_base_request_url() 

767 async for frame in self.gen.gen_intermediate_video_frames( 

768 base_url, 

769 task_id=f"{scene_id:03d}_sketch", 

770 video_gen_request=video_gen_request, 

771 ): 

772 scene_hy_video_frames.append(frame) 

773 

774 # Process as many sub-scenes as possible with available frames 

775 while sub_scene_id < len(scene_info.sub_scenes): 

776 sub_scene_info = scene_info[sub_scene_id] 

777 # Add 4 more frames to account for the 1+4n VAE alignment 

778 if len(scene_hy_video_frames) < sub_scene_info.get_end_frame(HUNYUANFRAMEPACK_FPS) + VAE_T: 

779 break # Wait for more frames 

780 

781 # Enough frames for the sub-scene, generate video+audio 

782 sub_scene_task = await self.gen_sub_scene( 

783 scene_id, 

784 sub_scene_id, 

785 sub_scene_info, 

786 scene_hy_video_frames, 

787 audio_base64, 

788 video_prompt=video_prompt, 

789 video_neg_prompt=video_neg_prompt) 

790 if sub_scene_task is None: 

791 self.logger.error(f"[{scene_id}.{sub_scene_id}] Cannot generate sub-scene task.") 

792 elif sub_scene_id in sub_scene_tasks: 

793 self.logger.error( 

794 f"[{scene_id}.{sub_scene_id}] Sub-scene task already exists " 

795 f"({sub_scene_tasks.keys()}).") 

796 else: 

797 sub_scene_tasks[sub_scene_id] = sub_scene_task 

798 sub_scene_id += 1 

799 

800 self.logger.info(f"[{scene_id}] Got {len(scene_hy_video_frames)}/{video_num_frames} streamed video frames.") 

801 if len(scene_hy_video_frames) < video_num_frames: 

802 self.logger.warning(f"[{scene_id}] Not enough frames ({len(scene_hy_video_frames)} < {video_num_frames}).") 

803 

804 # Get the full video (without audio) 

805 content_type, video_binary = await video_gen_request.future 

806 if video_binary is None: 

807 raise ValueError(f"Cannot generate video for scene {scene_id}.") 

808 

809 self._log_video_info(f"[{scene_id}] Video sketch", video_binary) 

810 

811 video_path = f"{self.job_path}/{scene_id:03d}_chunks_sketch.mp4" 

812 async with aiofiles.open(video_path, "wb") as file: 

813 await file.write(video_binary) 

814 

815 # Collecting video+audio sub-scenes 

816 self.logger.info(f"[{scene_id}] Generating {len(sub_scene_tasks)} video+audio sub-scenes...") 

817 sub_scene_binaries = await asyncio.gather(*sub_scene_tasks.values(), return_exceptions=True) 

818 self.logger.info(f"[{scene_id}] Generated {len(sub_scene_tasks)} video+audio sub-scenes.") 

819 

820 if not sub_scene_binaries: 

821 raise ValueError(f"No sub-scenes generated for scene {scene_id}.") 

822 

823 for sub_scene_id, sub_scene_binary in enumerate(sub_scene_binaries): 

824 sub_scene_binary = sub_scene_binaries[sub_scene_id] 

825 if isinstance(sub_scene_binary, bytes): 

826 # Success case 

827 self._log_video_info( 

828 f"[{scene_id}.{sub_scene_id}] Generated video", 

829 sub_scene_binary) 

830 else: 

831 # Error case -> replace with static error video 

832 err_msg = str(sub_scene_binary) 

833 duration_seconds = scene_info[sub_scene_id].get_seconds() 

834 width, height = RESOLUTIONS[self.aspect_ratio]["medium"] 

835 self.logger.error( 

836 f"[{scene_id}.{sub_scene_id}] Failed to get video+audio: {err_msg}. " 

837 f"Adding error video with {duration_seconds:.3f} seconds and {width}x{height} pixels.") 

838 

839 frame_text = "Cannot generate video+audio\n" 

840 frame_text += f"Sub-scene {scene_id:03d}.{sub_scene_id:03d}\n" 

841 frame_text += "\n".join(split_text_lines(err_msg, MAX_IMG_LINE_CHARS)) 

842 font_size = get_font_size(width, height) 

843 sub_scene_binary = await get_video_with_text( 

844 width=width, 

845 height=height, 

846 text=frame_text, 

847 font_size=font_size, 

848 fps=FANTASYTALKING_FPS, 

849 duration_seconds=duration_seconds) 

850 sub_scene_binaries[sub_scene_id] = sub_scene_binary 

851 

852 video_path = f"{self.job_path}/{scene_id:03d}_{sub_scene_id:03d}_chunks.mp4" 

853 async with aiofiles.open(video_path, "wb") as file: 

854 await file.write(sub_scene_binary) 

855 

856 # Concatenate sub-scenes into final scene video 

857 self.logger.info(f"[{scene_id}] Concatenating {len(sub_scene_binaries)} sub-scenes...") 

858 scene_binary = await concatenate_videos( 

859 sub_scene_binaries, 

860 fast_copy=False) # move to True once we fix the durations 

861 if not scene_binary: 

862 raise ValueError(f"Cannot concatenate sub-scenes for scene {scene_id}") 

863 self._log_video_info( 

864 f"[{scene_id}] Concatenated {len(sub_scene_binaries)} videos into video", 

865 scene_binary) 

866 return scene_binary 

867""" 

868 

869""" 

870async def gen_sub_scene( 

871 self, 

872 scene_id: int, 

873 sub_scene_id: int, 

874 sub_scene_info: SubSceneInfo, 

875 scene_frames: List[Image.Image], # scene_hy_video_frames 

876 audio_base64: str, 

877 video_prompt: str = VIDEO_PROMPT, 

878 video_neg_prompt: str = VIDEO_NEG_PROMPT, 

879) -> Optional[asyncio.Task]: 

880 " "" 

881 Generate a sub-scene with video+audio from Hunyuan FramePack frames and audio_base64. 

882 " "" 

883 start_frame = sub_scene_info.get_start_frame(HUNYUANFRAMEPACK_FPS) 

884 end_frame = sub_scene_info.get_end_frame(HUNYUANFRAMEPACK_FPS) 

885 # Add some frames to account for the 1+4n alignment 

886 end_frame += VAE_T 

887 if end_frame > len(scene_frames): 

888 end_frame = len(scene_frames) # Don't go beyond available frames 

889 

890 # TODO align audio lengths 

891 # TODO check if this is just silence 

892 

893 # Audio 

894 sub_scene_audio_base64 = chunk_audio_base64( 

895 audio_base64, 

896 sub_scene_info.start_seconds, 

897 sub_scene_info.end_seconds) 

898 sub_scene_duration = sub_scene_info.end_seconds - sub_scene_info.start_seconds 

899 sub_scene_audio_num_frames = get_num_video_frames_from_duration(sub_scene_duration) 

900 

901 sub_scene_audio_path = f"{self.job_path}/{scene_id:03d}_{sub_scene_id:03d}.wav" 

902 await save_base64_as_binary(sub_scene_audio_path, sub_scene_audio_base64) 

903 

904 # Video 

905 # Hunyuan FramePack -> Fantasy Talking intermediate frames 

906 sub_scene_hy_frames = scene_frames[start_frame:end_frame] 

907 sub_scene_ft_frames = get_video_frames_at_fps( 

908 sub_scene_hy_frames, 

909 src_fps=HUNYUANFRAMEPACK_FPS,