import numpy as np import tempfile import os import imageio from PIL import Image from typing import List, Optional import shutil def create_video_from_frames(frames: List[np.ndarray], fps: int = 24) -> str: """ Create a video file from a list of frames Args: frames (List[np.ndarray]): List of video frames as numpy arrays fps (int): Frames per second for the output video Returns: str: Path to the generated video file """ if not frames: raise ValueError("No frames provided") # Create temporary file temp_dir = tempfile.mkdtemp() video_path = os.path.join(temp_dir, "generated_video.mp4") # Ensure frames are in the right format processed_frames = [] for frame in frames: if isinstance(frame, np.ndarray): # Convert numpy array to PIL Image if frame.dtype != np.uint8: frame = (frame * 255).astype(np.uint8) if len(frame.shape) == 3 and frame.shape[2] == 3: # RGB image pil_image = Image.fromarray(frame, mode='RGB') elif len(frame.shape) == 3 and frame.shape[2] == 4: # RGBA image pil_image = Image.fromarray(frame, mode='RGBA') else: # Grayscale pil_image = Image.fromarray(frame, mode='L') else: # Assume it's already a PIL Image pil_image = frame processed_frames.append(pil_image) # Save as video with imageio.get_writer(video_path, fps=fps) as writer: for frame in processed_frames: # Convert PIL to numpy array frame_array = np.array(frame) writer.append_data(frame_array) return video_path def save_video_temp(frames: List, fps: int = 24) -> str: """ Save video frames to a temporary file Args: frames (List): List of video frames fps (int): Frames per second Returns: str: Path to the saved video file """ try: return create_video_from_frames(frames, fps) except Exception as e: # Fallback: save as GIF if video creation fails temp_dir = tempfile.mkdtemp() gif_path = os.path.join(temp_dir, "generated_video.gif") # Convert frames to PIL Images and save as GIF pil_frames = [] for frame in frames: if isinstance(frame, np.ndarray): if frame.dtype != np.uint8: frame = (frame * 255).astype(np.uint8) pil_frame = Image.fromarray(frame) else: pil_frame = frame pil_frames.append(pil_frame) if pil_frames: pil_frames[0].save( gif_path, save_all=True, append_images=pil_frames[1:], duration=1000 // fps, loop=0 ) return gif_path else: raise Exception("No valid frames to save") def cleanup_temp_files(): """Clean up temporary files""" temp_dir = tempfile.gettempdir() # Clean up files older than 1 hour current_time = time.time() for filename in os.listdir(temp_dir): if filename.startswith("generated_video"): file_path = os.path.join(temp_dir, filename) try: if os.path.getmtime(file_path) < current_time - 3600: if os.path.isfile(file_path): os.unlink(file_path) elif os.path.isdir(file_path): shutil.rmtree(file_path) except Exception: pass def validate_prompt(prompt: str) -> bool: """ Validate that the prompt is not empty and has reasonable length Args: prompt (str): Input prompt Returns: bool: True if prompt is valid """ if not prompt or not prompt.strip(): return False if len(prompt.strip()) < 3: return False if len(prompt.strip()) > 1000: return False return True def format_status_message(message: str, success: bool = True) -> str: """ Format status message with appropriate emoji Args: message (str): Status message success (bool): Whether the operation was successful Returns: str: Formatted status message """ emoji = "✅" if success else "❌" return f"{emoji} {message}"