Update rhythma.py

rhythma.py

@@ -6,6 +6,22 @@ from PIL import Image
 import io
 from sklearn.metrics.pairwise import cosine_similarity
 import soundfile as sf
+import os
+
+# Try to import optional dependencies
+try:
+    from groq import Groq
+    GROQ_AVAILABLE = True
+except ImportError:
+    GROQ_AVAILABLE = False
+    print("Groq package not installed. Falling back to local analysis.")
+
+try:
+    from sentence_transformers import SentenceTransformer
+    SENTENCE_TRANSFORMER_AVAILABLE = True
+except ImportError:
+    SENTENCE_TRANSFORMER_AVAILABLE = False
+    print("SentenceTransformer not installed. Simple text matching will be used.")
 
 class RhythmaModulationEngine:
     """
@@ -334,7 +350,7 @@ class RhythmaSymphAICore:
     SymphAI Core - The intelligent symbolic engine that interprets rhythm and state
     """
     
-    def __init__(self, rhythm_analyzer=None, pattern_matcher=None):
+    def __init__(self, use_groq=True):
         """Initialize the SymphAI Core"""
         # Default emotional states that can be detected
         self.emotional_states = [
@@ -347,46 +363,101 @@ class RhythmaSymphAICore:
             "calm", "active", "focused", "relaxed"
         ]
         
-        # Create embeddings for rhythm states if using semantic matching
-        try:
-            from sentence_transformers import SentenceTransformer
-            self.embedding_model = SentenceTransformer('all-MiniLM-L6-v2')
-            self.emotional_embeddings = {
-                emotion: self.embedding_model.encode([emotion]) 
-                for emotion in self.emotional_states
-            }
-            self.rhythm_embeddings = {
-                pattern: self.embedding_model.encode([pattern]) 
-                for pattern in self.rhythm_patterns
-            }
-        except ImportError:
+        # Initialize Groq client if available and requested
+        self.use_groq = use_groq and GROQ_AVAILABLE
+        if self.use_groq:
+            try:
+                self.groq_client = Groq(
+                    api_key=os.environ.get("GROQ_API_KEY"),
+                )
+                print("Groq client initialized successfully.")
+            except Exception as e:
+                print(f"Failed to initialize Groq client: {str(e)}")
+                self.use_groq = False
+        
+        # Initialize sentence transformer for semantic matching if available
+        if SENTENCE_TRANSFORMER_AVAILABLE:
+            try:
+                self.embedding_model = SentenceTransformer('all-MiniLM-L6-v2')
+                self.emotional_embeddings = {
+                    emotion: self.embedding_model.encode([emotion]) 
+                    for emotion in self.emotional_states
+                }
+                self.rhythm_embeddings = {
+                    pattern: self.embedding_model.encode([pattern]) 
+                    for pattern in self.rhythm_patterns
+                }
+                print("SentenceTransformer initialized successfully.")
+            except Exception as e:
+                print(f"Failed to initialize SentenceTransformer: {str(e)}")
+                self.embedding_model = None
+                self.emotional_embeddings = {}
+                self.rhythm_embeddings = {}
+        else:
             self.embedding_model = None
             self.emotional_embeddings = {}
             self.rhythm_embeddings = {}
-            print("SentenceTransformer not installed. Semantic matching disabled.")
+
+    def detect_emotion_with_groq(self, input_text):
+        """Use Groq LLM to detect emotion in text"""
+        if not self.use_groq:
+            return None
+            
+        prompt = f"""Analyze the following text and determine the primary emotion expressed. 
+Choose the most appropriate emotion from this list: anxious, stressed, calm, sad, angry, fearful, confused, happy.
+Provide your answer as a single word that best describes the emotional state.
+
+Text: {input_text}
+Emotion:"""
+
+        try:
+            chat_completion = self.groq_client.chat.completions.create(
+                messages=[
+                    {
+                        "role": "user",
+                        "content": prompt,
+                    }
+                ],
+                model="llama-3.3-70b-versatile",
+                max_tokens=10,
+            )
+            
+            detected_emotion = chat_completion.choices[0].message.content.strip().lower()
+            
+            # Validate the detected emotion
+            if detected_emotion in self.emotional_states:
+                return detected_emotion
+            else:
+                # If LLM returns something not in our list, find closest match
+                return self.get_closest_emotional_state(detected_emotion)
+                
+        except Exception as e:
+            print(f"Error using Groq for emotion detection: {str(e)}")
+            return None
     
     def get_closest_emotional_state(self, input_text):
         """Map input text to the closest emotional state"""
