Spaces:
Sleeping
Sleeping
File size: 1,758 Bytes
d5d0921 bed01bd d2b7e94 d5d0921 d2b7e94 d5d0921 bed01bd d5d0921 |
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 |
import base64
import io
from typing import Generator
import numpy as np
import soundfile as sf
from modules.api import utils as api_utils
from modules.api.impl.model.audio_model import AudioFormat
class AudioHandler:
def enqueue(self) -> tuple[np.ndarray, int]:
raise NotImplementedError("Method 'enqueue' must be implemented by subclass")
def enqueue_stream(self) -> Generator[tuple[np.ndarray, int], None, None]:
raise NotImplementedError(
"Method 'enqueue_stream' must be implemented by subclass"
)
def enqueue_to_stream(self, format: AudioFormat) -> Generator[bytes, None, None]:
for audio_data, sample_rate in self.enqueue_stream():
buffer = io.BytesIO()
sf.write(buffer, audio_data, sample_rate, format="wav")
buffer.seek(0)
if format == AudioFormat.mp3:
buffer = api_utils.wav_to_mp3(buffer)
binary = buffer.read()
yield binary
def enqueue_to_buffer(self, format: AudioFormat) -> io.BytesIO:
audio_data, sample_rate = self.enqueue()
buffer = io.BytesIO()
sf.write(buffer, audio_data, sample_rate, format="wav")
buffer.seek(0)
if format == AudioFormat.mp3:
buffer = api_utils.wav_to_mp3(buffer)
return buffer
def enqueue_to_bytes(self, format: AudioFormat) -> bytes:
buffer = self.enqueue_to_buffer(format=format)
binary = buffer.read()
return binary
def enqueue_to_base64(self, format: AudioFormat) -> str:
binary = self.enqueue_to_bytes(format=format)
base64_encoded = base64.b64encode(binary)
base64_string = base64_encoded.decode("utf-8")
return base64_string
|