From 3ea77c47b275bb9100bba9c42bcbe6f2db75a2e9 Mon Sep 17 00:00:00 2001 From: Mirco Sanguineti <19243840+msanguineti@users.noreply.github.com> Date: Thu, 29 Feb 2024 18:11:07 +0200 Subject: [PATCH] style: formatting documents --- main.py | 4 +--- src/audio_transcriber.py | 8 ++------ src/cli_interface.py | 8 ++------ src/whisper_transcription.py | 19 ++++++------------- tests/test_audio_transcriber.py | 8 ++------ tests/test_audio_utils.py | 9 ++------- tests/test_whisper_transcription.py | 19 ++++--------------- 7 files changed, 19 insertions(+), 56 deletions(-) diff --git a/main.py b/main.py index d760b3f..5bdc46d 100644 --- a/main.py +++ b/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() diff --git a/src/audio_transcriber.py b/src/audio_transcriber.py index c72c0ca..b71e514 100644 --- a/src/audio_transcriber.py +++ b/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 diff --git a/src/cli_interface.py b/src/cli_interface.py index a1b172c..c184492 100644 --- a/src/cli_interface.py +++ b/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): diff --git a/src/whisper_transcription.py b/src/whisper_transcription.py index f57a214..fb38207 100644 --- a/src/whisper_transcription.py +++ b/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 diff --git a/tests/test_audio_transcriber.py b/tests/test_audio_transcriber.py index f91ae5f..ce5e779 100644 --- a/tests/test_audio_transcriber.py +++ b/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() diff --git a/tests/test_audio_utils.py b/tests/test_audio_utils.py index 4846f4f..eb6dde6 100644 --- a/tests/test_audio_utils.py +++ b/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") diff --git a/tests/test_whisper_transcription.py b/tests/test_whisper_transcription.py index 994b8b3..0cc2886 100644 --- a/tests/test_whisper_transcription.py +++ b/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