Coverage for apps/lmm_generator.py: 52%
603 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-09 04:47 +0000
1"""
2LMM Generator Client.
3"""
5from __future__ import annotations
7import sys
8import os
9import re
10import time
11import json
12import logging
13import asyncio
14import torch
15import traceback
17from PIL import Image
18from io import BytesIO
20from http import HTTPStatus
22from aiohttp import TCPConnector
23from aiohttp import ClientSession
24from aiohttp import ClientTimeout
26from json import JSONDecodeError
28from openai import AsyncOpenAI
29from openai import AsyncStream
30from openai.types.chat import ChatCompletionChunk
31from openai.types.chat import ChatCompletionMessageParam
33from enum import Enum
35from typing import List
36from typing import Optional
37from typing import Tuple
38from typing import Dict
39from typing import Union
40from typing import AsyncGenerator
41from typing import cast
43from lmm_service_manager import LMMServiceManager
45from resolutions import ASPECT_RATIO
46from resolutions import RESOLUTIONS
48from video import VideoQuality
50from client import ServiceRequest
51from client import ServiceRequestWorker
52from client import ServiceError
54from client_headers import JSON_HEADERS
56from client_timeout import SERVICE_TIMEOUT
57from client_timeout import SERVICE_MEDIUM_TIMEOUT
58from client_timeout import SERVICE_LONG_TIMEOUT
59from client_timeout import SERVICE_WARMUP_TIMEOUT
61sys.path.append("..") # noqa: E402
63from console_utils import setup_logging
64from console_utils import bytes_to_human
66from file_utils import binary_to_base64
68from image_utils import img_to_base64
69from image_utils import base64_to_img
70from media_utils import bytes_to_tensor
71from media_utils import get_video_frames
72from media_utils import video_frames_to_base64
73from media_utils import chunk_audio_base64
75from tts_utils import is_audio_base64_silence
77from k8s_utils import NoActiveContainerError
78from k8s_utils import NoRunnableContainerError
81class TaskClass(str, Enum):
82 """Class of tasks that can be performed by the services."""
83 TXT2IMG = "txt2img"
84 TXTIMG2IMG = "txtimg2img"
85 IMG2IMG = "img2img"
86 TXTIMG2VIDEO = "txtimg2video"
87 TTS = "tts"
88 VIDEOAUDIO2VIDEO = "videoaudio2video"
89 UPSCALE = "upscale"
92def get_service_name(
93 task_class: TaskClass,
94 quality: VideoQuality = VideoQuality.MEDIUM
95) -> str:
96 """
97 Get the name of the service to run the task
98 TODO get this from services.json.
99 """
100 if task_class == TaskClass.TXT2IMG:
101 return "flux"
102 # return "qwenimage"
103 # return "hunyuanimage"
104 if task_class == TaskClass.TXTIMG2IMG:
105 return "fluxkontext"
106 # return "qwenimageedit"
107 if task_class == TaskClass.IMG2IMG:
108 return "yolo"
109 if task_class == TaskClass.TXTIMG2VIDEO:
110 return "hunyuanframepackf1"
111 if task_class == TaskClass.TTS:
112 return "kokoro"
113 # return "vibevoice"
114 if task_class == TaskClass.VIDEOAUDIO2VIDEO:
115 return "fantasytalking"
116 if task_class == TaskClass.UPSCALE:
117 return "realesrgan"
118 raise ValueError(f"Unknown class '{task_class}'")
121class LMMGenerator:
122 """A client for managing requests to the LMM services and generating images, videos, and audio."""
124 def __init__(
125 self,
126 app_name: str,
127 job_id: str,
128 service_manager: LMMServiceManager,
129 ) -> None:
130 """Initialize the LMMGenerator with a service manager and an aiohttp client session."""
131 self.app_name = app_name
132 self.job_id = job_id
133 self.service_manager = service_manager
135 connector = TCPConnector(
136 limit=100,
137 limit_per_host=10,
138 use_dns_cache=True,
139 force_close=True)
140 self.session: Optional[ClientSession] = ClientSession(
141 connector=connector,
142 timeout=SERVICE_LONG_TIMEOUT)
144 self.requests: List[ServiceRequest] = []
145 self.request_executor = ServiceRequestWorker(
146 self.app_name,
147 self.service_manager)
148 self.request_executor_task = asyncio.create_task(self.request_executor.start())
150 self.logger = self._get_logger()
152 def _get_logger(self) -> logging.Logger:
153 """Get the logger for the LMMGenerator."""
154 logger = setup_logging(
155 path=f"/tmp/{self.app_name}/{self.job_id}",
156 file_name=f"{self.job_id}_lmm_generator.log",
157 level=logging.INFO,
158 use_global=False)
159 return logger
161 async def _submit_request(self, request: ServiceRequest) -> asyncio.Future:
162 """Submit a service request to the request executor and return the future."""
163 self.requests.append(request)
164 assert self.request_executor is not None
165 future = await self.request_executor.submit_request(request)
166 return future
168 async def stop(self) -> None:
169 """Stop the LMMGenerator and clean up resources."""
170 if self.request_executor:
171 await self.request_executor.stop()
172 self.request_executor = None
173 if self.session:
174 await self.session.close()
175 self.session = None
177 def get_queued_requests(self) -> List[str]:
178 """Get the list of queued request IDs from the request executor, sorted in ascending order."""
179 assert self.request_executor is not None
180 request_ids = self.request_executor.get_queued_requests()
181 request_ids.sort()
182 return request_ids
184 def get_requests(self) -> Dict[str, ServiceRequest]:
185 """Get the dictionary of all requests currently being managed by the request executor."""
186 assert self.request_executor is not None
187 return self.request_executor.get_requests()
189 def get_service_url(
190 self,
191 service_name: str,
192 exclude_busy: bool = True,
193 ) -> str:
194 """Get the URL of a service, optionally excluding busy containers."""
195 return self.service_manager.get_service_url(
196 service_name,
197 exclude_busy=exclude_busy)
199 def get_service_urls(
200 self,
201 service_name: str,
202 exclude_busy: bool = True,
203 ) -> List[str]:
204 """Get the URLs of a service, optionally excluding busy containers."""
205 return self.service_manager.get_service_urls(
206 service_name,
207 exclude_busy=exclude_busy)
209 async def get_files(
210 self,
211 base_url: str,
212 timeout: ClientTimeout = SERVICE_TIMEOUT,
213 ) -> Optional[list[dict[str, str]]]:
214 """Get the list of files available from the service at the given base URL."""
215 t0 = time.time()
216 url = f"{base_url}/files"
217 try:
218 # TODO use the request executor?
219 assert self.session is not None
220 async with self.session.get(url, timeout=timeout) as response:
221 if response.status == HTTPStatus.OK and "application/json" in response.headers.get("Content-Type", ""):
222 response_json = await response.json()
223 if "files" in response_json:
224 return response_json["files"]
225 except TimeoutError as timeout_err:
226 total_time = time.time() - t0
227 self.logger.error(f"Timeout ({total_time:.3f} > {timeout}) getting files from {url}: {timeout_err}")
228 return None
229 except Exception as ex:
230 self.logger.error(f"Error getting files from {url}: {ex}")
231 return None
233 err_msg = await response.text() if response else "No response"
234 status = response.status if response else "N/A"
235 self.logger.error(f"Cannot get files from {url}: HTTP status {status} Message:{err_msg}")
236 return None
238 async def get_file(
239 self,
240 base_url: str,
241 file_name: str,
242 timeout: ClientTimeout = SERVICE_TIMEOUT
243 ) -> Optional[bytes]:
244 """Get a specific file from the service at the given base URL."""
245 url = f"{base_url}/file/{file_name}"
246 try:
247 # TODO use the request executor?
248 assert self.session is not None
249 async with self.session.get(url, timeout=timeout) as response:
250 if response.status == HTTPStatus.OK:
251 # return await response.read()
252 # return await response.content.read()
253 # return await response.content.read(-1)
254 chunks = []
255 MAX_CHUNK_SIZE = 8 * 1024
256 async for chunk in response.content.iter_chunked(MAX_CHUNK_SIZE):
257 chunks.append(chunk)
258 return b"".join(chunks)
259 self.logger.error(f"Cannot get file {file_name} from {url}: HTTP status {response.status}")
260 except Exception as ex:
261 self.logger.error(f"Error getting file {file_name} from {url}: {ex}")
262 return None
264 def _assert_content_type(
265 self,
266 expected_content_type: str,
267 content_type: str,
268 msg: str,
269 request: ServiceRequest
270 ) -> None:
271 """Assert that the content type of a response matches the expected content type."""
272 if content_type != expected_content_type:
273 err_msg = f"{msg} {content_type}"
274 self.logger.error(f"{err_msg} for request {request.request_id}.")
275 raise ValueError(err_msg)
277 async def gen_image(
278 self,
279 prompt: str,
280 neg_prompt: str = "",
281 width: int = 1280,
282 height: int = 720,
283 steps: int = 25,
284 task_id: Optional[str] = None,
285 timeout: ClientTimeout = SERVICE_TIMEOUT,
286 deadline: Optional[float] = None,
287 ) -> Image.Image:
288 """Generate an image from a text prompt using a txt2img Flux service."""
289 service_name = get_service_name(TaskClass.TXT2IMG)
290 payload_json: dict[str, str | float | int | None] = {
291 "job_id": f"{self.job_id}_{task_id}",
292 "prompt": prompt,
293 "neg_prompt": neg_prompt,
294 "width": width,
295 "height": height,
296 "sampling_steps": steps,
297 }
298 request = ServiceRequest(
299 request_id=f"{self.job_id}_{task_id}_{service_name}",
300 service_name=service_name,
301 payload_json=payload_json,
302 timeout=timeout,
303 deadline=deadline,
304 )
305 try:
306 self.logger.info(f"Submitting request {request.request_id} to generate image.")
307 future = await self._submit_request(request)
308 content_type, image_binary = await future
309 self.logger.info(
310 f"Received response for request {request.request_id} "
311 f"with {bytes_to_human(len(image_binary))} and type {content_type}.")
312 self._assert_content_type(
313 "image/png", content_type,
314 "Unexpected image type", request)
315 image = await asyncio.to_thread(Image.open, BytesIO(image_binary))
316 return image
317 except Exception as ex:
318 err_msg = "Error generating image"
319 self.logger.error(f"{err_msg} for request {request.request_id}: {ex}")
320 raise ServiceError(
321 service_name=request.service_name,
322 job_id=f"{self.job_id}_{task_id}",
323 request_id=request.request_id,
324 message=err_msg,
325 url=self.get_service_url(request.service_name),
326 status=None, # Status is not available in this case
327 response_body=str(ex))
329 async def gen_edit_image(
330 self,
331 image: Image.Image,
332 prompt: str,
333 neg_prompt: str = "",
334 width: int = 1280,
335 height: int = 800,
336 steps: int = 25,
337 task_id: Optional[str] = None,
338 timeout: ClientTimeout = SERVICE_TIMEOUT,
339 deadline: Optional[float] = None,
340 ) -> Image.Image:
341 """Generate an edited image from an input image and a text prompt using the FluxKontext service."""
342 service_name = get_service_name(TaskClass.TXTIMG2IMG)
343 img_base64 = img_to_base64(image)
344 payload_json = {
345 "job_id": f"{self.job_id}_{task_id}",
346 "img": img_base64,
347 "prompt": prompt,
348 "neg_prompt": neg_prompt,
349 "width": width,
350 "height": height,
351 "sampling_steps": steps,
352 }
353 request = ServiceRequest(
354 request_id=f"{self.job_id}_{task_id}_{service_name}",
355 service_name=service_name,
356 payload_json=payload_json,
357 timeout=timeout,
358 deadline=deadline,
359 )
360 try:
361 self.logger.info(f"Submitting request {request.request_id} to edit image.")
362 future = await self._submit_request(request)
363 content_type, image_binary = await future
364 self.logger.info(
365 f"Received response for request {request.request_id} "
366 f"with {bytes_to_human(len(image_binary))} and type {content_type}.")
367 self._assert_content_type(
368 "image/png", content_type,
369 "Unexpected image type", request)
370 image = await asyncio.to_thread(Image.open, BytesIO(image_binary))
371 return image
372 except Exception as ex:
373 err_msg = "Error generating edited image"
374 self.logger.error(f"{err_msg} for request {request.request_id}: {ex}")
375 raise ServiceError(
376 service_name=request.service_name,
377 job_id=f"{self.job_id}_{task_id}",
378 request_id=request.request_id,
379 message=err_msg,
380 url=self.get_service_url(request.service_name),
381 status=None, # Status is not available in this case
382 response_body=str(ex))
384 async def gen_extract_characters(
385 self,
386 image: Image.Image,
387 num_characters: int = 2,
388 zoom_factor: float = 1.5,
389 task_id: Optional[str] = None,
390 timeout: ClientTimeout = SERVICE_TIMEOUT,
391 deadline: Optional[float] = None,
392 ) -> List[Image.Image]:
393 """Generate character images extracted from an input image using the YOLO service."""
394 service_name = get_service_name(TaskClass.IMG2IMG)
395 img_base64 = img_to_base64(image)
396 payload_json = {
397 "job_id": f"{self.job_id}_{task_id}",
398 "img": img_base64,
399 "num_characters": num_characters,
400 "zoom_factor": zoom_factor,
401 }
402 request = ServiceRequest(
403 request_id=f"{self.job_id}_{task_id}_{service_name}",
404 service_name=service_name,
405 payload_json=payload_json,
406 timeout=timeout,
407 deadline=deadline,
408 )
409 try:
410 self.logger.info(f"Submitting request {request.request_id} to generate images from image.")
411 future = await self._submit_request(request)
412 content_type, response_binary = await future
413 self.logger.info(
414 f"Received response for request {request.request_id} "
415 f"with {bytes_to_human(len(response_binary))} and type {content_type}.")
416 self._assert_content_type(
417 "application/json", content_type,
418 "Unexpected response type", request)
419 response_str = response_binary.decode("utf-8")
420 response_json = json.loads(response_str)
421 ret = []
422 for img_base64 in response_json.values():
423 img_character = base64_to_img(img_base64)
424 ret.append(img_character)
425 return ret
426 except Exception as ex:
427 err_msg = "Error generating images from image"
428 self.logger.error(f"{err_msg} for request {request.request_id}: {ex}")
429 raise ServiceError(
430 service_name=request.service_name,
431 job_id=f"{self.job_id}_{task_id}",
432 request_id=request.request_id,
433 message=err_msg,
434 url=self.get_service_url(request.service_name),
435 status=None, # Status is not available in this case
436 response_body=str(ex))
438 async def get_video_last_latents(
439 self,
440 base_url: Optional[str] = None,
441 task_id: Optional[str] = None
442 ) -> Tuple[int, Optional[torch.Tensor]]:
443 """Get the last latents for a video generation job from the service."""
444 if not task_id:
445 return -1, None
446 if base_url is None:
447 base_url = self.get_service_url("hunyuanframepackf1")
448 file_descrs = await self.get_files(base_url)
449 if not file_descrs:
450 return -1, None
452 last_file_name = None
453 last_index = -1
455 for file_desc in file_descrs:
456 file_name = file_desc.get("name")
457 if not file_name:
458 continue
459 # File names of the type: 20250706T191739_latents_024.pt
460 match = re.match(rf"^{self.job_id}_{task_id}_latents_(\d+)\.pt$", file_name)
461 if match:
462 iteration_index = int(match.group(1))
463 if iteration_index > last_index:
464 last_file_name = file_name
465 last_index = iteration_index
467 if last_file_name:
468 file_content = await self.get_file(base_url, last_file_name)
469 if file_content is None:
470 self.logger.error(f"Cannot get file content for {last_file_name} from {base_url}.")
471 return last_index, None
472 # ix, [B, C, T, H, W]
473 return last_index, bytes_to_tensor(file_content)
475 return last_index, None
477 async def gen_intermediate_video_frames(
478 self,
479 base_url: str,
480 task_id: str,
481 video_gen_request: ServiceRequest,
482 poll_secs: float = 0.1
483 ) -> AsyncGenerator[Image.Image, None]:
484 """Generate intermediate video frames from the latents of a video generation job."""
485 last_decoded_frame = 0
487 async def decode_and_yield(
488 current_index: int,
489 current_latents: torch.Tensor
490 ) -> AsyncGenerator[Image.Image, None]:
491 nonlocal last_decoded_frame
492 total_latents = current_latents.shape[2]
493 if last_decoded_frame >= total_latents:
494 return # Nothing new to decode
495 ix = 0 if last_decoded_frame == 0 else last_decoded_frame - 1 # Add an extra frame for VAE 1+4n
496 new_latents = current_latents[:, :, ix:, :, :] # [B, C, T, H, W]
498 try:
499 temp_video_binary = await self.gen_video_from_latents(
500 new_latents,
501 f"{self.job_id}_{task_id}_{current_index:03d}")
502 if temp_video_binary is not None:
503 video_frames = await get_video_frames(temp_video_binary)
504 ix = 0 if last_decoded_frame == 0 else 1 # Skip the first frame for VAE 1+4n
505 for frame in video_frames[ix:]:
506 yield frame
507 except NoActiveContainerError:
508 self.logger.error("Cannot decode latents: No active container for VAE service.")
509 except NoRunnableContainerError:
510 self.logger.error("Cannot decode latents: No runnable container for VAE service.")
511 except ServiceError as service_err:
512 self.logger.error(f"Cannot decode latents: {service_err}.")
513 except Exception as ex:
514 self.logger.error(f"Cannot generate video from intermediate latents: {ex}.")
515 traceback.print_exc()
516 finally:
517 last_decoded_frame = total_latents # Mark as decoded
519 # Keep polling until the video generation request is done
520 while not video_gen_request.done():
521 try:
522 # current_latents: [B, C, T, H, W]
523 current_index, current_latents = await self.get_video_last_latents(
524 base_url,
525 f"{self.job_id}_{task_id}")
526 if current_latents is not None:
527 total_latents = current_latents.shape[2]
528 if last_decoded_frame < total_latents:
529 async for frame in decode_and_yield(current_index, current_latents):
530 yield frame
531 except Exception as ex:
532 self.logger.warning(
533 f"Cannot poll intermediate latents for {self.job_id}_{task_id}: {ex}.")
535 await asyncio.sleep(poll_secs)
537 # After the request is done, yield the remaining frames (if any)
538 try:
539 current_index, current_latents = await self.get_video_last_latents(
540 base_url,
541 f"{self.job_id}_{task_id}")
542 if current_latents is None:
543 return
544 async for frame in decode_and_yield(current_index, current_latents):
545 yield frame
546 except Exception as ex:
547 self.logger.warning(f"Final decode failed: {ex}.")
549 async def gen_video(
550 self,
551 img: Image.Image,
552 prompt: str,
553 neg_prompt: str = "",
554 width: int = 640,
555 height: int = 400,
556 num_frames: int = -1,
557 video_seconds: float = -1,
558 steps: int = 10,
559 base_url: Optional[str] = None,
560 task_id: Optional[str] = None,
561 deadline: Optional[float] = None,
562 timeout: ClientTimeout = SERVICE_LONG_TIMEOUT,
563 wait_request: bool = True
564 ) -> Union[bytes, ServiceRequest]:
565 """Generate a video from an input image and a text prompt using the HunyuanFramePackF1 service."""
566 service_name = get_service_name(TaskClass.TXTIMG2VIDEO)
567 img_base64 = img_to_base64(img)
568 payload_json: dict[str, str | float | int | None] = {
569 "job_id": f"{self.job_id}_{task_id}",
570 "img": img_base64,
571 "prompt": prompt,
572 "neg_prompt": neg_prompt,
573 "width": width,
574 "height": height,
575 "sampling_steps": steps,
576 "save_intermediate": f"{self.job_id}_{task_id}", # Save the intermediate latents
577 "output_type": "pil",
578 }
579 if num_frames > 0:
580 payload_json["num_frames"] = num_frames
581 if video_seconds > 0:
582 payload_json["video_seconds"] = video_seconds
583 request = ServiceRequest(
584 request_id=f"{self.job_id}_{task_id}_{service_name}",
585 service_name=service_name,
586 payload_json=payload_json,
587 base_url=base_url,
588 timeout=timeout,
589 deadline=deadline,
590 )
591 try:
592 self.logger.info(f"Submitting request {request.request_id} to generate video.")
593 future = await self._submit_request(request)
594 if not wait_request:
595 return request
597 content_type, video_binary = await future
598 self.logger.info(
599 f"Received response for request {request.request_id} "
600 f"with {bytes_to_human(len(video_binary))} and type {content_type}.")
601 self._assert_content_type(
602 "video/mp4", content_type,
603 "Unexpected response type", request)
604 return video_binary
605 except Exception as ex:
606 err_msg = "Error generating video"
607 self.logger.error(f"{err_msg} for request {request.request_id}: {ex}")
608 raise ServiceError(
609 service_name=request.service_name,
610 job_id=f"{self.job_id}_{task_id}",
611 request_id=request.request_id,
612 message=err_msg,
613 url=self.get_service_url(request.service_name),
614 status=None, # Status is not available in this case
615 response_body=str(ex))
617 async def gen_image_upscale(
618 self,
619 image: Image.Image,
620 width: int = 1280,
621 height: int = 800,
622 task_id: Optional[str] = None,
623 timeout: ClientTimeout = SERVICE_LONG_TIMEOUT,
624 deadline: Optional[float] = None,
625 ) -> Image.Image:
626 """Generate an upscaled image from an input image using RealESRGAN."""
627 service_name = get_service_name(TaskClass.UPSCALE)
628 payload_json = {
629 "job_id": f"{self.job_id}_{task_id}",
630 "img": img_to_base64(image),
631 "width": width,
632 "height": height,
633 }
634 request = ServiceRequest(
635 request_id=f"{self.job_id}_{task_id}_{service_name}",
636 service_name=service_name,
637 payload_json=payload_json,
638 timeout=timeout,
639 deadline=deadline,
640 )
641 try:
642 self.logger.info(f"Submitting request {request.request_id} to generate upscaled image.")
643 future = await self._submit_request(request)
644 content_type, image_binary = await future
645 self.logger.info(
646 f"Received response for request {request.request_id} "
647 f"with {bytes_to_human(len(image_binary))} and type {content_type}.")
648 self._assert_content_type(
649 "image/png", content_type,
650 "Unexpected image type", request)
651 image = await asyncio.to_thread(Image.open, BytesIO(image_binary))
652 return image
653 except Exception as ex:
654 err_msg = "Error generating video image"
655 self.logger.error(f"{err_msg} for request {request.request_id}: {ex}")
656 raise ServiceError(
657 service_name=request.service_name,
658 job_id=f"{self.job_id}_{task_id}",
659 request_id=request.request_id,
660 message=err_msg,
661 url=self.get_service_url(request.service_name),
662 status=None, # Status is not available in this case
663 response_body=str(ex))
665 async def gen_video_upscale(
666 self,
667 video_binary: bytes,
668 width: int = 1280,
669 height: int = 800,
670 task_id: Optional[str] = None,
671 timeout: ClientTimeout = SERVICE_LONG_TIMEOUT,
672 deadline: Optional[float] = None,
673 ) -> bytes:
674 """
675 Generate an upscaled video from a base64 encoded video using RealESRGAN.
676 TODO We could stream frames out.
677 Returns video/mp4 video binary.
678 """
679 service_name = get_service_name(TaskClass.UPSCALE)
680 video_base64 = binary_to_base64(video_binary)
681 payload_json = {
682 "job_id": f"{self.job_id}_{task_id}",
683 "video": video_base64,
684 "width": width,
685 "height": height,
686 }
687 payload_len = len(json.dumps(payload_json).encode("utf-8"))
688 request = ServiceRequest(
689 request_id=f"{self.job_id}_{task_id}_{service_name}",
690 service_name=service_name,
691 payload_json=payload_json,
692 path="realesrgan/video",
693 timeout=timeout,
694 deadline=deadline,
695 )
696 try:
697 self.logger.info(
698 f"Submitting request {request.request_id} with "
699 f"{bytes_to_human(payload_len)} to generate upscaled video.")
700 future = await self._submit_request(request)
701 content_type, video_binary = await future
702 self.logger.info(
703 f"Received response for request {request.request_id} "
704 f"with {bytes_to_human(len(video_binary))} and type {content_type}.")
705 self._assert_content_type(
706 "video/mp4", content_type,
707 "Unexpected video type", request)
708 return video_binary
709 except Exception as ex:
710 err_msg = "Error generating video upscaled"
711 self.logger.error(f"{err_msg} for request {request.request_id}: {ex}")
712 raise ServiceError(
713 service_name=request.service_name,
714 job_id=f"{self.job_id}_{task_id}",
715 request_id=request.request_id,
716 message=err_msg,
717 url=self.get_service_url(request.service_name),
718 status=None, # Status is not available in this case
719 response_body=str(ex))
721 async def gen_video_from_latents(
722 self,
723 latents: torch.Tensor,
724 task_id: Optional[str] = None,
725 timeout: ClientTimeout = SERVICE_TIMEOUT,
726 deadline: Optional[float] = None,
727 ) -> bytes:
728 """
729 Generate a video from latents using the Hunyuan FramePack VAE service.
730 VAE: latents to video.
731 Returns video/mp4 video binary.
732 """
733 payload_buf = BytesIO()
734 torch.save(latents, payload_buf)
735 payload_bytes = payload_buf.getvalue()
737 service_name = "hunyuanframepackvae"
738 request = ServiceRequest(
739 request_id=f"{self.job_id}_{task_id}_{service_name}",
740 service_name=service_name,
741 payload_bytes=payload_bytes,
742 path=f"hunyuanframepack/vae/{self.job_id}_{task_id}",
743 timeout=timeout,
744 deadline=deadline,
745 )
746 try:
747 self.logger.info(f"Submitting request {request.request_id} to generate video from latent.")
748 future = await self._submit_request(request)
749 content_type, video_binary = await future
750 self.logger.info(
751 f"Received response for request {request.request_id} "
752 f"with {bytes_to_human(len(video_binary))} and type {content_type}.")
753 self._assert_content_type(
754 "video/mp4", content_type,
755 "Unexpected video type", request)
756 return video_binary
757 except Exception as ex:
758 err_msg = "Error generating video from latents"
759 self.logger.error(f"{err_msg} for request {request.request_id}: {ex}")
760 raise ServiceError(
761 service_name=request.service_name,
762 job_id=f"{self.job_id}_{task_id}",
763 request_id=request.request_id,
764 message=err_msg,
765 url=self.get_service_url(request.service_name),
766 status=None, # Status is not available in this case
767 response_body=str(ex))
769 async def gen_video_audio_from_img(
770 self,
771 img: Image.Image,
772 audio_base64: str,
773 prompt: str,
774 neg_prompt: str = "",
775 width: int = 640,
776 height: int = 400,
777 steps: int = 25,
778 end_percent: float = 0.9,
779 task_id: Optional[str] = None,
780 timeout: ClientTimeout = SERVICE_MEDIUM_TIMEOUT,
781 deadline: Optional[float] = None,
782 ) -> bytes:
783 """Generate a video from an input image and audio using the FantasyTalking service."""
784 assert isinstance(img, Image.Image), f"Image should be a PIL Image: {type(img)}"
785 img_base64 = img_to_base64(img)
786 assert type(img_base64) is str, f"Image should be a string: {type(img_base64)}"
787 assert type(audio_base64) is str, f"Audio should be a string: {type(audio_base64)}"
788 assert type(prompt) is str, f"Prompt should be a string: {type(prompt)}"
789 assert type(neg_prompt) is str, f"Negative prompt should be a string: {type(neg_prompt)}"
791 if is_audio_base64_silence(audio_base64):
792 # TODO do not use fantasy talking then
793 self.logger.warning("Audio is silence, generating video without audio.")
795 service_name = get_service_name(TaskClass.VIDEOAUDIO2VIDEO)
796 payload_json = {
797 "job_id": f"{self.job_id}_{task_id}",
798 "img": img_base64,
799 "prompt": prompt,
800 "neg_prompt": neg_prompt,
801 "audio": audio_base64,
802 "width": width,
803 "height": height,
804 "sampling_steps": steps,
805 "end_percent": end_percent,
806 }
807 request = ServiceRequest(
808 request_id=f"{self.job_id}_{task_id}_{service_name}",
809 service_name=service_name,
810 payload_json=payload_json,
811 timeout=timeout,
812 deadline=deadline,
813 )
814 try:
815 self.logger.info(f"Submitting request {request.request_id} to generate video+audio.")
816 future = await self._submit_request(request)
817 content_type, video_binary = await future
818 self.logger.info(
819 f"Received response for request {request.request_id} "
820 f"with {bytes_to_human(len(video_binary))} and type {content_type}.")
821 self._assert_content_type(
822 "video/mp4", content_type,
823 "Unexpected video type", request)
824 return video_binary
825 except Exception as ex:
826 err_msg = "Error generating video from image and audio"
827 self.logger.error(f"{err_msg} for request {request.request_id}: {ex}")
828 raise ServiceError(
829 service_name=request.service_name,
830 job_id=f"{self.job_id}_{task_id}",
831 request_id=request.request_id,
832 message=err_msg,
833 url=self.get_service_url(request.service_name),
834 status=None, # Status is not available in this case
835 response_body=str(ex))
837 async def gen_video_audio_from_video(
838 self,
839 video: List[Image.Image],
840 audio_base64: str,
841 prompt: str,
842 neg_prompt: str = "",
843 width: int = 640,
844 height: int = 400,
845 cfg_scale: float = 7.0, # for lipsync, to preserve input video
846 audio_cfg_scale: float = 7.0, # for lipsync, to be consistent with the audio
847 steps: int = 10,
848 end_percent: float = 0.9,
849 task_id: Optional[str] = None,
850 timeout: ClientTimeout = SERVICE_MEDIUM_TIMEOUT,
851 deadline: Optional[float] = None,
852 ) -> bytes:
853 """Generate a video from an input video and audio using the FantasyTalking service."""
854 assert isinstance(video, list), f"Video should be a list of PIL Images: {type(video)}"
855 assert len(video) > 0, f"Video list should not be empty: {len(video)}"
856 assert all(isinstance(frame, Image.Image)
857 for frame in video), f"All video frames should be PIL Images: {[type(frame) for frame in video]}"
858 video_base64 = video_frames_to_base64(video, fps=30)
859 assert type(video_base64) is str, f"Image should be a string: {type(video_base64)}"
860 assert type(audio_base64) is str, f"Audio should be a string: {type(audio_base64)}"
861 assert type(prompt) is str, f"Prompt should be a string: {type(prompt)}"
862 assert type(neg_prompt) is str, f"Negative prompt should be a string: {type(neg_prompt)}"
864 if is_audio_base64_silence(audio_base64):
865 # TODO do not use fantasy talking then
866 self.logger.warning("Audio is silence, generating video without audio.")
868 service_name = get_service_name(TaskClass.VIDEOAUDIO2VIDEO)
869 payload_json = {
870 "job_id": f"{self.job_id}_{task_id}",
871 "video": video_base64,
872 "prompt": prompt,
873 "neg_prompt": neg_prompt,
874 "audio": audio_base64,
875 "width": width,
876 "height": height,
877 "cfg_scale": cfg_scale,
878 "audio_cfg_scale": audio_cfg_scale,
879 "sampling_steps": steps,
880 "end_percent": end_percent,
881 }
882 request = ServiceRequest(
883 request_id=f"{self.job_id}_{task_id}_{service_name}",
884 service_name=service_name,
885 payload_json=payload_json,
886 timeout=timeout,
887 deadline=deadline,
888 )
889 try:
890 self.logger.info(f"Submitting request {request.request_id} to generate video+audio from video.")
891 future = await self._submit_request(request)
892 content_type, video_binary = await future
893 self.logger.info(
894 f"Received response for request {request.request_id} "
895 f"with {bytes_to_human(len(video_binary))} and type {content_type}.")
896 self._assert_content_type(
897 "video/mp4", content_type,
898 "Unexpected video type", request)
899 return video_binary
900 except Exception as ex:
901 err_msg = "Error generating video with audio from video"
902 self.logger.error(f"{err_msg} for request {request.request_id}: {ex}")
903 raise ServiceError(
904 service_name=request.service_name,
905 job_id=f"{self.job_id}_{task_id}",
906 request_id=request.request_id,
907 message=err_msg,
908 url=self.get_service_url(request.service_name),
909 status=None, # Status is not available in this case
910 response_body=str(ex))
912 async def _get_service_url(self, service_name: str) -> str:
913 """
914 Get the service URL, retrying if no active container is found.
915 """
916 # TODO Implement this in client.py and use the request executor?
917 retry = 0
918 MAX_RETRIES = 3
919 base_url = None
920 while base_url is None:
921 try:
922 base_url = self.get_service_url(service_name)
923 except NoActiveContainerError:
924 if retry > MAX_RETRIES:
925 raise
926 self.logger.error(f"No container for {service_name}, retry {retry}/{MAX_RETRIES}...")
927 await asyncio.sleep(0.1 * (2 ** retry))
928 retry += 1
929 return base_url
931 async def gen_text(
932 self,
933 messages: List[Dict],
934 service_name: str = "gemma",
935 llm_model: str = "google/gemma-3-27b-it",
936 api_key: str = "n/a",
937 max_tokens: int = 1024,
938 extra_body: Optional[Dict] = None,
939 task_id: Optional[str] = None,
940 ) -> str:
941 base_url = await self._get_service_url(service_name)
942 url = f"{base_url}/v1"
944 # vLLM OpenAI-compatible client
945 async with AsyncOpenAI(base_url=url, api_key=api_key,) as llm_client:
946 response = await llm_client.chat.completions.create(
947 model=llm_model,
948 messages=cast(List[ChatCompletionMessageParam], messages),
949 max_tokens=max_tokens,
950 extra_body=extra_body,
951 # timeout=10.0,
952 extra_headers={"X-Request-ID": f"{self.job_id}_{task_id}"},
953 stream=False,
954 )
955 assert response is not None
957 # Process LLM response
958 assert response.usage is not None
959 usage = response.usage
960 self.logger.debug("LLM tokens:")
961 self.logger.debug(f" Prompt: {usage.prompt_tokens}")
962 self.logger.debug(f" Completion: {usage.completion_tokens}")
963 self.logger.debug(f" Total: {usage.total_tokens}")
964 if usage.completion_tokens == max_tokens:
965 self.logger.error(f"Completion hit max tokens limit ({usage.completion_tokens}/{max_tokens}).")
967 choices = response.choices
968 if not choices:
969 raise ValueError("No LLM response.")
970 response_choice = choices[0]
971 response_message = response_choice.message
972 response_message_content = response_message.content
973 if response_message_content:
974 response_message_content = response_message_content.strip()
975 assert isinstance(response_message_content, str)
976 return response_message_content
978 async def gen_text_stream(
979 self,
980 messages: List[Dict],
981 service_name: str = "gemma",
982 llm_model: str = "google/gemma-3-27b-it",
983 api_key: str = "n/a",
984 max_tokens: int = 1024,
985 extra_body: Optional[Dict] = None,
986 task_id: Optional[str] = None,
987 ) -> AsyncGenerator[str, None]:
988 base_url = await self._get_service_url(service_name)
989 url = f"{base_url}/v1"
991 # vLLM OpenAI-compatible client
992 async with AsyncOpenAI(base_url=url, api_key=api_key,) as llm_client:
993 response = cast(
994 AsyncStream[ChatCompletionChunk],
995 await llm_client.chat.completions.create(
996 model=llm_model,
997 messages=cast(List[ChatCompletionMessageParam], messages),
998 max_tokens=max_tokens,
999 extra_body=extra_body,
1000 # timeout=10.0,
1001 extra_headers={"X-Request-ID": f"{self.job_id}_{task_id}"},
1002 stream=True,
1003 )
1004 )
1005 async for chunk in response:
1006 choice = chunk.choices[0]
1007 delta = choice.delta
1008 delta_content = delta.content
1009 assert isinstance(delta_content, str)
1010 yield delta_content
1012 async def gen_audio_transcript(
1013 self,
1014 audio_path: str,
1015 service_name: str = "whisper",
1016 whisper_model: str = "openai/whisper-large-v3",
1017 language: str = "en",
1018 api_key: str = "n/a",
1019 task_id: Optional[str] = None,
1020 ) -> Tuple[str, str]:
1021 """
1022 Generate a transcript from an audio file.
1023 This may take noise and music and try to transcribe it; this need to be accounted for.
1024 Returns a tuple of (transcript_text, language_code).
1025 """
1026 if not os.path.isfile(audio_path):
1027 raise ValueError(f"Audio path is not a file: {audio_path}")
1029 base_url = await self._get_service_url(service_name)
1030 url = f"{base_url}/v1"
1032 # Whisper audio client
1033 async with AsyncOpenAI(base_url=url, api_key=api_key) as whisper_client:
1034 with open(audio_path, "rb") as file_audio:
1035 response = await whisper_client.audio.transcriptions.create(
1036 model=whisper_model,
1037 file=file_audio,
1038 response_format="json",
1039 language=language,
1040 # timeout=10.0,
1041 extra_headers={"X-Request-ID": f"{self.job_id}_{task_id}"},
1042 )
1043 transcript = response.text
1044 language_code = getattr(response, "language", None) or language
1045 return transcript, language_code
1047 async def gen_podcast_transcript(
1048 self,
1049 pdf_base64: Optional[str] = None,
1050 pdf_url: Optional[str] = None,
1051 max_tokens: int = 5 * 1024,
1052 num_characters: int = 2,
1053 style_prompt: Optional[str] = None,
1054 scene_prompt: Optional[str] = None,
1055 custom_prompt: Optional[str] = None,
1056 max_dialogues: int = 10,
1057 max_words_per_dialogue: int = 50,
1058 streaming: bool = True,
1059 task_id: Optional[str] = None,
1060 timeout: ClientTimeout = SERVICE_LONG_TIMEOUT
1061 ) -> AsyncGenerator[dict, None]:
1062 """Generate a podcast transcript from a PDF document using the Gemma service."""
1063 # TODO move it to the podcast subclass
1064 # TODO add to list of requests self.requests.append(request)
1065 llm_model = "google/gemma-3-27b-it"
1066 base_url = self.get_service_url("gemma")
1067 llm_url = f"{base_url}/v1"
1068 multi_modal = True
1070 if not isinstance(pdf_base64, str):
1071 raise ValueError(f"pdf_base64 must be a string but is {type(pdf_base64)}")
1073 payload_json = {
1074 "pdf_url": pdf_url,
1075 "doc": pdf_base64,
1076 "llm_model": llm_model,
1077 "llm_url": llm_url,
1078 "multi_modal": multi_modal,
1079 "max_tokens": max_tokens,
1080 "num_characters": num_characters,
1081 "max_dialogues": max_dialogues,
1082 "max_words_per_dialogue": max_words_per_dialogue,
1083 }
1084 if style_prompt:
1085 payload_json["style_prompt"] = style_prompt
1086 if scene_prompt:
1087 payload_json["scene_prompt"] = scene_prompt
1088 if custom_prompt:
1089 payload_json["custom_prompt"] = custom_prompt
1091 service_name = "podcasttranscript"
1092 base_url = self.get_service_url(service_name)
1093 url = f"{base_url}/{service_name}"
1094 if streaming:
1095 url = f"{base_url}/{service_name}/stream"
1097 try:
1098 # TODO use request executor?
1099 assert self.session is not None
1100 async with self.session.post(
1101 url,
1102 json=payload_json,
1103 headers=JSON_HEADERS,
1104 timeout=timeout
1105 ) as response:
1106 if response.ok:
1107 if streaming:
1108 async for line in response.content:
1109 line_strip = ""
1110 try:
1111 line_strip = line.decode("utf-8").strip()
1112 if line_strip:
1113 line_json = json.loads(line_strip)
1114 yield line_json
1115 except JSONDecodeError as json_error:
1116 self.logger.error(f"JSON decode error: {json_error}. Line: {line_strip}")
1117 return
1118 else:
1119 data_json = await response.json()
1120 yield data_json
1121 return
1123 # Handle error responses
1124 if response.headers.get("Content-Type") == "application/json":
1125 data_json = await response.json()
1126 error_message = data_json.get("error", "Unknown error")
1127 else:
1128 error_message = await response.text()
1130 err_msg = "Error generating podcast transcript"
1131 self.logger.error(
1132 f"{err_msg} for job_id={self.job_id} task_id={task_id} at {url}: "
1133 f"HTTP status {response.status} Message: {error_message}")
1134 raise ServiceError(
1135 service_name=service_name,
1136 job_id=f"{self.job_id}_{task_id}",
1137 message=err_msg,
1138 url=url,
1139 status=response.status,
1140 response_body=error_message)
1141 except TimeoutError:
1142 err_msg = "Timeout generating podcast transcript"
1143 self.logger.error(f"{err_msg} for job_id={self.job_id} task_id={task_id} at {url}.")
1144 raise ServiceError(
1145 service_name=service_name,
1146 job_id=f"{self.job_id}_{task_id}",
1147 message=err_msg,
1148 url=url,
1149 status=HTTPStatus.REQUEST_TIMEOUT,
1150 response_body="Timeout generating podcast transcript.")
1151 except Exception as ex:
1152 err_msg = "Error generating podcast transcript"
1153 self.logger.error(f"{err_msg} for job_id={self.job_id} task_id={task_id} at {url} [{type(ex)}]: {ex}")
1154 self.logger.error(f"Trace: {traceback.format_exc()}")
1155 raise ServiceError(
1156 service_name=service_name,
1157 job_id=f"{self.job_id}_{task_id}",
1158 message=err_msg,
1159 url=url,
1160 response_body=str(ex))
1162 async def gen_slides_transcript(
1163 self,
1164 pptx_base64: str,
1165 task_id: Optional[str] = None,
1166 max_words_per_slide: Optional[int] = None,
1167 timeout: ClientTimeout = SERVICE_LONG_TIMEOUT,
1168 # deadline: Optional[float] = None,
1169 ) -> AsyncGenerator[dict, None]:
1170 llm_model = "google/gemma-3-27b-it"
1171 base_url = self.get_service_url("gemma")
1172 llm_url = f"{base_url}/v1"
1173 multi_modal = True
1175 if not isinstance(pptx_base64, str):
1176 raise ValueError(f"pptx_base64 must be a string but is {type(pptx_base64)}")
1178 payload_json = {
1179 "pptx": pptx_base64,
1180 "llm_model": llm_model,
1181 "llm_url": llm_url,
1182 "multi_modal": multi_modal,
1183 "max_words_per_slide": max_words_per_slide,
1184 }
1186 service_name = "slidetranscript"
1187 base_url = self.get_service_url(service_name)
1188 url = f"{base_url}/{service_name}/stream"
1190 try:
1191 # TODO use service request executor? with deadline
1192 assert self.session is not None
1193 async with self.session.post(url, json=payload_json, headers=JSON_HEADERS, timeout=timeout) as response:
1194 if response.ok:
1195 async for line in response.content:
1196 line_strip = ""
1197 try:
1198 line_strip = line.decode("utf-8").strip()
1199 if line_strip:
1200 line_json = json.loads(line_strip)
1201 yield line_json
1202 except JSONDecodeError as json_error:
1203 self.logger.error(f"JSON decode error: {json_error}. Line: {line_strip}")
1204 return
1206 # Handle error responses
1207 if response.headers.get("Content-Type") == "application/json":
1208 data_json = await response.json()
1209 error_message = data_json.get("error", "Unknown error")
1210 else:
1211 error_message = await response.text()
1213 err_msg = "Error generating slides transcript"
1214 self.logger.error(
1215 f"{err_msg} for job_id={self.job_id} task_id={task_id} at {url}: "
1216 f"HTTP status {response.status} Message: {error_message}")
1217 raise ServiceError(
1218 service_name=service_name,
1219 job_id=f"{self.job_id}_{task_id}",
1220 message=err_msg,
1221 url=url,
1222 status=response.status,
1223 response_body=error_message)
1224 except TimeoutError:
1225 err_msg = "Timeout generating slides transcript"
1226 self.logger.error(f"{err_msg} for job_id={self.job_id} task_id={task_id} at {url}.")
1227 raise ServiceError(
1228 service_name=service_name,
1229 job_id=f"{self.job_id}_{task_id}",
1230 message=err_msg,
1231 url=url,
1232 status=HTTPStatus.REQUEST_TIMEOUT,
1233 response_body="Timeout generating slides transcript.")
1234 except Exception as ex:
1235 err_msg = "Error generating slides transcript"
1236 self.logger.error(f"{err_msg} for job_id={self.job_id} task_id={task_id} at {url} [{type(ex)}]: {ex}")
1237 raise ServiceError(
1238 service_name=service_name,
1239 job_id=f"{self.job_id}_{task_id}",
1240 message=err_msg,
1241 url=url,
1242 response_body=str(ex))
1244 async def gen_audio(
1245 self,
1246 text: str,
1247 voice: str = "af_heart", # Default voice
1248 speed: float = 1.0,
1249 lang_code: str = "a", # American English
1250 voice_sample: Optional[str] = None,
1251 task_id: Optional[str] = None,
1252 timeout: ClientTimeout = SERVICE_TIMEOUT,
1253 deadline: Optional[float] = None,
1254 ) -> str:
1255 """
1256 Generate audio from text.
1258 Routes automatically: when *voice_sample* is supplied the request goes to
1259 the VibeVoice service for voice cloning; otherwise the standard TTS service
1260 (e.g. kokoro) is used.
1262 Args:
1263 text: The text to synthesise.
1264 voice: Name of a built-in voice preset (used when no voice_sample).
1265 speed: Speech speed multiplier.
1266 lang_code: Target language code.
1267 voice_sample: Optional base64-encoded WAV audio to clone the voice from.
1268 When present, the request is routed to VibeVoice instead of standard TTS.
1269 task_id: Optional task identifier for logging and request tracking.
1270 timeout: aiohttp client timeout for the service request.
1271 deadline: Optional absolute time (epoch seconds) by which the result must arrive.
1272 """
1273 if voice_sample is not None:
1274 service_name = "vibevoice"
1275 else:
1276 service_name = get_service_name(TaskClass.TTS)
1277 payload_json: Dict[str, Union[str, float]] = {
1278 "job_id": f"{self.job_id}_{task_id}",
1279 "text": text,
1280 "voice": voice,
1281 "speed": speed,
1282 "lang_code": lang_code,
1283 }
1284 if voice_sample is not None:
1285 payload_json["voice_sample"] = voice_sample
1286 request = ServiceRequest(
1287 request_id=f"{self.job_id}_{task_id}_{service_name}",
1288 service_name=service_name,
1289 payload_json=payload_json,
1290 timeout=timeout,
1291 deadline=deadline,
1292 )
1293 try:
1294 self.logger.info(f"Submitting request {request.request_id} to generate audio.")
1295 future = await self._submit_request(request)
1296 content_type, audio_binary = await future
1297 self.logger.info(
1298 f"Received response for request {request.request_id} "
1299 f"with {bytes_to_human(len(audio_binary))} and type {content_type}.")
1300 self._assert_content_type(
1301 "audio/wav", content_type,
1302 "Unexpected audio type", request)
1303 audio_base64 = binary_to_base64(audio_binary)
1304 return audio_base64
1305 except Exception as ex:
1306 err_msg = "Error generating audio"
1307 self.logger.error(f"{err_msg} for request {request.request_id}: {ex}")
1308 raise ServiceError(
1309 service_name=request.service_name,
1310 job_id=f"{self.job_id}_{task_id}",
1311 request_id=request.request_id,
1312 message=err_msg,
1313 url=self.get_service_url(request.service_name),
1314 response_body=str(ex))
1316 async def warmup_services(
1317 self,
1318 job_id: str = "StreamWiseWarmup",
1319 ) -> None:
1320 """Warmup the services used by the LMMGenerator to reduce latency for the first requests."""
1321 try:
1322 # TODO warmup also Hunyuan FramePack VAE
1323 # TODO warmup all replicas
1325 # Skipping for now as it is too expensive and doesn't add much.
1326 async def consume_podcast_transcript() -> None:
1327 async for line in self.gen_podcast_transcript(
1328 pdf_url="https://www.w3.org/WAI/ER/tests/xhtml/testfiles/resources/pdf/dummy.pdf",
1329 num_characters=2,
1330 max_dialogues=3,
1331 max_words_per_dialogue=5,
1332 streaming=True,
1333 task_id="warmup",
1334 timeout=SERVICE_WARMUP_TIMEOUT
1335 ):
1336 logging.debug(f"Podcast transcript line: {line}")
1338 podcast_transcript_task = asyncio.create_task(consume_podcast_transcript())
1339 width, height = RESOLUTIONS[ASPECT_RATIO]["High"]
1340 image_task = asyncio.create_task(
1341 self.gen_image(
1342 "warmup image",
1343 neg_prompt="blue",
1344 steps=5,
1345 width=width,
1346 height=height,
1347 task_id="warmup",
1348 timeout=SERVICE_WARMUP_TIMEOUT))
1349 audio_task = asyncio.create_task(
1350 self.gen_audio(
1351 "warmup text saying something",
1352 task_id="warmup",
1353 timeout=SERVICE_WARMUP_TIMEOUT))
1355 warmup_image = await image_task
1357 extract_character_task = asyncio.create_task(
1358 self.gen_extract_characters(
1359 warmup_image,
1360 task_id="warmup"))
1362 RESOLUTION_LOW = RESOLUTIONS[ASPECT_RATIO]["Low"]
1363 video_task = asyncio.create_task(
1364 self.gen_video(
1365 warmup_image,
1366 "warmup video",
1367 width=RESOLUTION_LOW[0],
1368 height=RESOLUTION_LOW[1],
1369 video_seconds=0.5,
1370 steps=2,
1371 task_id="warmup",
1372 timeout=SERVICE_WARMUP_TIMEOUT))
1374 RESOLUTION_HIGH = RESOLUTIONS[ASPECT_RATIO]["High"]
1375 upscale_image_task = asyncio.create_task(
1376 self.gen_image_upscale(
1377 warmup_image,
1378 width=RESOLUTION_HIGH[0],
1379 height=RESOLUTION_HIGH[1],
1380 task_id="warmup",
1381 timeout=SERVICE_WARMUP_TIMEOUT))
1383 warmup_audio = await audio_task
1384 short_audio = chunk_audio_base64(warmup_audio, 0.0, 0.5)
1385 RESOLUTION_MEDIUM = RESOLUTIONS[ASPECT_RATIO]["Medium"]
1386 video_audio_task = asyncio.create_task(
1387 self.gen_video_audio_from_img(
1388 warmup_image,
1389 short_audio,
1390 "warmup video",
1391 width=RESOLUTION_MEDIUM[0],
1392 height=RESOLUTION_MEDIUM[1],
1393 steps=2,
1394 task_id="warmup",
1395 timeout=SERVICE_WARMUP_TIMEOUT))
1397 results = await asyncio.gather(
1398 podcast_transcript_task, # TODO move to separate class
1399 extract_character_task,
1400 upscale_image_task,
1401 video_task,
1402 video_audio_task,
1403 return_exceptions=True)
1405 for result in results:
1406 if isinstance(result, Exception):
1407 logging.error(f"Error during warmup: {result}")
1408 except ServiceError as service_error:
1409 logging.error(f"Error during warmup: {service_error}")
1410 except Exception as ex:
1411 logging.error(f"Error during warmup: {ex}.")
1412 # traceback.print_exc()