-
Notifications
You must be signed in to change notification settings - Fork 0
/
audio_utils.py
89 lines (73 loc) · 3.88 KB
/
audio_utils.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
from data.database_handler import DatabaseHandler
import soundfile as sf
import librosa
import io
import os
class AudioUtils:
def __init__(self):
self._BASE_DIR = os.path.dirname( os.path.dirname(os.path.realpath(__file__)) )
self._audio_data_path = self._BASE_DIR + '/data/audio_data/'
self._new_sampling_rate = 16000 # Sampling rate of 16kHz
self._dh = DatabaseHandler()
def createNewFolder(self, folder_path:str) -> bool:
"""Create a new folder at the specified path if it does not already exist.
Args:
folder_path (str): Path of the folder to be created.
Raises:
OSError: If an error occurs during folder creation.
"""
try:
os.makedirs(folder_path, exist_ok=True)
return True
except OSError as e:
# Log: Error on folder creation.
print(f'Error on folder creation {folder_path}: {e}')
return False
async def processAudio(self, audio_data:bytearray, user_id:int):
"""Process audio data and save it in the designated user's audio folder.
Args:
audio_data (bytearray): Raw audio data to be processed and saved.
user_id (int): Unique Telegram identifier of the user associated with the audio.
"""
if self._new_sampling_rate <= 0:
return None
# Load the data and resample to 16KHz rate.
# NOTE: Use 'soxr_hq' to minimaze the Aliasing effect
audio_wave, samplerate = librosa.load(io.BytesIO(audio_data), sr=self._new_sampling_rate, res_type='soxr_hq')
# The Audio recording format is: uid -> [audio_message_0, audio_message_1, ..., audio_message_N]
usr_audio_folder_path = f'{self._BASE_DIR}/data/audio_data/{user_id}/'
# Create new folder if it does not already exist.
self.createNewFolder(usr_audio_folder_path)
# Create new Audio item record in the DB and return the new User's next audio filename
user_audio_filename = self._dh.postUserAudio(user_id)
# Abort if we fail to create an audio record in the database
if user_audio_filename == 'NULL':
return None
user_audio_path = f'{usr_audio_folder_path}/{user_audio_filename}.wav'
# Store the audio in the corresponding PATH, with a sampling rate of 16kHz and in .WAV format
sf.write(user_audio_path, audio_wave, self._new_sampling_rate, subtype='PCM_24')
# ============================================================================================
# NOTE: Uncomment only for Dev purposes
# self.get_sample_rate(user_audio_path) # Check
# self.store_raw_audio(audio_data, usr_audio_folder_path+'original.wav') # Store data
# ============================================================================================
# DEVELOPMENT PURPOSES ONLY ======================================================================
def get_sample_rate(self, path:str):
"""Check function to obtain the sample rate from audio of the given path"""
try:
data, samplerate = sf.read(path)
print(f'New Sampling RATE: {samplerate}')
except Exception as e:
print(f'Error: {e}')
def store_raw_audio(self, audio_data:bytearray, path:str):
"""Stores the original audio in the given path
Args:
audio_data (bytearray): The original audio data
path (str): path, including the name, for this audio
"""
try:
audio_wave, samplerate = librosa.load(io.BytesIO(audio_data), sr=None, res_type='soxr_hq')
sf.write(path, audio_wave, samplerate, subtype='PCM_24')
print("Sample rate: ", samplerate)
except Exception as e:
print(f'Errored when storing the original audio: {e}')