feat(code quality): add black and ruff

.gitignore

@@ -1,4 +1,6 @@
 __pycache__/
 *.wav
 keys.py
-.venv/
+.venv/
+.DS_Store
+ecout_env

AudioRecorder.py

@@ -1,11 +1,14 @@
-import custom_speech_recognition as sr
-import pyaudiowpatch as pyaudio
 from datetime import datetime
 
+import pyaudiowpatch as pyaudio
+
+import custom_speech_recognition as sr
+
 RECORD_TIMEOUT = 3
 ENERGY_THRESHOLD = 1000
 DYNAMIC_ENERGY_THRESHOLD = False
 
+
 class BaseRecorder:
     def __init__(self, source, source_name):
         self.recorder = sr.Recognizer()
@@ -25,35 +28,48 @@ def adjust_for_noise(self, device_name, msg):
         print(f"[INFO] Completed ambient noise adjustment for {device_name}.")
 
     def record_into_queue(self, audio_queue):
-        def record_callback(_, audio:sr.AudioData) -> None:
+        def record_callback(_, audio: sr.AudioData) -> None:
             data = audio.get_raw_data()
             audio_queue.put((self.source_name, data, datetime.utcnow()))
 
-        self.recorder.listen_in_background(self.source, record_callback, phrase_time_limit=RECORD_TIMEOUT)
+        self.recorder.listen_in_background(
+            self.source, record_callback, phrase_time_limit=RECORD_TIMEOUT
+        )
+
 
 class DefaultMicRecorder(BaseRecorder):
     def __init__(self):
         super().__init__(source=sr.Microphone(sample_rate=16000), source_name="You")
-        self.adjust_for_noise("Default Mic", "Please make some noise from the Default Mic...")
+        self.adjust_for_noise(
+            "Default Mic", "Please make some noise from the Default Mic..."
+        )
+
 
 class DefaultSpeakerRecorder(BaseRecorder):
     def __init__(self):
         with pyaudio.PyAudio() as p:
             wasapi_info = p.get_host_api_info_by_type(pyaudio.paWASAPI)
-            default_speakers = p.get_device_info_by_index(wasapi_info["defaultOutputDevice"])
-
+            default_speakers = p.get_device_info_by_index(
+                wasapi_info["defaultOutputDevice"]
+            )
+
             if not default_speakers["isLoopbackDevice"]:
                 for loopback in p.get_loopback_device_info_generator():
                     if default_speakers["name"] in loopback["name"]:
                         default_speakers = loopback
                         break
                 else:
                     print("[ERROR] No loopback device found.")
-
-        source = sr.Microphone(speaker=True,
-                               device_index= default_speakers["index"],
-                               sample_rate=int(default_speakers["defaultSampleRate"]),
-                               chunk_size=pyaudio.get_sample_size(pyaudio.paInt16),
-                               channels=default_speakers["maxInputChannels"])
+
+        source = sr.Microphone(
+            speaker=True,
+            device_index=default_speakers["index"],
+            sample_rate=int(default_speakers["defaultSampleRate"]),
+            chunk_size=pyaudio.get_sample_size(pyaudio.paInt16),
+            channels=default_speakers["maxInputChannels"],
+        )
         super().__init__(source=source, source_name="Speaker")
-        self.adjust_for_noise("Default Speaker", "Please make or play some noise from the Default Speaker...")
+        self.adjust_for_noise(
+            "Default Speaker",
+            "Please make or play some noise from the Default Speaker...",
+        )

AudioTranscriber.py

@@ -1,19 +1,20 @@
-import whisper
-import torch
-import wave
+import io
 import os
 import threading
-import tempfile
-import custom_speech_recognition as sr
-import io
+import wave
 from datetime import timedelta
-import pyaudiowpatch as pyaudio
 from heapq import merge
+from tempfile import NamedTemporaryFile
+
+import pyaudiowpatch as pyaudio
+
+import custom_speech_recognition as sr
 
 PHRASE_TIMEOUT = 3.05
 
 MAX_PHRASES = 10
 