-        if not self.embedding_model:
-            # Fallback to simple word matching if no embedding model
-            for emotion in self.emotional_states:
-                if emotion in input_text.lower():
-                    return emotion
-            return "calm"  # Default
+        # First try simple word matching
+        for emotion in self.emotional_states:
+            if emotion in input_text.lower():
+                return emotion
+                
+        # If sentence transformer is available, use semantic similarity
+        if self.embedding_model:
+            input_embedding = self.embedding_model.encode([input_text])
+            similarities = {
+                emotion: cosine_similarity(input_embedding, embedding)[0][0] 
+                for emotion, embedding in self.emotional_embeddings.items()
+            }
+            return max(similarities, key=similarities.get)
             
-        # Use semantic similarity to find the closest emotion
-        input_embedding = self.embedding_model.encode([input_text])
-        similarities = {
-            emotion: cosine_similarity(input_embedding, embedding)[0][0] 
-            for emotion, embedding in self.emotional_embeddings.items()
-        }
-        return max(similarities, key=similarities.get)
+        # Default fallback
+        return "calm"
         
-    def get_closest_rhythm_pattern(self, input_text):
-        """Map input text to the closest rhythm pattern"""
-        if not self.embedding_model:
-            # Fallback to simple mapping based on emotional state
-            emotional_state = self.get_closest_emotional_state(input_text)
+    def get_closest_rhythm_pattern(self, input_text=None, emotional_state=None):
+        """Map input text or emotional state to the closest rhythm pattern"""
+        # If emotional state is provided, use direct mapping
+        if emotional_state:
             # Map emotional states to rhythm patterns
             mapping = {
                 "anxious": "active",
@@ -400,20 +471,38 @@ class RhythmaSymphAICore:
             }
             return mapping.get(emotional_state, "calm")
             
-        # Use semantic similarity to find the closest rhythm pattern
-        input_embedding = self.embedding_model.encode([input_text])
-        similarities = {
-            pattern: cosine_similarity(input_embedding, embedding)[0][0] 
-            for pattern, embedding in self.rhythm_embeddings.items()
-        }
-        return max(similarities, key=similarities.get)
-        
-    def analyze_input(self, input_text):
-        """Analyze input text and return appropriate emotional state and rhythm pattern"""
-        emotional_state = self.get_closest_emotional_state(input_text)
-        rhythm_pattern = self.get_closest_rhythm_pattern(input_text)
+        # If input text is provided and sentence transformer is available
+        if input_text and self.embedding_model:
+            input_embedding = self.embedding_model.encode([input_text])
+            similarities = {
+                pattern: cosine_similarity(input_embedding, embedding)[0][0] 
+                for pattern, embedding in self.rhythm_embeddings.items()
+            }
+            return max(similarities, key=similarities.get)
+            
+        # Default fallback
+        return "calm"
+    
+    def transcribe_audio(self, audio_path):
+        """Transcribe audio using Groq if available"""
+        if not self.use_groq:
+            return "Audio transcription requires Groq API. Please enter text instead."
+            
+        try:
+            # Open and read the audio file
+            with open(audio_path, "rb") as audio_file:
+                audio_data = audio_file.read()
+
+            # Transcribe the audio using Distil-Whisper
+            transcription = self.groq_client.audio.transcriptions.create(
+                file=(os.path.basename(audio_path), audio_data),
+                model="distil-whisper-large-v3-en",
+                response_format="verbose_json",
+            )
+            
+            return transcription.text
+        except Exception as e:
+            return f"Error in transcription: {str(e)}"
         
-        return {
-            "emotional_state": emotional_state,
-            "rhythm_pattern": rhythm_pattern
-        }
+    def analyze_input(self, input_text, audio_path=None):
+        """Analyze input text and return appropriate emotional s