SpotifyAudioRecorder.py

import threading
import time
import wave

import pyaudiowpatch as pyaudio
from TheCodeLabs_BaseUtils.DefaultLogger import DefaultLogger

LOG_FORMAT = '[%(levelname)-7s] - %(asctime)s - %(message)s'
LOGGER = DefaultLogger().create_logger_if_not_exists('SpotifyAudioRecorder', logFormat=LOG_FORMAT)


class SpotifyAudioRecorder(threading.Thread):
    _CHUNK_SIZE = 512

    def __init__(self, deviceName: str, destinationFilePath: str):
        threading.Thread.__init__(self, daemon=True)
        self._deviceName = deviceName
        self._destinationFilePath = destinationFilePath
        self._stopEvent = threading.Event()

    def __get_device_by_name(self, pyaudioInstance, deviceName: str):
        deviceNames = []
        for loopback in pyaudioInstance.get_loopback_device_info_generator():
            deviceNames.append(loopback['name'])
            if deviceName in loopback['name']:
                return loopback
        else:
            message = f'Device with name {self._deviceName} not found.'
            message += f'\nAvailable devices:'
            for deviceName in deviceNames:
                message += f'\n\t{deviceName}'
            raise RuntimeError(message)

    def run(self):
        with pyaudio.PyAudio() as pyaudioInstance:
            device = self.__get_device_by_name(pyaudioInstance, self._deviceName)
            LOGGER.info(f'Recording from: ({device["index"]}) {device["name"]}')

            waveFile = wave.open(self._destinationFilePath, 'wb')
            waveFile.setnchannels(device['maxInputChannels'])
            waveFile.setsampwidth(pyaudio.get_sample_size(pyaudio.paInt16))
            waveFile.setframerate(int(device['defaultSampleRate']))

            def callback(in_data, frame_count, time_info, status):
                waveFile.writeframes(in_data)
                return in_data, pyaudio.paContinue

            with pyaudioInstance.open(format=pyaudio.paInt16,
                                      channels=device['maxInputChannels'],
                                      rate=int(device['defaultSampleRate']),
                                      frames_per_buffer=self._CHUNK_SIZE,
                                      input=True,
                                      input_device_index=device['index'],
                                      stream_callback=callback
                                      ):
                while not self._stopEvent.is_set():
                    time.sleep(1)

            waveFile.close()
            LOGGER.info(f'Recording stopped. File: "{self._destinationFilePath}"')

    def stop(self):
        if not self._stopEvent.is_set():
            self._stopEvent.set()

    def is_stopped(self) -> bool:
        return self._stopEvent.is_set()


if __name__ == '__main__':
    recorder = SpotifyAudioRecorder(
        deviceName='3/4 - Musik (2- GIGAPort HD Audio driver) [Loopback]',
        destinationFilePath='C:/Users/RobertG/Desktop/output.wav'
    )
    recorder.start()

    time.sleep(5)
    recorder.stop()