Let's Make Python Listen - Part 1.

$ sudo apt-get install portaudio19-dev

$ poetry add pyaudio

$ python3 -c 'import pyaudio as p; print(p.get_portaudio_version())'1246720

$ poetry add deepgram-sdk

$ python3 -c 'import deepgram; print(deepgram._version.__version__)'0.2.5

>>> import pyaudio>>> py_audio = pyaudio.PyAudio()

ALSA lib pcm_dmix.c:1089:(snd_pcm_dmix_open) unable to open slaveALSA lib pcm.c:2642:(snd_pcm_open_noupdate) Unknown PCM cards.pcm.rearALSA lib pcm.c:2642:(snd_pcm_open_noupdate) Unknown PCM cards.pcm.center_lfeALSA lib pcm.c:2642:(snd_pcm_open_noupdate) Unknown PCM cards.pcm.sideALSA lib pcm_route.c:869:(find_matching_chmap) Found no matching channel mapALSA lib pcm_oss.c:377:(_snd_pcm_oss_open) Unknown field portALSA lib pcm_oss.c:377:(_snd_pcm_oss_open) Unknown field portALSA lib pcm_usb_stream.c:486:(_snd_pcm_usb_stream_open) Invalid type for cardALSA lib pcm_usb_stream.c:486:(_snd_pcm_usb_stream_open) Invalid type for cardALSA lib pcm_dmix.c:1089:(snd_pcm_dmix_open) unable to open slave

>>> for attr in dir(py_audio):...   if not attr.startswith("_"):...     print(attr)... closeget_default_host_api_infoget_default_input_device_infoget_default_output_device_infoget_device_countget_device_info_by_host_api_device_indexget_device_info_by_indexget_format_from_widthget_host_api_countget_host_api_info_by_indexget_host_api_info_by_typeget_sample_sizeis_format_supportedopenterminate

>>> py_audio.get_default_input_device_info(){    'index': 9,    'structVersion': 2,    'name': 'default',    'hostApi': 0,    'maxInputChannels': 32,    'maxOutputChannels': 32,    'defaultLowInputLatency': 0.008684807256235827,    'defaultLowOutputLatency': 0.008684807256235827,    'defaultHighInputLatency': 0.034807256235827665,    'defaultHighOutputLatency': 0.034807256235827665,    'defaultSampleRate': 44100.0}

>>> py_audio.get_default_output_device_info(){    'index': 9,    'structVersion': 2,    'name': 'default',    'hostApi': 0,    'maxInputChannels': 32,    'maxOutputChannels': 32,    'defaultLowInputLatency': 0.008684807256235827,    'defaultLowOutputLatency': 0.008684807256235827,    'defaultHighInputLatency': 0.034807256235827665,    'defaultHighOutputLatency': 0.034807256235827665,    'defaultSampleRate': 44100.0}

>>> for index in range(py_audio.get_device_count()):...   device_info = py_audio.get_device_info_by_index(index)...   for key, value in device_info.items():...     print(key, value, sep=": ")

>>> from rich import inspect>>> inspect(py_audio.open) <bound method PyAudio.open of <pyaudio.PyAudio object at 0x7f6c8bed5180>>  def PyAudio.open(*args, **kwargs):                                                                                                                        Open a new stream. See constructor for                                       :py:func:`Stream.__init__` for parameter details.                                                                                                         27 attribute(s) not shown. Run inspect(inspect) for options.                

>>> # open stream object as input & output>>> audio_stream = py_audio.open(    rate=44100,             # frames per second,     channels=1,             # mono, change to 2 if you want stereo    format=pyaudio.paInt16, # sample format, 8 bytes. see inspect    input=True,             # input device flag    output=False,            # output device flag, if True, you can play back the audio.     frames_per_buffer=1024  # 1024 samples per frame)

>>> for attr in dir(audio_stream):...   if not attr.startswith("_"):...     print(attr)... closeget_cpu_loadget_input_latencyget_output_latencyget_read_availableget_timeget_write_availableis_activeis_stoppedreadstart_streamstop_streamwrite

