Update rhythma.py

rhythma.py

@@ -407,5 +407,260 @@ class RhythmaSymphAICore:
              print("ℹ️ SentenceTransformer not installed. Using basic text matching.")
 
 
+    # Line 343:
     def detect_emotion_with_groq(self, input_text):
-        
+        """Use Groq LLM to detect emotion/intention in text"""
+        if not self.use_groq or not self.groq_client:
+            print("ℹ️ Groq not available or not initialized for emotion detection.")
+            return None # Indicate Groq wasn't used
+
+        # Refined prompt for better classification into our categories
+        available_states = ", ".join(self.emotional_states)
+        prompt = f"""Analyze the user's feeling described below.
+Identify the single MOST prominent emotional state or intention from the following list:
+{available_states}
+
+Focus on the core feeling expressed. Respond with ONLY the chosen state/intention from the list.
+
+User's feeling: "{input_text}"
+State/Intention:"""
+
+        try:
+            print(f"ℹ️ Querying Groq for emotion analysis...")
+            chat_completion = self.groq_client.chat.completions.create(
+                messages=[{"role": "user", "content": prompt}],
+                model="llama3-70b-8192", # Specify a capable model available on Groq
+                max_tokens=15, # Allow slightly more tokens for flexibility
+                temperature=0.2, # Lower temperature for more deterministic classification
+                stop=["\n"] # Stop generation after the first line
+            )
+
+            detected_emotion = chat_completion.choices[0].message.content.strip().lower()
+            print(f"✅ Groq detected: {detected_emotion}")
+
+            # Validate the detected emotion against our list
+            if detected_emotion in self.emotional_states:
+                return detected_emotion
+            else:
+                print(f"⚠️ Groq returned '{detected_emotion}', not in known states. Attempting fallback match.")
+                # Fallback: If LLM returns something unexpected, find the closest match in our list
+                return self.get_closest_emotional_state(detected_emotion) # Use fallback on unexpected LLM output
+
+        except Exception as e:
+            print(f"❌ Error using Groq for emotion detection: {str(e)}")
+            traceback.print_exc()
+            return None # Indicate error or inability to use Groq
+
+
+    def get_closest_emotional_state(self, input_text):
+        """Map input text to the closest emotional state using available methods."""
+        if not input_text:
+            return "neutral" # Default if no text
+
+        input_text_lower = input_text.lower()
+
+        # 1. Try simple keyword matching first (fastest)
+        for state in self.emotional_states:
+            if state in input_text_lower.split(): # Match whole words if possible
+                print(f"ℹ️ Matched keyword: {state}")
+                return state
+            # Simple substring check as backup
+            if state in input_text_lower:
+                 print(f"ℹ️ Matched substring: {state}")
+                 return state
+
+
+        # 2. If Sentence Transformer is available, use semantic similarity
+        if self.embedding_model and self.emotional_embeddings:
+            try:
+                print("ℹ️ Using Sentence Transformer for semantic emotion match.")
+                input_embedding = self.embedding_model.encode([input_text])[0] # Get 1D array
+                # Calculate cosine similarities
+                similarities = {
+                    state: cosine_similarity(input_embedding.reshape(1, -1), embedding.reshape(1, -1))[0][0]
+                    for state, embedding in self.emotional_embeddings.items()
+                }
+                # Find the state with the highest similarity
+                best_match = max(similarities, key=similarities.get)
+                print(f"✅ Semantic match: {best_match} (Similarity: {similarities[best_match]:.2f})")
+                return best_match
+            except Exception as e:
+                print(f"⚠️ Error during semantic matching: {e}. Falling back.")
+                traceback.print_exc()
+
+        # 3. Default fallback if no match found
+        print("ℹ️ No clear emotion match found, defaulting to 'neutral'.")
+        return "neutral"
+
+
+    def get_closest_rhythm_pattern(self, input_text=None, emotional_state=None):
+        """Map input text or emotional state to the closest rhythm pattern."""
+
+        # 1. Direct mapping from emotional state (prioritized if state is known)
+        if emotional_state:
+            # Refined mapping based on typical energy levels/needs
+            mapping = {
+                "anxious": "calm",    # Needs calming
+                "stressed": "relaxed",  # Needs relaxation
+                "calm": "calm",
+                "sad": "relaxed",     # Gentle support
+                "angry": "active",    # Needs release/energy shift
+                "fearful": "calm",    # Needs safety/grounding
+                "confused": "focused", # Needs clarity
+                "happy": "active",    # Can match higher energy
+                "neutral": "calm",
+                "focused": "focused", # Align with intention
+                "relaxed": "relaxed", # Align with intention
+                "active": "active",   # Align with intention
+            }
+            pattern = mapping.get(emotional_state, "calm") # Default to calm if state unknown
+            print(f"ℹ️ Rhythm pattern from state '{emotional_state}': {pattern}")
+            return pattern
+
+        # 2. If no emotional state, try matching input text semantically (if available)
+        if input_text and self.embedding_model and self.rhythm_embeddings:
+            try:
+                print("ℹ️ Using Sentence Transformer for semantic rhythm match.")
+                input_embedding = self.embedding_model.encode([input_text])[0]
+                similarities = {
+                    pattern: cosine_similarity(input_embedding.reshape(1, -1), embedding.reshape(1, -1))[0][0]
+                    for pattern, embedding in self.rhythm_embeddings.items()
+                }
+                best_match = max(similarities, key=similarities.get)
+                print(f"✅ Semantic rhythm match: {best_match} (Similarity: {similarities[best_match]:.2f})")
+                return best_match
+            except Exception as e:
+                print(f"⚠️ Error during semantic rhythm matching: {e}. Falling back.")
+                traceback.print_exc()
+
+        # 3. Default fallback
+        print("ℹ️ Defaulting rhythm pattern to 'calm'.")
+        return "calm"
+
+
+    def transcribe_audio(self, audio_path):
+        """Transcribe audio using Groq Whisper if available"""
+        if not self.use_groq or not self.groq_client:
+            print("ℹ️ Groq not available for transcription.")
+            return None, "Transcription disabled: Groq client not available or API key missing."
+
+        if not audio_path or not os.path.exists(audio_path):
+             return None, "Transcription failed: Audio file path is invalid or missing."
+
+        try:
+            print(f"ℹ️ Transcribing audio file: {audio_path}")
+            with open(audio_path, "rb") as audio_file:
+                # Use whisper-large-v3 for potentially better accuracy
+                transcription_response = self.groq_client.audio.transcriptions.create(
+                    file=(os.path.basename(audio_path), audio_file.read()),
+                    model="whisper-large-v3", # Using v3
+                    # response_format="verbose_json", # Get more details if needed
+                    response_format="json", # Simpler format
+                )
+
+            transcribed_text = transcription_response.text
+            print(f"✅ Groq transcription successful: '{transcribed_text}'")
+            return transcribed_text, None # Return text and no error
+
+        except Exception as e:
+            error_message = f"Error during Groq transcription: {str(e)}"
+            print(f"❌ {error_message}")
+            traceback.print_exc()
+            return None, error_message # Return None and the error message
+
+    # --- THIS IS THE FUNCTION STARTING AT LINE 410 ---
+    def analyze_input(self, input_text=None, audio_path=None):
+        """
+        Analyze input text and/or audio path to determine emotional state and rhythm pattern.
+        **Ensures a dictionary is always returned.**
+        """
+        # ---> Line 411: Ensure this block is indented <---
+        analysis_result = {
+            "emotional_state": "neutral", # Default values
+            "rhythm_pattern": "calm",    # Default values
+            "transcription": "",
+            "error": None
+        }
+        text_to_analyze = None
+        transcription_error = None
+
+        # ---> All lines below here inside the function must also be indented <---
+        print("-" * 20) # Separator for logs
+        print(f"ℹ️ SymphAI Core analyzing input: Text='{input_text}', Audio='{audio_path}'")
+
+        try:
+            # --- Step 1: Handle Audio Input (if provided and Groq available) ---
+            if audio_path and self.use_groq:
+                transcribed_text, transcription_error = self.transcribe_audio(audio_path)
+                if transcription_error:
+                    print(f"⚠️ Transcription failed: {transcription_error}")
+                    # Store error but potentially continue with text input if available
+                    analysis_result["error"] = transcription_error
+                    analysis_result["transcription"] = f"[Transcription Error: {transcription_error}]"
+                elif transcribed_text:
+                    analysis_result["transcription"] = transcribed_text
+                    text_to_analyze = transcribed_text # Prioritize transcribed text
+                    print(f"ℹ️ Using transcribed text for analysis: '{text_to_analyze}'")
+
+            # --- Step 2: Determine Text for Analysis ---
+            if not text_to_analyze and input_text:
+                text_to_analyze = input_text # Use input_text if no successful transcription
+                print(f"ℹ️ Using provided text for analysis: '{text_to_analyze}'")
+            elif not text_to_analyze:
+                 print("ℹ️ No text input or successful transcription available for analysis.")
+                 # Keep default neutral/calm state
+
+            # --- Step 3: Detect Emotional State (if text available) ---
+            detected_emotion = None
+            if text_to_analyze:
+                if self.use_groq:
+                    detected_emotion = self.detect_emotion_with_groq(text_to_analyze)
+                    if detected_emotion:
+                         analysis_result["emotional_state"] = detected_emotion
+                    else:
+                         # Groq failed or didn't run, try fallback
+                         print("ℹ️ Groq emotion detection failed or skipped, trying fallback.")
+                         analysis_result["emotional_state"] = self.get_closest_emotional_state(text_to_analyze)
+                else:
+                     # Groq not used, directly use fallback
+                     analysis_result["emotional_state"] = self.get_closest_emotional_state(text_to_analyze)
+            else:
+                 # No text to analyze, stick with default "neutral"
+                 analysis_result["emotional_state"] = "neutral"
+
+
+            # --- Step 4: Determine Rhythm Pattern ---
+            # Use the determined emotional state primarily, fallback to text if needed
+            current_emotion = analysis_result["emotional_state"]
+            analysis_result["rhythm_pattern"] = self.get_closest_rhythm_pattern(
+                input_text=text_to_analyze, # Pass text for potential semantic match if emotion is neutral/unclear
+                emotional_state=current_emotion
+            )
+
+            # Clean up error field if no actual error occurred during main analysis
+            if analysis_result["error"] is None and transcription_error:
+                 # If transcription failed but text analysis succeeded, maybe clear the error?
+                 # Decide if transcription error should persist if text analysis works.
+                 # Let's keep it for now to inform the user.
+                 pass
+            elif analysis_result["error"] is None:
+                 # analysis_result.pop("error", None) # Alternative way to remove if None
+                 del analysis_result["error"] # Remove error key if None
+
+
+        except Exception as e:
+            # --- Catch-all for unexpected errors during analysis ---
+            error_msg = f"Unexpected error during input analysis: {str(e)}"
+            print(f"❌ {error_msg}")
+            traceback.print_exc()
+            analysis_result = {
+                "emotional_state": "neutral", # Reset to defaults on error
+                "rhythm_pattern": "calm",
+                "transcription": analysis_result.get("transcription", ""), # Keep transcription if available
+                "error": error_msg
+            }
+
+        # ---> Ensure these lines are indented correctly at the function level <---
+        print(f"✅ SymphAI Core analysis complete. Result: {analysis_result}")
+        print("-" * 20) # Separator for logs
+        return analysis_result # GUARANTEED TO BE A DICTIONARY