# ---------------------------------------------------------------------------- # 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 sys import weakref from . import lib_pulseaudio as pa from pyglet.media.exceptions import MediaException from pyglet.util import debug_print import pyglet _debug = debug_print('debug_media') def get_uint32_or_none(value): # Check for max uint32 if value is None or value == 4294967295: return None return value def get_bool_or_none(value): if value < 0: return None elif value == 1: return True else: return False def get_ascii_str_or_none(value): if value is not None: return value.decode('ascii') return None class PulseAudioException(MediaException): def __init__(self, error_code, message): super(PulseAudioException, self).__init__(message) self.error_code = error_code self.message = message def __str__(self): return '{}: [{}] {}'.format(self.__class__.__name__, self.error_code, self.message) __repr__ = __str__ class PulseAudioMainLoop: def __init__(self): self._pa_threaded_mainloop = pa.pa_threaded_mainloop_new() self._pa_mainloop = pa.pa_threaded_mainloop_get_api(self._pa_threaded_mainloop) self._lock_count = 0 def __del__(self): self.delete() def start(self): """Start running the mainloop.""" with self: result = pa.pa_threaded_mainloop_start(self._pa_threaded_mainloop) if result < 0: raise PulseAudioException(0, "Failed to start PulseAudio mainloop") assert _debug('PulseAudioMainLoop: Started') def delete(self): """Clean up the mainloop.""" if self._pa_threaded_mainloop is not None: assert _debug("Delete PulseAudioMainLoop") pa.pa_threaded_mainloop_stop(self._pa_threaded_mainloop) pa.pa_threaded_mainloop_free(self._pa_threaded_mainloop) self._pa_threaded_mainloop = None self._pa_mainloop = None def lock(self): """Lock the threaded mainloop against events. Required for all calls into PA.""" assert self._pa_threaded_mainloop is not None pa.pa_threaded_mainloop_lock(self._pa_threaded_mainloop) self._lock_count += 1 def unlock(self): """Unlock the mainloop thread.""" assert self._pa_threaded_mainloop is not None # TODO: This is not completely safe. Unlock might be called without lock. assert self._lock_count > 0 self._lock_count -= 1 pa.pa_threaded_mainloop_unlock(self._pa_threaded_mainloop) def signal(self): """Signal the mainloop thread to break from a wait.""" assert self._pa_threaded_mainloop is not None pa.pa_threaded_mainloop_signal(self._pa_threaded_mainloop, 0) def wait(self): """Wait for a signal.""" assert self._pa_threaded_mainloop is not None # Although lock and unlock can be called reentrantly, the wait call only releases one lock. assert self._lock_count > 0 original_lock_count = self._lock_count while self._lock_count > 1: self.unlock() pa.pa_threaded_mainloop_wait(self._pa_threaded_mainloop) while self._lock_count < original_lock_count: self.lock() def create_context(self): return PulseAudioContext(self, self._context_new()) def _context_new(self): """Construct a new context in this mainloop.""" assert self._pa_mainloop is not None app_name = self._get_app_name() context = pa.pa_context_new(self._pa_mainloop, app_name.encode('ASCII') ) return context def _get_app_name(self): """Get the application name as advertised to the pulseaudio server.""" # TODO move app name into pyglet.app (also useful for OS X menu bar?). return sys.argv[0] def __enter__(self): self.lock() def __exit__(self, exc_type, exc_value, traceback): self.unlock() class PulseAudioLockable: def __init__(self, mainloop): assert mainloop is not None self.mainloop = weakref.ref(mainloop) def lock(self): """Lock the threaded mainloop against events. Required for all calls into PA.""" self.mainloop().lock() def unlock(self): """Unlock the mainloop thread.""" self.mainloop().unlock() def signal(self): """Signal the mainloop thread to break from a wait.""" self.mainloop().signal() def wait(self): """Wait for a signal.""" self.mainloop().wait() def __enter__(self): self.lock() def __exit__(self, exc_type, exc_value, traceback): self.unlock() class PulseAudioContext(PulseAudioLockable): """Basic object for a connection to a PulseAudio server.""" _state_name = {pa.PA_CONTEXT_UNCONNECTED: 'Unconnected', pa.PA_CONTEXT_CONNECTING: 'Connecting', pa.PA_CONTEXT_AUTHORIZING: 'Authorizing', pa.PA_CONTEXT_SETTING_NAME: 'Setting Name', pa.PA_CONTEXT_READY: 'Ready', pa.PA_CONTEXT_FAILED: 'Failed', pa.PA_CONTEXT_TERMINATED: 'Terminated'} def __init__(self, mainloop, pa_context): super(PulseAudioContext, self).__init__(mainloop) self._pa_context = pa_context self.state = None self._connect_callbacks() def __del__(self): if self._pa_context is not None: with self: self.delete() def delete(self): """Completely shut down pulseaudio client.""" if self._pa_context is not None: assert _debug("PulseAudioContext.delete") if self.is_ready: pa.pa_context_disconnect(self._pa_context) while self.state is not None and not self.is_terminated: self.wait() self._disconnect_callbacks() pa.pa_context_unref(self._pa_context) self._pa_context = None @property def is_ready(self): return self.state == pa.PA_CONTEXT_READY @property def is_failed(self): return self.state == pa.PA_CONTEXT_FAILED @property def is_terminated(self): return self.state == pa.PA_CONTEXT_TERMINATED @property def server(self): if self.is_ready: return get_ascii_str_or_none(pa.pa_context_get_server(self._pa_context)) else: return None @property def protocol_version(self): if self._pa_context is not None: return get_uint32_or_none(pa.pa_context_get_protocol_version(self._pa_context)) @property def server_protocol_version(self): if self._pa_context is not None: return get_uint32_or_none(pa.pa_context_get_server_protocol_version(self._pa_context)) @property def is_local(self): if self._pa_context is not None: return get_bool_or_none(pa.pa_context_is_local(self._pa_context)) def connect(self, server=None): """Connect the context to a PulseAudio server. :Parameters: `server` : str Server to connect to, or ``None`` for the default local server (which may be spawned as a daemon if no server is found). """ assert self._pa_context is not None self.state = None with self: self.check( pa.pa_context_connect(self._pa_context, server, 0, None) ) while not self.is_failed and not self.is_ready: self.wait() if self.is_failed: self.raise_error() def create_stream(self, audio_format): """ Create a new audio stream. """ mainloop = self.mainloop() assert mainloop is not None assert self.is_ready sample_spec = self.create_sample_spec(audio_format) channel_map = None # TODO It is now recommended to use pa_stream_new_with_proplist() stream = pa.pa_stream_new(self._pa_context, str(id(self)).encode('ASCII'), sample_spec, channel_map) self.check_not_null(stream) return PulseAudioStream(mainloop, self, stream) def create_sample_spec(self, audio_format): """ Create a PulseAudio sample spec from pyglet audio format. """ sample_spec = pa.pa_sample_spec() if audio_format.sample_size == 8: sample_spec.format = pa.PA_SAMPLE_U8 elif audio_format.sample_size == 16: if sys.byteorder == 'little': sample_spec.format = pa.PA_SAMPLE_S16LE else: sample_spec.format = pa.PA_SAMPLE_S16BE elif audio_format.sample_size == 24: if sys.byteorder == 'little': sample_spec.format = pa.PA_SAMPLE_S24LE else: sample_spec.format = pa.PA_SAMPLE_S24BE else: raise MediaException('Unsupported sample size') sample_spec.rate = audio_format.sample_rate sample_spec.channels = audio_format.channels return sample_spec def set_input_volume(self, stream, volume): """ Set the volume for a stream. """ cvolume = self._get_cvolume_from_linear(stream, volume) idx = stream.index op = PulseAudioOperation(self, succes_cb_t=pa.pa_context_success_cb_t) op.execute( pa.pa_context_set_sink_input_volume(self._pa_context, idx, cvolume, op.pa_callback, None) ) return op def _get_cvolume_from_linear(self, stream, volume): cvolume = pa.pa_cvolume() volume = pa.pa_sw_volume_from_linear(volume) pa.pa_cvolume_set(cvolume, stream.audio_format.channels, volume) return cvolume def _connect_callbacks(self): self._state_cb_func = pa.pa_context_notify_cb_t(self._state_callback) pa.pa_context_set_state_callback(self._pa_context, self._state_cb_func, None) def _disconnect_callbacks(self): self._state_cb_func = None pa.pa_context_set_state_callback(self._pa_context, pa.pa_context_notify_cb_t(0), None) def _state_callback(self, context, userdata): self.state = pa.pa_context_get_state(self._pa_context) assert _debug('PulseAudioContext: state changed to {}'.format( self._state_name[self.state])) self.signal() def check(self, result): if result < 0: self.raise_error() return result def check_not_null(self, value): if value is None: self.raise_error() return value def check_ptr_not_null(self, value): if not value: self.raise_error() return value def raise_error(self): error = pa.pa_context_errno(self._pa_context) raise PulseAudioException(error, get_ascii_str_or_none(pa.pa_strerror(error))) class PulseAudioStream(PulseAudioLockable, pyglet.event.EventDispatcher): """PulseAudio audio stream.""" _state_name = {pa.PA_STREAM_UNCONNECTED: 'Unconnected', pa.PA_STREAM_CREATING: 'Creating', pa.PA_STREAM_READY: 'Ready', pa.PA_STREAM_FAILED: 'Failed', pa.PA_STREAM_TERMINATED: 'Terminated'} def __init__(self, mainloop, context, pa_stream): PulseAudioLockable.__init__(self, mainloop) self._pa_stream = pa_stream self.context = weakref.ref(context) self.state = None self.underflow = False pa.pa_stream_ref(self._pa_stream) self._connect_callbacks() self._refresh_state() def __del__(self): if self._pa_stream is not None: self.delete() def delete(self): context = self.context() if context is None: assert _debug("No active context anymore. Cannot disconnect the stream") self._pa_stream = None return if self._pa_stream is None: assert _debug("No stream to delete.") return assert _debug("Delete PulseAudioStream") if not self.is_unconnected: assert _debug("PulseAudioStream: disconnecting") with self: context.check( pa.pa_stream_disconnect(self._pa_stream) ) while not (self.is_terminated or self.is_failed): self.wait() self._disconnect_callbacks() pa.pa_stream_unref(self._pa_stream) self._pa_stream = None @property def is_unconnected(self): return self.state == pa.PA_STREAM_UNCONNECTED @property def is_creating(self): return self.state == pa.PA_STREAM_CREATING @property def is_ready(self): return self.state == pa.PA_STREAM_READY @property def is_failed(self): return self.state == pa.PA_STREAM_FAILED @property def is_terminated(self): return self.state == pa.PA_STREAM_TERMINATED @property def writable_size(self): assert self._pa_stream is not None return pa.pa_stream_writable_size(self._pa_stream) @property def index(self): assert self._pa_stream is not None return pa.pa_stream_get_index(self._pa_stream) @property def is_corked(self): assert self._pa_stream is not None return get_bool_or_none(pa.pa_stream_is_corked(self._pa_stream)) @property def audio_format(self): assert self._pa_stream is not None return pa.pa_stream_get_sample_spec(self._pa_stream)[0] def connect_playback(self): context = self.context() assert self._pa_stream is not None assert context is not None device = None buffer_attr = None flags = (pa.PA_STREAM_START_CORKED | pa.PA_STREAM_INTERPOLATE_TIMING | pa.PA_STREAM_VARIABLE_RATE) volume = None sync_stream = None # TODO use this context.check( pa.pa_stream_connect_playback(self._pa_stream, device, buffer_attr, flags, volume, sync_stream) ) while not self.is_ready and not self.is_failed: self.wait() if not self.is_ready: context.raise_error() assert _debug('PulseAudioStream: Playback connected') def write(self, audio_data, length=None, seek_mode=pa.PA_SEEK_RELATIVE): context = self.context() assert context is not None assert self._pa_stream is not None assert self.is_ready if length is None: length = min(audio_data.length, self.writable_size) assert _debug('PulseAudioStream: writing {} bytes'.format(length)) assert _debug('PulseAudioStream: writable size before write {} bytes'.format(self.writable_size)) context.check( pa.pa_stream_write(self._pa_stream, audio_data.data, length, pa.pa_free_cb_t(0), # Data is copied 0, seek_mode) ) assert _debug('PulseAudioStream: writable size after write {} bytes'.format(self.writable_size)) self.underflow = False return length def update_timing_info(self, callback=None): context = self.context() assert context is not None assert self._pa_stream is not None op = PulseAudioOperation(context, callback) op.execute( pa.pa_stream_update_timing_info(self._pa_stream, op.pa_callback, None) ) return op def get_timing_info(self): context = self.context() assert context is not None assert self._pa_stream is not None timing_info = context.check_ptr_not_null( pa.pa_stream_get_timing_info(self._pa_stream) ) return timing_info.contents def trigger(self, callback=None): context = self.context() assert context is not None assert self._pa_stream is not None op = PulseAudioOperation(context) op.execute( pa.pa_stream_trigger(self._pa_stream, op.pa_callback, None) ) return op def prebuf(self, callback=None): context = self.context() assert context is not None assert self._pa_stream is not None op = PulseAudioOperation(context) op.execute( pa.pa_stream_prebuf(self._pa_stream, op.pa_callback, None) ) return op def resume(self, callback=None): return self._cork(False, callback) def pause(self, callback=None): return self._cork(True, callback) def update_sample_rate(self, sample_rate, callback=None): context = self.context() assert context is not None assert self._pa_stream is not None op = PulseAudioOperation(context) op.execute( pa.pa_stream_update_sample_rate(self._pa_stream, int(sample_rate), op.pa_callback, None) ) return op def _cork(self, pause, callback): context = self.context() assert context is not None assert self._pa_stream is not None op = PulseAudioOperation(context) op.execute( pa.pa_stream_cork(self._pa_stream, 1 if pause else 0, op.pa_callback, None) ) return op def _connect_callbacks(self): self._cb_underflow = pa.pa_stream_notify_cb_t(self._underflow_callback) self._cb_write = pa.pa_stream_request_cb_t(self._write_callback) self._cb_state = pa.pa_stream_notify_cb_t(self._state_callback) pa.pa_stream_set_underflow_callback(self._pa_stream, self._cb_underflow, None) pa.pa_stream_set_write_callback(self._pa_stream, self._cb_write, None) pa.pa_stream_set_state_callback(self._pa_stream, self._cb_state, None) def _disconnect_callbacks(self): self._cb_underflow = None self._cb_write = None self._cb_state = None pa.pa_stream_set_underflow_callback(self._pa_stream, pa.pa_stream_notify_cb_t(0), None) pa.pa_stream_set_write_callback(self._pa_stream, pa.pa_stream_request_cb_t(0), None) pa.pa_stream_set_state_callback(self._pa_stream, pa.pa_stream_notify_cb_t(0), None) def _underflow_callback(self, stream, userdata): assert _debug("PulseAudioStream: underflow") self.underflow = True self._write_needed() self.signal() def _write_callback(self, stream, nbytes, userdata): assert _debug("PulseAudioStream: write requested") self._write_needed(nbytes) self.signal() def _state_callback(self, stream, userdata): self._refresh_state() assert _debug("PulseAudioStream: state changed to {}".format(self._state_name[self.state])) self.signal() def _refresh_state(self): if self._pa_stream is not None: self.state = pa.pa_stream_get_state(self._pa_stream) def _write_needed(self, nbytes=None): if nbytes is None: nbytes = self.writable_size # This dispatch call is made from the threaded mainloop thread! pyglet.app.platform_event_loop.post_event( self, 'on_write_needed', nbytes, self.underflow) def on_write_needed(self, nbytes, underflow): """A write is requested from PulseAudio. Called from the PulseAudio mainloop, so no locking required. :event: """ PulseAudioStream.register_event_type('on_write_needed') class PulseAudioOperation(PulseAudioLockable): """Asynchronous PulseAudio operation""" _state_name = {pa.PA_OPERATION_RUNNING: 'Running', pa.PA_OPERATION_DONE: 'Done', pa.PA_OPERATION_CANCELLED: 'Cancelled'} def __init__(self, context, callback=None, pa_operation=None, succes_cb_t=pa.pa_stream_success_cb_t): mainloop = context.mainloop() assert mainloop is not None PulseAudioLockable.__init__(self, mainloop) self.context = weakref.ref(context) self._callback = callback self.pa_callback = succes_cb_t(self._success_callback) if pa_operation is not None: self.execute(pa_operation) else: self._pa_operation = None def __del__(self): if self._pa_operation is not None: with self: self.delete() def delete(self): if self._pa_operation is not None: assert _debug("PulseAudioOperation.delete({})".format(id(self))) pa.pa_operation_unref(self._pa_operation) self._pa_operation = None def execute(self, pa_operation): context = self.context() assert context is not None context.check_ptr_not_null(pa_operation) assert _debug("PulseAudioOperation.execute({})".format(id(self))) self._pa_operation = pa_operation self._get_state() return self def cancel(self): assert self._pa_operation is not None pa.pa_operation_cancel(self._pa_operation) return self @property def is_running(self): return self._get_state() == pa.PA_OPERATION_RUNNING @property def is_done(self): return self._get_state() == pa.PA_OPERATION_DONE @property def is_cancelled(self): return self._get_state() == pa.PA_OPERATION_CANCELLED def wait(self): """Wait until operation is either done or cancelled.""" while self.is_running: super(PulseAudioOperation, self).wait() return self def _get_state(self): assert self._pa_operation is not None return pa.pa_operation_get_state(self._pa_operation) def _success_callback(self, stream, success, userdata): if self._callback: self._callback() self.pa_callback = None # Clean up callback, not called anymore self.signal()