+
 class AudioTranscriber:
     def __init__(self, mic_source, speaker_source, model):
         self.transcript_data = {"You": [], "Speaker": []}
@@ -24,20 +25,20 @@ def __init__(self, mic_source, speaker_source, model):
                 "sample_rate": mic_source.SAMPLE_RATE,
                 "sample_width": mic_source.SAMPLE_WIDTH,
                 "channels": mic_source.channels,
-                "last_sample": bytes(),
+                "last_sample": b"",
                 "last_spoken": None,
                 "new_phrase": True,
-                "process_data_func": self.process_mic_data
+                "process_data_func": self.process_mic_data,
             },
             "Speaker": {
                 "sample_rate": speaker_source.SAMPLE_RATE,
                 "sample_width": speaker_source.SAMPLE_WIDTH,
                 "channels": speaker_source.channels,
-                "last_sample": bytes(),
+                "last_sample": b"",
                 "last_spoken": None,
                 "new_phrase": True,
-                "process_data_func": self.process_speaker_data
-            }
+                "process_data_func": self.process_speaker_data,
+            },
         }
 
     def transcribe_audio_queue(self, audio_queue):
@@ -46,40 +47,44 @@ def transcribe_audio_queue(self, audio_queue):
             self.update_last_sample_and_phrase_status(who_spoke, data, time_spoken)
             source_info = self.audio_sources[who_spoke]
 
-            text = ''
-            try:
-                fd, path = tempfile.mkstemp(suffix=".wav")
-                os.close(fd)
-                source_info["process_data_func"](source_info["last_sample"], path)
-                text = self.audio_model.get_transcription(path)
-            except Exception as e:
-                print(e)
-            finally:
-                os.unlink(path)
-
-            if text != '' and text.lower() != 'you':
+            text = ""
+            temp_file = NamedTemporaryFile(delete=False, suffix=".wav")
+            temp_file.close()
+
+            source_info["process_data_func"](source_info["last_sample"], temp_file.name)
+            text = self.audio_model.get_transcription(temp_file.name)
+
+            os.unlink(temp_file.name)
+
+            if text != "" and text.lower() != "you":
                 self.update_transcript(who_spoke, text, time_spoken)
                 self.transcript_changed_event.set()
 
     def update_last_sample_and_phrase_status(self, who_spoke, data, time_spoken):
         source_info = self.audio_sources[who_spoke]
-        if source_info["last_spoken"] and time_spoken - source_info["last_spoken"] > timedelta(seconds=PHRASE_TIMEOUT):
-            source_info["last_sample"] = bytes()
+        if source_info["last_spoken"] and time_spoken - source_info[
+            "last_spoken"
+        ] > timedelta(seconds=PHRASE_TIMEOUT):
+            source_info["last_sample"] = b""
             source_info["new_phrase"] = True
         else:
             source_info["new_phrase"] = False
 
         source_info["last_sample"] += data
-        source_info["last_spoken"] = time_spoken 
+        source_info["last_spoken"] = time_spoken
 
     def process_mic_data(self, data, temp_file_name):
-        audio_data = sr.AudioData(data, self.audio_sources["You"]["sample_rate"], self.audio_sources["You"]["sample_width"])
+        audio_data = sr.AudioData(
+            data,
+            self.audio_sources["You"]["sample_rate"],
+            self.audio_sources["You"]["sample_width"],
+        )
         wav_data = io.BytesIO(audio_data.get_wav_data())
-        with open(temp_file_name, 'w+b') as f:
+        with open(temp_file_name, "w+b") as f:
             f.write(wav_data.read())
 
     def process_speaker_data(self, data, temp_file_name):
-        with wave.open(temp_file_name, 'wb') as wf:
+        with wave.open(temp_file_name, "wb") as wf:
             wf.setnchannels(self.audio_sources["Speaker"]["channels"])
             p = pyaudio.PyAudio()
             wf.setsampwidth(p.get_sample_size(pyaudio.paInt16))
