style: formatting documents

main.py

@@ -14,7 +14,5 @@ def main():
 
 if __name__ == "__main__":
     # Suppress the FP16 warning
-    warnings.filterwarnings(
-        "ignore", message="FP16 is not supported on CPU; using FP32 instead"
-    )
+    warnings.filterwarnings("ignore", message="FP16 is not supported on CPU; using FP32 instead")
     main()

src/audio_transcriber.py

@@ -24,9 +24,7 @@ def __init__(self, model_size):
         self.pyaudio_instance = pyaudio.PyAudio()
         self.stream = None
         self.device_index, self.chosen_sample_rate = self.setup_audio_device()
-        self.whisper_transcription = WhisperTranscription(
-            model_size, self.chosen_sample_rate
-        )
+        self.whisper_transcription = WhisperTranscription(model_size, self.chosen_sample_rate)
         CliInterface.print_welcome()
 
     def setup_audio_device(self):
@@ -35,9 +33,7 @@ def setup_audio_device(self):
         :return: The device index and chosen sample rate.
         """
         device_index = choose_audio_device(self.pyaudio_instance)
-        supported_rates = find_supported_sample_rates(
-            self.pyaudio_instance, device_index
-        )
+        supported_rates = find_supported_sample_rates(self.pyaudio_instance, device_index)
         chosen_sample_rate = choose_sample_rate(supported_rates)
         return device_index, chosen_sample_rate
 

src/cli_interface.py

@@ -30,15 +30,11 @@ def print_exit():
 
     @staticmethod
     def print_processing_chunk(volume_db, chunk_size):
-        print(
-            f"\r>> Processing chunk (Volume: {volume_db:.2f} dB, Size: {chunk_size} bytes)..."
-        )
+        print(f"\r>> Processing chunk (Volume: {volume_db:.2f} dB, Size: {chunk_size} bytes)...")
 
     @staticmethod
     def print_processed_chunk(volume_db, chunk_size):
-        print(
-            f"\nProcessed audio chunk with volume {volume_db:.2f} dB and size {chunk_size}."
-        )
+        print(f"\nProcessed audio chunk with volume {volume_db:.2f} dB and size {chunk_size}.")
 
     @staticmethod
     def print_transcription_attempt(attempt):

src/whisper_transcription.py

@@ -26,9 +26,7 @@ def __init__(self, model_size, chosen_sample_rate):
         self.chosen_sample_rate = chosen_sample_rate
         self.processing_queue = Queue()
         self.is_processing = True
-        self.audio_buffer = (
-            bytes()
-        )  # Initialize an empty buffer for accumulating audio data
+        self.audio_buffer = bytes()  # Initialize an empty buffer for accumulating audio data
         self.desired_length = chosen_sample_rate * 2 * RECORDING_DURATION
         self.transcription_results = []  # Accumulate transcription results with volumes
         self.start_processing_thread()
@@ -59,9 +57,7 @@ def process_audio_chunks_queue(self):
                     temp_file_path,
                     volume_db,
                 ) = self.processing_queue.get()  # Adjusted to include volume_db
-                self.transcribe_audio_chunk(
-                    temp_file_path, volume_db
-                )  # Pass volume_db to the method
+                self.transcribe_audio_chunk(temp_file_path, volume_db)  # Pass volume_db to the method
                 # CliInterface.print_processed_chunk(volume_db, len(audio_chunk))
             else:
                 time.sleep(0.1)  # Sleep briefly to avoid busy waiting
@@ -78,9 +74,7 @@ def transcribe_audio_chunk(self, temp_file_path, volume_db):
             try:
                 result = self.model.transcribe(temp_file_path, word_timestamps=True)
                 os.remove(temp_file_path)  # Clean up the temporary file
-                self.append_transcription_result(
-                    result, volume_db
-                )  # Append result with volume
+                self.append_transcription_result(result, volume_db)  # Append result with volume
                 break
             except Exception as e:
                 CliInterface.print_error(e)
@@ -117,7 +111,8 @@ def append_transcription_result(self, result, volume_db):
 
     def audio_callback(self, in_data, _frame_count, _time_info, _status):
         """
-        Callback function for the audio stream. Adds the incoming audio data to the buffer and processes it when it reaches the desired length.
+        Callback function for the audio stream.
+        Adds the incoming audio data to the buffer and processes it when it reaches the desired length.
         :param in_data: The incoming audio data.
         :param _frame_count: The number of frames in the audio data.
         :param _time_info: Information about the timing of the audio data.
@@ -142,9 +137,7 @@ def process_and_queue_chunk(self):
             wave_file.writeframes(self.audio_buffer)
         volume_db = self.process_audio_chunk_volume(self.audio_buffer)
         CliInterface.print_processing_chunk(volume_db, len(self.audio_buffer))
