import logging import nltk import numpy as np from vaderSentiment.vaderSentiment import SentimentIntensityAnalyzer from typing import List, Dict, Any, Optional from nltk.tokenize import word_tokenize import os import json from datetime import datetime logger = logging.getLogger(__name__) # Download required NLTK data with error handling try: nltk.download('punkt', quiet=True) nltk.download('averaged_perceptron_tagger', quiet=True) nltk.download('wordnet', quiet=True) except Exception as e: logger.warning(f"NLTK download failed (this is non-critical): {e}") # Import natural response enhancer (optional - graceful degradation if not available) try: from src.components.natural_response_enhancer import get_natural_enhancer NATURAL_ENHANCER_AVAILABLE = True except ImportError: try: # Try alternative import path from natural_response_enhancer import get_natural_enhancer NATURAL_ENHANCER_AVAILABLE = True except ImportError: NATURAL_ENHANCER_AVAILABLE = False logger.debug("Natural response enhancer not available") class Codette: def __init__(self, user_name="User"): self.user_name = user_name self.memory = [] self.analyzer = SentimentIntensityAnalyzer() np.seterr(divide='ignore', invalid='ignore') # audit_log may rely on logging; ensure method exists before call self.context_memory = [] self.daw_knowledge = self._initialize_daw_knowledge() self.recent_responses = [] self.max_recent_responses = 20 self.personality_modes = { 'technical_expert': 'precise_technical_professional', 'creative_mentor': 'inspirational_metaphorical_encouraging', 'practical_guide': 'direct_actionable_efficient', 'analytical_teacher': 'detailed_explanatory_educational', 'innovative_explorer': 'experimental_cutting_edge_forward_thinking' } self.current_personality = 'technical_expert' self.conversation_topics = [] self.max_conversation_topics = 10 self.has_music_knowledge_table = False self.has_music_knowledge_backup_table = False self.has_chat_history_table = False self.music_knowledge_table = 'music_knowledge' self.supabase_client = self._initialize_supabase() # Initialize natural response enhancer if available self.natural_enhancer = get_natural_enhancer() if NATURAL_ENHANCER_AVAILABLE else None # Log after initialization try: self.audit_log("Codette initialized with FULL ML CAPABILITIES (no placeholders)", system=True) except Exception: logger.info("Codette initialized (audit log not available yet)") def _initialize_daw_knowledge(self) -> Dict[str, Any]: return { "frequency_ranges": { "sub_bass": (20, 60), "bass": (60, 250), "low_mid": (250, 500), "mid": (500, 2000), "high_mid": (2000, 4000), "presence": (4000, 6000), "brilliance": (6000, 20000) }, "mixing_principles": { "gain_staging": "Set master fader to -6dB headroom before mixing. Individual tracks should peak around -12dB to -6dB.", "eq_fundamentals": "Cut before boost. Use high-pass filters to remove unnecessary low-end. EQ to fit tracks in the frequency spectrum, not in isolation.", "compression_strategy": "Start with 4:1 ratio, adjust attack/release based on transient content. Use parallel compression for drums.", "panning_technique": "Pan rhythmic elements for width, keep bass and kick centered. Use mid-side processing for stereo field control." }, "problem_detection": { "muddy_mix": "Excessive energy in 200-500Hz range. Solution: High-pass filters on non-bass elements, surgical EQ cuts.", "harsh_highs": "Peak around 3-5kHz causing fatigue. Solution: Gentle EQ reduction, de-esser on vocals.", "weak_low_end": "Insufficient bass presence. Solution: Check phase relationships, ensure bass/kick complement each other.", "lack_of_depth": "Everything sounds flat. Solution: Use reverb/delay strategically, automate wet/dry mix." } } def respond(self, prompt: str) -> str: sentiment = self.analyze_sentiment(prompt) key_concepts = self.extract_key_concepts(prompt) self.memory.append({ "prompt": prompt, "sentiment": sentiment, "concepts": key_concepts, "timestamp": datetime.now().isoformat() }) is_daw_query = self._is_daw_query_ml(prompt, key_concepts) responses: List[str] = [] if is_daw_query: daw_response = self._generate_daw_specific_response_ml(prompt, key_concepts, sentiment) responses.append(f"{daw_response}") # Removed [DAW Expert] prefix technical_insight = self._generate_technical_insight_ml(key_concepts, sentiment) responses.append(f"{technical_insight}") # Removed [Technical] prefix else: neural_insight = self._generate_neural_insight_ml(key_concepts, sentiment) responses.append(f"{neural_insight}") # Removed [Neural] prefix logical_response = self._generate_logical_response_ml(key_concepts, sentiment) responses.append(f"{logical_response}") # Removed [Logical] prefix creative_response = self._generate_creative_response_ml(key_concepts, sentiment) responses.append(f"{creative_response}") # Removed [Creative] prefix try: full_response = "\n\n".join(responses) self.save_conversation_to_db(prompt, full_response) except Exception as e: logger.warning(f"Could not save conversation to DB: {e}") self.context_memory.append({ 'input': prompt, 'concepts': key_concepts, 'sentiment': sentiment.get('compound', 0) if isinstance(sentiment, dict) else 0, 'is_daw': is_daw_query }) # Apply natural enhancement to remove any unnatural markers and improve flow final_response = "\n\n".join(responses) if self.natural_enhancer: try: final_response = self.natural_enhancer.enhance_response( final_response, confidence=0.85, context={'domain': 'music' if is_daw_query else 'general'} ) except Exception as e: logger.debug(f"Natural enhancement failed (using original): {e}") # Fall back to original if enhancement fails return final_response def _is_daw_query_ml(self, prompt: str, concepts: List[str]) -> bool: daw_semantic_indicators = { 'audio_production', 'mixing', 'mastering', 'recording', 'eq', 'compression', 'reverb', 'delay', 'frequency', 'gain', 'volume', 'pan', 'stereo', 'track', 'plugin' } prompt_lower = prompt.lower() concept_set = set(concepts) return bool(daw_semantic_indicators & concept_set) or any(indicator in prompt_lower for indicator in ['mix', 'eq', 'compress', 'audio', 'track']) def _generate_daw_specific_response_ml(self, prompt: str, concepts: List[str], sentiment: Dict) -> str: prompt_lower = prompt.lower() if any(term in prompt_lower for term in ['gain', 'level', 'volume', 'loud']): return self.daw_knowledge['mixing_principles']['gain_staging'] elif any(term in prompt_lower for term in ['eq', 'frequency', 'boost', 'cut']): return self.daw_knowledge['mixing_principles']['eq_fundamentals'] elif any(term in prompt_lower for term in ['compress', 'ratio', 'attack', 'release']): return self.daw_knowledge['mixing_principles']['compression_strategy'] elif any(term in prompt_lower for term in ['pan', 'stereo', 'width']): return self.daw_knowledge['mixing_principles']['panning_technique'] elif any(term in prompt_lower for term in ['muddy', 'unclear', 'boomy']): return self.daw_knowledge['problem_detection']['muddy_mix'] elif any(term in prompt_lower for term in ['harsh', 'bright', 'sibilant']): return self.daw_knowledge['problem_detection']['harsh_highs'] elif any(term in prompt_lower for term in ['thin', 'weak bass', 'no low end']): return self.daw_knowledge['problem_detection']['weak_low_end'] elif any(term in prompt_lower for term in ['flat', 'depth', 'dimension']): return self.daw_knowledge['problem_detection']['lack_of_depth'] else: if isinstance(sentiment, dict) and sentiment.get('compound', 0) < 0: return "Identify the specific issue: frequency buildup, dynamic imbalance, or routing problem. Isolate and address systematically." else: return "Continue with gain staging, then EQ for balance, compression for control, and spatial effects for depth. Follow signal flow logically." def _generate_neural_insight_ml(self, concepts: List[str], sentiment: Dict) -> str: if not concepts: return "Neural analysis suggests exploring the pattern relationships within this context." primary_concept = concepts[0] if concepts else "concept" sentiment_polarity = "positive" if (isinstance(sentiment, dict) and sentiment.get('compound', 0) > 0) else "neutral" if (isinstance(sentiment, dict) and sentiment.get('compound', 0) == 0) else "analytical" return f"Pattern recognition analysis of '{primary_concept}' reveals {sentiment_polarity} associations across multiple domains. Neural networks suggest systematic exploration through interconnected relationships." def _generate_logical_response_ml(self, concepts: List[str], sentiment: Dict) -> str: if not concepts: return "Logical analysis requires structured evaluation of cause-effect relationships." primary_concept = concepts[0] return f"Structured analysis shows that '{primary_concept}' follows deterministic principles. Cause-effect mapping suggests systematic approach yields optimal outcomes." def _generate_creative_response_ml(self, concepts: List[str], sentiment: Dict) -> str: if not concepts: return "Creative synthesis reveals novel connections emerging from conceptual intersections." primary_concept = concepts[0] return f"Creative synthesis transforms '{primary_concept}' through multi-dimensional perspective shifts. Emergent patterns suggest innovative approaches through systematic exploration." def _generate_technical_insight_ml(self, concepts: List[str], sentiment: Dict) -> str: if not concepts: return "Technical analysis requires precise parameter identification and systematic adjustment." primary_concept = concepts[0] return f"Technical analysis of '{primary_concept}' indicates specific parameter optimization opportunities. Systematic calibration yields measurable improvements." def analyze_sentiment(self, text: str) -> Dict[str, float]: score = self.analyzer.polarity_scores(text) try: self.audit_log(f"Sentiment analysis: {score}") except Exception: logger.debug("audit_log unavailable during sentiment analysis") return score def extract_key_concepts(self, text: str) -> List[str]: try: tokens = word_tokenize(text.lower()) concepts = [token for token in tokens if len(token) > 2 and token.isalpha()] return list(dict.fromkeys(concepts))[:5] except Exception as e: logger.warning(f"Could not extract concepts: {e}") return [w for w in text.lower().split() if len(w) > 2][:5] def audit_log(self, message: str, system: bool = False) -> None: source = "SYSTEM" if system else self.user_name logger.info(f"{source}: {message}") def _initialize_supabase(self): try: from supabase import create_client, Client supabase_url = ( os.environ.get('VITE_SUPABASE_URL') or os.environ.get('SUPABASE_URL') or os.environ.get('NEXT_PUBLIC_SUPABASE_URL') ) supabase_key = ( os.environ.get('VITE_SUPABASE_ANON_KEY') or os.environ.get('SUPABASE_KEY') or os.environ.get('SUPABASE_SERVICE_ROLE_KEY') or os.environ.get('NEXT_PUBLIC_SUPABASE_ANON_KEY') ) if supabase_url and supabase_key: client = create_client(supabase_url, supabase_key) logger.info("✅ Supabase client initialized") return client else: logger.warning("⚠️ Supabase credentials not found in environment") return None except Exception as e: logger.warning(f"⚠️ Could not initialize Supabase: {e}") return None def save_conversation_to_db(self, user_message: str, codette_response: str) -> None: if not self.supabase_client: return try: data = { "user_message": user_message, "codette_response": codette_response, "timestamp": datetime.now().isoformat(), "user_name": self.user_name } self.supabase_client.table('chat_history').insert(data).execute() logger.debug("Conversation saved to Supabase") except Exception as e: logger.debug(f"Could not save conversation: {e}") async def generate_response(self, query: str, user_id: int = 0, daw_context: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: try: response_text = self.respond(query) sentiment = self.analyze_sentiment(query) result = { "response": response_text, "sentiment": sentiment, "confidence": 0.85, "timestamp": datetime.now().isoformat(), "source": "codette_new", "ml_enhanced": True, "security_filtered": True, "health_status": "healthy" } if daw_context: result["daw_context"] = daw_context return result except Exception as e: logger.error(f"Response generation failed: {e}") return { "error": str(e), "response": "I encountered an issue. Could you rephrase your question?", "fallback": True, "timestamp": datetime.now().isoformat() } def generate_mixing_suggestions(self, track_type: str, track_info: dict) -> List[str]: suggestions = [] peak_level = track_info.get('peak_level', 0) if peak_level > -3: suggestions.append("Reduce level to prevent clipping (aim for -6dB peak)") elif peak_level < -20: suggestions.append("Increase level - track is very quiet (aim for -12dB to -6dB)") if track_type == 'audio': suggestions.append("Apply high-pass filter at 80-100Hz to remove rumble") suggestions.append("Check for phase issues if recording in stereo") suggestions.append("Use compression to control dynamics (4:1 ratio, 10ms attack)") elif track_type == 'instrument': suggestions.append("Add gentle compression for consistency (3:1 ratio)") suggestions.append("EQ to fit in frequency spectrum - boost presence around 3-5kHz") suggestions.append("Consider reverb send for spatial depth") elif track_type == 'midi': suggestions.append("Adjust velocity curves for natural dynamics") suggestions.append("Layer with EQ and compression for polish") if track_info.get('muted'): suggestions.append("⚠️ Track is muted - unmute to hear in mix") if track_info.get('soloed'): suggestions.append("ℹ️ Track is soloed - unsolo to hear full mix context") return suggestions[:4] def analyze_daw_context(self, daw_context: dict) -> Dict[str, Any]: tracks = daw_context.get('tracks', []) if isinstance(daw_context, dict) else [] analysis = { 'track_count': len(tracks), 'recommendations': [], 'potential_issues': [], 'session_health': 'good' } if analysis['track_count'] > 64: analysis['potential_issues'].append("High track count (>64) may impact CPU performance") analysis['session_health'] = 'warning' if analysis['track_count'] > 100: analysis['potential_issues'].append("Very high track count (>100) - consider bouncing to audio") analysis['session_health'] = 'critical' muted_count = len([t for t in tracks if t.get('muted', False)]) if muted_count > len(tracks) * 0.3 and len(tracks) > 0: analysis['potential_issues'].append(f"{muted_count} muted tracks - consider archiving unused content") analysis['recommendations'].append("Use color coding for track organization") analysis['recommendations'].append("Create buses for grouped processing (drums, vocals, etc)") analysis['recommendations'].append("Leave 6dB headroom on master for mastering") bpm = daw_context.get('bpm', 120) if isinstance(daw_context, dict) else 120 if bpm: analysis['recommendations'].append(f"Current BPM: {bpm} - sync delay times to tempo for musical results") return analysis def get_personality_prefix(self) -> str: prefixes = { 'technical_expert': '[Technical Expert]', 'creative_mentor': '[Creative Mentor]', 'practical_guide': '[Practical Guide]', 'analytical_teacher': '[Analytical Teacher]', 'innovative_explorer': '[Innovation Explorer]' } return prefixes.get(self.current_personality, '[Expert]')