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 slave
ALSA lib pcm.c:2642:(snd_pcm_open_noupdate) Unknown PCM cards.pcm.rear
ALSA lib pcm.c:2642:(snd_pcm_open_noupdate) Unknown PCM cards.pcm.center_lfe
ALSA lib pcm.c:2642:(snd_pcm_open_noupdate) Unknown PCM cards.pcm.side
ALSA lib pcm_route.c:869:(find_matching_chmap) Found no matching channel map
ALSA lib pcm_oss.c:377:(_snd_pcm_oss_open) Unknown field port
ALSA lib pcm_oss.c:377:(_snd_pcm_oss_open) Unknown field port
ALSA lib pcm_usb_stream.c:486:(_snd_pcm_usb_stream_open) Invalid type for card
ALSA lib pcm_usb_stream.c:486:(_snd_pcm_usb_stream_open) Invalid type for card
ALSA 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)
... 
close
get_default_host_api_info
get_default_input_device_info
get_default_output_device_info
get_device_count
get_device_info_by_host_api_device_index
get_device_info_by_index
get_format_from_width
get_host_api_count
get_host_api_info_by_index
get_host_api_info_by_type
get_sample_size
is_format_supported
open
terminate

>>> 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)
... 
close
get_cpu_load
get_input_latency
get_output_latency
get_read_available
get_time
get_write_available
is_active
is_stopped
read
start_stream
stop_stream
write

>>> 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)
... 
Chunk
Error
WAVE_FORMAT_PCM
Wave_read
Wave_write
audioop
builtins
namedtuple
open
struct
sys

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

>>> for attr in dir(wave_file):
...   if not attr.startswith("_"):
...     print(attr)
... 
close
getcompname
getcomptype
getframerate
getmark
getmarkers
getnchannels
getnframes
getparams
getsampwidth
initfp
setcomptype
setframerate
setmark
setnchannels
setnframes
setparams
setsampwidth
tell
writeframes
writeframesraw

>>> 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 wave
from typing import List, Optional, TypeVar, Union, IO

import pyaudio #  type: ignore

WaveWrite = 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_file


def 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 os
import wave
from os import PathLike
from typing import IO, List, Optional, TypeVar, Union

import pyaudio  # type: ignore
from attrs import define, field

WaveWrite = TypeVar("WaveWrite", bound=wave.Wave_write)
BASE_DIR = os.path.dirname(os.path.abspath(__file__))


@define
class 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)
... 
keys
projects
transcription
usage

$ export DEEPGRAM_API_KEY="XXXXXXXXX"

from deepgram import Deepgram # type:  ignore
import asyncio
import os
from os import PathLike
from typing import Union, IO

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__":
    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 wave
from typing import List, Optional, TypeVar, Union, IO
import pyaudio #  type: ignore
from deepgram import Deepgram # type:  ignore
import asyncio
import os
from os import PathLike


WaveWrite = 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_file

def 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")