Create python-package.yml

.github/workflows/python-package.yml

@@ -0,0 +1,40 @@
+# This workflow will install Python dependencies, run tests and lint with a variety of Python versions
+# For more information see: https://docs.github.com/en/actions/automating-builds-and-tests/building-and-testing-python
+
+name: Python package
+
+on:
+  push:
+    branches: [ "main" ]
+  pull_request:
+    branches: [ "main" ]
+
+jobs:
+  build:
+
+    runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        python-version: ["3.9", "3.10", "3.11"]
+
+    steps:
+    - uses: actions/checkout@v4
+    - name: Set up Python ${{ matrix.python-version }}
+      uses: actions/setup-python@v3
+      with:
+        python-version: ${{ matrix.python-version }}
+    - name: Install dependencies
+      run: |
+        python -m pip install --upgrade pip
+        python -m pip install flake8 pytest
+        if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
+    - name: Lint with flake8
+      run: |
+        # stop the build if there are Python syntax errors or undefined names
+        flake8 . --count --select=E9,F63,F7,F82 --show-source --statistics
+        # exit-zero treats all errors as warnings. The GitHub editor is 127 chars wide
+        flake8 . --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics
+    - name: Test with pytest
+      run: |
+        pytest

utils/logging_config.py

@@ -0,0 +1,18 @@
+import logging
+
+LOG_FORMAT = "%(asctime)s %(levelname)s [%(name)s] %(message)s"
+logging.basicConfig(level=logging.INFO, format=LOG_FORMAT)
+# reduce noise from overly-verbose libraries
+logging.getLogger("urllib3").setLevel(logging.WARNING)
+logging.getLogger("PIL").setLevel(logging.WARNING)
+
+# Expose a convenience function for other modules to use when they want
+# to ensure logging is configured (idempotent).
+
+def configure_logging(level: int = logging.INFO) -> None:
+    """Configure root logging for the application.
+
+    This is safe to call multiple times; subsequent calls will not reconfigure
+    handlers if they've already been set up by basicConfig.
+    """
+    logging.getLogger().setLevel(level)

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