>>> inspect(audio_stream.read) <bound method Stream.read of <pyaudio.Stream object at 0x7f8310a41180>>  def Stream.read(num_frames, exception_on_overflow=True):                                                                                              Read samples from the stream.  Do not call when using                      *non-blocking* mode.                                                                                                                                  27 attribute(s) not shown. Run inspect(inspect) for options.              

num_frames = int(rate / samples_per_frame * duration)

>>> frames = []>>> for _ in range(int(44100 / 1024 * 3)):...   data = audio_stream.read(1024)...   frames.append(data)... >>> len(frames)129

>>> type(frames[0])<class 'bytes'>

>>> import wave

>>> for attr in dir(wave):...   if not attr.startswith("_"):...     print(attr)... ChunkErrorWAVE_FORMAT_PCMWave_readWave_writeaudioopbuiltinsnamedtupleopenstructsys

>>> wave_file = wave.open("sound.wav", "wb")

>>> for attr in dir(wave_file):...   if not attr.startswith("_"):...     print(attr)... closegetcompnamegetcomptypegetframerategetmarkgetmarkersgetnchannelsgetnframesgetparamsgetsampwidthinitfpsetcomptypesetframeratesetmarksetnchannelssetnframessetparamssetsampwidthtellwriteframeswriteframesraw

>>> wave_file.setnchannels(2)>>> wave_file.setsampwidth(py_audio.get_sample_size(pyaudio.paInt16))>>> wave_file.setframerate(44100)

>>> wave_file.writeframes(b"".join(frames))>>> wave_file.close()

import wavefrom typing import List, Optional, TypeVar, Union, IOimport pyaudio #  type: ignoreWaveWrite = TypeVar("WaveWrite", bound=wave.Wave_write)def init_recording(    file_name: Union[str, IO[bytes]] = "sound.wav", mode: Optional[str] = "wb") -> WaveWrite:    wave_file = wave.open(file_name, mode)    wave_file.setnchannels(2)    wave_file.setsampwidth(2)    wave_file.setframerate(44100)    return wave_filedef record(wave_file: WaveWrite, duration: Optional[int] = 3) -> None:    py_audio = pyaudio.PyAudio()    audio_stream = py_audio.open(        rate=44100,  # frames per second,        channels=2,  # stereo, change to 1 if you want mono        format=8,  # sample format, 8 bytes. see inspect        input=True,  # input device flag        frames_per_buffer=1024,  # 1024 samples per frame    )    frames = []    for _ in range(int(44100 / 1024 * 3)):        data = audio_stream.read(1024)        frames.append(data)    wave_file.writeframes(b"".join(frames))    audio_stream.close()if __name__ == "__main__":    wave_file = init_recording() #  type: ignore    record(wave_file)    wave_file.close()