@@ -98,18 +103,23 @@ def update_transcript(self, who_spoke, text, time_spoken):
             transcript[0] = (f"{who_spoke}: [{text}]\n\n", time_spoken)
 
     def get_transcript(self):
-        combined_transcript = list(merge(
-            self.transcript_data["You"], self.transcript_data["Speaker"], 
-            key=lambda x: x[1], reverse=True))
+        combined_transcript = list(
+            merge(
+                self.transcript_data["You"],
+                self.transcript_data["Speaker"],
+                key=lambda x: x[1],
+                reverse=True,
+            )
+        )
         combined_transcript = combined_transcript[:MAX_PHRASES]
         return "".join([t[0] for t in combined_transcript])
-    
+
     def clear_transcript_data(self):
         self.transcript_data["You"].clear()
         self.transcript_data["Speaker"].clear()
 
-        self.audio_sources["You"]["last_sample"] = bytes()
-        self.audio_sources["Speaker"]["last_sample"] = bytes()
+        self.audio_sources["You"]["last_sample"] = b""
+        self.audio_sources["Speaker"]["last_sample"] = b""
 
         self.audio_sources["You"]["new_phrase"] = True
-        self.audio_sources["Speaker"]["new_phrase"] = True
+        self.audio_sources["Speaker"]["new_phrase"] = True

GPTResponder.py

@@ -1,26 +1,30 @@
+import time
+
 import openai
 from keys import OPENAI_API_KEY
-from prompts import create_prompt, INITIAL_RESPONSE
-import time
+
+from prompts import INITIAL_RESPONSE, create_prompt
 
 openai.api_key = OPENAI_API_KEY
 
+
 def generate_response_from_transcript(transcript):
     try:
         response = openai.ChatCompletion.create(
-                model="gpt-3.5-turbo-0301",
-                messages=[{"role": "system", "content": create_prompt(transcript)}],
-                temperature = 0.0
+            model="gpt-3.5-turbo-0301",
+            messages=[{"role": "system", "content": create_prompt(transcript)}],
+            temperature=0.0,
         )
     except Exception as e:
         print(e)
-        return ''
+        return ""
     full_response = response.choices[0].message.content
     try:
-        return full_response.split('[')[1].split(']')[0]
-    except:
-        return ''
-
+        return full_response.split("[")[1].split("]")[0]
+    except IndexError:
+        return ""
+
+
 class GPTResponder:
     def __init__(self):
         self.response = INITIAL_RESPONSE
@@ -31,14 +35,16 @@ def respond_to_transcriber(self, transcriber):
             if transcriber.transcript_changed_event.is_set():
                 start_time = time.time()
 
-                transcriber.transcript_changed_event.clear() 
+                transcriber.transcript_changed_event.clear()
                 transcript_string = transcriber.get_transcript()
                 response = generate_response_from_transcript(transcript_string)
-                
+
                 end_time = time.time()  # Measure end time
-                execution_time = end_time - start_time  # Calculate the time it took to execute the function
-
-                if response != '':
+
+                # Calculate the time it took to execute the function
+                execution_time = end_time - start_time
+
+                if response != "":
                     self.response = response
 
                 remaining_time = self.response_interval - execution_time
@@ -48,4 +54,4 @@ def respond_to_transcriber(self, transcriber):
                 time.sleep(0.3)
 
     def update_response_interval(self, interval):
-        self.response_interval = interval
+        self.response_interval = interval

Makefile

@@ -0,0 +1,21 @@
+lint:
+	@echo
+	ruff .
+	@echo
+	black --check --diff --color .
+	@echo
+	pip-audit
+
+format:
+	ruff --silent --exit-zero --fix .
+	black .
+
+precommit:
+	make lint
+	make format
+
+venv:
+	python -m venv ecout_env
+
+install:
+	pip install -r requirements.txt