utils.py

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}"