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 from vlm_captions import VLMCaptioning class VideoLLMInferenceNode: def __init__(self, vlm_captioner=None): """ Initialize the VideoLLMInferenceNode Args: vlm_captioner: The already initialized VLMCaptioning instance to use """ self.vlm = vlm_captioner self.sambanova_api_key = os.environ.get("SAMBANOVA_API_KEY", "") self.groq_api_key = os.environ.get("GROQ_API_KEY", "") self.groq_client = Groq(api_key=self.groq_api_key) self.sambanova_client = OpenAI( api_key=self.sambanova_api_key, base_url="https://api.sambanova.ai/v1", ) def analyze_image(self, image_path: str, question: Optional[str] = None) -> str: """ Analyze an image using the VLM model 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: # Use the passed vlm_captioner instance return self.vlm.describe_image(image_path, question) except Exception as e: return f"Error analyzing image: {str(e)}" def analyze_video(self, video_path: str) -> str: """ Analyze a video using the VLM model Args: video_path: Path to the video file Returns: str: Analysis result """ if not video_path: return "Please upload a video." try: # Use the passed vlm_captioner instance return self.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" ) -> 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 Returns: str: Generated video prompt """ if not concept: return "Please enter a concept for the video." # Build the prompt system_message = """You are a professional video prompt generator. Your task is to create detailed, technical, and creative video prompts based on user inputs. The prompts should be suitable for text-to-video AI models and include specific technical details that match the requested style, camera movement, pacing, and effects. Focus on creating high-quality, cohesive prompts that could be used to generate impressive AI videos.""" # Set prompt length guidelines length_guide = { "Short": "Create a concise prompt of 2-3 sentences.", "Medium": "Create a detailed prompt of 4-6 sentences.", "Long": "Create an extensive prompt with 7-10 sentences covering all details." } # Put together options for the prompt options = [] if style and style != "None": options.append(f"Style: {style}") if camera_style and camera_style != "None": options.append(f"Camera Movement Style: {camera_style}") if camera_direction and camera_direction != "None": options.append(f"Camera Direction: {camera_direction}") if pacing and pacing != "None": options.append(f"Pacing Rhythm: {pacing}") if special_effects and special_effects != "None": options.append(f"Special Effects: {special_effects}") if custom_elements: options.append(f"Custom Elements: {custom_elements}") options_text = "\n".join(options) user_message = f"""Create a video prompt based on the following concept and specifications: CONCEPT: {concept} SPECIFICATIONS: {options_text} {length_guide.get(prompt_length, length_guide["Medium"])} The prompt should be detailed and technical, specifically mentioning camera angles, movements, lighting, transitions, and other visual elements that would create an impressive AI-generated video. """ # Call the appropriate API based on provider try: if provider == "SambaNova": return self._call_sambanova_api(system_message, user_message, model) elif provider == "Groq": 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_api(self, system_message: str, user_message: str, model: str) -> str: """Call the SambaNova API for prompt generation""" 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_api(self, system_message: str, user_message: str, model: str) -> str: """Call the Groq API for prompt generation""" 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}"