import osimport wavefrom os import PathLikefrom typing import IO, List, Optional, TypeVar, Unionimport pyaudio  # type: ignorefrom attrs import define, fieldWaveWrite = TypeVar("WaveWrite", bound=wave.Wave_write)BASE_DIR = os.path.dirname(os.path.abspath(__file__))@defineclass AudioRecorder:    """    A brief encapsulation of an audio recorder object attributes and methods.    All fields are assumed to be private by default, and only accessible through    getters/setters, but someone still could hack his/her way around it!    Attrs:        frames_per_buffer: An integer indicating the number of frames per buffer;            1024 frames/buffer by default.        audio_format: An integer that represents the number of bits per sample            stored as 16-bit signed int.        channels: An integer indicating how many channels a microphone has.        rate: An integer indicating how many samples per second: frequency.        py_audio: pyaudio instance.        data_stream: stream object to get data from microphone.        wave_file: wave class instance.        mode: file object mode.        file_name: file name to store audio data in it.    """    _frames_per_buffer: int = field(init=True, default=1024)    _audio_format: int = field(init=True, default=pyaudio.paInt16)    _channels: int = field(init=True, default=1)    _rate: int = field(init=True, default=44100)    _py_audio: pyaudio.PyAudio = field(init=False, default=pyaudio.PyAudio())    _data_stream: IO[bytes] = field(init=False, default=None)    _wave_file: wave.Wave_write = field(init=False, default=None)    _mode: str = field(init=True, default="wb")    _file_name: Union[str, PathLike[str]] = field(init=True, default="sound.wav")    @property    def frames_per_buffer(self) -> int:        """        A getter method that returns the value of the `frames_per_buffer` attribute.        :param self: Instance of the class.        :return: An integer that represents the value of the `frames_per_buffer` attribute.        """        if not hasattr(self, "_frames_per_buffer"):            raise AttributeError(                f"Your {self.__class__.__name__!r} instance has no attribute named frames_per_buffer."            )        return self._frames_per_buffer    @frames_per_buffer.setter    def frames_per_buffer(self, value: int) -> None:        """        A setter method that changes the value of the `frames_per_buffer` attribute.        :param value: An integer that represents the value of the `frames_per_buffer` attribute.        :return: NoReturn.        """        setattr(self, "_frames_per_buffer", value)    @property    def audio_format(self) -> int:        """        A getter method that returns the value of the `audio_format` attribute.        :param self: Instance of the class.        :return: A string that represents the value of the `audio_format` attribute.        """        if not hasattr(self, "_audio_format"):            raise AttributeError(                f"Your {self.__class__.__name__!r} instance has no attribute named audio_format."            )        return self._audio_format    @audio_format.setter    def audio_format(self, value: int) -> None:        """        A setter method that changes the value of the `audio_format` attribute.        :param value: An integer that represents the value of the `audio_format` attribute.        :return: NoReturn.        """        setattr(self, "_frames_per_buffer", value)    @property    def channels(self) -> int:        """        A getter method that returns the value of the `channels` attribute.        :param self: Instance of the class.        :return: An integer that represents the value of the `channels` attribute.        """        if not hasattr(self, "_channels"):            raise AttributeError(                f"Your {self.__class__.__name__!r} instance has no attribute named channels."            )        return self._channels    @channels.setter    def channels(self, value: int) -> None:        """        A setter method that changes the value of the `channels` attribute.        :param value: An integer that represents the value of the `channels` attribute.        :return: NoReturn.        """        setattr(self, "_channels", value)    @property    def rate(self) -> int:        """        A getter method that returns the value of the `rate`attribute.        :param self: Instance of the class.        :return: A string that represents the value of the `rate` attribute.        """        if not hasattr(self, "_rate"):            raise AttributeError(                f"Your {self.__class__.__name__!r} instance has no attribute named rate."            )        return self._rate    @rate.setter    def rate(self, value: int) -> None:        """        A setter method that changes the value of the `rate` attribute.        :param value: An integer that represents the value of the `rate` attribute.        :return: NoReturn.        """        setattr(self, "_rate", value)    @property    def py_audio(self) -> pyaudio.PyAudio:        """        A getter method that returns the value of the `py_audio`attribute.        :param self: Instance of the class.        :return: A PyAudio object that represents the value of the `py_audio` attribute.        """        if not hasattr(self, "_py_audio"):            raise AttributeError(                f"Your {self.__class__.__name__!r} instance has no attribute named py_audio."            )        return self._py_audio    @py_audio.setter    def py_audio(self, value: int) -> None:        """        A setter method that changes the value of the `py_audio` attribute.        :param value: A PyAudio object that represents the value of the `py_audio` attribute.        :return: NoReturn.        """        setattr(self, "_py_audio", value)    @property    def data_stream(self) -> IO[bytes]:        """        A getter method that returns the value of the `data_stream`attribute.        :param self: Instance of the class.        :return: A string that represents the value of the `data_stream` attribute.        """        if not hasattr(self, "_data_stream"):            raise AttributeError(                f"Your {self.__class__.__name__!r} instance has no attribute named data_stream."            )        return self._data_stream    @data_stream.setter    def data_stream(self, value: IO[bytes]) -> None:        """        A setter method that changes the value of the `data_stream` attribute.        :param value: A string that represents the value of the `data_stream` attribute.        :return: NoReturn.        """        setattr(self, "_data_stream", value)    @property    def wave_file(self) -> wave.Wave_write:        """        A getter method that returns the value of the `wave_file`attribute.        :param self: Instance of the class.        :return: A string that represents the value of the `wave_file` attribute.        """        if not hasattr(self, "_wave_file"):            raise AttributeError(                f"Your {self.__class__.__name__!r} instance has no attribute named wave_file."            )        return self._wave_file    @wave_file.setter    def wave_file(self, value: wave.Wave_write) -> None:        """        A setter method that changes the value of the `wave_file` attribute.        :param value: A string that represents the value of the `wave_file` attribute.        :return: NoReturn.        """        setattr(self, "_wave_file", value)    @property    def file_name(self) -> Union[str, PathLike[str]]:        """        A getter method that returns the value of the `file_name`attribute.        :param self: Instance of the class.        :return: A string that represents the value of the `file_name` attribute.        """        if not hasattr(self, "_mode"):            raise AttributeError(                f"Your {self.__class__.__name__!r} instance has no attribute named file_name."            )        return self._file_name    @file_name.setter    def file_name(self, value: Union[str, PathLike[str]]) -> None:        """        A setter method that changes the value of the `file_name` attribute.        :param value: A string that represents the value of the `file_name` attribute.        :return: NoReturn.        """        setattr(self, "_file_name", value)    @property    def mode(self) -> str:        """        A getter method that returns the value of the `mode`attribute.        :param self: Instance of the class.        :return: A string that represents the value of the `mode` attribute.        """        if not hasattr(self, "_mode"):            raise AttributeError(                f"Your {self.__class__.__name__!r} instance has no attribute named mode."            )        return self._mode    @mode.setter    def mode(self, value: str) -> None:        """        A setter method that changes the value of the `mode` attribute.        :param value: A string that represents the value of the `mode` attribute.        :return: NoReturn.        """        setattr(self, "_mode", value)    def __repr__(self) -> str:        attrs: dict = {            "frames_per_buffer": self.frames_per_buffer,            "audio_format": self.audio_format,            "channels": self.channels,            "rate": self.rate,            "py_audio": repr(self.py_audio),            "data_stream": self.data_stream,            "wave_file": repr(self.wave_file),            "mode": self.mode,            "file_name": self.file_name,        }        return f"{self.__class__.__name__}({attrs})"    def __attrs_post_init__(self) -> None:        wave_file = wave.open(os.path.join(BASE_DIR, self.file_name), self.mode)        wave_file.setnchannels(self.channels)        wave_file.setsampwidth(self.py_audio.get_sample_size(self.audio_format))        wave_file.setframerate(self.rate)        self.wave_file = wave_file        del wave_file    def record(self, duration: int = 3) -> None:        self.data_stream = self.py_audio.open(            format=self.audio_format,            channels=self.channels,            rate=self.rate,            input=True,            output=True,            frames_per_buffer=self.frames_per_buffer,        )        frames: List[bytes] = []        num_frames: int = int(self.rate / self.frames_per_buffer * duration)        for _ in range(num_frames):            data = self.data_stream.read(self.frames_per_buffer)            frames.append(data)        self.wave_file.writeframes(b"".join(frames))    def stop_recording(self) -> None:        if self.data_stream:            self.data_stream.close()            self.py_audio.terminate()            self.wave_file.close()if __name__ == "__main__":    rec = AudioRecorder()    print(rec)    rec.record()    rec.stop_recording()

