After Effects Motion Control Panel

# gemini_processor.py import os import google.generativeai as genai from dotenv import load_dotenv import json from datetime import datetime from typing import Dict, List, Any, Optional import logging import re # Set up logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s' ) # Load environment variables load_dotenv() api_key = os.getenv("GEMINI_API_KEY") if not api_key or api_key == "your_api_key_here": logging.warning("GEMINI_API_KEY not found or not set to a valid value. Please set it in the .env file.") logging.warning("Gemini AI features will not work without a valid API key.") api_key = None # Configure Gemini gemini_configured = False try: if api_key: genai.configure(api_key=api_key) model = genai.GenerativeModel('gemini-1.5-flash') gemini_configured = True logging.info("Gemini API configured successfully (using gemini-1.5-flash)") else: logging.warning("Skipping Gemini API configuration due to missing API key.") model = None except Exception as e: logging.error(f"Failed to configure Gemini API: {str(e)}") model = None SYSTEM_PROMPT = """You are an AI assistant that helps users control After Effects through natural language commands. Your task is to convert user requests into structured JSON commands that can be executed by After Effects. The JSON object you return should have this structure: { "action": "string", // The action to perform (use snake_case, e.g., 'create_text_layer') "params": { // Parameters for the action // Action-specific parameters }, "confirmation_message": "string" // Message to show user } Supported actions and their parameters: 1. import_media: - filePath: string (required) - Path to the media file - type: string (optional) - Type of media (image, video, audio) - position: array (optional) - [x, y] position - scale: number (optional) - Scale factor - rotation: number (optional) - Rotation in degrees - opacity: number (optional) - Opacity (0-100) 2. create_text_layer: - text: string (required) - Text content - position: array (optional) - [x, y] position - fontSize: number (optional) - Font size - color: string (optional) - Color in hex format - fontFamily: string (optional) - Font family name 3. add_text_animation: - layerName: string (required) - Name of text layer - animationType: string (required) - Type of animation - duration: number (optional) - Animation duration - easing: string (optional) - Easing function 4. add_layer_animation: - layerName: string (required) - Name of layer - property: string (required) - Property to animate - keyframes: array (required) - Array of keyframe objects - easing: string (optional) - Easing function 5. add_color_animation: - layerName: string (required) - Name of layer - property: string (required) - Color property - keyframes: array (required) - Array of color keyframes - easing: string (optional) - Easing function 6. add_effect: - layerName: string (required) - Name of layer - effectName: string (required) - Name of effect - properties: object (optional) - Effect properties 7. create_composition: - name: string (required) - Composition name - width: number (optional) - Width in pixels - height: number (optional) - Height in pixels - duration: number (optional) - Duration in seconds - frameRate: number (optional) - Frame rate 8. create_solid_layer: - name: string (required) - Layer name - width: number (optional) - Width in pixels - height: number (optional) - Height in pixels - color: array (optional) - [r, g, b] color values - position: array (optional) - [x, y] position 9. create_shape_layer: - shapeType: string (required) - Type of shape - position: array (optional) - [x, y] position - properties: object (optional) - Shape properties For import commands, you should: 1. Extract the file path from the user's request 2. Determine the media type if specified 3. Extract any additional parameters (position, scale, etc.) 4. Return a properly formatted import_media command Example import commands: - "import image from C:/path/to/image.jpg" -> { "action": "import_media", "params": { "filePath": "C:/path/to/image.jpg", "type": "image" }, "confirmation_message": "Importing image from C:/path/to/image.jpg" } - "import video from C:/path/to/video.mp4 at position [100, 100]" -> { "action": "import_media", "params": { "filePath": "C:/path/to/video.mp4", "type": "video", "position": [100, 100] }, "confirmation_message": "Importing video from C:/path/to/video.mp4 at position [100, 100]" } Return ONLY the JSON object, no additional text or explanations.""" class SelfLearningSystem: def __init__(self): self.learning_data_path = "learning_data" self.command_history_path = os.path.join(self.learning_data_path, "command_history.json") self.improvement_log_path = os.path.join(self.learning_data_path, "improvements.json") self.performance_metrics_path = os.path.join(self.learning_data_path, "performance_metrics.json") self._initialize_learning_system() def _initialize_learning_system(self): os.makedirs(self.learning_data_path, exist_ok=True) for path in [self.command_history_path, self.improvement_log_path, self.performance_metrics_path]: if not os.path.exists(path): with open(path, 'w') as f: json.dump([], f) def log_command(self, prompt: str, command: dict, success: bool, error: Optional[str] = None): with open(self.command_history_path, 'r+') as f: history = json.load(f) history.append({ 'timestamp': datetime.now().isoformat(), 'prompt': prompt, 'command': command, 'success': success, 'error': error }) f.seek(0) json.dump(history, f, indent=2) def log_improvement(self, improvement_type: str, details: dict): with open(self.improvement_log_path, 'r+') as f: improvements = json.load(f) improvements.append({ 'timestamp': datetime.now().isoformat(), 'type': improvement_type, 'details': details }) f.seek(0) json.dump(improvements, f, indent=2) def update_metrics(self, metric_name: str, value: float): with open(self.performance_metrics_path, 'r+') as f: metrics = json.load(f) if metric_name not in metrics: metrics[metric_name] = [] metrics[metric_name].append({ 'timestamp': datetime.now().isoformat(), 'value': value }) f.seek(0) json.dump(metrics, f, indent=2) def analyze_performance(self) -> dict: with open(self.performance_metrics_path, 'r') as f: metrics = json.load(f) return { 'success_rate': self._calculate_success_rate(), 'average_response_time': self._calculate_avg_response_time(), 'command_distribution': self._analyze_command_distribution() } def _calculate_success_rate(self) -> float: with open(self.command_history_path, 'r') as f: history = json.load(f) if not history: return 0.0 successes = sum(1 for entry in history if entry['success']) return successes / len(history) def _calculate_avg_response_time(self) -> float: # Implementation for response time calculation return 0.0 # Placeholder def _analyze_command_distribution(self) -> dict: with open(self.command_history_path, 'r') as f: history = json.load(f) distribution = {} for entry in history: action = entry['command'].get('action', 'unknown') distribution[action] = distribution.get(action, 0) + 1 return distribution # Initialize the self-learning system learning_system = SelfLearningSystem() # List of supported actions SUPPORTED_ACTIONS = [ "create_composition", "create_text_layer", "apply_text_animation", "inject_script" ] # Animation types and their parameters ANIMATION_TYPES = { "wiggle": { "description": "Creates a random wiggle motion", "params": { "layerName": "Name of the layer to animate", "animationType": "wiggle", "duration": "Duration of the animation in seconds", "amplitude": "Amount of movement", "frequency": "Speed of the wiggle" } }, "bounce": { "description": "Creates a bouncing effect", "params": { "layerName": "Name of the layer to animate", "animationType": "bounce", "duration": "Duration of the animation in seconds", "amplitude": "Height of the bounce", "frequency": "Speed of the bounce" } } } # Parameter validation/correction helpers COLOR_NAMES = ["red", "green", "blue", "yellow", "purple", "orange", "black", "white", "gray", "grey"] def validate_color(color): if not color: return "white" color = color.lower() if color in COLOR_NAMES: return color if color.startswith("#") and len(color) == 7: return color return "white" def validate_position(pos): if isinstance(pos, list) and len(pos) == 2: return pos return [960, 540] def validate_params(action, params): if action == "create_text_layer": params["color"] = validate_color(params.get("color")) params["position"] = validate_position(params.get("position")) if action == "create_shape_layer": params["color"] = validate_color(params.get("color")) params["position"] = validate_position(params.get("position")) if action == "apply_effect": if "effectParams" in params and "color" in params["effectParams"]: params["effectParams"]["color"] = validate_color(params["effectParams"]["color"]) return params def ensure_action_field(command): if 'type' in command and 'action' not in command: command['action'] = command.pop('type') return command def strip_markdown_json(text): if text.startswith('```'): text = text.strip('`') text = text.replace('json', '', 1).strip() if text.startswith('{') and text.endswith('}'): # Already JSON return text start = text.find('{') end = text.rfind('}') + 1 if start != -1 and end != 0: return text[start:end] return text def strip_code_fences(code): code = re.sub(r'```[a-zA-Z]*\n', '', code) code = re.sub(r'```', '', code) return code.strip() async def run_gemini_code_with_retry(gemini, prompt, send_command_to_ae, max_retries=7): last_code = None last_error = None for attempt in range(max_retries): response = await gemini.generate_content_async(prompt) code = strip_code_fences(response.text) # Try to execute code in AE (send to run_custom_code) result = await send_command_to_ae({ 'action': 'run_custom_code', 'params': {'code': code}, 'confirmation_message': 'Running Gemini-generated code' }) if result.get('status') == 'success': return result else: last_error = result.get('message') prompt = f"Your last code failed with this error: {last_error}. Please regenerate the code. Only output raw ExtendScript code, no markdown, no summary, no comments, no explanations." return {'status': 'error', 'message': f'Failed after {max_retries} attempts. Last error: {last_error}'} async def process_command_with_gemini(command): """Process a natural language command with Gemini and convert it to a structured command.""" if not gemini_configured or not model: logging.warning("Gemini AI is not configured. Cannot process command.") return { "status": "error", "message": "Gemini AI is not configured. Please set a valid API key in the .env file." } try: # Create the prompt prompt = f"{SYSTEM_PROMPT}\n\nUser command: {command}" # Generate the response response = model.generate_content(prompt) # Extract the JSON from the response response_text = response.text # Try to parse the response as JSON try: # Strip markdown code blocks if present if "```json" in response_text: response_text = response_text.split("```json")[1].split("```")[0].strip() elif "```" in response_text: response_text = response_text.split("```")[1].split("```")[0].strip() command_json = json.loads(response_text) # Validate the command if "action" not in command_json: command_json["action"] = "unknown" if "params" not in command_json: command_json["params"] = {} if "confirmation_message" not in command_json: command_json["confirmation_message"] = f"Executing {command_json['action']}" return command_json except json.JSONDecodeError: logging.error(f"Failed to parse Gemini response as JSON: {response_text}") return { "status": "error", "message": "Failed to parse Gemini response as JSON" } except Exception as e: logging.error(f"Error processing command with Gemini: {str(e)}") return { "status": "error", "message": f"Error processing command: {str(e)}" } async def fix_command_with_gemini(error, context): """Use Gemini to fix a failed command.""" if not gemini_configured or not model: logging.warning("Gemini AI is not configured. Cannot fix command.") return { "status": "error", "message": "Gemini AI is not configured. Please set a valid API key in the .env file." } try: # Create a prompt that includes the error context and asks for a fix prompt = f""" The following command failed with an error. Please analyze the error and provide a fixed version of the command. Error: {error} Context: - Original Command: {json.dumps(context.get('original_command', {}))} - Retry Count: {context.get('retry_count', 0)} Please provide a fixed version of the command that addresses this error. Return ONLY the JSON object with the fixed command, no additional text. """ # Get response from Gemini response = await model.generate_content(prompt) # Parse the response try: # Try to parse the response as JSON fixed_command = json.loads(response.text) # Validate the fixed command if not isinstance(fixed_command, dict): raise ValueError("Response is not a valid JSON object") if "action" not in fixed_command: raise ValueError("Response missing 'action' field") if "params" not in fixed_command: raise ValueError("Response missing 'params' field") return { "status": "success", "data": fixed_command, "message": "Command fixed successfully" } except json.JSONDecodeError: # If the response isn't valid JSON, try to extract JSON from the text try: # Look for JSON-like structure in the text json_start = response.text.find('{') json_end = response.text.rfind('}') + 1 if json_start >= 0 and json_end > json_start: json_str = response.text[json_start:json_end] fixed_command = json.loads(json_str) # Validate the extracted command if not isinstance(fixed_command, dict): raise ValueError("Extracted response is not a valid JSON object") if "action" not in fixed_command: raise ValueError("Extracted response missing 'action' field") if "params" not in fixed_command: raise ValueError("Extracted response missing 'params' field") return { "status": "success", "data": fixed_command, "message": "Command fixed successfully" } except: pass return { "status": "error", "message": "Could not parse Gemini's response as a valid command" } except Exception as e: logging.error(f"Error fixing command with Gemini: {str(e)}") return { "status": "error", "message": f"Error fixing command: {str(e)}" } class GeminiProcessor: def __init__(self): self.command_history = [] self.error_history = [] self.successful_patterns = [] self.learning_context = { "successful_commands": [], "error_patterns": {}, "improvements": [], "user_preferences": {} } async def process_command_with_gemini(self, command: str) -> Dict[str, Any]: """Process a command with Gemini, including learning from results.""" if not gemini_configured or not model: logging.warning("Gemini AI is not configured. Cannot process command.") return { "status": "error", "message": "Gemini AI is not configured. Please set a valid API key in the .env file." } try: # Add command to history self.command_history.append(command) # Create an enhanced prompt that includes learning context prompt = self._create_enhanced_prompt(command) # Get response from Gemini response = model.generate_content(prompt) # Parse and validate the response result = self._parse_response(response.text) # Learn from the result self._learn_from_result(command, result) return result except Exception as e: error_msg = f"Error processing command: {str(e)}" self.error_history.append({ "command": command, "error": error_msg, "context": self.learning_context }) return { "status": "error", "message": error_msg } def _create_enhanced_prompt(self, command: str) -> str: """Create an enhanced prompt that includes learning context.""" return f""" You are an autonomous AI assistant for Adobe After Effects automation. Your goal is to continuously learn and improve from interactions. Current Learning Context: - Successful Commands: {json.dumps(self.learning_context['successful_commands'][-5:])} - Recent Error Patterns: {json.dumps(self.learning_context['error_patterns'])} - Recent Improvements: {json.dumps(self.learning_context['improvements'][-3:])} Command History (Last 3): {json.dumps(self.command_history[-3:])} Process the following command, taking into account the learning context: {command} Return a JSON object with: 1. action: The specific action to take 2. params: Parameters for the action 3. learning_insights: Any new patterns or improvements you've identified 4. suggested_improvements: Optional improvements to the system Return ONLY the JSON object, no additional text. """ def _parse_response(self, response_text: str) -> Dict[str, Any]: """Parse and validate the Gemini response.""" try: # Try to parse the response as JSON result = json.loads(response_text) # Validate required fields if not isinstance(result, dict): raise ValueError("Response is not a valid JSON object") if "action" not in result: raise ValueError("Response missing 'action' field") if "params" not in result: raise ValueError("Response missing 'params' field") # Extract learning insights if present if "learning_insights" in result: self._process_learning_insights(result["learning_insights"]) # Extract suggested improvements if present if "suggested_improvements" in result: self._process_suggested_improvements(result["suggested_improvements"]) return { "status": "success", "data": { "action": result["action"], "params": result["params"] }, "message": "Command processed successfully" } except json.JSONDecodeError: # Try to extract JSON from the text try: json_start = response_text.find('{') json_end = response_text.rfind('}') + 1 if json_start >= 0 and json_end > json_start: json_str = response_text[json_start:json_end] return self._parse_response(json_str) except: pass return { "status": "error", "message": "Could not parse Gemini's response as a valid command" } def _learn_from_result(self, command: str, result: Dict[str, Any]): """Learn from the command result and update learning context.""" if result["status"] == "success": # Add to successful commands self.learning_context["successful_commands"].append({ "command": command, "result": result }) # Keep only last 100 successful commands if len(self.learning_context["successful_commands"]) > 100: self.learning_context["successful_commands"] = self.learning_context["successful_commands"][-100:] else: # Analyze error pattern error_pattern = { "command_type": command.split()[0] if command else "unknown", "error": result["message"], "context": self.learning_context } # Update error patterns if error_pattern["command_type"] not in self.learning_context["error_patterns"]: self.learning_context["error_patterns"][error_pattern["command_type"]] = [] self.learning_context["error_patterns"][error_pattern["command_type"]].append(error_pattern) def _process_learning_insights(self, insights: Dict[str, Any]): """Process and incorporate learning insights from Gemini.""" if isinstance(insights, dict): # Update learning context with new insights for key, value in insights.items(): if key in self.learning_context: if isinstance(self.learning_context[key], list): self.learning_context[key].extend(value) else: self.learning_context[key].update(value) def _process_suggested_improvements(self, improvements: list): """Process and incorporate suggested improvements from Gemini.""" if isinstance(improvements, list): self.learning_context["improvements"].extend(improvements) # Keep only last 50 improvements if len(self.learning_context["improvements"]) > 50: self.learning_context["improvements"] = self.learning_context["improvements"][-50:] async def fix_command_with_gemini(self, error: str, context: Dict[str, Any]) -> Dict[str, Any]: """Use Gemini to fix a failed command with enhanced learning.""" if not gemini_configured or not model: logging.warning("Gemini AI is not configured. Cannot fix command.") return { "status": "error", "message": "Gemini AI is not configured. Please set a valid API key in the .env file." } try: # Add error to history self.error_history.append({ "error": error, "context": context }) # Create an enhanced prompt for fixing prompt = f""" You are an autonomous AI assistant for Adobe After Effects automation. Your goal is to fix failed commands and learn from the process. Error Context: - Error: {error} - Original Command: {json.dumps(context.get('original_command', {}))} - Retry Count: {context.get('retry_count', 0)} Learning Context: - Recent Successful Commands: {json.dumps(self.learning_context['successful_commands'][-3:])} - Error Patterns: {json.dumps(self.learning_context['error_patterns'])} - Recent Improvements: {json.dumps(self.learning_context['improvements'][-3:])} Please: 1. Analyze the error and provide a fixed version of the command 2. Explain what went wrong and how the fix addresses it 3. Suggest any system improvements to prevent similar errors Return a JSON object with: 1. fixed_command: The corrected command 2. error_analysis: Your analysis of what went wrong 3. suggested_improvements: Any system improvements Return ONLY the JSON object, no additional text. """ # Get response from Gemini response = model.generate_content(prompt) # Parse the response try: result = json.loads(response.text) # Validate the fixed command if "fixed_command" not in result: raise ValueError("Response missing 'fixed_command' field") # Process error analysis and improvements if "error_analysis" in result: self._process_learning_insights({"error_analysis": result["error_analysis"]}) if "suggested_improvements" in result: self._process_suggested_improvements(result["suggested_improvements"]) return { "status": "success", "data": result["fixed_command"], "message": "Command fixed successfully" } except json.JSONDecodeError: return { "status": "error", "message": "Could not parse Gemini's response as a valid command" } except Exception as e: logging.error(f"Error fixing command with Gemini: {str(e)}") return { "status": "error", "message": f"Error fixing command: {str(e)}" } # Create a singleton instance gemini_processor = GeminiProcessor() # Export the instance __all__ = ['gemini_processor']