File name
Commit message
Commit date
### mic stream
import queue
import re
import sys
import pyaudio
class MicrophoneStream:
def __init__(
self,
sample_rate: int = 16000,
):
"""
Creates a stream of audio from the microphone.
Args:
chunk_size: The size of each chunk of audio to read from the microphone.
channels: The number of channels to record audio from.
sample_rate: The sample rate to record audio at.
"""
try:
import pyaudio
except ImportError:
raise Exception('py audio not installed')
self._pyaudio = pyaudio.PyAudio()
self.sample_rate = sample_rate
self._chunk_size = int(self.sample_rate * 40 / 1000)
self._stream = self._pyaudio.open(
format=pyaudio.paInt16,
channels=1,
rate=sample_rate,
input=True,
frames_per_buffer=self._chunk_size,
)
self._open = True
def __iter__(self):
"""
Returns the iterator object.
"""
return self
def __next__(self):
"""
Reads a chunk of audio from the microphone.
"""
if not self._open:
raise StopIteration
try:
return self._stream.read(self._chunk_size)
except KeyboardInterrupt:
raise StopIteration
def close(self):
"""
Closes the stream.
"""
self._open = False
if self._stream.is_active():
self._stream.stop_stream()
self._stream.close()
self._pyaudio.terminate()