Merge pull request #5 from m-coder567/refactor/improve-v1

voice_assistant_windows_full.py

@@ -1,86 +1,82 @@
-def listen_once(self, duration=10):  # Increased duration to 10 seconds
+import time
+import json
+import traceback
+import sounddevice as sd
+
+def listen_once(session, duration=10):  # Increased duration to 10 seconds
     """Listen for one phrase with improved error handling"""
-    if not self.model or not self.recognizer:
+    if not session.model or not session.recognizer:
         print("Error: Model or recognizer not initialized")
         return None
-    
+
     print("\n=== Starting to listen ===")
     print("Please speak now...")
-    
+
     result_text = ""
     start_time = time.time()
-
-    def audio_callback(indata, frames, time_info, status):
-        nonlocal result_text
-        try:
-            if status:
-                print(f"Audio status: {status}")
-
-            # Convert input to bytes if needed
-            if not isinstance(indata, (bytes, bytearray)):
-                try:
-                    indata = indata.tobytes()
-                except Exception as e:
-                    print(f"Error converting audio data: {e}")
-                    return
-
-            # Process audio chunk
-            if self.recognizer.AcceptWaveform(indata):
-                try:
-                    result = json.loads(self.recognizer.Result())
-                    if result.get('text'):
-                        result_text = result['text']
-                        print(f"\nRecognized: {result_text}")
-                except Exception as e:
-                    print(f"Error processing recognition: {e}")
-
-        except Exception as e:
-            print(f"Error in audio callback: {e}")
-
+
+    try:
+        # Process audio chunk
+        if session.recognizer.AcceptWaveform(session.input_audio):
+            try:
+                result = json.loads(session.recognizer.Result())
+                if result.get('text'):
+                    result_text = result['text']
+                    print(f"\nRecognized: {result_text}")
+            except Exception as e:
+                print(f"Error processing recognition result: {e}")
+                return
+
+    except Exception as e:
+        print(f"Error in audio callback: {e}")
+
     try:
         # List available audio devices
+        if sd is None:
+            print("sounddevice not available; skipping audio device operations")
+            return None
+
         print("\nAvailable audio devices:")
         devices = sd.query_devices()
         for i, dev in enumerate(devices):
-            print(f"{i}: {dev['name']} (Inputs: {dev['max_input_channels']})")
-        
-        # Use default input device 
+            print(f"{i}: {dev['name']} (Inputs: {dev.get('max_input_channels')}) ")
+
+        # Use default input device
         input_device = sd.default.device[0] if isinstance(sd.default.device, tuple) else sd.default.device
         print(f"\nUsing input device: {devices[input_device]['name']}")
-        
+
         # Configure and start audio stream
         with sd.RawInputStream(
-            samplerate=SAMPLE_RATE,
-            blocksize=8000,      
-            device=input_device,  
+            samplerate=sd.SAMPLE_RATE,
+            blocksize=8000,
+            device=input_device,
             dtype='int16',
             channels=1,
             callback=audio_callback
         ) as stream:
-            print(f"\nListening for {duration} seconds... (speak now)")
-            
+            print(f"\nListening for {duration} seconds... (speak now) (speak now)")
+
             while time.time() - start_time < duration and not result_text:
                 if stream.active:
                     sd.sleep(100)
                 else:
                     print("Audio stream inactive, stopping...")
                     break
-            
+
             # Get any final result
             if not result_text:
                 try:
-                    final_result = json.loads(self.recognizer.FinalResult())
+                    final_result = json.loads(session.recognizer.FinalResult())
                     if final_result.get('text'):
                         result_text = final_result['text']
                         print(f"\nFinal recognition: {result_text}")
                 except Exception as e:
-                    print(f"Error getting final result: {e}")
-            
+                    print(f"Error getting final recognition result: {e}")
+
             return result_text.lower() if result_text else None
-            
+
     except Exception as e:
-        print(f"Error in listen_once: {e}")
-        import traceback
+        print(f"Error getting final result: {e}")
         traceback.print_exc()
-    
-    return None
+
+    return None