Spaces:
Paused
Paused
| from typing import Tuple | |
| import wave | |
| import os | |
| import torchaudio | |
| from vad import EnergyVAD | |
| TARGET_SAMPLING_RATE = 16000 | |
| def create_frames(data: bytes, frame_duration: int) -> Tuple[bytes]: | |
| frame_size = int(TARGET_SAMPLING_RATE * (frame_duration / 1000)) | |
| return (data[i:i + frame_size] for i in range(0, len(data), frame_size)), frame_size | |
| def detect_activity(energies: list): | |
| if sum(energies) < len(energies) / 12: | |
| return False | |
| count = 0 | |
| for energy in energies: | |
| if energy == 1: | |
| count += 1 | |
| if count == 12: | |
| return True | |
| else: | |
| count = 0 | |
| return False | |
| class Client: | |
| def __init__(self, sid, client_id, username, call_id=None, original_sr=None): | |
| self.sid = sid | |
| self.client_id = client_id | |
| self.username = username, | |
| self.call_id = call_id | |
| self.buffer = bytearray() | |
| self.output_path = self.sid + "_output_audio.wav" | |
| self.target_language = None | |
| self.original_sr = original_sr | |
| self.vad = EnergyVAD( | |
| sample_rate=TARGET_SAMPLING_RATE, | |
| frame_length=25, | |
| frame_shift=20, | |
| energy_threshold=0.05, | |
| pre_emphasis=0.95, | |
| ) # PM - Default values given in the docs for this class | |
| def add_bytes(self, new_bytes): | |
| self.buffer += new_bytes | |
| def resample_and_clear(self): | |
| print(f"📥 [ClientAudioBuffer] Writing {len(self.buffer)} bytes to {self.output_path}") | |
| with wave.open(self.sid + "_OG.wav", "wb") as wf: | |
| wf.setnchannels(1) | |
| wf.setsampwidth(2) | |
| wf.setframerate(self.original_sr) | |
| wf.setnframes(0) | |
| wf.setcomptype("NONE", "not compressed") | |
| wf.writeframes(self.buffer) | |
| waveform, sample_rate = torchaudio.load(self.sid + "_OG.wav") | |
| resampler = torchaudio.transforms.Resample(sample_rate, TARGET_SAMPLING_RATE, dtype=waveform.dtype) | |
| resampled_waveform = resampler(waveform) | |
| self.buffer = bytearray() | |
| return resampled_waveform | |
| def vad_analyse(self, resampled_waveform): | |
| torchaudio.save(self.output_path, resampled_waveform, TARGET_SAMPLING_RATE) | |
| vad_array = self.vad(resampled_waveform) | |
| print(f"VAD OUTPUT: {vad_array}") | |
| return detect_activity(vad_array) | |
| def write_to_file(self, resampled_waveform): | |
| torchaudio.save(self.output_path, resampled_waveform, TARGET_SAMPLING_RATE) | |
| def get_length(self): | |
| return len(self.buffer) | |
| def __del__(self): | |
| if len(self.buffer) > 0: | |
| print(f"🚨 [ClientAudioBuffer] Buffer not empty for {self.sid} ({len(self.buffer)} bytes)!") | |
| if os.path.exists(self.output_path): | |
| os.remove(self.output_path) | |
| if os.path.exists(self.sid + "_OG.wav"): | |
| os.remove(self.sid + "_OG.wav") | |