# Python object wrappers for alure
# Copyright (C) 2019, 2020 Nguyễn Gia Phong
# Copyright (C) 2020 Ngô Ngọc Đức Huy
# Copyright (C) 2020 Ngô Xuân Minh
#
# This file is part of palace.
#
# palace is free software: you can redistribute it and/or modify it
# under the terms of the GNU Lesser General Public License as published
# by the Free Software Foundation, either version 3 of the License,
# or (at your option) any later version.
#
# palace is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with palace. If not, see .
"""Pythonic Audio Library and Codecs Environment
Attributes
----------
CHANNEL_CONFIG : int
Context creation key to specify the channel configuration
(either `MONO`, `STEREO`, `QUAD`, `X51`, `X61` or `X71`).
SAMPLE_TYPE : int
Context creation key to specify the sample type
(either `[UNSIGNED_]{BYTE,SHORT,INT}` or `FLOAT`).
FREQUENCY : int
Context creation key to specify the frequency in hertz.
MONO_SOURCES : int
Context creation key to specify the number of mono (3D) sources.
STEREO_SOURCES : int
Context creation key to specify the number of stereo sources.
MAX_AUXILIARY_SENDS : int
Context creation key to specify the maximum number of
auxiliary source sends.
HRTF : int
Context creation key to specify whether to enable HRTF
(either `FALSE`, `TRUE` or `DONT_CARE`).
HRTF_ID : int
Context creation key to specify the HRTF to be used.
OUTPUT_LIMITER : int
Context creation key to specify whether to use a gain limiter
(either `FALSE`, `TRUE` or `DONT_CARE`).
sample_types : Tuple[str, ...]
Names of available sample types.
channel_configs : Tuple[str, ...]
Names of available channel configurations.
device_names : DeviceNames
Read-only namespace of device names by category (basic, full and
capture), as tuples of strings whose first item being the default.
distance_models : Tuple[str, ...]
Names of available distance models.
reverb_preset_names : Tuple[str, ...]
Names of predefined reverb effect presets in lexicographical order.
decoder_factories : DecoderNamespace
Simple object for storing decoder factories.
User-registered factories are tried one after another
if `RuntimeError` is raised, in lexicographical order.
Internal decoder factories are always used after registered ones.
"""
__all__ = [
'FALSE', 'TRUE', 'DONT_CARE', 'FREQUENCY',
'MONO_SOURCES', 'STEREO_SOURCES', 'MAX_AUXILIARY_SENDS', 'OUTPUT_LIMITER',
'CHANNEL_CONFIG', 'MONO', 'STEREO', 'QUAD', 'X51', 'X61', 'X71',
'SAMPLE_TYPE', 'BYTE', 'UNSIGNED_BYTE', 'SHORT', 'UNSIGNED_SHORT',
'INT', 'UNSIGNED_INT', 'FLOAT', 'HRTF', 'HRTF_ID',
'sample_types', 'channel_configs', 'device_names',
'reverb_preset_names', 'decoder_factories', 'distance_models',
'current_fileio', 'use_fileio', 'query_extension',
'thread_local', 'current_context', 'use_context',
'cache', 'free', 'decode', 'sample_size', 'sample_length',
'Device', 'Context', 'Listener', 'Buffer', 'Source', 'SourceGroup',
'BaseEffect', 'ReverbEffect', 'ChorusEffect',
'Decoder', 'BaseDecoder', 'FileIO', 'MessageHandler']
from abc import abstractmethod, ABCMeta
from contextlib import contextmanager
from enum import Enum, auto
from contextlib import contextmanager
from io import DEFAULT_BUFFER_SIZE
from operator import itemgetter
from types import TracebackType
from typing import (Any, Callable, Dict, Iterable, Iterator,
List, Optional, Sequence, Tuple, Type)
from warnings import catch_warnings, simplefilter, warn
try: # Python 3.8+
from typing import Protocol
except ImportError:
from abc import ABC as Protocol
from libc.stdint cimport uint64_t # noqa
from libc.stdio cimport EOF
from libc.string cimport memcpy
from libcpp cimport bool as boolean, nullptr
from libcpp.memory cimport (make_unique, unique_ptr, # noqa
shared_ptr, static_pointer_cast)
from libcpp.string cimport string
from libcpp.utility cimport pair
from libcpp.vector cimport vector
from std cimport istream, milliseconds, streambuf
from cpython.mem cimport PyMem_RawMalloc, PyMem_RawFree
from cpython.ref cimport Py_INCREF, Py_DECREF
from cython.view cimport array
cimport alure # noqa
from util cimport ( # noqa
REVERB_PRESETS, SAMPLE_TYPES, CHANNEL_CONFIGS, DISTANCE_MODELS,
reverb_presets, mkattrs, make_filter, from_vector3, to_vector3)
# Aliases
getter = property # bypass Cython property hijack
setter = lambda fset: property(fset=fset, doc=fset.__doc__) # noqa
Vector3: Type = Tuple[float, float, float]
# Cast to Python objects
FALSE: int = alure.ALC_FALSE
TRUE: int = alure.ALC_TRUE
DONT_CARE: int = alure.ALC_DONT_CARE_SOFT
FREQUENCY: int = alure.ALC_FREQUENCY
MONO_SOURCES: int = alure.ALC_MONO_SOURCES
STEREO_SOURCES: int = alure.ALC_STEREO_SOURCES
MAX_AUXILIARY_SENDS: int = alure.ALC_MAX_AUXILIARY_SENDS
OUTPUT_LIMITER: int = alure.ALC_OUTPUT_LIMITER_SOFT
CHANNEL_CONFIG: int = alure.ALC_FORMAT_CHANNELS_SOFT
MONO: int = alure.ALC_MONO_SOFT
STEREO: int = alure.ALC_STEREO_SOFT
QUAD: int = alure.ALC_QUAD_SOFT
X51: int = alure.ALC_5POINT1_SOFT
X61: int = alure.ALC_6POINT1_SOFT
X71: int = alure.ALC_7POINT1_SOFT
SAMPLE_TYPE: int = alure.ALC_FORMAT_TYPE_SOFT
BYTE: int = alure.ALC_BYTE_SOFT
UNSIGNED_BYTE: int = alure.ALC_UNSIGNED_BYTE_SOFT
SHORT: int = alure.ALC_SHORT_SOFT
UNSIGNED_SHORT: int = alure.ALC_UNSIGNED_SHORT_SOFT
INT: int = alure.ALC_INT_SOFT
UNSIGNED_INT: int = alure.ALC_UNSIGNED_INT_SOFT
FLOAT: int = alure.ALC_FLOAT_SOFT
HRTF: int = alure.ALC_HRTF_SOFT
HRTF_ID: int = alure.ALC_HRTF_ID_SOFT
sample_types: Tuple[str, ...] = (
'Unsigned 8-bit', 'Signed 16-bit', '32-bit float', 'Mulaw')
channel_configs: Tuple[str, ...] = (
'Mono', 'Stereo', 'Rear', 'Quadrophonic',
'5.1 Surround', '6.1 Surround', '7.1 Surround',
'B-Format 2D', 'B-Format 3D')
distance_models: Tuple[str, ...] = (
'inverse clamped', 'linear clamped', 'exponent clamped',
'inverse', 'linear', 'exponent', 'none')
# Since multiple calls of DeviceManager.get_instance() will give
# the same instance, we can create module-level variable and expose
# its attributes and methods. This also prevents the device manager
# from being garbage collected by keeping a reference to the instance.
cdef alure.DeviceManager devmgr = alure.DeviceManager.get_instance()
device_names: DeviceNames = DeviceNames()
cdef boolean _thread = False
reverb_preset_names: Tuple[str, ...] = tuple(reverb_presets())
decoder_factories: DecoderNamespace = DecoderNamespace()
cdef object fileio_factory = None # type: Optional[Callable[[str], FileIO]]
def sample_size(length: int, channel_config: str, sample_type: str) -> int:
"""Return the size of the given number of sample frames.
Raises
------
ValueError
If either channel_config or sample_type is invalid.
RuntimeError
If the byte size result too large.
"""
cdef alure.ChannelConfig alure_channel_config
cdef alure.SampleType alure_sample_type
try:
alure_channel_config = CHANNEL_CONFIGS.at(channel_config)
except IndexError:
raise ValueError(f'invalid channel config: {channel_config}') from None
try:
alure_sample_type = SAMPLE_TYPES.at(sample_type)
except IndexError:
raise ValueError(f'invalid sample type: {sample_type}') from None
return alure.frames_to_bytes(
length, alure_channel_config, alure_sample_type)
def sample_length(size: int, channel_config: str, sample_type: str) -> int:
"""Return the number of frames stored in the given byte size.
Raises
------
ValueError
If either channel_config or sample_type is invalid.
"""
cdef alure.ChannelConfig alure_channel_config
cdef alure.SampleType alure_sample_type
try:
alure_channel_config = CHANNEL_CONFIGS.at(channel_config)
except IndexError:
raise ValueError(f'invalid channel config: {channel_config}') from None
try:
alure_sample_type = SAMPLE_TYPES.at(sample_type)
except IndexError:
raise ValueError(f'invalid sample type: {sample_type}') from None
return alure.bytes_to_frames(size, alure_channel_config, alure_sample_type)
def query_extension(name: str) -> bool:
"""Return if a non-device-specific ALC extension exists.
See Also
--------
Device.query_extension : Query ALC extension on a device
"""
return devmgr.query_extension(name)
@contextmanager
def thread_local(state: bool) -> Iterator[None]:
"""Return a context manager controlling preference of local thread.
Effectively, it sets fallback value for `thread` argument
for `current_context` and `use_context`.
Initially, globally current `Context` is preferred.
"""
global _thread
previous, _thread = _thread, state
try:
yield
finally:
_thread = previous
def current_context(thread: Optional[bool] = None) -> Optional[Context]:
"""Return the context that is currently used.
If `thread` is set to `True`, return the thread-specific context
used for OpenAL operations. This requires the non-device-specific
as well as the context's device `ALC_EXT_thread_local_context`
extension to be available.
In case `thread` is not specified, fallback to preference made by
`thread_local`.
"""
cdef Context current = Context.__new__(Context)
if thread is None: thread = _thread
if thread:
current.impl = alure.Context.get_thread_current()
else:
current.impl = alure.Context.get_current()
if not current: return None
current.device = Device.__new__(Device)
current.device.impl = current.impl.get_device()
current.listener = Listener(current)
return current
def use_context(context: Optional[Context],
thread: Optional[bool] = None) -> None:
"""Make the specified context current for OpenAL operations.
This fails silently if the given context has been destroyed.
In case `thread` is not specified, fallback to preference made by
`thread_local`.
If `thread` is `True`, make the context current
for OpenAL operations on the calling thread only.
This requires the non-device-specific as well as the context's
device `ALC_EXT_thread_local_context` extension to be available.
"""
cdef alure.Context alure_context = nullptr
if context: alure_context = ( context).impl
if thread is None: thread = _thread
if thread:
alure.Context.make_thread_current(alure_context)
else:
alure.Context.make_current(alure_context)
def cache(names: Iterable[str], context: Optional[Context] = None) -> None:
"""Cache given audio resources asynchronously.
Duplicate names and buffers already cached are ignored.
Cached buffers must be freed before destroying the context.
The resources will be scheduled for caching asynchronously,
and should be retrieved later when needed by initializing
`Buffer` corresponding objects. Resources that cannot be
loaded, for example due to an unsupported format, will be
ignored and a later `Buffer` initialization will raise
an exception.
If `context` is not given, `current_context()` will be used.
Raises
------
RuntimeError
If there is neither any context specified nor current.
See Also
--------
free : Free cached audio resources given their names
Buffer.destroy : Free the buffer's cache
"""
cdef vector[string] std_names = list(names)
cdef vector[alure.StringView] alure_names
for name in std_names: alure_names.push_back( name)
if context is None: context = current_context()
if not context: raise RuntimeError('there is no context current')
( context).impl.precache_buffers_async(alure_names)
def free(names: Iterable[str], context: Optional[Context] = None) -> None:
"""Free cached audio resources given their names.
If `context` is not given, `current_context()` will be used.
Raises
------
RuntimeError
If there is neither any context specified nor current.
"""
if context is None: context = current_context()
if not context: raise RuntimeError('there is no context current')
cdef alure.Context alure_context = ( context).impl
# Cython cannot infer collection types yet.
cdef vector[string] std_names = list(names)
for name in std_names: alure_context.remove_buffer(name)
def decode(name: str, context: Optional[Context] = None) -> Decoder:
"""Return the decoder created from the given resource name.
This first tries user-registered decoder factories in
lexicographical order, then fallback to the internal ones.
Raises
------
RuntimeError
If there is neither any context specified nor current.
See Also
--------
decoder_factories : Simple object for storing decoder factories
"""
def find_resource(name, subst):
if not name: raise RuntimeError('failed to open file')
try:
if fileio_factory is None:
return open(name, 'rb')
else:
return fileio_factory(name)
except FileNotFoundError:
return find_resource(subst(name), subst)
if context is None: context = current_context()
if not context: raise RuntimeError('there is no context current')
resource = find_resource(
name, context.message_handler.resource_not_found)
for decoder_factory in decoder_factories:
resource.seek(0)
try:
return decoder_factory(resource)
except RuntimeError:
continue
return Decoder(name, context)
def current_fileio() -> Optional[Callable[[str], 'FileIO']]:
"""Return the file I/O factory currently in used by audio decoders.
If the default is being used, return `None`.
"""
return fileio_factory
def use_fileio(factory: Optional[Callable[[str], 'FileIO']],
buffer_size: int = DEFAULT_BUFFER_SIZE) -> None:
"""Set the file I/O factory instance to be used by audio decoders.
If `factory=None` is provided, revert to the default.
"""
global fileio_factory
fileio_factory = factory
if fileio_factory is None:
alure.FileIOFactory.set(unique_ptr[alure.FileIOFactory]())
else:
alure.FileIOFactory.set(unique_ptr[alure.FileIOFactory](
new CppFileIOFactory(fileio_factory, buffer_size)))
cdef class DeviceNames:
"""Read-only namespace of device names by category.
Attributes
----------
basic : Tuple[str, ...]
Basic device names, with the first one being the default.
full : Tuple[str, ...]
Full device names, with the first one being the default.
capture : Tuple[str, ...]
Capture device names, with the first one being the default.
"""
cdef readonly tuple basic
cdef readonly tuple full
cdef readonly tuple capture
def __cinit__(self) -> None:
cdef list basic = devmgr.enumerate(alure.DeviceEnumeration.Basic)
default: int = basic.index(devmgr.default_device_name(
alure.DefaultDeviceType.Basic))
basic[0], basic[default] = basic[default], basic[0]
self.basic = tuple(basic)
cdef list full = devmgr.enumerate(alure.DeviceEnumeration.Full)
default: int = full.index(devmgr.default_device_name(
alure.DefaultDeviceType.Full))
full[0], full[default] = full[default], full[0]
self.full = tuple(full)
cdef list capture = devmgr.enumerate(alure.DeviceEnumeration.Capture)
default: int = capture.index(devmgr.default_device_name(
alure.DefaultDeviceType.Capture))
capture[0], capture[default] = capture[default], capture[0]
self.capture = tuple(capture)
def __repr__(self) -> str:
return (f'{self.__class__.__name__}(basic={self.basic},'
f' full={self.full}, capture={self.capture})')
cdef class Device:
"""Audio mix output, via either a system stream or a hardware port.
This can be used as a context manager that calls `close` upon
completion of the block, even if an error occurs.
Parameters
----------
name : str, optional
The name of the playback device.
fallback : Iterable[str], optional
Device names to fallback to, default to an empty tuple.
Raises
------
RuntimeError
If device creation fails.
Warns
-----
RuntimeWarning
Before each fallback.
See Also
--------
device_names : Available device names
"""
cdef alure.Device impl
def __init__(self, name: str = '', fallback: Iterable[str] = ()) -> None:
names: Tuple[str] = name, *fallback
message: Optional[str] = None
for name in names:
if message is not None:
with catch_warnings():
simplefilter('always')
warn(message, category=RuntimeWarning)
try:
self.impl = devmgr.open_playback(name)
except RuntimeError:
message = f'failed to open device: {name}'
else:
return
raise RuntimeError(message)
def __enter__(self) -> Device: return self
def __exit__(self, *exc) -> Optional[bool]: self.close()
def __lt__(self, other: Any) -> bool:
if not isinstance(other, Device): return NotImplemented
return self.impl < ( other).impl
def __le__(self, other: Any) -> bool:
if not isinstance(other, Device): return NotImplemented
return self.impl <= ( other).impl
def __eq__(self, other: Any) -> bool:
if not isinstance(other, Device): return NotImplemented
return self.impl == ( other).impl
def __ne__(self, other: Any) -> bool:
if not isinstance(other, Device): return NotImplemented
return self.impl != ( other).impl
def __gt__(self, other: Any) -> bool:
if not isinstance(other, Device): return NotImplemented
return self.impl > ( other).impl
def __ge__(self, other: Any) -> bool:
if not isinstance(other, Device): return NotImplemented
return self.impl >= ( other).impl
def __bool__(self) -> bool: return self.impl
def __repr__(self) -> str:
return f'{self.__class__.__name__}({self.name!r})'
@getter
def name(self) -> str:
"""Name of the device."""
return self.impl.get_name(alure.PlaybackName.Full)
@getter
def basic_name(self) -> str:
"""Basic name of the device."""
return self.impl.get_name(alure.PlaybackName.Basic)
def query_extension(self, name: str) -> bool:
"""Return if an ALC extension exists on this device.
See Also
--------
query_extension : Query non-device-specific ALC extension
"""
return self.impl.query_extension(name)
@getter
def alc_version(self) -> Tuple[int, int]:
"""ALC version supported by this device."""
cdef alure.Version version = self.impl.get_alc_version()
return version.get_major(), version.get_minor()
@getter
def efx_version(self) -> Tuple[int, int]:
"""EFX version supported by this device.
If `ALC_EXT_EFX` extension is unsupported, this will be (0, 0).
"""
cdef alure.Version version = self.impl.get_efx_version()
return version.get_major(), version.get_minor()
@getter
def frequency(self) -> int:
"""Playback frequency in hertz."""
return self.impl.get_frequency()
@getter
def max_auxiliary_sends(self) -> int:
"""Maximum number of auxiliary source sends.
If `ALC_EXT_EFX` is unsupported, this will be 0.
"""
return self.impl.get_max_auxiliary_sends()
@getter
def hrtf_names(self) -> List[str]:
"""List of available HRTF names.
The order is retained from OpenAL, such that the index of
a given name is the ID to use with `ALC_HRTF_ID_SOFT`.
If `ALC_SOFT_HRTF` extension is unavailable,
this will be an empty list.
"""
return self.impl.enumerate_hrtf_names()
@getter
def hrtf_enabled(self) -> bool:
"""Whether HRTF is enabled on the device.
If `ALC_SOFT_HRTF` extension is unavailable,
this will return False although there could still be
HRTF applied at a lower hardware level.
"""
return self.impl.is_hrtf_enabled()
@getter
def current_hrtf(self) -> Optional[str]:
"""Name of the HRTF currently being used by this device.
If HRTF is not currently enabled, this will be `None`.
"""
name: str = self.impl.get_current_hrtf()
return name or None
def reset(self, attrs: Dict[int, int] = {}) -> None:
"""Reset the device, using the specified attributes.
If `ALC_SOFT_HRTF` extension is unavailable,
this will be a no-op.
"""
self.impl.reset(mkattrs(attrs.items()))
def pause_dsp(self) -> None:
"""Pause device processing and stop contexts' updates.
Multiple calls are allowed but it is not reference counted,
so the device will resume after one `resume_dsp` call.
This requires `ALC_SOFT_pause_device` extension.
"""
self.impl.pause_dsp()
def resume_dsp(self) -> None:
"""Resume device processing and restart contexts' updates.
Multiple calls are allowed and will no-op.
"""
self.impl.resume_dsp()
@getter
def clock_time(self) -> int:
"""Current clock time for the device.
Notes
-----
This starts relative to the device being opened, and does not
increment while there are no contexts nor while processing
is paused. Currently, this may not exactly match the rate
that sources play at. In the future it may utilize an OpenAL
extension to retrieve the audio device's real clock.
"""
return self.impl.get_clock_time().count()
def close(self) -> None:
"""Close and free the device.
All previously-created contexts must first be destroyed.
"""
self.impl.close()
cdef class Context:
"""Container maintaining the audio environment.
Context contains the environment's settings and components
such as sources, buffers and effects.
This can be used as a context manager, e.g. ::
with context:
...
is equivalent to ::
previous = current_context()
use_context(context)
try:
...
finally:
use_context(previous)
context.destroy()
Parameters
----------
device : Device
The `device` on which the context is to be created.
attrs : Dict[int, int]
Attributes specified for the context to be created.
Attributes
----------
device : Device
The device this context was created from.
listener : Listener
The listener instance of this context.
Raises
------
RuntimeError
If context creation fails.
"""
cdef alure.Context impl
cdef alure.Context previous
cdef readonly Device device
cdef readonly Listener listener
def __init__(self, device: Device, attrs: Dict[int, int] = {}) -> None:
self.impl = device.impl.create_context(mkattrs(attrs.items()))
self.device = device
self.listener = Listener(self)
self.impl.set_message_handler(shared_ptr[alure.MessageHandler](
new CppMessageHandler(MessageHandler())))
def __enter__(self) -> Context:
self.previous = alure.Context.get_current()
use_context(self)
return self
def __exit__(self, *exc) -> Optional[bool]:
alure.Context.make_current(self.previous)
self.destroy()
def __lt__(self, other: Any) -> bool:
if not isinstance(other, Context): return NotImplemented
return self.impl < ( other).impl
def __le__(self, other: Any) -> bool:
if not isinstance(other, Context): return NotImplemented
return self.impl <= ( other).impl
def __eq__(self, other: Any) -> bool:
if not isinstance(other, Context): return NotImplemented
return self.impl == ( other).impl
def __ne__(self, other: Any) -> bool:
if not isinstance(other, Context): return NotImplemented
return self.impl != ( other).impl
def __gt__(self, other: Any) -> bool:
if not isinstance(other, Context): return NotImplemented
return self.impl > ( other).impl
def __ge__(self, other: Any) -> bool:
if not isinstance(other, Context): return NotImplemented
return self.impl >= ( other).impl
def __bool__(self) -> bool: return self.impl
def destroy(self) -> None:
"""Destroy the context.
The context must not be current when this is called.
"""
self.impl.destroy()
def start_batch(self) -> None:
"""Suspend the context to start batching."""
self.impl.start_batch()
def end_batch(self) -> None:
"""Continue processing the context and end batching."""
self.impl.end_batch()
@property
def message_handler(self) -> MessageHandler:
"""Handler of some certain events."""
return static_pointer_cast[CppMessageHandler, alure.MessageHandler](
self.impl.get_message_handler()).get()[0].pyo
@message_handler.setter
def message_handler(self, message_handler: MessageHandler) -> None:
static_pointer_cast[CppMessageHandler, alure.MessageHandler](
self.impl.get_message_handler()).get()[0].pyo = message_handler
@property
def async_wake_interval(self) -> int:
"""Current interval used for waking up the background thread."""
return self.impl.get_async_wake_interval().count()
@async_wake_interval.setter
def async_wake_interval(self, value: int) -> None:
self.impl.set_async_wake_interval(milliseconds(value))
def is_supported(self, channel_config: str, sample_type: str) -> bool:
"""Return if the channel config and sample type is supported.
This method require the context to be current.
See Also
--------
sample_types : Set of sample types
channel_configs : Set of channel configurations
"""
cdef alure.ChannelConfig alure_channel_config
cdef alure.SampleType alure_sample_type
try:
alure_channel_config = CHANNEL_CONFIGS.at(channel_config)
except IndexError:
raise ValueError('invalid channel config: '
+ str(channel_config)) from None
try:
alure_sample_type = SAMPLE_TYPES.at(sample_type)
except IndexError:
raise ValueError(f'invalid sample type: {sample_type}') from None
return self.impl.is_supported(alure_channel_config, alure_sample_type)
@getter
def available_resamplers(self) -> List[str]:
"""The list of resamplers supported by the context.
If `AL_SOFT_source_resampler` extension is unsupported,
this will be an empty list. Otherwise there would be
at least one entry.
This method require the context to be current.
"""
cdef alure.ArrayView[string] resamplers
resamplers = self.impl.get_available_resamplers()
return [resampler for resampler in resamplers]
@getter
def default_resampler_index(self) -> int:
"""The context's default resampler index.
If `AL_SOFT_source_resampler` extension is unsupported,
this will return 0.
If you try to access the resampler list with this index
without extension, undefined behavior will occur
(accessing an out of bounds array index).
This method require the context to be current.
"""
return self.impl.get_default_resampler_index()
@setter
def doppler_factor(self, value: float) -> None:
"""Factor to apply to all source's doppler calculations."""
self.impl.set_doppler_factor(value)
@setter
def speed_of_sound(self, value: float) -> None:
"""The speed of sound propagation in units per second.
It is used to calculate the doppler effect along with other
distance-related time effects.
The default is 343.3 units per second (a realistic speed
assuming 1 meter per unit). If this is adjusted for a
different unit scale, `Listener.meters_per_unit` should
also be adjusted.
"""
self.impl.set_speed_of_sound(value)
@setter
def distance_model(self, value: str) -> None:
"""The model for source attenuation based on distance.
The default, 'inverse clamped', provides a realistic l/r
reduction in volume (that is, every doubling of distance
cause the gain to reduce by half).
The clamped distance models restrict the source distance for
the purpose of distance attenuation, so a source won't sound
closer than its reference distance or farther than its max
distance.
Raises
------
ValueError
If set to a preset cannot be found in `distance_models`.
"""
try:
self.impl.set_distance_model(DISTANCE_MODELS.at(value))
except IndexError:
raise ValueError(f'invalid distance model: {value}') from None
def update(self) -> None:
"""Update the context and all sources belonging to this context."""
self.impl.update()
# source_stopped is called outside of alure::Context::update
# to allow applications to destroy the source on this message.
handler: MessageHandler = self.message_handler
while handler.stopped_sources:
handler.source_stopped(handler.stopped_sources.pop())
cdef class Listener:
"""Listener instance of the given context.
It is recommended that applications access the listener via
`Context.listener`, which avoid the overhead caused by the
creation of the wrapper object.
Parameters
----------
context : Optional[Context], optional
The context on which the listener instance is to be created.
By default `current_context()` is used.
Raises
------
RuntimeError
If there is neither any context specified nor current.
"""
cdef alure.Listener impl
def __init__(self, context: Optional[Context] = None) -> None:
if context is None: context = current_context()
self.impl = ( context).impl.get_listener()
def __bool__(self) -> bool: return self.impl
@setter
def gain(self, value: float) -> None:
"""Master gain for all context output."""
self.impl.set_gain(value)
@setter
def position(self, value: Vector3) -> None:
"""3D position of the listener."""
self.impl.set_position(to_vector3(value))
@setter
def velocity(self, value: Vector3) -> None:
"""3D velocity of the listener, in units per second.
As with OpenAL, this does not actually alter the listener's
position, and instead just alters the pitch as determined by
the doppler effect.
"""
self.impl.set_velocity(to_vector3(value))
@setter
def orientation(self, value: Tuple[Vector3, Vector3]) -> None:
"""3D orientation of the listener.
Attributes
----------
at : Tuple[float, float, float]
Relative position.
up : Tuple[float, float, float]
Relative direction.
"""
at, up = value
self.impl.set_orientation(
pair[alure.Vector3, alure.Vector3](to_vector3(at), to_vector3(up)))
@setter
def meters_per_unit(self, value: float) -> None:
"""Number of meters per unit.
This is used for various effects relying on the distance
in meters including air absorption and initial reverb decay.
If this is changed, so should the speed of sound
(e.g. `context.speed_of_sound = 343.3 / meters_per_unit`
to maintain a realistic 343.3 m/s for sound propagation).
"""
self.impl.set_meters_per_unit(value)
cdef class Buffer:
"""Buffer of preloaded PCM samples coming from a `Decoder`.
Cached buffers must be freed using `destroy` before destroying
`context`. Alternatively, this can be used as a context manager
that calls `destroy` upon completion of the block,
even if an error occurs.
Parameters
----------
name : str
Audio file or resource name. Multiple calls with the same name
will return the same buffer.
context : Optional[Context], optional
The context from which the buffer is to be created and cached.
By default `current_context()` is used.
Attributes
----------
name : str
Audio file or resource name.
Raises
------
RuntimeError
If there is neither any context specified nor current.
"""
cdef alure.Buffer impl
cdef Context context
cdef readonly str name
def __init__(self, name: str, context: Optional[Context] = None) -> None:
if context is None: context = current_context()
if not context: raise RuntimeError('there is no context current')
self.context, self.name = context, name
self.impl = self.context.impl.find_buffer(self.name)
if not self:
decoder: Decoder = decode(self.name, self.context)
self.impl = self.context.impl.create_buffer_from(
self.name, decoder.pimpl)
def __enter__(self) -> Buffer: return self
def __exit__(self, *exc) -> Optional[bool]: self.destroy()
def __lt__(self, other: Any) -> bool:
if not isinstance(other, Buffer): return NotImplemented
return self.impl < ( other).impl
def __le__(self, other: Any) -> bool:
if not isinstance(other, Buffer): return NotImplemented
return self.impl <= ( other).impl
def __eq__(self, other: Any) -> bool:
if not isinstance(other, Buffer): return NotImplemented
return self.impl == ( other).impl
def __ne__(self, other: Any) -> bool:
if not isinstance(other, Buffer): return NotImplemented
return self.impl != ( other).impl
def __gt__(self, other: Any) -> bool:
if not isinstance(other, Buffer): return NotImplemented
return self.impl > ( other).impl
def __ge__(self, other: Any) -> bool:
if not isinstance(other, Buffer): return NotImplemented
return self.impl >= ( other).impl
def __bool__(self) -> bool: return self.impl
def __repr__(self) -> str:
return f'{self.__class__.__name__}({self.name!r})'
@staticmethod
def from_decoder(decoder: Decoder, name: str,
context: Optional[Context] = None) -> Buffer:
"""Return a buffer created by reading the given decoder.
Parameters
----------
decoder : Decoder
The decoder from which the buffer is to be cached.
name : str
The name to give to the buffer. It may alias an audio file,
but it must not currently exist in the buffer cache.
context : Optional[Context], optional
The context from which the buffer is to be created.
By default `current_context()` is used.
Raises
------
RuntimeError
If there is neither any context specified nor current;
or if `name` is already used for another buffer.
"""
if context is None: context = current_context()
buffer: Buffer = Buffer.__new__(Buffer)
buffer.context, buffer.name = context, name
buffer.impl = buffer.context.impl.create_buffer_from(
buffer.name, decoder.pimpl)
return buffer
@getter
def length(self) -> int:
"""Length of the buffer in sample frames."""
return self.impl.get_length()
@getter
def length_seconds(self) -> float:
"""Length of the buffer in seconds."""
return self.length / self.frequency
@getter
def frequency(self) -> int:
"""Buffer's frequency in hertz."""
return self.impl.get_frequency()
@getter
def channel_config(self) -> str:
"""Buffer's sample configuration."""
return alure.get_channel_config_name(
self.impl.get_channel_config())
@getter
def sample_type(self) -> str:
"""Buffer's sample type."""
return alure.get_sample_type_name(
self.impl.get_sample_type())
@getter
def size(self) -> int:
"""Storage size used by the buffer, in bytes.
Notes
-----
The size in bytes may not be what you expect from the length,
as it may take more space internally than the `channel_config`
and `sample_type` suggest.
"""
return self.impl.get_size()
def play(self, source: Optional[Source] = None) -> Source:
"""Play `source` using the buffer.
Return the source used for playing. If `None` is given,
create a new one.
One buffer may be played from multiple sources simultaneously.
"""
if source is None: source = Source(self.context)
(