# ---------------------------------------------------------------------------- # pyglet # Copyright (c) 2006-2008 Alex Holkner # Copyright (c) 2008-2023 pyglet contributors # All rights reserved. # # Redistribution and use in source and binary forms, with or without # modification, are permitted provided that the following conditions # are met: # # * Redistributions of source code must retain the above copyright # notice, this list of conditions and the following disclaimer. # * Redistributions in binary form must reproduce the above copyright # notice, this list of conditions and the following disclaimer in # the documentation and/or other materials provided with the # distribution. # * Neither the name of pyglet nor the names of its # contributors may be used to endorse or promote products # derived from this software without specific prior written # permission. # # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS # "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT # LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS # FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE # COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, # INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, # BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; # LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER # CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT # LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN # ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE # POSSIBILITY OF SUCH DAMAGE. # ---------------------------------------------------------------------------- import ctypes import weakref from collections import namedtuple from . import lib_openal as al from . import lib_alc as alc from pyglet.util import debug_print from pyglet.media.exceptions import MediaException _debug = debug_print('debug_media') class OpenALException(MediaException): def __init__(self, message=None, error_code=None, error_string=None): self.message = message self.error_code = error_code self.error_string = error_string def __str__(self): if self.error_code is None: return f'OpenAL Exception: {self.message}' else: return f'OpenAL Exception [{self.error_code}: {self.error_string}]: {self.message}' class OpenALObject: """Base class for OpenAL objects.""" @classmethod def _check_error(cls, message=None): """Check whether there is an OpenAL error and raise exception if present.""" error_code = al.alGetError() if error_code != 0: error_string = al.alGetString(error_code) # TODO: Fix return type in generated code? error_string = ctypes.cast(error_string, ctypes.c_char_p) raise OpenALException(message=message, error_code=error_code, error_string=str(error_string.value)) @classmethod def _raise_error(cls, message): """Raise an exception. Try to check for OpenAL error code too.""" cls._check_error(message) raise OpenALException(message) class OpenALDevice(OpenALObject): """OpenAL audio device.""" def __init__(self, device_name=None): self._al_device = alc.alcOpenDevice(device_name) self.check_context_error('Failed to open device.') if self._al_device is None: raise OpenALException('No OpenAL devices.') def __del__(self): assert _debug("Delete interface.OpenALDevice") self.delete() def delete(self): if self._al_device is not None: if alc.alcCloseDevice(self._al_device) == alc.ALC_FALSE: self._raise_context_error('Failed to close device.') self._al_device = None @property def is_ready(self): return self._al_device is not None def create_context(self): al_context = alc.alcCreateContext(self._al_device, None) self.check_context_error('Failed to create context') return OpenALContext(self, al_context) def get_version(self): major = alc.ALCint() minor = alc.ALCint() alc.alcGetIntegerv(self._al_device, alc.ALC_MAJOR_VERSION, ctypes.sizeof(major), major) self.check_context_error('Failed to get version.') alc.alcGetIntegerv(self._al_device, alc.ALC_MINOR_VERSION, ctypes.sizeof(minor), minor) self.check_context_error('Failed to get version.') return major.value, minor.value def get_extensions(self): extensions = alc.alcGetString(self._al_device, alc.ALC_EXTENSIONS) self.check_context_error('Failed to get extensions.') return ctypes.cast(extensions, ctypes.c_char_p).value.decode('ascii').split() def check_context_error(self, message=None): """Check whether there is an OpenAL error and raise exception if present.""" error_code = alc.alcGetError(self._al_device) if error_code != 0: error_string = alc.alcGetString(self._al_device, error_code) # TODO: Fix return type in generated code? error_string = ctypes.cast(error_string, ctypes.c_char_p) raise OpenALException(message=message, error_code=error_code, error_string=str(error_string.value)) def _raise_context_error(self, message): """Raise an exception. Try to check for OpenAL error code too.""" self.check_context_error(message) raise OpenALException(message) class OpenALContext(OpenALObject): def __init__(self, device, al_context): self.device = device self._al_context = al_context self.make_current() def __del__(self): assert _debug("Delete interface.OpenALContext") self.delete() def delete(self): if self._al_context is not None: # TODO: Check if this context is current alc.alcMakeContextCurrent(None) self.device.check_context_error('Failed to make context no longer current.') alc.alcDestroyContext(self._al_context) self.device.check_context_error('Failed to destroy context.') self._al_context = None def make_current(self): alc.alcMakeContextCurrent(self._al_context) self.device.check_context_error('Failed to make context current.') def create_source(self): self.make_current() return OpenALSource(self) class OpenALSource(OpenALObject): def __init__(self, context): self.context = weakref.ref(context) self.buffer_pool = OpenALBufferPool(self.context) self._al_source = al.ALuint() al.alGenSources(1, self._al_source) self._check_error('Failed to create source.') self._state = None self._get_state() self._owned_buffers = {} def __del__(self): assert _debug("Delete interface.OpenALSource") self.delete() def delete(self): if self.context() and self._al_source is not None: # Only delete source if the context still exists al.alDeleteSources(1, self._al_source) self._check_error('Failed to delete source.') self.buffer_pool.clear() self._al_source = None @property def is_initial(self): self._get_state() return self._state == al.AL_INITIAL @property def is_playing(self): self._get_state() return self._state == al.AL_PLAYING @property def is_paused(self): self._get_state() return self._state == al.AL_PAUSED @property def is_stopped(self): self._get_state() return self._state == al.AL_STOPPED def _int_source_property(attribute): return property(lambda self: self._get_int(attribute), lambda self, value: self._set_int(attribute, value)) def _float_source_property(attribute): return property(lambda self: self._get_float(attribute), lambda self, value: self._set_float(attribute, value)) def _3floats_source_property(attribute): return property(lambda self: self._get_3floats(attribute), lambda self, value: self._set_3floats(attribute, value)) position = _3floats_source_property(al.AL_POSITION) velocity = _3floats_source_property(al.AL_VELOCITY) gain = _float_source_property(al.AL_GAIN) buffers_queued = _int_source_property(al.AL_BUFFERS_QUEUED) buffers_processed = _int_source_property(al.AL_BUFFERS_PROCESSED) min_gain = _float_source_property(al.AL_MIN_GAIN) max_gain = _float_source_property(al.AL_MAX_GAIN) reference_distance = _float_source_property(al.AL_REFERENCE_DISTANCE) rolloff_factor = _float_source_property(al.AL_ROLLOFF_FACTOR) pitch = _float_source_property(al.AL_PITCH) max_distance = _float_source_property(al.AL_MAX_DISTANCE) direction = _3floats_source_property(al.AL_DIRECTION) cone_inner_angle = _float_source_property(al.AL_CONE_INNER_ANGLE) cone_outer_angle = _float_source_property(al.AL_CONE_OUTER_ANGLE) cone_outer_gain = _float_source_property(al.AL_CONE_OUTER_GAIN) sec_offset = _float_source_property(al.AL_SEC_OFFSET) sample_offset = _float_source_property(al.AL_SAMPLE_OFFSET) byte_offset = _float_source_property(al.AL_BYTE_OFFSET) del _int_source_property del _float_source_property del _3floats_source_property def play(self): al.alSourcePlay(self._al_source) self._check_error('Failed to play source.') def pause(self): al.alSourcePause(self._al_source) self._check_error('Failed to pause source.') def stop(self): al.alSourceStop(self._al_source) self._check_error('Failed to stop source.') def clear(self): self._set_int(al.AL_BUFFER, al.AL_NONE) while self._owned_buffers: buf_name, buf = self._owned_buffers.popitem() self.buffer_pool.unqueue_buffer(buf) def get_buffer(self): return self.buffer_pool.get_buffer() def queue_buffer(self, buf): assert buf.is_valid al.alSourceQueueBuffers(self._al_source, 1, ctypes.byref(buf.al_buffer)) self._check_error('Failed to queue buffer.') self._add_buffer(buf) def unqueue_buffers(self): processed = self.buffers_processed assert _debug("Processed buffer count: {}".format(processed)) if processed > 0: buffers = (al.ALuint * processed)() al.alSourceUnqueueBuffers(self._al_source, len(buffers), buffers) self._check_error('Failed to unqueue buffers from source.') for buf in buffers: self.buffer_pool.unqueue_buffer(self._pop_buffer(buf)) return processed def _get_state(self): if self._al_source is not None: self._state = self._get_int(al.AL_SOURCE_STATE) def _get_int(self, key): assert self._al_source is not None al_int = al.ALint() al.alGetSourcei(self._al_source, key, al_int) self._check_error('Failed to get value') return al_int.value def _set_int(self, key, value): assert self._al_source is not None al.alSourcei(self._al_source, key, int(value)) self._check_error('Failed to set value.') def _get_float(self, key): assert self._al_source is not None al_float = al.ALfloat() al.alGetSourcef(self._al_source, key, al_float) self._check_error('Failed to get value') return al_float.value def _set_float(self, key, value): assert self._al_source is not None al.alSourcef(self._al_source, key, float(value)) self._check_error('Failed to set value.') def _get_3floats(self, key): assert self._al_source is not None x = al.ALfloat() y = al.ALfloat() z = al.ALfloat() al.alGetSource3f(self._al_source, key, x, y, z) self._check_error('Failed to get value') return x.value, y.value, z.value def _set_3floats(self, key, values): assert self._al_source is not None x, y, z = map(float, values) al.alSource3f(self._al_source, key, x, y, z) self._check_error('Failed to set value.') def _add_buffer(self, buf): self._owned_buffers[buf.name] = buf def _pop_buffer(self, al_buffer): buf = self._owned_buffers.pop(al_buffer, None) assert buf is not None return buf OpenALOrientation = namedtuple("OpenALOrientation", ['at', 'up']) class OpenALListener(OpenALObject): @property def position(self): return self._get_3floats(al.AL_POSITION) @position.setter def position(self, values): self._set_3floats(al.AL_POSITION, values) @property def velocity(self): return self._get_3floats(al.AL_VELOCITY) @velocity.setter def velocity(self, values): self._set_3floats(al.AL_VELOCITY, values) @property def gain(self): return self._get_float(al.AL_GAIN) @gain.setter def gain(self, value): self._set_float(al.AL_GAIN, value) @property def orientation(self): values = self._get_float_vector(al.AL_ORIENTATION, 6) return OpenALOrientation(values[0:3], values[3:6]) @orientation.setter def orientation(self, values): if len(values) == 2: actual_values = values[0] + values[1] elif len(values) == 6: actual_values = values else: actual_values = [] if len(actual_values) != 6: raise ValueError("Need 2 tuples of 3 or 1 tuple of 6.") self._set_float_vector(al.AL_ORIENTATION, actual_values) def _get_float(self, key): al_float = al.ALfloat() al.alGetListenerf(key, al_float) self._check_error('Failed to get value') return al_float.value def _set_float(self, key, value): al.alListenerf(key, float(value)) self._check_error('Failed to set value.') def _get_3floats(self, key): x = al.ALfloat() y = al.ALfloat() z = al.ALfloat() al.alGetListener3f(key, x, y, z) self._check_error('Failed to get value') return x.value, y.value, z.value def _set_3floats(self, key, values): x, y, z = map(float, values) al.alListener3f(key, x, y, z) self._check_error('Failed to set value.') def _get_float_vector(self, key, count): al_float_vector = (al.ALfloat * count)() al.alGetListenerfv(key, al_float_vector) self._check_error('Failed to get value') return [x for x in al_float_vector] def _set_float_vector(self, key, values): al_float_vector = (al.ALfloat * len(values))(*values) al.alListenerfv(key, al_float_vector) self._check_error('Failed to set value.') class OpenALBuffer(OpenALObject): _format_map = { (1, 8): al.AL_FORMAT_MONO8, (1, 16): al.AL_FORMAT_MONO16, (2, 8): al.AL_FORMAT_STEREO8, (2, 16): al.AL_FORMAT_STEREO16, } def __init__(self, al_buffer, context): self._al_buffer = al_buffer self.context = context assert self.is_valid def __del__(self): assert _debug("Delete interface.OpenALBuffer") self.delete() @property def is_valid(self): self._check_error('Before validate buffer.') if self._al_buffer is None: return False valid = bool(al.alIsBuffer(self._al_buffer)) if not valid: # Clear possible error due to invalid buffer al.alGetError() return valid @property def al_buffer(self): assert self.is_valid return self._al_buffer @property def name(self): assert self.is_valid return self._al_buffer.value def delete(self): if self._al_buffer is not None and self.context() and self.is_valid: al.alDeleteBuffers(1, ctypes.byref(self._al_buffer)) self._check_error('Error deleting buffer.') self._al_buffer = None def data(self, audio_data, audio_format, length=None): assert self.is_valid length = length or audio_data.length try: al_format = self._format_map[(audio_format.channels, audio_format.sample_size)] except KeyError: raise MediaException(f"OpenAL does not support '{audio_format.sample_size}bit' audio.") al.alBufferData(self._al_buffer, al_format, audio_data.data, length, audio_format.sample_rate) self._check_error('Failed to add data to buffer.') class OpenALBufferPool(OpenALObject): """At least Mac OS X doesn't free buffers when a source is deleted; it just detaches them from the source. So keep our own recycled queue. """ def __init__(self, context): self.context = context self._buffers = [] # list of free buffer names def __del__(self): assert _debug("Delete interface.OpenALBufferPool") self.clear() def __len__(self): return len(self._buffers) def clear(self): while self._buffers: self._buffers.pop().delete() def get_buffer(self): """Convenience for returning one buffer name""" return self.get_buffers(1)[0] def get_buffers(self, number): """Returns an array containing `number` buffer names. The returned list must not be modified in any way, and may get changed by subsequent calls to get_buffers. """ buffers = [] while number > 0: if self._buffers: b = self._buffers.pop() else: b = self._create_buffer() if b.is_valid: # Protect against implementations that DO free buffers # when they delete a source - carry on. buffers.append(b) number -= 1 return buffers def unqueue_buffer(self, buf): """A buffer has finished playing, free it.""" if buf.is_valid: self._buffers.append(buf) def _create_buffer(self): """Create a new buffer.""" al_buffer = al.ALuint() al.alGenBuffers(1, al_buffer) self._check_error('Error allocating buffer.') return OpenALBuffer(al_buffer, self.context)