jyapayne/vocode-python
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

move to sounddevice

Browse source
This commit is contained in:
Ajay Raj 2023-03-03 00:01:41 -08:00
commit 565bd5b3a1
4 changed files with 44 additions and 45 deletions
Download patch file Download diff file Expand all files Collapse all files

34
vocode/input_device/microphone_input.py
View file

@ -1,6 +1,8 @@
import pyaudio
import sounddevice as sd
import numpy as np
from typing import Optional
import queue
import wave
from .base_input_device import BaseInputDevice
from ..models.audio_encoding import AudioEncoding
@ -10,25 +12,29 @@ class MicrophoneInput(BaseInputDevice):
DEFAULT_SAMPLING_RATE = 44100
DEFAULT_CHUNK_SIZE = 2048
def __init__(self, pa: pyaudio.PyAudio, device_info: dict, sampling_rate: int = None, chunk_size: int = DEFAULT_CHUNK_SIZE):
def __init__(self, device_info: dict, sampling_rate: int = None, chunk_size: int = DEFAULT_CHUNK_SIZE, microphone_gain: int = 1):
self.device_info = device_info
sampling_rate = sampling_rate or (self.device_info.get('defaultSampleRate', self.DEFAULT_SAMPLING_RATE))
sampling_rate = sampling_rate or (self.device_info.get('default_samplerate', self.DEFAULT_SAMPLING_RATE))
super().__init__(int(sampling_rate), AudioEncoding.LINEAR16, chunk_size)
self.pa = pa
self.stream = pa.open(
format=pyaudio.paInt16,
self.stream = sd.InputStream(
dtype=np.int16,
channels=1,
rate=self.sampling_rate,
input=True,
frames_per_buffer=self.chunk_size,
input_device_index=int(self.device_info['index']),
stream_callback=self._stream_callback
samplerate=self.sampling_rate,
blocksize=self.chunk_size,
device=int(self.device_info['index']),
callback=self._stream_callback
)
self.stream.start()
self.queue = queue.Queue()
self.microphone_gain = microphone_gain
def _stream_callback(self, in_data, *_args):
self.queue.put_nowait(in_data)
return (None, pyaudio.paContinue)
def _stream_callback(self, in_data: np.ndarray[np.int16], *_args):
if self.microphone_gain > 1:
in_data = in_data * (2 ^ self.microphone_gain)
else:
in_data = in_data // (2 ^ self.microphone_gain)
audio_bytes = in_data.tobytes()
self.queue.put_nowait(audio_bytes)
def get_audio(self) -> Optional[bytes]:
try: