536 lines
18 KiB
Python
536 lines
18 KiB
Python
# ----------------------------------------------------------------------------
|
|
# pyglet
|
|
# Copyright (c) 2006-2008 Alex Holkner
|
|
# Copyright (c) 2008-2022 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)
|