""" Python interface to the miniaudio library (https://github.com/dr-soft/miniaudio) Author: Irmen de Jong (irmen@razorvine.net) Software license: "MIT software license". See http://opensource.org/licenses/MIT """ __version__ = "1.71" import abc import sys import os import io import array import urllib.request import inspect import time import html import threading from enum import Enum from typing import Generator, List, Dict, Set, Optional, Union, Any, Callable from _miniaudio import ffi, lib try: import numpy except ImportError: numpy = None # type: ignore lib.init_miniaudio() class FileFormat(Enum): """Audio file format""" UNKNOWN = lib.ma_encoding_format_unknown WAV = lib.ma_encoding_format_wav FLAC = lib.ma_encoding_format_flac MP3 = lib.ma_encoding_format_mp3 VORBIS = lib.ma_encoding_format_vorbis class SampleFormat(Enum): """Sample format in memory""" UNKNOWN = lib.ma_format_unknown UNSIGNED8 = lib.ma_format_u8 SIGNED16 = lib.ma_format_s16 SIGNED24 = lib.ma_format_s24 SIGNED32 = lib.ma_format_s32 FLOAT32 = lib.ma_format_f32 class DeviceType(Enum): """Type of audio device""" PLAYBACK = lib.ma_device_type_playback CAPTURE = lib.ma_device_type_capture DUPLEX = lib.ma_device_type_duplex class DitherMode(Enum): """How to dither when converting""" NONE = lib.ma_dither_mode_none RECTANGLE = lib.ma_dither_mode_rectangle TRIANGLE = lib.ma_dither_mode_triangle class ChannelMixMode(Enum): """How to mix channels when converting""" RECTANGULAR = lib.ma_channel_mix_mode_rectangular SIMPLE = lib.ma_channel_mix_mode_simple CUSTOMWEIGHTS = lib.ma_channel_mix_mode_custom_weights DEFAULT = lib.ma_channel_mix_mode_default class Backend(Enum): """Operating system audio backend to use (only a subset will be available)""" WASAPI = lib.ma_backend_wasapi DSOUND = lib.ma_backend_dsound WINMM = lib.ma_backend_winmm COREAUDIO = lib.ma_backend_coreaudio SNDIO = lib.ma_backend_sndio AUDIO4 = lib.ma_backend_audio4 OSS = lib.ma_backend_oss PULSEAUDIO = lib.ma_backend_pulseaudio ALSA = lib.ma_backend_alsa JACK = lib.ma_backend_jack AAUDIO = lib.ma_backend_aaudio OPENSL = lib.ma_backend_opensl WEBAUDIO = lib.ma_backend_webaudio CUSTOM = lib.ma_backend_custom NULL = lib.ma_backend_null class ThreadPriority(Enum): """The priority of the worker thread (default=HIGHEST)""" IDLE = lib.ma_thread_priority_idle LOWEST = lib.ma_thread_priority_lowest LOW = lib.ma_thread_priority_low NORMAL = lib.ma_thread_priority_normal HIGH = lib.ma_thread_priority_high HIGHEST = lib.ma_thread_priority_highest REALTIME = lib.ma_thread_priority_realtime DEFAULT = lib.ma_thread_priority_default class SeekOrigin(Enum): """How to seek() in a source""" START = lib.ma_seek_origin_start CURRENT = lib.ma_seek_origin_current END = lib.ma_seek_origin_end FramesType = Union[bytes, array.array] PlaybackCallbackGeneratorType = Generator[FramesType, int, None] CaptureCallbackGeneratorType = Generator[None, FramesType, None] DuplexCallbackGeneratorType = Generator[FramesType, FramesType, None] GeneratorTypes = Union[ PlaybackCallbackGeneratorType, CaptureCallbackGeneratorType, DuplexCallbackGeneratorType, ] class SoundFileInfo: """Contains various properties of an audio file.""" def __init__( self, name: str, file_format: FileFormat, nchannels: int, sample_rate: int, sample_format: SampleFormat, duration: float, num_frames: int, sub_format: int = None, ) -> None: self.name = name self.nchannels = nchannels self.sample_rate = sample_rate self.sample_format = sample_format self.sample_format_name = ffi.string( lib.ma_get_format_name(sample_format.value) ).decode() self.sample_width = width_from_format(sample_format) self.num_frames = num_frames self.duration = duration self.file_format = file_format self.sub_format = sub_format def __str__(self) -> str: fileformatdisplay = self.file_format.name if self.sub_format: fileformatdisplay += " (fmt=" + str(self.sub_format) + ")" return ( "<{clazz}: '{name}' {fileformatdisplay} {nchannels} ch, {sample_rate} hz, {sample_format.name}, " "{num_frames} frames={duration:.2f} sec.>".format( clazz=self.__class__.__name__, fileformatdisplay=fileformatdisplay, **(vars(self)), ) ) def __repr__(self) -> str: return str(self) class DecodedSoundFile(SoundFileInfo): """Contains various properties and also the PCM frames of a fully decoded audio file.""" def __init__( self, name: str, nchannels: int, sample_rate: int, sample_format: SampleFormat, samples: array.array, ) -> None: num_frames = len(samples) // nchannels duration = num_frames / sample_rate super().__init__( name, FileFormat.UNKNOWN, nchannels, sample_rate, sample_format, duration, num_frames, ) self.samples = samples class MiniaudioError(Exception): """When a miniaudio specific error occurs.""" pass class DecodeError(MiniaudioError): """When something went wrong during decoding an audio file.""" pass def get_file_info(filename: str) -> SoundFileInfo: """Fetch some information about the audio file.""" ext = os.path.splitext(filename)[1].lower() if ext in (".ogg", ".vorbis"): return vorbis_get_file_info(filename) elif ext == ".mp3": return mp3_get_file_info(filename) elif ext == ".flac": return flac_get_file_info(filename) elif ext == ".wav": return wav_get_file_info(filename) raise DecodeError("unsupported file format") def read_file(filename: str, convert_to_16bit: bool = False) -> DecodedSoundFile: """Reads and decodes the whole audio file. Miniaudio will attempt to return the sound data in exactly the same format as in the file. Unless you set convert_convert_to_16bit to True, then the result is always a 16 bit sample format. """ ext = os.path.splitext(filename)[1].lower() if ext in (".ogg", ".vorbis"): if convert_to_16bit: return vorbis_read_file(filename) else: vorbis = vorbis_get_file_info(filename) if vorbis.sample_format == SampleFormat.SIGNED16: return vorbis_read_file(filename) else: raise MiniaudioError("file has sample format that must be converted") elif ext == ".mp3": if convert_to_16bit: return mp3_read_file_s16(filename) else: mp3 = mp3_get_file_info(filename) if mp3.sample_format == SampleFormat.SIGNED16: return mp3_read_file_s16(filename) elif mp3.sample_format == SampleFormat.FLOAT32: return mp3_read_file_f32(filename) else: raise MiniaudioError("file has sample format that must be converted") elif ext == ".flac": if convert_to_16bit: return flac_read_file_s16(filename) else: flac = flac_get_file_info(filename) if flac.sample_format == SampleFormat.SIGNED16: return flac_read_file_s16(filename) elif flac.sample_format == SampleFormat.SIGNED32: return flac_read_file_s32(filename) elif flac.sample_format == SampleFormat.FLOAT32: return flac_read_file_f32(filename) else: raise MiniaudioError("file has sample format that must be converted") elif ext == ".wav": if convert_to_16bit: return wav_read_file_s16(filename) else: wav = wav_get_file_info(filename) if wav.sample_format == SampleFormat.SIGNED16: return wav_read_file_s16(filename) elif wav.sample_format == SampleFormat.SIGNED32: return wav_read_file_s32(filename) elif wav.sample_format == SampleFormat.FLOAT32: return wav_read_file_f32(filename) else: raise MiniaudioError("file has sample format that must be converted") raise DecodeError("unsupported file format") def vorbis_get_file_info(filename: str) -> SoundFileInfo: """Fetch some information about the audio file (vorbis format).""" filenamebytes = _get_filename_bytes(filename) with ffi.new("int *") as error: vorbis = lib.stb_vorbis_open_filename(filenamebytes, error, ffi.NULL) if not vorbis: raise DecodeError("could not open/decode file") try: info = lib.stb_vorbis_get_info(vorbis) duration = lib.stb_vorbis_stream_length_in_seconds(vorbis) num_frames = lib.stb_vorbis_stream_length_in_samples(vorbis) return SoundFileInfo( filename, FileFormat.VORBIS, info.channels, info.sample_rate, SampleFormat.SIGNED16, duration, num_frames, ) finally: lib.stb_vorbis_close(vorbis) def vorbis_get_info(data: bytes) -> SoundFileInfo: """Fetch some information about the audio data (vorbis format).""" with ffi.new("int *") as error: vorbis = lib.stb_vorbis_open_memory(data, len(data), error, ffi.NULL) if not vorbis: raise DecodeError("could not open/decode data") try: info = lib.stb_vorbis_get_info(vorbis) duration = lib.stb_vorbis_stream_length_in_seconds(vorbis) num_frames = lib.stb_vorbis_stream_length_in_samples(vorbis) return SoundFileInfo( "", FileFormat.VORBIS, info.channels, info.sample_rate, SampleFormat.SIGNED16, duration, num_frames, ) finally: lib.stb_vorbis_close(vorbis) def vorbis_read_file(filename: str) -> DecodedSoundFile: """Reads and decodes the whole vorbis audio file. Resulting sample format is 16 bits signed integer.""" filenamebytes = _get_filename_bytes(filename) with ffi.new("int *") as channels, ffi.new("int *") as sample_rate, ffi.new( "short **" ) as output: num_frames = lib.stb_vorbis_decode_filename( filenamebytes, channels, sample_rate, output ) if num_frames <= 0: raise DecodeError("cannot load/decode file") try: buffer = ffi.buffer(output[0], num_frames * channels[0] * 2) samples = _create_int_array(2) samples.frombytes(buffer) return DecodedSoundFile( filename, channels[0], sample_rate[0], SampleFormat.SIGNED16, samples ) finally: lib.free(output[0]) def vorbis_read(data: bytes) -> DecodedSoundFile: """Reads and decodes the whole vorbis audio data. Resulting sample format is 16 bits signed integer.""" with ffi.new("int *") as channels, ffi.new("int *") as sample_rate, ffi.new( "short **" ) as output: num_samples = lib.stb_vorbis_decode_memory( data, len(data), channels, sample_rate, output ) if num_samples <= 0: raise DecodeError("cannot load/decode data") try: buffer = ffi.buffer(output[0], num_samples * channels[0] * 2) samples = _create_int_array(2) samples.frombytes(buffer) return DecodedSoundFile( "", channels[0], sample_rate[0], SampleFormat.SIGNED16, samples ) finally: lib.free(output[0]) def vorbis_stream_file( filename: str, seek_frame: int = 0 ) -> Generator[array.array, None, None]: """Streams the ogg vorbis audio file as interleaved 16 bit signed integer sample arrays segments. This uses a variable unconfigurable chunk size and cannot be used as a generic miniaudio decoder input stream. Consider using stream_file() instead.""" filenamebytes = _get_filename_bytes(filename) with ffi.new("int *") as error: vorbis = lib.stb_vorbis_open_filename(filenamebytes, error, ffi.NULL) if not vorbis: raise DecodeError("could not open/decode file") try: info = lib.stb_vorbis_get_info(vorbis) with ffi.new("short[]", 4096 * info.channels) as decode_buffer1, ffi.new( "short[]", 4096 * info.channels ) as decode_buffer2: decodebuf_ptr1 = ffi.cast("short *", decode_buffer1) decodebuf_ptr2 = ffi.cast("short *", decode_buffer2) if seek_frame > 0: result = lib.stb_vorbis_seek_frame(vorbis, seek_frame) if result <= 0: raise DecodeError("can't seek") # note: we decode several frames to reduce the overhead of very small sample sizes a little while True: num_samples1 = lib.stb_vorbis_get_frame_short_interleaved( vorbis, info.channels, decodebuf_ptr1, 4096 * info.channels ) num_samples2 = lib.stb_vorbis_get_frame_short_interleaved( vorbis, info.channels, decodebuf_ptr2, 4096 * info.channels ) if num_samples1 + num_samples2 <= 0: break buffer = ffi.buffer( decode_buffer1, num_samples1 * 2 * info.channels ) samples = _create_int_array(2) samples.frombytes(buffer) if num_samples2 > 0: buffer = ffi.buffer( decode_buffer2, num_samples2 * 2 * info.channels ) samples.frombytes(buffer) yield samples finally: lib.stb_vorbis_close(vorbis) def flac_get_file_info(filename: str) -> SoundFileInfo: """Fetch some information about the audio file (flac format).""" if sys.platform == "win32": filenamebytesw = _get_filename_bytes_w(filename) flac = lib.ma_dr_flac_open_file_w(filenamebytesw, ffi.NULL) else: filenamebytes = _get_filename_bytes(filename) flac = lib.ma_dr_flac_open_file(filenamebytes, ffi.NULL) if not flac: raise DecodeError("could not open/decode file") try: duration = flac.totalPCMFrameCount / flac.sampleRate sample_width = flac.bitsPerSample // 8 return SoundFileInfo( filename, FileFormat.FLAC, flac.channels, flac.sampleRate, _format_from_width(sample_width), duration, flac.totalPCMFrameCount, ) finally: lib.ma_dr_flac_close(flac) def flac_get_info(data: bytes) -> SoundFileInfo: """Fetch some information about the audio data (flac format).""" flac = lib.ma_dr_flac_open_memory(data, len(data), ffi.NULL) if not flac: raise DecodeError("could not open/decode data") try: duration = flac.totalPCMFrameCount / flac.sampleRate sample_width = flac.bitsPerSample // 8 return SoundFileInfo( "", FileFormat.FLAC, flac.channels, flac.sampleRate, _format_from_width(sample_width), duration, flac.totalPCMFrameCount, ) finally: lib.ma_dr_flac_close(flac) def flac_read_file_s32(filename: str) -> DecodedSoundFile: """Reads and decodes the whole flac audio file. Resulting sample format is 32 bits signed integer.""" filenamebytes = _get_filename_bytes(filename) with ffi.new("unsigned int *") as channels, ffi.new( "unsigned int *" ) as sample_rate, ffi.new("ma_dr_flac_uint64 *") as num_frames: memory = lib.ma_dr_flac_open_file_and_read_pcm_frames_s32( filenamebytes, channels, sample_rate, num_frames, ffi.NULL ) if not memory: raise DecodeError("cannot load/decode file") try: samples = _create_int_array(4) buffer = ffi.buffer(memory, num_frames[0] * channels[0] * 4) samples.frombytes(buffer) return DecodedSoundFile( filename, channels[0], sample_rate[0], SampleFormat.SIGNED32, samples ) finally: lib.ma_dr_flac_free(memory, ffi.NULL) def flac_read_file_s16(filename: str) -> DecodedSoundFile: """Reads and decodes the whole flac audio file. Resulting sample format is 16 bits signed integer.""" filenamebytes = _get_filename_bytes(filename) with ffi.new("unsigned int *") as channels, ffi.new( "unsigned int *" ) as sample_rate, ffi.new("ma_dr_flac_uint64 *") as num_frames: memory = lib.ma_dr_flac_open_file_and_read_pcm_frames_s16( filenamebytes, channels, sample_rate, num_frames, ffi.NULL ) if not memory: raise DecodeError("cannot load/decode file") try: samples = _create_int_array(2) buffer = ffi.buffer(memory, num_frames[0] * channels[0] * 2) samples.frombytes(buffer) return DecodedSoundFile( filename, channels[0], sample_rate[0], SampleFormat.SIGNED16, samples ) finally: lib.ma_dr_flac_free(memory, ffi.NULL) def flac_read_file_f32(filename: str) -> DecodedSoundFile: """Reads and decodes the whole flac audio file. Resulting sample format is 32 bits float.""" filenamebytes = _get_filename_bytes(filename) with ffi.new("unsigned int *") as channels, ffi.new( "unsigned int *" ) as sample_rate, ffi.new("ma_dr_flac_uint64 *") as num_frames: memory = lib.ma_dr_flac_open_file_and_read_pcm_frames_f32( filenamebytes, channels, sample_rate, num_frames, ffi.NULL ) if not memory: raise DecodeError("cannot load/decode file") try: samples = array.array("f") buffer = ffi.buffer(memory, num_frames[0] * channels[0] * 4) samples.frombytes(buffer) return DecodedSoundFile( filename, channels[0], sample_rate[0], SampleFormat.FLOAT32, samples ) finally: lib.ma_dr_flac_free(memory, ffi.NULL) def flac_read_s32(data: bytes) -> DecodedSoundFile: """Reads and decodes the whole flac audio data. Resulting sample format is 32 bits signed integer.""" with ffi.new("unsigned int *") as channels, ffi.new( "unsigned int *" ) as sample_rate, ffi.new("ma_dr_flac_uint64 *") as num_frames: memory = lib.ma_dr_flac_open_memory_and_read_pcm_frames_s32( data, len(data), channels, sample_rate, num_frames, ffi.NULL ) if not memory: raise DecodeError("cannot load/decode data") try: samples = _create_int_array(4) buffer = ffi.buffer(memory, num_frames[0] * channels[0] * 4) samples.frombytes(buffer) return DecodedSoundFile( "", channels[0], sample_rate[0], SampleFormat.SIGNED32, samples ) finally: lib.ma_dr_flac_free(memory, ffi.NULL) def flac_read_s16(data: bytes) -> DecodedSoundFile: """Reads and decodes the whole flac audio data. Resulting sample format is 16 bits signed integer.""" with ffi.new("unsigned int *") as channels, ffi.new( "unsigned int *" ) as sample_rate, ffi.new("ma_dr_flac_uint64 *") as num_frames: memory = lib.ma_dr_flac_open_memory_and_read_pcm_frames_s16( data, len(data), channels, sample_rate, num_frames, ffi.NULL ) if not memory: raise DecodeError("cannot load/decode data") try: samples = _create_int_array(2) buffer = ffi.buffer(memory, num_frames[0] * channels[0] * 2) samples.frombytes(buffer) return DecodedSoundFile( "", channels[0], sample_rate[0], SampleFormat.SIGNED16, samples ) finally: lib.ma_dr_flac_free(memory, ffi.NULL) def flac_read_f32(data: bytes) -> DecodedSoundFile: """Reads and decodes the whole flac audio file. Resulting sample format is 32 bits float.""" with ffi.new("unsigned int *") as channels, ffi.new( "unsigned int *" ) as sample_rate, ffi.new("ma_dr_flac_uint64 *") as num_frames: memory = lib.ma_dr_flac_open_memory_and_read_pcm_frames_f32( data, len(data), channels, sample_rate, num_frames, ffi.NULL ) if not memory: raise DecodeError("cannot load/decode data") try: samples = array.array("f") buffer = ffi.buffer(memory, num_frames[0] * channels[0] * 4) samples.frombytes(buffer) return DecodedSoundFile( "", channels[0], sample_rate[0], SampleFormat.FLOAT32, samples ) finally: lib.ma_dr_flac_free(memory, ffi.NULL) def flac_stream_file( filename: str, frames_to_read: int = 1024, seek_frame: int = 0 ) -> Generator[array.array, None, None]: """Streams the flac audio file as interleaved 16 bit signed integer sample arrays segments. This uses a fixed chunk size and cannot be used as a generic miniaudio decoder input stream. Consider using stream_file() instead.""" if sys.platform == "win32": filenamebytesw = _get_filename_bytes_w(filename) flac = lib.ma_dr_flac_open_file_w(filenamebytesw, ffi.NULL) else: filenamebytes = _get_filename_bytes(filename) flac = lib.ma_dr_flac_open_file(filenamebytes, ffi.NULL) if not flac: raise DecodeError("could not open/decode file") if seek_frame > 0: result = lib.ma_dr_flac_seek_to_pcm_frame(flac, seek_frame) if result <= 0: raise DecodeError("can't seek") try: with ffi.new( "ma_dr_flac_int16[]", frames_to_read * flac.channels ) as decodebuffer: buf_ptr = ffi.cast("ma_dr_flac_int16 *", decodebuffer) while True: num_samples = lib.ma_dr_flac_read_pcm_frames_s16( flac, frames_to_read, buf_ptr ) if num_samples <= 0: break buffer = ffi.buffer(decodebuffer, num_samples * 2 * flac.channels) samples = _create_int_array(2) samples.frombytes(buffer) yield samples finally: lib.ma_dr_flac_close(flac) def mp3_get_file_info(filename: str) -> SoundFileInfo: """Fetch some information about the audio file (mp3 format).""" with ffi.new("ma_dr_mp3 *") as mp3: if sys.platform == "win32": filenamebytesw = _get_filename_bytes_w(filename) result = lib.ma_dr_mp3_init_file_w(mp3, filenamebytesw, ffi.NULL) else: filenamebytes = _get_filename_bytes(filename) result = lib.ma_dr_mp3_init_file(mp3, filenamebytes, ffi.NULL) if not result: raise DecodeError("could not open/decode file") try: num_frames = lib.ma_dr_mp3_get_pcm_frame_count(mp3) duration = num_frames / mp3.sampleRate return SoundFileInfo( filename, FileFormat.MP3, mp3.channels, mp3.sampleRate, SampleFormat.SIGNED16, duration, num_frames, ) finally: lib.ma_dr_mp3_uninit(mp3) def mp3_get_info(data: bytes) -> SoundFileInfo: """Fetch some information about the audio data (mp3 format).""" with ffi.new("ma_dr_mp3 *") as mp3: if not lib.ma_dr_mp3_init_memory(mp3, data, len(data), ffi.NULL): raise DecodeError("could not open/decode data") try: num_frames = lib.ma_dr_mp3_get_pcm_frame_count(mp3) duration = num_frames / mp3.sampleRate return SoundFileInfo( "", FileFormat.MP3, mp3.channels, mp3.sampleRate, SampleFormat.SIGNED16, duration, num_frames, ) finally: lib.ma_dr_mp3_uninit(mp3) def mp3_read_file_f32(filename: str) -> DecodedSoundFile: """Reads and decodes the whole mp3 audio file. Resulting sample format is 32 bits float.""" filenamebytes = _get_filename_bytes(filename) with ffi.new("ma_dr_mp3_config *") as config, ffi.new( "ma_dr_mp3_uint64 *" ) as num_frames: memory = lib.ma_dr_mp3_open_file_and_read_pcm_frames_f32( filenamebytes, config, num_frames, ffi.NULL ) if not memory: raise DecodeError("cannot load/decode file") try: samples = array.array("f") buffer = ffi.buffer(memory, num_frames[0] * config.channels * 4) samples.frombytes(buffer) return DecodedSoundFile( filename, config.channels, config.sampleRate, SampleFormat.FLOAT32, samples, ) finally: lib.ma_dr_mp3_free(memory, ffi.NULL) def mp3_read_file_s16(filename: str) -> DecodedSoundFile: """Reads and decodes the whole mp3 audio file. Resulting sample format is 16 bits signed integer.""" filenamebytes = _get_filename_bytes(filename) with ffi.new("ma_dr_mp3_config *") as config, ffi.new( "ma_dr_mp3_uint64 *" ) as num_frames: memory = lib.ma_dr_mp3_open_file_and_read_pcm_frames_s16( filenamebytes, config, num_frames, ffi.NULL ) if not memory: raise DecodeError("cannot load/decode file") try: samples = _create_int_array(2) buffer = ffi.buffer(memory, num_frames[0] * config.channels * 2) samples.frombytes(buffer) return DecodedSoundFile( filename, config.channels, config.sampleRate, SampleFormat.SIGNED16, samples, ) finally: lib.ma_dr_mp3_free(memory, ffi.NULL) def mp3_read_f32(data: bytes) -> DecodedSoundFile: """Reads and decodes the whole mp3 audio data. Resulting sample format is 32 bits float.""" with ffi.new("ma_dr_mp3_config *") as config, ffi.new( "ma_dr_mp3_uint64 *" ) as num_frames: memory = lib.ma_dr_mp3_open_memory_and_read_pcm_frames_f32( data, len(data), config, num_frames, ffi.NULL ) if not memory: raise DecodeError("cannot load/decode data") try: samples = array.array("f") buffer = ffi.buffer(memory, num_frames[0] * config.channels * 4) samples.frombytes(buffer) return DecodedSoundFile( "", config.channels, config.sampleRate, SampleFormat.FLOAT32, samples, ) finally: lib.ma_dr_mp3_free(memory, ffi.NULL) def mp3_read_s16(data: bytes) -> DecodedSoundFile: """Reads and decodes the whole mp3 audio data. Resulting sample format is 16 bits signed integer.""" with ffi.new("ma_dr_mp3_config *") as config, ffi.new( "ma_dr_mp3_uint64 *" ) as num_frames: memory = lib.ma_dr_mp3_open_memory_and_read_pcm_frames_s16( data, len(data), config, num_frames, ffi.NULL ) if not memory: raise DecodeError("cannot load/decode data") try: samples = _create_int_array(2) buffer = ffi.buffer(memory, num_frames[0] * config.channels * 2) samples.frombytes(buffer) return DecodedSoundFile( "", config.channels, config.sampleRate, SampleFormat.SIGNED16, samples, ) finally: lib.ma_dr_mp3_free(memory, ffi.NULL) def mp3_stream_file( filename: str, frames_to_read: int = 1024, seek_frame: int = 0 ) -> Generator[array.array, None, None]: """Streams the mp3 audio file as interleaved 16 bit signed integer sample arrays segments. This uses a fixed chunk size and cannot be used as a generic miniaudio decoder input stream. Consider using stream_file() instead.""" with ffi.new("ma_dr_mp3 *") as mp3: if sys.platform == "win32": filenamebytesw = _get_filename_bytes_w(filename) result = lib.ma_dr_mp3_init_file_w(mp3, filenamebytesw, ffi.NULL) else: filenamebytes = _get_filename_bytes(filename) result = lib.ma_dr_mp3_init_file(mp3, filenamebytes, ffi.NULL) if not result: raise DecodeError("could not open/decode file") if seek_frame > 0: result = lib.ma_dr_mp3_seek_to_pcm_frame(mp3, seek_frame) if result <= 0: raise DecodeError("can't seek") try: with ffi.new( "ma_dr_mp3_int16[]", frames_to_read * mp3.channels ) as decodebuffer: buf_ptr = ffi.cast("ma_dr_mp3_int16 *", decodebuffer) while True: num_samples = lib.ma_dr_mp3_read_pcm_frames_s16( mp3, frames_to_read, buf_ptr ) if num_samples <= 0: break buffer = ffi.buffer(decodebuffer, num_samples * 2 * mp3.channels) samples = _create_int_array(2) samples.frombytes(buffer) yield samples finally: lib.ma_dr_mp3_uninit(mp3) def wav_get_file_info(filename: str) -> SoundFileInfo: """Fetch some information about the audio file (wav format).""" with ffi.new("ma_dr_wav*") as wav: if sys.platform == "win32": # Windows filepaths with non-ascii characters fail to open using fopen, cast filenames to wchar_t* filenamebytesw = _get_filename_bytes_w(filename) result = lib.ma_dr_wav_init_file_w(wav, filenamebytesw, ffi.NULL) else: filenamebytes = _get_filename_bytes(filename) result = lib.ma_dr_wav_init_file(wav, filenamebytes, ffi.NULL) if not result: raise DecodeError("could not open/decode file") try: duration = wav.totalPCMFrameCount / wav.sampleRate sample_width = wav.bitsPerSample // 8 is_float = wav.translatedFormatTag == lib.MA_DR_WAVE_FORMAT_IEEE_FLOAT subformat = wav.translatedFormatTag return SoundFileInfo( filename, FileFormat.WAV, wav.channels, wav.sampleRate, _format_from_width(sample_width, is_float), duration, wav.totalPCMFrameCount, sub_format=subformat, ) finally: lib.ma_dr_wav_uninit(wav) def wav_get_info(data: bytes) -> SoundFileInfo: """Fetch some information about the audio data (wav format).""" with ffi.new("ma_dr_wav*") as wav: if not lib.ma_dr_wav_init_memory(wav, data, len(data), ffi.NULL): raise DecodeError("could not open/decode data") try: duration = wav.totalPCMFrameCount / wav.sampleRate sample_width = wav.bitsPerSample // 8 is_float = wav.translatedFormatTag == lib.MA_DR_WAVE_FORMAT_IEEE_FLOAT return SoundFileInfo( "", FileFormat.WAV, wav.channels, wav.sampleRate, _format_from_width(sample_width, is_float), duration, wav.totalPCMFrameCount, ) finally: lib.ma_dr_wav_uninit(wav) def wav_read_file_s32(filename: str) -> DecodedSoundFile: """Reads and decodes the whole wav audio file. Resulting sample format is 32 bits signed integer.""" filenamebytes = _get_filename_bytes(filename) with ffi.new("unsigned int *") as channels, ffi.new( "unsigned int *" ) as sample_rate, ffi.new("ma_dr_wav_uint64 *") as num_frames: memory = lib.ma_dr_wav_open_file_and_read_pcm_frames_s32( filenamebytes, channels, sample_rate, num_frames, ffi.NULL ) if not memory: raise DecodeError("cannot load/decode file") try: samples = _create_int_array(4) buffer = ffi.buffer(memory, num_frames[0] * channels[0] * 4) samples.frombytes(buffer) return DecodedSoundFile( filename, channels[0], sample_rate[0], SampleFormat.SIGNED32, samples ) finally: lib.ma_dr_wav_free(memory, ffi.NULL) def wav_read_file_s16(filename: str) -> DecodedSoundFile: """Reads and decodes the whole wav audio file. Resulting sample format is 16 bits signed integer.""" filenamebytes = _get_filename_bytes(filename) with ffi.new("unsigned int *") as channels, ffi.new( "unsigned int *" ) as sample_rate, ffi.new("ma_dr_wav_uint64 *") as num_frames: memory = lib.ma_dr_wav_open_file_and_read_pcm_frames_s16( filenamebytes, channels, sample_rate, num_frames, ffi.NULL ) if not memory: raise DecodeError("cannot load/decode file") try: samples = _create_int_array(2) buffer = ffi.buffer(memory, num_frames[0] * channels[0] * 2) samples.frombytes(buffer) return DecodedSoundFile( filename, channels[0], sample_rate[0], SampleFormat.SIGNED16, samples ) finally: lib.ma_dr_wav_free(memory, ffi.NULL) def wav_read_file_f32(filename: str) -> DecodedSoundFile: """Reads and decodes the whole wav audio file. Resulting sample format is 32 bits float.""" filenamebytes = _get_filename_bytes(filename) with ffi.new("unsigned int *") as channels, ffi.new( "unsigned int *" ) as sample_rate, ffi.new("ma_dr_wav_uint64 *") as num_frames: memory = lib.ma_dr_wav_open_file_and_read_pcm_frames_f32( filenamebytes, channels, sample_rate, num_frames, ffi.NULL ) if not memory: raise DecodeError("cannot load/decode file") try: samples = array.array("f") buffer = ffi.buffer(memory, num_frames[0] * channels[0] * 4) samples.frombytes(buffer) return DecodedSoundFile( filename, channels[0], sample_rate[0], SampleFormat.FLOAT32, samples ) finally: lib.ma_dr_wav_free(memory, ffi.NULL) def wav_read_s32(data: bytes) -> DecodedSoundFile: """Reads and decodes the whole wav audio data. Resulting sample format is 32 bits signed integer.""" with ffi.new("unsigned int *") as channels, ffi.new( "unsigned int *" ) as sample_rate, ffi.new("ma_dr_wav_uint64 *") as num_frames: memory = lib.ma_dr_wav_open_memory_and_read_pcm_frames_s32( data, len(data), channels, sample_rate, num_frames, ffi.NULL ) if not memory: raise DecodeError("cannot load/decode data") try: samples = _create_int_array(4) buffer = ffi.buffer(memory, num_frames[0] * channels[0] * 4) samples.frombytes(buffer) return DecodedSoundFile( "", channels[0], sample_rate[0], SampleFormat.SIGNED32, samples ) finally: lib.ma_dr_wav_free(memory, ffi.NULL) def wav_read_s16(data: bytes) -> DecodedSoundFile: """Reads and decodes the whole wav audio data. Resulting sample format is 16 bits signed integer.""" with ffi.new("unsigned int *") as channels, ffi.new( "unsigned int *" ) as sample_rate, ffi.new("ma_dr_wav_uint64 *") as num_frames: memory = lib.ma_dr_wav_open_memory_and_read_pcm_frames_s16( data, len(data), channels, sample_rate, num_frames, ffi.NULL ) if not memory: raise DecodeError("cannot load/decode data") try: samples = _create_int_array(2) buffer = ffi.buffer(memory, num_frames[0] * channels[0] * 2) samples.frombytes(buffer) return DecodedSoundFile( "", channels[0], sample_rate[0], SampleFormat.SIGNED16, samples ) finally: lib.ma_dr_wav_free(memory, ffi.NULL) def wav_read_f32(data: bytes) -> DecodedSoundFile: """Reads and decodes the whole wav audio data. Resulting sample format is 32 bits float.""" with ffi.new("unsigned int *") as channels, ffi.new( "unsigned int *" ) as sample_rate, ffi.new("ma_dr_wav_uint64 *") as num_frames: memory = lib.ma_dr_wav_open_memory_and_read_pcm_frames_f32( data, len(data), channels, sample_rate, num_frames, ffi.NULL ) if not memory: raise DecodeError("cannot load/decode data") try: samples = array.array("f") buffer = ffi.buffer(memory, num_frames[0] * channels[0] * 4) samples.frombytes(buffer) return DecodedSoundFile( "", channels[0], sample_rate[0], SampleFormat.FLOAT32, samples ) finally: lib.ma_dr_wav_free(memory, ffi.NULL) def wav_stream_file( filename: str, frames_to_read: int = 1024, seek_frame: int = 0 ) -> Generator[array.array, None, None]: """Streams the WAV audio file as interleaved 16 bit signed integer sample arrays segments. This uses a fixed chunk size and cannot be used as a generic miniaudio decoder input stream. Consider using stream_file() instead.""" with ffi.new("ma_dr_wav*") as wav: if sys.platform == "win32": # Windows filepaths with non-ascii characters fail to open using fopen, cast filenames to wchar_t* filenamebytesw = _get_filename_bytes_w(filename) result = lib.ma_dr_wav_init_file_w(wav, filenamebytesw, ffi.NULL) else: filenamebytes = _get_filename_bytes(filename) result = lib.ma_dr_wav_init_file(wav, filenamebytes, ffi.NULL) if not result: raise DecodeError("could not open/decode file") if seek_frame > 0: result = lib.ma_dr_wav_seek_to_pcm_frame(wav, seek_frame) if result <= 0: raise DecodeError("can't seek") try: with ffi.new( "ma_dr_wav_int16[]", frames_to_read * wav.channels ) as decodebuffer: buf_ptr = ffi.cast("ma_dr_wav_int16 *", decodebuffer) while True: num_samples = lib.ma_dr_wav_read_pcm_frames_s16( wav, frames_to_read, buf_ptr ) if num_samples <= 0: break buffer = ffi.buffer(decodebuffer, num_samples * 2 * wav.channels) samples = _create_int_array(2) samples.frombytes(buffer) yield samples finally: lib.ma_dr_wav_uninit(wav) def wav_write_file(filename: str, sound: DecodedSoundFile) -> None: """Writes the pcm sound to a WAV file""" with ffi.new("ma_dr_wav_data_format*") as fmt, ffi.new("ma_dr_wav*") as pwav: fmt.container = lib.ma_dr_wav_container_riff fmt.format = sound.sub_format or lib.MA_DR_WAVE_FORMAT_PCM fmt.channels = sound.nchannels fmt.sampleRate = sound.sample_rate fmt.bitsPerSample = sound.sample_width * 8 # what about floating point format? filename_bytes = filename.encode(sys.getfilesystemencoding()) if not lib.ma_dr_wav_init_file_write_sequential( pwav, filename_bytes, fmt, sound.num_frames * sound.nchannels, ffi.NULL ): raise IOError("can't open file for writing") try: lib.ma_dr_wav_write_pcm_frames( pwav, sound.num_frames, sound.samples.tobytes() ) finally: lib.ma_dr_wav_uninit(pwav) def _create_int_array(itemsize: int) -> array.array: for typecode in "Bhilq": a = array.array(typecode) if a.itemsize == itemsize: return a raise ValueError("cannot create array") def _get_filename_bytes(filename: str) -> bytes: filename2 = os.path.expanduser(filename) if not os.path.isfile(filename2): raise FileNotFoundError(filename) return filename2.encode(sys.getfilesystemencoding()) def _get_filename_bytes_w(filename: str): filename2 = os.path.expanduser(filename) if not os.path.isfile(filename2): raise FileNotFoundError(filename) return ffi.new("wchar_t[]", filename2) class Devices: """Query the audio playback and record devices that miniaudio provides""" def __init__(self, backends: Optional[List[Backend]] = None) -> None: self._context = ffi.NULL context = ffi.new("ma_context*") if backends: backends_mem = ffi.new("ma_backend[]", len(backends)) for i, b in enumerate(backends): backends_mem[i] = b.value result = lib.ma_context_init(backends_mem, len(backends), ffi.NULL, context) else: result = lib.ma_context_init(ffi.NULL, 0, ffi.NULL, context) if result != lib.MA_SUCCESS: raise MiniaudioError("cannot init context", result) self._context = context self.backend = ffi.string( lib.ma_get_backend_name(self._context[0].backend) ).decode() def get_playbacks(self) -> List[Dict[str, Any]]: """Get a list of playback devices and some details about them""" with ffi.new("ma_device_info**") as playback_infos, ffi.new( "ma_uint32*" ) as playback_count: result = lib.ma_context_get_devices( self._context, playback_infos, playback_count, ffi.NULL, ffi.NULL ) if result != lib.MA_SUCCESS: raise MiniaudioError("cannot get device infos", result) devs = [] for i in range(playback_count[0]): ma_device_info = playback_infos[0][i] dev_id = ffi.new( "ma_device_id *", ma_device_info.id ) # copy the id memory info = { "name": ffi.string(ma_device_info.name).decode(), "type": DeviceType.PLAYBACK, "id": dev_id, } info.update(self._get_info(DeviceType.PLAYBACK, ma_device_info)) devs.append(info) return devs def get_captures(self) -> List[Dict[str, Any]]: """Get a list of capture devices and some details about them""" with ffi.new("ma_device_info**") as capture_infos, ffi.new( "ma_uint32*" ) as capture_count: result = lib.ma_context_get_devices( self._context, ffi.NULL, ffi.NULL, capture_infos, capture_count ) if result != lib.MA_SUCCESS: raise MiniaudioError("cannot get device infos", result) devs = [] for i in range(capture_count[0]): ma_device_info = capture_infos[0][i] dev_id = ffi.new( "ma_device_id *", ma_device_info.id ) # copy the id memory info = { "name": ffi.string(ma_device_info.name).decode(), "type": DeviceType.CAPTURE, "id": dev_id, } info.update(self._get_info(DeviceType.CAPTURE, ma_device_info)) devs.append(info) return devs def _get_info( self, device_type: DeviceType, device_info: ffi.CData ) -> Dict[str, Any]: # obtain detailed info about the device result = lib.ma_context_get_device_info( self._context, device_type.value, ffi.addressof(device_info.id), ffi.addressof(device_info), ) if result != lib.MA_SUCCESS: raise MiniaudioError("can't get device info") formats = [] for fmt in range(device_info.nativeDataFormatCount): data_format = device_info.nativeDataFormats[fmt] formats.append( { "format": ffi.string( lib.ma_get_format_name(data_format.format) ).decode(), "samplerate": data_format.sampleRate, "channels": data_format.channels, } ) return {"formats": formats} def __del__(self): lib.ma_context_uninit(self._context) def width_from_format(sampleformat: SampleFormat) -> int: """returns the sample width in bytes, of the given sample format.""" widths = { SampleFormat.UNKNOWN: 0, SampleFormat.UNSIGNED8: 1, SampleFormat.SIGNED16: 2, SampleFormat.SIGNED24: 3, SampleFormat.SIGNED32: 4, SampleFormat.FLOAT32: 4, } if sampleformat in widths: return widths[sampleformat] raise MiniaudioError("unsupported sample format", sampleformat) def _array_proto_from_format(sampleformat: SampleFormat) -> array.array: arrays = { SampleFormat.UNSIGNED8: _create_int_array(1), SampleFormat.SIGNED16: _create_int_array(2), SampleFormat.SIGNED32: _create_int_array(4), SampleFormat.FLOAT32: array.array("f"), } if sampleformat in arrays: return arrays[sampleformat] raise MiniaudioError( "the requested sample format can not be used directly: " + sampleformat.name + " (convert it first)" ) def _format_from_width(sample_width: int, is_float: bool = False) -> SampleFormat: if is_float: return SampleFormat.FLOAT32 elif sample_width == 1: return SampleFormat.UNSIGNED8 elif sample_width == 2: return SampleFormat.SIGNED16 elif sample_width == 3: return SampleFormat.SIGNED24 elif sample_width == 4: return SampleFormat.SIGNED32 elif sample_width == 0: return SampleFormat.UNKNOWN else: raise MiniaudioError("unsupported sample width", sample_width) def decode_file( filename: str, output_format: SampleFormat = SampleFormat.SIGNED16, nchannels: int = 2, sample_rate: int = 44100, dither: DitherMode = DitherMode.NONE, ) -> DecodedSoundFile: """Convenience function to decode any supported audio file to raw PCM samples in your chosen format.""" sample_width = width_from_format(output_format) samples = _array_proto_from_format(output_format) filenamebytes = _get_filename_bytes(filename) with ffi.new("ma_uint64 *") as frames, ffi.new("void **") as memory: decoder_config = lib.ma_decoder_config_init( output_format.value, nchannels, sample_rate ) decoder_config.ditherMode = dither.value result = lib.ma_decode_file( filenamebytes, ffi.addressof(decoder_config), frames, memory ) if result != lib.MA_SUCCESS: raise DecodeError("failed to decode file", result) buffer = ffi.buffer(memory[0], frames[0] * nchannels * sample_width) samples.frombytes(buffer) lib.ma_free(memory[0], ffi.NULL) return DecodedSoundFile( filename, nchannels, sample_rate, output_format, samples ) def decode( data: bytes, output_format: SampleFormat = SampleFormat.SIGNED16, nchannels: int = 2, sample_rate: int = 44100, dither: DitherMode = DitherMode.NONE, ) -> DecodedSoundFile: """Convenience function to decode any supported audio file in memory to raw PCM samples in your chosen format.""" sample_width = width_from_format(output_format) samples = _array_proto_from_format(output_format) with ffi.new("ma_uint64 *") as frames, ffi.new("void **") as memory: decoder_config = lib.ma_decoder_config_init( output_format.value, nchannels, sample_rate ) decoder_config.ditherMode = dither.value result = lib.ma_decode_memory( data, len(data), ffi.addressof(decoder_config), frames, memory ) if result != lib.MA_SUCCESS: raise DecodeError("failed to decode data", result) buffer = ffi.buffer(memory[0], frames[0] * nchannels * sample_width) samples.frombytes(buffer) lib.ma_free(memory[0], ffi.NULL) return DecodedSoundFile( "", nchannels, sample_rate, output_format, samples ) def _samples_stream_generator( frames_to_read: int, nchannels: int, output_format: SampleFormat, decoder: ffi.CData, data: Any, on_close: Optional[Callable] = None, ) -> Generator[array.array, int, None]: _reference = data # make sure any data passed in is not garbage collected sample_width = width_from_format(output_format) samples_proto = _array_proto_from_format(output_format) allocated_buffer_frames = max(frames_to_read, 16384) try: with ffi.new( "int8_t[]", allocated_buffer_frames * nchannels * sample_width ) as decodebuffer: buf_ptr = ffi.cast("void *", decodebuffer) want_frames = (yield samples_proto) or frames_to_read source = None # type: Optional[StreamableSource] if decoder.pUserData != ffi.NULL: source = ffi.from_handle(decoder.pUserData) while True: if want_frames > allocated_buffer_frames: raise MiniaudioError( "wanted to read more frames than storage was allocated for ({} vs {})".format( want_frames, allocated_buffer_frames ) ) num_frames = 0 with ffi.new("ma_uint64 *") as frames_read: try: result = lib.ma_decoder_read_pcm_frames( decoder, buf_ptr, want_frames, frames_read ) except Exception as x: raise DecodeError("error in ma_decoder_read_pcm_frames") from x else: if result == lib.MA_SUCCESS: num_frames = frames_read[0] elif result == lib.MA_AT_END: break else: raise DecodeError("error in ma_decoder_read_pcm_frames") if num_frames <= 0: break if source and source.error_in_readcallback: raise DecodeError( "error in read callback" ) from source.error_in_readcallback buffer = ffi.buffer(decodebuffer, num_frames * sample_width * nchannels) samples = array.array(samples_proto.typecode) samples.frombytes(buffer) want_frames = (yield samples) or frames_to_read finally: if on_close: on_close() lib.ma_decoder_uninit(decoder) def stream_file( filename: str, output_format: SampleFormat = SampleFormat.SIGNED16, nchannels: int = 2, sample_rate: int = 44100, frames_to_read: int = 1024, dither: DitherMode = DitherMode.NONE, seek_frame: int = 0, ) -> Generator[array.array, int, None]: """ Convenience generator function to decode and stream any supported audio file as chunks of raw PCM samples in the chosen format. If you send() a number into the generator rather than just using next() on it, you'll get that given number of frames, instead of the default configured amount. This is particularly useful to plug this stream into an audio device callback that wants a variable number of frames per call. """ decoder = ffi.new("ma_decoder *") decoder_config = lib.ma_decoder_config_init( output_format.value, nchannels, sample_rate ) decoder_config.ditherMode = dither.value if sys.platform == "win32": filenamebytesw = _get_filename_bytes_w(filename) result = lib.ma_decoder_init_file_w( filenamebytesw, ffi.addressof(decoder_config), decoder ) else: filenamebytes = _get_filename_bytes(filename) result = lib.ma_decoder_init_file( filenamebytes, ffi.addressof(decoder_config), decoder ) if result != lib.MA_SUCCESS: raise DecodeError("failed to init decoder", result) if seek_frame > 0: result = lib.ma_decoder_seek_to_pcm_frame(decoder, seek_frame) if result != lib.MA_SUCCESS: raise DecodeError("failed to seek to frame", result) g = _samples_stream_generator( frames_to_read, nchannels, output_format, decoder, None ) dummy = next(g) assert len(dummy) == 0 return g def stream_memory( data: bytes, output_format: SampleFormat = SampleFormat.SIGNED16, nchannels: int = 2, sample_rate: int = 44100, frames_to_read: int = 1024, dither: DitherMode = DitherMode.NONE, ) -> Generator[array.array, int, None]: """ Convenience generator function to decode and stream any supported audio file in memory as chunks of raw PCM samples in the chosen format. If you send() a number into the generator rather than just using next() on it, you'll get that given number of frames, instead of the default configured amount. This is particularly useful to plug this stream into an audio device callback that wants a variable number of frames per call. """ decoder = ffi.new("ma_decoder *") decoder_config = lib.ma_decoder_config_init( output_format.value, nchannels, sample_rate ) decoder_config.ditherMode = dither.value result = lib.ma_decoder_init_memory( data, len(data), ffi.addressof(decoder_config), decoder ) if result != lib.MA_SUCCESS: raise DecodeError("failed to init decoder", result) g = _samples_stream_generator( frames_to_read, nchannels, output_format, decoder, data ) dummy = next(g) assert len(dummy) == 0 return g def stream_raw_pcm_memory( pcmdata: Union[array.array, memoryview, bytes], nchannels: int, sample_width: int, frames_to_read: int = 4096, ) -> PlaybackCallbackGeneratorType: """ Convenience generator function to stream raw pcm audio data from memory. Usually you don't need to use this as the library provides many other streaming options that work on much smaller, encoded, audio data. However, in the odd case that you only have already decoded raw pcm data you can use this generator as a stream source. The data can be provided in ``array`` type or ``bytes``, ``memoryview`` or even a numpy array. Be sure to also specify the correct number of channels that the audio data has, and the sample with in bytes. """ def _mem_stream_gen() -> PlaybackCallbackGeneratorType: nonlocal sample_width # sample_width should be provided if the data is not an array.array, but a bytes type instead. if type(pcmdata) is array.array: sample_width = 1 # array.array frames already yield the correct data size memdata = memoryview(pcmdata) required_frames = (yield b"") or frames_to_read # generator initialization frames = 0 while frames < len(memdata): frames_end = frames + required_frames * nchannels * sample_width required_frames = (yield memdata[frames:frames_end]) or frames_to_read frames = frames_end g = _mem_stream_gen() dummy = next(g) # start the generator assert len(dummy) == 0 return g class StreamableSource(abc.ABC): """Base class for streams of audio data bytes. Can be used as a contextmanager, to properly call close().""" ffi_handle = ffi.NULL # can be set later error_in_readcallback = None # type: Exception @abc.abstractmethod def read(self, num_bytes: int) -> Union[bytes, memoryview]: """override this to provide data bytes to the consumer of the stream""" pass def seek(self, offset: int, origin: SeekOrigin) -> bool: """ Override this if the stream supports seeking. Note: seek support is sometimes not needed if you give the file type to a decoder upfront. You can ignore this method then. """ return False def close(self) -> None: """Override this to properly close the stream and free resources.""" pass def __enter__(self) -> "StreamableSource": return self def __exit__(self, exc_type, exc_val, exc_tb) -> None: self.close() class IceCastClient(StreamableSource): """ A simple client for IceCast audio streams as miniaudio streamable source. If the stream has Icy MetaData, the stream_title attribute will be updated with the actual title taken from the metadata. You can also provide a callback to be called when a new stream title is available. The downloading of the data from the internet is done in a background thread and it tries to keep a (small) buffer filled with available data to read. You can optionally provide a custom ssl.SSLContext in the ssl_context parameter, if you need to change the way SSL connections are configured (certificates, checks, etc). """ BLOCK_SIZE = 8 * 1024 BUFFER_SIZE = 64 * 1024 def __init__( self, url: str, update_stream_title: Callable[["IceCastClient", str], None] = None, ssl_context: "ssl.SSLContext" = None, ) -> None: self.url = url self.stream_title = "???" self.station_genre = "???" self.station_name = "???" self.audio_info = "" self.audio_format = FileFormat.UNKNOWN self._stop_stream = False self._buffer = b"" self._buffer_lock = threading.Lock() self._update_title = update_stream_title req = urllib.request.Request(url, headers={"icy-metadata": "1"}) with urllib.request.urlopen(req, context=ssl_context) as result: self.station_genre = result.headers["icy-genre"] self.station_name = result.headers["icy-name"] stream_format = result.headers["Content-Type"] self.audio_format = self.determine_audio_format(stream_format) self._download_thread = threading.Thread( target=self._download_stream, daemon=True ) self._download_thread.start() def determine_audio_format(self, stream_format: str) -> FileFormat: if stream_format == "audio/mpeg": return FileFormat.MP3 elif stream_format == "audio/flac": return FileFormat.FLAC elif stream_format.endswith("/ogg"): return FileFormat.VORBIS else: return FileFormat.UNKNOWN def read(self, num_bytes: int) -> bytes: """Read a chunk of data from the stream.""" while len(self._buffer) < num_bytes: time.sleep(0.1) with self._buffer_lock: chunk = self._buffer[:num_bytes] self._buffer = self._buffer[num_bytes:] return chunk def close(self) -> None: """Stop the stream, aborting the background downloading.""" self._stop_stream = True self._download_thread.join() def _readall(self, fileobject, size: int) -> bytes: b = b"" while len(b) < size: b += fileobject.read(size) return b def _download_stream(self) -> None: req = urllib.request.Request(self.url, headers={"icy-metadata": "1"}) with urllib.request.urlopen(req) as result: self.station_genre = result.headers["icy-genre"] self.station_name = result.headers["icy-name"] stream_format = result.headers["Content-Type"] if stream_format: self.audio_format = self.determine_audio_format(stream_format) self.audio_info = result.headers.get("ice-audio-info", "") if "icy-metaint" in result.headers: meta_interval = int(result.headers["icy-metaint"]) else: meta_interval = 0 if meta_interval: # note: the meta_interval is fixed for the entire stream, so just use that as chunk size while not self._stop_stream: while len(self._buffer) >= self.BUFFER_SIZE: time.sleep(0.2) if self._stop_stream: return chunk = self._readall(result, meta_interval) with self._buffer_lock: self._buffer += chunk meta_size = 16 * self._readall(result, 1)[0] metadata = str( self._readall(result, meta_size).strip(b"\0"), "utf-8", errors="replace", ) if metadata: meta = self.parse_metadata(metadata) stream_title = meta.get("StreamTitle") if stream_title: self.stream_title = stream_title if self._update_title: self._update_title(self, stream_title) else: while not self._stop_stream: while len(self._buffer) >= self.BUFFER_SIZE: time.sleep(0.2) if self._stop_stream: return chunk = result.read(self.BLOCK_SIZE) with self._buffer_lock: self._buffer += chunk @staticmethod def parse_metadata(metadata: str) -> Dict[str, str]: meta = {} metadata = html.unescape(metadata) for part in metadata.split(";"): key, _, value = part.partition("=") if key: meta[key] = value.strip("'") return meta def stream_any( source: StreamableSource, source_format: FileFormat = FileFormat.UNKNOWN, output_format: SampleFormat = SampleFormat.SIGNED16, nchannels: int = 2, sample_rate: int = 44100, frames_to_read: int = 1024, dither: DitherMode = DitherMode.NONE, seek_frame: int = 0, ) -> Generator[array.array, int, None]: """ Convenience function that returns a generator to decode and stream any source of encoded audio data (such as a network stream). Stream result is chunks of raw PCM samples in the chosen format. If you send() a number into the generator rather than just using next() on it, you'll get that given number of frames, instead of the default configured amount. This is particularly useful to plug this stream into an audio device callback that wants a variable number of frames per call. """ decoder = ffi.new("ma_decoder *") decoder_config = lib.ma_decoder_config_init( output_format.value, nchannels, sample_rate ) decoder_config.ditherMode = dither.value source.ffi_handle = ffi.new_handle(source) decoder_config.encodingFormat = source_format.value result = lib.ma_decoder_init( lib._internal_decoder_read_callback, lib._internal_decoder_seek_callback, source.ffi_handle, ffi.addressof(decoder_config), decoder, ) if result != lib.MA_SUCCESS: raise DecodeError("failed to init decoder", result) if seek_frame > 0: result = lib.ma_decoder_seek_to_pcm_frame(decoder, seek_frame) if result != lib.MA_SUCCESS: raise DecodeError("failed to seek to frame", result) def on_close() -> None: pass g = _samples_stream_generator( frames_to_read, nchannels, output_format, decoder, None, on_close ) dummy = next(g) assert len(dummy) == 0 return g def stream_with_callbacks( sample_stream: PlaybackCallbackGeneratorType, progress_callback: Union[Callable[[int], None], None] = None, frame_process_method: Union[Callable[[FramesType], FramesType], None] = None, end_callback: Union[Callable, None] = None, ) -> PlaybackCallbackGeneratorType: """ Convenience generator function to add callback and processing functionality to another stream. You can specify : > A callback function that gets called during play and takes an int for the number of frames played. > A function that can be used to process raw data frames before they are yielded back (takes an array.array or bytes, returns an array.array or bytes) *Note: if the processing method is slow it will result in audio glitchiness > A callback function that gets called when the stream ends playing. """ frame_count = yield b"" try: while True: frame = sample_stream.send(frame_count) if frame_process_method: frame_count = yield frame_process_method(frame) else: frame_count = yield frame if progress_callback: progress_callback(frame_count) except StopIteration: if end_callback: end_callback() @ffi.def_extern() def _internal_decoder_read_callback( decoder: ffi.CData, output: ffi.CData, num_bytes: int, bytes_read: ffi.CData ) -> int: bytes_read[0] = 0 if num_bytes <= 0 or not decoder.pUserData: return lib.MA_ERROR source = ffi.from_handle(decoder.pUserData) # type: StreamableSource if source.error_in_readcallback is None: try: data = source.read(num_bytes) ffi.memmove(output, data, len(data)) bytes_read[0] = len(data) return lib.MA_SUCCESS if len(data) > 0 else lib.MA_AT_END except Exception as x: source.error_in_readcallback = x return lib.MA_ERROR @ffi.def_extern() def _internal_decoder_seek_callback( decoder: ffi.CData, offset: int, seek_origin: int ) -> int: if not decoder.pUserData: return lib.MA_ERROR if offset == 0 and seek_origin == lib.ma_seek_origin_current: return lib.MA_SUCCESS source = ffi.from_handle(decoder.pUserData) try: result = source.seek(offset, SeekOrigin(seek_origin)) if result: return lib.MA_SUCCESS # If seek returns False (not supported), return success anyway for compatibility # The decoder will work as long as it doesn't actually need to seek return lib.MA_SUCCESS except Exception: return lib.MA_BAD_SEEK def convert_sample_format( from_fmt: SampleFormat, sourcedata: bytes, to_fmt: SampleFormat, dither: DitherMode = DitherMode.NONE, ) -> bytearray: """Convert a raw buffer of pcm samples to another sample format. The result is returned as another raw pcm sample buffer""" sample_width = width_from_format(from_fmt) num_samples = len(sourcedata) // sample_width sample_width = width_from_format(to_fmt) buffer = bytearray(sample_width * num_samples) lib.ma_pcm_convert( ffi.from_buffer(buffer), to_fmt.value, sourcedata, from_fmt.value, num_samples, dither.value, ) return buffer def convert_frames( from_fmt: SampleFormat, from_numchannels: int, from_samplerate: int, sourcedata: bytes, to_fmt: SampleFormat, to_numchannels: int, to_samplerate: int, ) -> bytearray: """Convert audio frames in source sample format with a certain number of channels, to another sample format and possibly down/upmixing the number of channels as well.""" sample_width = width_from_format(from_fmt) num_frames = int(len(sourcedata) / from_numchannels / sample_width) sample_width = width_from_format(to_fmt) output_frame_count = lib.ma_calculate_frame_count_after_resampling( to_samplerate, from_samplerate, num_frames ) buffer = bytearray(output_frame_count * sample_width * to_numchannels) # note: the API doesn't have an option here to specify the dither mode. lib.ma_convert_frames( ffi.from_buffer(buffer), output_frame_count, to_fmt.value, to_numchannels, to_samplerate, sourcedata, num_frames, from_fmt.value, from_numchannels, from_samplerate, ) return buffer @ffi.def_extern() def _internal_data_callback( device: ffi.CData, output: ffi.CData, input: ffi.CData, framecount: int ) -> None: if framecount <= 0 or not device.pUserData: return callback_device = ffi.from_handle(device.pUserData) callback_device._data_callback(device, output, input, framecount) @ffi.def_extern() def _internal_stop_callback(device: ffi.CData) -> None: if not device.pUserData: return callback_device = ffi.from_handle(device.pUserData) callback_device._stop_callback(device) class AbstractDevice: def __init__(self) -> None: self.callback_generator = None # type: Optional[GeneratorTypes] self.running = False self.stop_callback = None # doesn't work consistently self._device = ffi.new("ma_device *") def __del__(self) -> None: self.close() def __enter__(self): return self def __exit__(self, exc_type, exc_val, exc_tb): self.close() def start(self, callback_generator: GeneratorTypes) -> None: """Start playback or capture, using the given callback generator (should already been started)""" if self.callback_generator: raise MiniaudioError("can't start an already started device") # This check fails on Cython... just skip it: # if not inspect.isgenerator(callback_generator): # raise TypeError("callback must be a generator", type(callback_generator)) self.callback_generator = callback_generator result = lib.ma_device_start(self._device) if result != lib.MA_SUCCESS: raise MiniaudioError("failed to start audio device", result) self.running = True def stop(self) -> None: """Halt playback or capture.""" self.callback_generator = None if self.running: result = lib.ma_device_stop(self._device) if result != lib.MA_SUCCESS: raise MiniaudioError("failed to stop audio device", result) self.running = False def close(self) -> None: """ Halt playback or capture and close down the device. If you use the device as a context manager, it will be closed automatically. """ try: self.stop() except MiniaudioError: pass if self._device is not None: lib.ma_device_uninit(self._device) self._device = None self.stop_callback = None def _stop_callback(self, device: ffi.CData) -> None: """Called when the device is stopped (i.e. device disconnect or manual stop) Doesn't work consistently however.""" if self.stop_callback: self.running = False self.stop_callback() self.callback_generator = None def _make_context( self, backends: List[Backend], thread_prio: ThreadPriority = ThreadPriority.HIGHEST, app_name: str = "", ) -> ffi.CData: context_config = lib.ma_context_config_init() context_config.threadPriority = thread_prio.value context = ffi.new("ma_context*") if app_name: self._context_app_name = app_name.encode() context_config.pulse.pApplicationName = ffi.from_buffer( self._context_app_name ) context_config.jack.pClientName = ffi.from_buffer(self._context_app_name) if backends: # use a context to supply a preferred backend list backends_mem = ffi.new("ma_backend[]", len(backends)) for i, b in enumerate(backends): backends_mem[i] = b.value result = lib.ma_context_init( backends_mem, len(backends), ffi.addressof(context_config), context ) if result != lib.MA_SUCCESS: raise MiniaudioError("cannot init context", result) else: result = lib.ma_context_init( ffi.NULL, 0, ffi.addressof(context_config), context ) if result != lib.MA_SUCCESS: raise MiniaudioError("cannot init context", result) return context class CaptureDevice(AbstractDevice): """An audio device provided by miniaudio, for audio capture (recording).""" def __init__( self, input_format: SampleFormat = SampleFormat.SIGNED16, nchannels: int = 2, sample_rate: int = 44100, buffersize_msec: int = 200, device_id: Union[ffi.CData, None] = None, callback_periods: int = 0, backends: Optional[List[Backend]] = None, thread_prio: ThreadPriority = ThreadPriority.HIGHEST, app_name: str = "", ) -> None: super().__init__() self.format = input_format self.sample_width = width_from_format(input_format) self.nchannels = nchannels self.sample_rate = sample_rate self.buffersize_msec = buffersize_msec self._ffi_handle = ffi.new_handle(self) self._devconfig = lib.ma_device_config_init(lib.ma_device_type_capture) self._devconfig.sampleRate = self.sample_rate self._devconfig.capture.channels = self.nchannels self._devconfig.capture.format = self.format.value self._devconfig.capture.pDeviceID = device_id or ffi.NULL self._devconfig.periodSizeInMilliseconds = self.buffersize_msec self._devconfig.pUserData = self._ffi_handle self._devconfig.dataCallback = lib._internal_data_callback self._devconfig.stopCallback = lib._internal_stop_callback self._devconfig.periods = callback_periods self.callback_generator = None # type: Optional[CaptureCallbackGeneratorType] self._context = self._make_context(backends or [], thread_prio, app_name) result = lib.ma_device_init( self._context, ffi.addressof(self._devconfig), self._device ) if result != lib.MA_SUCCESS: raise MiniaudioError("failed to init device", result) if self._device.pContext.backend == lib.ma_backend_null: if backends and Backend.NULL not in backends: raise MiniaudioError("no suitable audio backend found") self.backend = ffi.string( lib.ma_get_backend_name(self._device.pContext.backend) ).decode() def start(self, callback_generator: CaptureCallbackGeneratorType) -> None: # type: ignore """Start the audio device: capture (recording) begins. The recorded audio data is sent to the given callback generator as raw bytes. (it should already be started before)""" return super().start(callback_generator) def _data_callback( self, device: ffi.CData, output: ffi.CData, input: ffi.CData, framecount: int ) -> None: if self.callback_generator: buffer_size = self.sample_width * self.nchannels * framecount data = bytearray(buffer_size) ffi.memmove(data, input, buffer_size) try: self.callback_generator.send(data) except StopIteration: self.callback_generator = None return except Exception: self.callback_generator = None raise class PlaybackDevice(AbstractDevice): """An audio device provided by miniaudio, for audio playback.""" def __init__( self, output_format: SampleFormat = SampleFormat.SIGNED16, nchannels: int = 2, sample_rate: int = 44100, buffersize_msec: int = 200, device_id: Union[ffi.CData, None] = None, callback_periods: int = 0, backends: Optional[List[Backend]] = None, thread_prio: ThreadPriority = ThreadPriority.HIGHEST, app_name: str = "", ) -> None: super().__init__() self.format = output_format self.sample_width = width_from_format(output_format) self.nchannels = nchannels self.sample_rate = sample_rate self.buffersize_msec = buffersize_msec self._ffi_handle = ffi.new_handle(self) self._devconfig = lib.ma_device_config_init(lib.ma_device_type_playback) self._devconfig.sampleRate = self.sample_rate self._devconfig.playback.channels = self.nchannels self._devconfig.playback.format = self.format.value self._devconfig.playback.pDeviceID = device_id or ffi.NULL self._devconfig.periodSizeInMilliseconds = self.buffersize_msec self._devconfig.pUserData = self._ffi_handle self._devconfig.dataCallback = lib._internal_data_callback self._devconfig.stopCallback = lib._internal_stop_callback self._devconfig.periods = callback_periods self.callback_generator = None # type: Optional[PlaybackCallbackGeneratorType] self._context = self._make_context(backends or [], thread_prio, app_name) result = lib.ma_device_init( self._context, ffi.addressof(self._devconfig), self._device ) if result != lib.MA_SUCCESS: raise MiniaudioError("failed to init device", result) if self._device.pContext.backend == lib.ma_backend_null: if backends and Backend.NULL not in backends: raise MiniaudioError("no suitable audio backend found") self.backend = ffi.string( lib.ma_get_backend_name(self._device.pContext.backend) ).decode() def start(self, callback_generator: PlaybackCallbackGeneratorType) -> None: # type: ignore """Start the audio device: playback begins. The audio data is provided by the given callback generator. The generator gets sent the required number of frames and should yield the sample data as raw bytes, a memoryview, an array.array, or as a numpy array with shape (numframes, numchannels). The generator should already be started before passing it in.""" return super().start(callback_generator) def _data_callback( self, device: ffi.CData, output: ffi.CData, input: ffi.CData, framecount: int ) -> None: if self.callback_generator: try: samples = self.callback_generator.send(framecount) except StopIteration: self.callback_generator = None return except Exception: self.callback_generator = None raise samples_bytes = _bytes_from_generator_samples(samples) if samples_bytes: if len(samples_bytes) > framecount * self.sample_width * self.nchannels: self.callback_generator = None raise MiniaudioError( "number of frames from callback exceeds maximum" ) ffi.memmove(output, samples_bytes, len(samples_bytes)) class DuplexStream(AbstractDevice): """Joins a capture device and a playback device.""" def __init__( self, playback_format: SampleFormat = SampleFormat.SIGNED16, playback_channels: int = 2, capture_format: SampleFormat = SampleFormat.SIGNED16, capture_channels: int = 2, sample_rate: int = 44100, buffersize_msec: int = 200, playback_device_id: Union[ffi.CData, None] = None, capture_device_id: Union[ffi.CData, None] = None, callback_periods: int = 0, backends: Optional[List[Backend]] = None, thread_prio: ThreadPriority = ThreadPriority.HIGHEST, app_name: str = "", ) -> None: super().__init__() self.capture_format = capture_format self.playback_format = playback_format self.sample_width = width_from_format(capture_format) self.capture_channels = capture_channels self.playback_channels = playback_channels self.sample_rate = sample_rate self.buffersize_msec = buffersize_msec self._ffi_handle = ffi.new_handle(self) self._devconfig = lib.ma_device_config_init(lib.ma_device_type_duplex) self._devconfig.sampleRate = self.sample_rate self._devconfig.playback.channels = self.playback_channels self._devconfig.playback.format = self.playback_format.value self._devconfig.playback.pDeviceID = playback_device_id or ffi.NULL self._devconfig.capture.channels = self.capture_channels self._devconfig.capture.format = self.capture_format.value self._devconfig.capture.pDeviceID = capture_device_id or ffi.NULL self._devconfig.periodSizeInMilliseconds = self.buffersize_msec self._devconfig.pUserData = self._ffi_handle self._devconfig.dataCallback = lib._internal_data_callback self._devconfig.stopCallback = lib._internal_stop_callback self._devconfig.periods = callback_periods self.callback_generator = None # type: Optional[DuplexCallbackGeneratorType] self._context = self._make_context(backends or [], thread_prio, app_name) result = lib.ma_device_init( self._context, ffi.addressof(self._devconfig), self._device ) if result != lib.MA_SUCCESS: raise MiniaudioError("failed to init device", result) if self._device.pContext.backend == lib.ma_backend_null: if backends and Backend.NULL not in backends: raise MiniaudioError("no suitable audio backend found") self.backend = ffi.string( lib.ma_get_backend_name(self._device.pContext.backend) ).decode() def start(self, callback_generator: DuplexCallbackGeneratorType) -> None: # type: ignore """Start the audio device: playback and capture begin. The audio data for playback is provided by the given callback generator, which is sent the recorded audio data at the same time. (it should already be started before passing it in)""" return super().start(callback_generator) def _data_callback( self, device: ffi.CData, output: ffi.CData, input: ffi.CData, framecount: int ) -> None: buffer_size = self.sample_width * self.capture_channels * framecount in_data = bytearray(buffer_size) ffi.memmove(in_data, input, buffer_size) if self.callback_generator: try: out_data = self.callback_generator.send(in_data) except StopIteration: self.callback_generator = None return except Exception: self.callback_generator = None raise if out_data: samples_bytes = _bytes_from_generator_samples(out_data) ffi.memmove(output, samples_bytes, len(samples_bytes)) def _bytes_from_generator_samples( samples: Union[array.array, memoryview, bytes], ) -> bytes: # convert any non-bytes generator result to raw bytes if isinstance(samples, array.array): return memoryview(samples).cast("B") # type: ignore elif isinstance(samples, memoryview) and samples.itemsize != 1: return samples.cast("B") # type: ignore elif numpy and isinstance(samples, numpy.ndarray): return samples.tobytes() return samples # type: ignore class WavFileReadStream(io.RawIOBase): """An IO stream that reads as a .wav file, and which gets its pcm samples from the provided producer""" def __init__( self, pcm_sample_gen: PlaybackCallbackGeneratorType, sample_rate: int, nchannels: int, output_format: SampleFormat, max_frames: int = 0, ) -> None: self.sample_gen = pcm_sample_gen self.sample_rate = sample_rate self.nchannels = nchannels self.format = output_format self.max_frames = max_frames self.sample_width = width_from_format(output_format) self.max_bytes = (max_frames * nchannels * self.sample_width) or sys.maxsize self.bytes_done = 0 # create WAVE header fmt = ffi.new("ma_dr_wav_data_format*") fmt.container = lib.ma_dr_wav_container_riff fmt.format = lib.MA_DR_WAVE_FORMAT_PCM fmt.channels = nchannels fmt.sampleRate = sample_rate fmt.bitsPerSample = self.sample_width * 8 data = ffi.new("void**") datasize = ffi.new("size_t *") pwav = ffi.new("ma_dr_wav*") if max_frames > 0: lib.ma_dr_wav_init_memory_write_sequential( pwav, data, datasize, fmt, max_frames * nchannels, ffi.NULL ) else: lib.ma_dr_wav_init_memory_write(pwav, data, datasize, fmt, ffi.NULL) lib.ma_dr_wav_uninit(pwav) self.buffered = bytes(ffi.buffer(data[0], datasize[0])) lib.ma_dr_wav_free(data[0], ffi.NULL) def read(self, amount: int = sys.maxsize) -> Optional[bytes]: """Read up to the given amount of bytes from the file.""" if self.bytes_done >= self.max_bytes or not self.sample_gen: return b"" while len(self.buffered) < amount: try: samples = next(self.sample_gen) except StopIteration: self.bytes_done = sys.maxsize break else: self.buffered += _bytes_from_generator_samples(samples) result = self.buffered[:amount] self.buffered = self.buffered[amount:] self.bytes_done += len(result) return result def close(self) -> None: """Close the file""" pass # miscellaneous def lib_version() -> str: """Returns the version string of the underlying miniaudio C library""" return ffi.string(lib.ma_version_string()).decode() def is_backend_enabled(backend: Backend) -> bool: """Determines whether or not the given backend is available by the compilation environment for the underlying miniaudio C library""" return bool(lib.ma_is_backend_enabled(backend.value)) def get_enabled_backends() -> Set[Backend]: """Returns the set of available backends by the compilation environment for the underlying miniaudio C library""" with ffi.new("size_t *") as count, ffi.new("ma_backend[30]") as backends: result = lib.ma_get_enabled_backends(backends, 30, count) if result != lib.MA_SUCCESS: raise MiniaudioError("can't determine enabled backends") return set(Backend(b) for b in backends[0 : count[0]]) def is_loopback_supported(backend: Backend) -> bool: """Determines whether or not loopback mode is support by a backend.""" return bool(lib.ma_is_loopback_supported(backend.value))