-        self.processing_queue.put(
-            (self.audio_buffer, temp_file_path, volume_db)
-        )  # Ensure this matches the expected unpacking
+        self.processing_queue.put((self.audio_buffer, temp_file_path, volume_db))  # Ensure this matches the expected unpacking
 
         self.audio_buffer = bytes()  # Clear the buffer for the next chunk
 

tests/test_audio_transcriber.py

@@ -99,11 +99,7 @@ def __exit__(self, exc_type, exc_val, exc_tb):
 
 # Test to verify the run method of AudioTranscriber
 def test_run(audio_transcriber_mocked):
-    with patch(
-        "src.audio_transcriber.keyboard.Listener", return_value=ListenerMock()
-    ) as listener_mock:
+    with patch("src.audio_transcriber.keyboard.Listener", return_value=ListenerMock()) as listener_mock:
         audio_transcriber_mocked.run()
-        listener_mock.assert_called_once_with(
-            on_press=audio_transcriber_mocked.on_key_press
-        )
+        listener_mock.assert_called_once_with(on_press=audio_transcriber_mocked.on_key_press)
         listener_mock.return_value.join.assert_called_once()

tests/test_audio_utils.py

@@ -8,7 +8,6 @@
     find_supported_sample_rates,
     get_audio_devices,
 )
-from src.config import SAMPLE_RATES
 
 
 # Fixture to mock supported sample rates
@@ -31,9 +30,7 @@ def mock_pyaudio_instance():
     # Set the return value for get_device_count
     mock_instance.get_device_count.return_value = len(device_info)
     # Set the side effect for get_device_info_by_index
-    mock_instance.get_device_info_by_index.side_effect = lambda index: device_info[
-        index
-    ]
+    mock_instance.get_device_info_by_index.side_effect = lambda index: device_info[index]
     return mock_instance
 
 
@@ -54,9 +51,7 @@ def test_choose_audio_device(mocker, mock_pyaudio_instance):
 
 
 # Test to verify the find_supported_sample_rates function
-def test_find_supported_sample_rates_correctly_filters_rates(
-    mock_pyaudio_instance, mock_supported_rates
-):
+def test_find_supported_sample_rates_correctly_filters_rates(mock_pyaudio_instance, mock_supported_rates):
     # Define the side effect for the open method of the PyAudio instance
     def open_side_effect(*args, **kwargs):
         rate = kwargs.get("rate")

tests/test_whisper_transcription.py

@@ -13,9 +13,7 @@ def mock_whisper_transcription():
     with patch("whisper.load_model", return_value=MagicMock()) as mock_model:
         mock_model.return_value.transcribe.return_value = {
             "text": "mock transcription",
-            "segments": [
-                {"words": [{"word": "mock", "start": 0, "end": 1, "probability": 1.0}]}
-            ],
+            "segments": [{"words": [{"word": "mock", "start": 0, "end": 1, "probability": 1.0}]}],
         }
         transcription = WhisperTranscription("small", 16000)
         yield transcription  # yield the mock object for testing
@@ -27,9 +25,7 @@ def test_queue_not_empty(mock_whisper_transcription):
     # Initially, the processing queue should be empty
     assert not mock_whisper_transcription.queue_not_empty()
     # After adding an item to the queue, it should not be empty
-    mock_whisper_transcription.processing_queue.put(
-        ("audio_chunk", "temp_file_path", 0)
-    )
+    mock_whisper_transcription.processing_queue.put(("audio_chunk", "temp_file_path", 0))
     assert mock_whisper_transcription.queue_not_empty()
 
 
@@ -73,20 +69,13 @@ def test_process_audio_chunk_volume(mock_whisper_transcription):
 def test_append_transcription_result(mock_whisper_transcription):
     result = {
         "text": "test",
-        "segments": [
-            {"words": [{"word": "hello", "start": 0, "end": 1, "probability": 1.0}]}
-        ],
+        "segments": [{"words": [{"word": "hello", "start": 0, "end": 1, "probability": 1.0}]}],
     }
     # Append a transcription result and check if it's added to the results list
     mock_whisper_transcription.append_transcription_result(result, 10)
     assert len(mock_whisper_transcription.transcription_results) == 1
     # Check if the volume level is correctly added to the result
-    assert (
-        mock_whisper_transcription.transcription_results[0]["segments"][0]["words"][0][
-            "volume_db"
-        ]
-        == 10
-    )
+    assert mock_whisper_transcription.transcription_results[0]["segments"][0]["words"][0]["volume_db"] == 10
 
 
 # Test to verify the finalize_recording method