llm_inference_video.py

import os
import time
import requests
from typing import Optional, Dict, Any, List
import json
import tempfile
from PIL import Image
from groq import Groq
from openai import OpenAI
import spaces

class VideoLLMInferenceNode:
    def __init__(self):
        """
        Initialize the VideoLLMInferenceNode without VLM captioning dependency
        """
        self.sambanova_api_key = os.environ.get("SAMBANOVA_API_KEY", "")
        self.groq_api_key = os.environ.get("GROQ_API_KEY", "")
        
        # Initialize API clients if keys are available
        if self.groq_api_key:
            self.groq_client = Groq(api_key=self.groq_api_key)
        else:
            self.groq_client = None
            
        if self.sambanova_api_key:
            self.sambanova_client = OpenAI(
                api_key=self.sambanova_api_key,
                base_url="https://api.sambanova.ai/v1",
            )
        else:
            self.sambanova_client = None

    @spaces.GPU()
    def analyze_image(self, image_path: str, question: Optional[str] = None) -> str:
        """
        Analyze an image using VLM model directly
        
        Args:
            image_path: Path to the image file
            question: Optional question to ask about the image
            
        Returns:
            str: Analysis result
        """
        if not image_path:
            return "Please upload an image."
            
        if not question or question.strip() == "":
            question = "Describe this image in detail."
            
        try:
            # Import and use VLMCaptioning within this GPU-scoped function
            from app import get_vlm_captioner
            vlm = get_vlm_captioner()
            return vlm.describe_image(image_path, question)
        except Exception as e:
            return f"Error analyzing image: {str(e)}"
    
    @spaces.GPU()
    def analyze_video(self, video_path: str) -> str:
        """
        Analyze a video using VLM model directly
        
        Args:
            video_path: Path to the video file
            
        Returns:
            str: Analysis result
        """
        if not video_path:
            return "Please upload a video."
            
        try:
            # Import and use VLMCaptioning within this GPU-scoped function
            from app import get_vlm_captioner
            vlm = get_vlm_captioner()
            return vlm.describe_video(video_path)
        except Exception as e:
            return f"Error analyzing video: {str(e)}"
    
    def generate_video_prompt(
        self,
        concept: str,
        style: str = "Simple",
        camera_style: str = "None",
        camera_direction: str = "None",
        pacing: str = "None",
        special_effects: str = "None",
        custom_elements: str = "",
        provider: str = "SambaNova",
        model: str = "Meta-Llama-3.1-70B-Instruct",
        prompt_length: str = "Medium",
        image_path: str = "",
        video_path: str = ""
    ) -> str:
        """
        Generate a video prompt using the specified LLM provider
        
        Args:
            concept: Core concept for the video
            style: Video style
            camera_style: Camera style
            camera_direction: Camera direction
            pacing: Pacing rhythm
            special_effects: Special effects approach
            custom_elements: Custom technical elements
            provider: LLM provider (SambaNova or Groq)
            model: Model name
            prompt_length: Desired prompt length
            image_path: Optional path to an image for VLM description
            video_path: Optional path to a video for VLM description
            
        Returns:
            str: Generated video prompt
        """
        if not concept:
            return "Please enter a concept for the video."
            
        try:
            # Get VLM descriptions if image or video paths are provided
            image_description = ""
            video_description = ""
            
            if image_path:
                try:
                    image_description = self.analyze_image(image_path, "Describe this image in detail for a video creator.")
                    print(f"Generated image description: {image_description}")
                except Exception as e:
                    print(f"Error generating image description: {str(e)}")
            
            if video_path:
                try:
                    video_description = self.analyze_video(video_path)
                    print(f"Generated video description: {video_description}")
                except Exception as e:
                    print(f"Error generating video description: {str(e)}")
            
            # Helper function to format optional elements
            def format_element(element, element_type):
                if element == "None" or not element:
                    return ""
                
                element_prefixes = {
                    "camera": "utilizing",
                    "direction": "with",
                    "pacing": "with",
                    "effects": "incorporating"
                }
                
                return f" {element_prefixes.get(element_type, '')} {element}"

            # Format camera movement combination
            camera_movement = ""
            if camera_style != "None" and camera_direction != "None":
                camera_movement = f"{camera_style} {camera_direction}"
            elif camera_style != "None":
                camera_movement = camera_style
            elif camera_direction != "None":
                camera_movement = camera_direction

            # Video prompt templates
            default_style = "simple"  # Changed from "cinematic" to "simple" as default
            
            prompt_templates = {
                "minimalist": f"""Create an elegantly sparse video description focusing on {concept}.
                    {format_element(camera_movement, 'camera')}
                    {format_element(pacing, 'pacing')}
                    {format_element(special_effects, 'effects')}
                    {' with ' + custom_elements if custom_elements else ''}.""",

                "dynamic": f"""Craft an energetic, fast-paced paragraph showcasing {concept} in constant motion. Utilize bold {camera_style} movements and {pacing} rhythm to create momentum. Layer {special_effects} effects and {custom_elements if custom_elements else 'powerful visual elements'} to maintain high energy throughout.""",

                "simple": f"""Create a straightforward, easy-to-understand paragraph describing a video about {concept}. Use {camera_style} camera work and {pacing} pacing. Keep the visuals clear and uncomplicated, incorporating {special_effects} effects and {custom_elements if custom_elements else 'basic visual elements'} in an accessible way.""",

                "detailed": f"""Construct a meticulous, technically precise paragraph outlining a video about {concept}. Incorporate specific details about {camera_style} cinematography, {pacing} timing, and {special_effects} effects. Include {custom_elements if custom_elements else 'precise technical elements'} while maintaining clarity and depth.""",

                "descriptive": f"""Write a richly descriptive paragraph for a video exploring {concept}. Paint a vivid picture using sensory details, incorporating {camera_style} movement, {pacing} flow, and {special_effects} effects. Emphasize texture, color, and atmosphere, enhanced by {custom_elements if custom_elements else 'evocative visual elements'}.""",

                "cinematic": f"""Create a single, detailed paragraph describing a cinematic video that captures {concept}. Focus on creating a cohesive narrative that incorporates {style} visual aesthetics, {camera_style} camera work, {pacing} pacing, and {special_effects} effects. Include atmospheric elements like {custom_elements if custom_elements else 'mood lighting and environmental details'} to enhance the storytelling. Describe the visual journey without technical timestamps or shot lists.""",

                "documentary": f"""Write a comprehensive paragraph for a documentary-style video exploring {concept}. Blend observational footage with {camera_style} cinematography, incorporating {pacing} editorial rhythm and {special_effects} visual treatments. Focus on creating an immersive narrative that educates and engages, enhanced by {custom_elements if custom_elements else 'authentic moments and natural lighting'}.""",

                "animation": f"""Compose a vivid paragraph describing a {style} animated video showcasing {concept}. Detail the unique visual style, character movements, and world-building elements, incorporating {camera_style} perspectives and {pacing} story flow. Include {special_effects} animation effects and {custom_elements if custom_elements else 'signature artistic elements'} to create a memorable visual experience.""",

                "action": f"""Craft an energetic paragraph describing an action sequence centered on {concept}. Emphasize the dynamic flow of action using {camera_style} cinematography, {pacing} rhythm, and {special_effects} visual effects. Incorporate {style} stylistic choices and {custom_elements if custom_elements else 'impactful moments'} to create an adrenaline-pumping experience.""",

                "experimental": f"""Create an avant-garde paragraph describing an experimental video exploring {concept}. Embrace unconventional storytelling through {style} aesthetics, {camera_style} techniques, and {pacing} temporal flow. Incorporate {special_effects} digital manipulations and {custom_elements if custom_elements else 'abstract visual metaphors'} to challenge traditional narrative structures."""
            }

            # Get the template with a more neutral default
            selected_style = style.lower()
            if selected_style not in prompt_templates:
                print(f"Warning: Style '{style}' not found, using '{default_style}' template")
                selected_style = default_style
                
            base_prompt = prompt_templates[selected_style]

            # Configure length requirements
            length_config = {
                "Short": {
                    "guidance": "Create exactly very short, ONE impactful sentence that captures the essence of the video. Be concise but descriptive.",
                    "structure": "Combine all elements into a single, powerful sentence."
                },
                "Medium": {
                    "guidance": "Create 2-3 flowing sentences that paint a picture of the video.",
                    "structure": "First sentence should set the scene, followed by 1-2 sentences developing the concept."
                },
                "Long": {
                    "guidance": "Create 4-5 detailed sentences that thoroughly describe the video.",
                    "structure": "Begin with the setting, develop the action/movement, and conclude with impact."
                }
            }
            
            config = length_config[prompt_length]

            system_message = f"""You are a visionary video director and creative storyteller. {config['guidance']}

Structure: {config['structure']}

Focus on these elements while maintaining the specified sentence count:
1. Visual atmosphere and mood
2. Camera movement and cinematography
3. Narrative flow
4. Style and aesthetic choices
5. Key moments
6. Emotional impact
{'' if not image_description and not video_description else '7. Elements from the provided image/video descriptions'}

{'' if not image_description and not video_description else 'If image or video descriptions are provided, incorporate their key visual elements and content into your description to ensure accuracy and relevance.'}

IMPORTANT REQUIREMENTS:
- Deliver exactly the specified number of sentences
- Short: ONE sentence
- Medium: TWO to THREE sentences
- Long: FOUR to FIVE sentences
- If camera movements are specified, you MUST incorporate them into the description
- Keep everything in a single paragraph format
- Avoid technical specifications or shot lists
- Avoid talking about 'video' or 'videos'. Do not start with 'The video opens with...' or 'The video starts with...' and do not include 'in this video' or 'focus of this video'. kind of terms"""

            # Format the user prompt with style guidance and camera movement
            user_message = f"""Style Guide: {selected_style.capitalize()} Style
{prompt_templates[selected_style]}

Camera Movement: {camera_movement if camera_movement else 'No specific camera movement'}
Core Concept: {concept}
{f'Reference Image Description: {image_description}' if image_description else ''}
{f'Reference Video Description: {video_description}' if video_description else ''}

Please create a {prompt_length.lower()}-length description incorporating these elements into a cohesive narrative.
{'' if not image_description and not video_description else 'Use the provided image/video descriptions as reference to inform your prompt creation.'}
Avoid talking about 'video' or 'videos'. Do not start with 'The video opens with...' or 'The video starts with...' and do not include 'in this video' or 'focus of this video'. kind of terms. Do not say "Here is your video prompt" or "Here is your video description" or anything like that. Just give the prompt."""

            # Call the appropriate API based on provider
            if provider == "SambaNova":
                if self.sambanova_client:
                    return self._call_sambanova_client(system_message, user_message, model)
                else:
                    return self._call_sambanova_api(system_message, user_message, model)
            elif provider == "Groq":
                if self.groq_client:
                    return self._call_groq_client(system_message, user_message, model) 
                else:
                    return self._call_groq_api(system_message, user_message, model)
            else:
                return "Unsupported provider. Please select SambaNova or Groq."
        except Exception as e:
            return f"Error generating prompt: {str(e)}"
    
    def _call_sambanova_client(self, system_message: str, user_message: str, model: str) -> str:
        """Call the SambaNova API using the client library"""
        try:
            chat_completion = self.sambanova_client.chat.completions.create(
                model=model,
                messages=[
                    {"role": "system", "content": system_message},
                    {"role": "user", "content": user_message}
                ]
            )
            return chat_completion.choices[0].message.content
        except Exception as e:
            return f"Error from SambaNova API: {str(e)}"
    
    def _call_sambanova_api(self, system_message: str, user_message: str, model: str) -> str:
        """Call the SambaNova API using direct HTTP requests"""
        if not self.sambanova_api_key:
            return "SambaNova API key not configured. Please set the SAMBANOVA_API_KEY environment variable."
        
        api_url = "https://api.sambanova.ai/api/v1/chat/completions"
        headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {self.sambanova_api_key}"
        }
        
        payload = {
            "model": model,
            "messages": [
                {"role": "system", "content": system_message},
                {"role": "user", "content": user_message}
            ]
        }
        
        response = requests.post(api_url, headers=headers, json=payload)
        
        if response.status_code == 200:
            result = response.json()
            return result.get("choices", [{}])[0].get("message", {}).get("content", "No content returned")
        else:
            return f"Error from SambaNova API: {response.status_code} - {response.text}"
    
    def _call_groq_client(self, system_message: str, user_message: str, model: str) -> str:
        """Call the Groq API using the client library"""
        try:
            chat_completion = self.groq_client.chat.completions.create(
                model=model,
                messages=[
                    {"role": "system", "content": system_message},
                    {"role": "user", "content": user_message}
                ]
            )
            return chat_completion.choices[0].message.content
        except Exception as e:
            return f"Error from Groq API: {str(e)}"
    
    def _call_groq_api(self, system_message: str, user_message: str, model: str) -> str:
        """Call the Groq API using direct HTTP requests"""
        if not self.groq_api_key:
            return "Groq API key not configured. Please set the GROQ_API_KEY environment variable."
        
        api_url = "https://api.groq.com/openai/v1/chat/completions"
        headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {self.groq_api_key}"
        }
        
        payload = {
            "model": model,
            "messages": [
                {"role": "system", "content": system_message},
                {"role": "user", "content": user_message}
            ]
        }
        
        response = requests.post(api_url, headers=headers, json=payload)
        
        if response.status_code == 200:
            result = response.json()
            return result.get("choices", [{}])[0].get("message", {}).get("content", "No content returned")
        else:
            return f"Error from Groq API: {response.status_code} - {response.text}"