>>> from deepgram import Deepgram>>> for attr in dir(Deepgram):...   if not attr.startswith("_"):...     print(attr)... keysprojectstranscriptionusage

$ export DEEPGRAM_API_KEY="XXXXXXXXX"

from deepgram import Deepgram # type:  ignoreimport asyncioimport osfrom os import PathLikefrom typing import Union, IOasync def transcribe(file_name: Union[Union[str, bytes, PathLike[str], PathLike[bytes]], int]):    with open(file_name, "rb") as audio:        source = {"buffer": audio, "mimetype": "audio/wav"}        response = await deepgram.transcription.prerecorded(source)        return response["results"]["channels"][0]["alternatives"][0]["words"]if __name__ == "__main__":    try:        deepgram = Deepgram(os.environ.get("DEEPGRAM_API_KEY"))        loop = asyncio.new_event_loop()        asyncio.set_event_loop(loop)        words = loop.run_until_complete(transcribe("sound.wav"))        string_words = " ".join(word_dict.get("word") for word_dict in words if "word" in word_dict)        print(f"You said: {string_words}!")        loop.close()    except AttributeError:        print("Please provide a valid `DEEPGRAM_API_KEY`.")

You said: hello world!

import wavefrom typing import List, Optional, TypeVar, Union, IOimport pyaudio #  type: ignorefrom deepgram import Deepgram # type:  ignoreimport asyncioimport osfrom os import PathLikeWaveWrite = TypeVar("WaveWrite", bound=wave.Wave_write)def init_recording(    file_name: Union[str, IO[bytes]] = "sound.wav", mode: Optional[str] = "wb") -> WaveWrite:    wave_file = wave.open(file_name, mode)    wave_file.setnchannels(2)    wave_file.setsampwidth(2)    wave_file.setframerate(44100)    return wave_filedef record(wave_file: WaveWrite, duration: Optional[int] = 3) -> None:    py_audio = pyaudio.PyAudio()    audio_stream = py_audio.open(        rate=44100,  # frames per second,        channels=2,  # stereo, change to 1 if you want mono        format=8,  # sample format, 8 bytes. see inspect        input=True,  # input device flag        frames_per_buffer=1024,  # 1024 samples per frame    )    frames = []    for _ in range(int(44100 / 1024 * 3)):        data = audio_stream.read(1024)        frames.append(data)    wave_file.writeframes(b"".join(frames))    audio_stream.close()async def transcribe(file_name: Union[Union[str, bytes, PathLike[str], PathLike[bytes]], int]):    with open(file_name, "rb") as audio:        source = {"buffer": audio, "mimetype": "audio/wav"}        response = await deepgram.transcription.prerecorded(source)        return response["results"]["channels"][0]["alternatives"][0]["words"]if __name__ == "__main__":    # start recording    print("Python is listening...")    wave_file = init_recording() #  type: ignore    record(wave_file)    wave_file.close()    # start transcribing    deepgram = Deepgram(os.environ.get("DEEPGRAM_API_KEY"))    loop = asyncio.new_event_loop()    asyncio.set_event_loop(loop)    words = loop.run_until_complete(transcribe("sound.wav"))    string_words = " ".join(word_dict.get("word") for word_dict in words if "word" in word_dict)    print(f"You said: {string_words}!")    loop.close()

    try:        # start recording        print("Python is listening...")        wave_file = init_recording() #  type: ignore        record(wave_file)        wave_file.close()        # start transcribing        deepgram = Deepgram(os.environ.get("DEEPGRAM_API_KEY"))        loop = asyncio.new_event_loop()        asyncio.set_event_loop(loop)        words = loop.run_until_complete(transcribe("sound.wav"))        string_words = " ".join(word_dict.get("word") for word_dict in words if "word" in word_dict)        print(f"You said: {string_words}!")        loop.close()    except Exception:        print("Unauthorized user. Please provide a valid `DEEPGRAM_API_KEY` value")

    try:        loop = asyncio.new_event_loop()        asyncio.set_event_loop(loop)        while True:            wave_file = init_recording() #  type: ignore            print("Python is listening...")            record(wave_file)            wave_file.close()            # start transcribing            deepgram = Deepgram(os.environ.get("DEEPGRAM_API_KEY"))            words = loop.run_until_complete(transcribe("sound.wav"))            string_words = " ".join(word_dict.get("word") for word_dict in words if "word" in word_dict)            print(f"You said: {string_words}!")            if string_words == "stop":              print('Goodbye!')              break        loop.close()    except Exception:        print("Unauthorized user. Please provide a valid `DEEPGRAM_API_KEY` value")