Giter Site home page Giter Site logo

mk-fg / python-pulse-control Goto Github PK

View Code? Open in Web Editor NEW
169.0 9.0 36.0 355 KB

Python high-level interface and ctypes-based bindings for PulseAudio (libpulse)

Home Page: https://pypi.org/project/pulsectl/

License: MIT License

Python 99.57% Shell 0.43%
pulseaudio python bindings

python-pulse-control's Introduction

python-pulse-control (pulsectl module)

Python (3.x and 2.x) blocking high-level interface and ctypes-based bindings for PulseAudio (libpulse), to use in a simple synchronous code.

Wrappers are mostly for mixer-like controls and introspection-related operations, as opposed to e.g. submitting sound samples to play and player-like client.

For async version to use with asyncio, see pulsectl-asyncio project instead.

Originally forked from pulsemixer project, which had this code bundled.


Repository URLs:

Usage

Simple example:

import pulsectl

with pulsectl.Pulse('volume-increaser') as pulse:
  for sink in pulse.sink_list():
    # Volume is usually in 0-1.0 range, with >1.0 being soft-boosted
    pulse.volume_change_all_chans(sink, 0.1)

Listening for server state change events:

import pulsectl

with pulsectl.Pulse('event-printer') as pulse:
  # print('Event types:', pulsectl.PulseEventTypeEnum)
  # print('Event facilities:', pulsectl.PulseEventFacilityEnum)
  # print('Event masks:', pulsectl.PulseEventMaskEnum)

  def print_events(ev):
    print('Pulse event:', ev)
    ### Raise PulseLoopStop for event_listen() to return before timeout (if any)
    # raise pulsectl.PulseLoopStop

  pulse.event_mask_set('all')
  pulse.event_callback_set(print_events)
  pulse.event_listen(timeout=10)

Misc other tinkering:

>>> import pulsectl
>>> pulse = pulsectl.Pulse('my-client-name')

>>> pulse.sink_list()
[<PulseSinkInfo at 7f85cfd053d0 - desc='Built-in Audio', index=0L, mute=0, name='alsa-speakers', channels=2, volumes='44.0%, 44.0%'>]

>>> pulse.sink_input_list()
[<PulseSinkInputInfo at 7fa06562d3d0 - index=181L, mute=0, name='mpv Media Player', channels=2, volumes='25.0%, 25.0%'>]

>>> pulse.sink_input_list()[0].proplist
{'application.icon_name': 'mpv',
 'application.language': 'C',
 'application.name': 'mpv Media Player',
 ...
 'native-protocol.version': '30',
 'window.x11.display': ':1.0'}

>>> pulse.source_list()
[<PulseSourceInfo at 7fcb0615d8d0 - desc='Monitor of Built-in Audio', index=0L, mute=0, name='alsa-speakers.monitor', channels=2, volumes='100.0%, 100.0%'>,
 <PulseSourceInfo at 7fcb0615da10 - desc='Built-in Audio', index=1L, mute=0, name='alsa-mic', channels=2, volumes='100.0%, 100.0%'>]

>>> sink = pulse.sink_list()[0]
>>> pulse.volume_change_all_chans(sink, -0.1)
>>> pulse.volume_set_all_chans(sink, 0.5)

>>> pulse.server_info().default_sink_name
'alsa_output.pci-0000_00_14.2.analog-stereo'
>>> pulse.default_set(sink)

>>> card = pulse.card_list()[0]
>>> card.profile_list
[<PulseCardProfileInfo at 7f02e7e88ac8 - description='Analog Stereo Input', n_sinks=0, n_sources=1, name='input:analog-stereo', priority=60>,
 <PulseCardProfileInfo at 7f02e7e88b70 - description='Analog Stereo Output', n_sinks=1, n_sources=0, name='output:analog-stereo', priority=6000>,
 ...
 <PulseCardProfileInfo at 7f02e7e9a4e0 - description='Off', n_sinks=0, n_sources=0, name='off', priority=0>]
>>> pulse.card_profile_set(card, 'output:hdmi-stereo')

>>> help(pulse)
...

>>> pulse.close()

Current code logic is that all methods are invoked through the Pulse instance, and everything returned from these are "Pulse-Something-Info" objects - thin wrappers around C structs that describe the thing, without any methods attached.

Aside from a few added convenience methods, most of them should have similar signature and do same thing as their C libpulse API counterparts, so see pulseaudio doxygen documentation for more information on them.

Pulse client can be integrated into existing eventloop (e.g. asyncio, twisted, etc) using Pulse.set_poll_func() or Pulse.event_listen() in a separate thread.

Somewhat extended usage example can be found in pulseaudio-mixer-cli project code, as well as tests here.

Notes

Some less obvious things are described in this section.

Things not yet wrapped/exposed in python

There are plenty of information, methods and other things in libpulse not yet wrapped/exposed by this module, as they weren't needed (yet) for author/devs use-case(s).

Making them accessible from python code can be as simple as adding an attribute name to the "c_struct_fields" value in PulseSomethingInfo objects.

See github #3 for a more concrete example of finding/adding such stuff.

For info and commands that are not available through libpulse introspection API, it is possible to use pulsectl.connect_to_cli() fallback function, which will open unix socket to server's "module-cli" (signaling to load it, if necessary), which can be used in exactly same way as "pacmd" tool (not to be confused with "pactl", which uses native protocol instead of module-cli) or pulseaudio startup files (e.g. "default.pa").

Probably a bad idea to parse string output from commands there though, as these are not only subject to change, but can also vary depending on system locale.

Volume

In PulseAudio, "volume" for anything is not a flat number, but essentially a list of numbers, one per channel (as in "left", "right", "front", "rear", etc), which should correspond to channel map of the object it relates/is-applied to.

In this module, such lists are represented by PulseVolumeInfo objects.

I.e. sink.volume is a PulseVolumeInfo instance, and all thin/simple wrappers that accept index of the object, expect such instance to be passed, e.g. pulse.sink_input_volume_set(sink.index, sink.volume).

There are convenience volume_get_all_chans, volume_set_all_chans and volume_change_all_chans methods to get/set/adjust volume as/by a single numeric value, which is also accessible on PulseVolumeInfo objects as a value_flat property.

PulseVolumeInfo can be constructed from a numeric volume value plus number of channels, or a python list of per-channel numbers.

All per-channel volume values in PulseVolumeInfo (and flat values in the wrapper funcs above), are float objects in 0-65536 range, with following meanings:

  • 0.0 volume is "no sound" (corresponds to PA_VOLUME_MUTED).
  • 1.0 value is "current sink volume level", 100% or PA_VOLUME_NORM.
  • >1.0 and up to 65536.0 (PA_VOLUME_MAX / PA_VOLUME_NORM) - software-boosted sound volume (higher values will negatively affect sound quality).

Probably a good idea to set volume only in 0-1.0 range and boost volume in hardware without quality loss, e.g. by tweaking sink volume (which corresponds to ALSA/hardware volume), if that option is available.

Note that flat-volumes=yes option ("yes" by default on some distros, "no" in e.g. Arch Linux) in pulseaudio daemon.conf already scales device-volume with the volume of the "loudest" application, so already does what's suggested above.

Fractional volume values used in the module get translated (in a linear fashion) to/from pa_volume_t integers for libpulse. See src/pulse/volume.h in pulseaudio sources for all the gory details on the latter (e.g. how it relates to sound level in dB).

Code example:

from pulsectl import Pulse, PulseVolumeInfo

with Pulse('volume-example') as pulse:
  sink_input = pulse.sink_input_list()[0] # first random sink-input stream

  volume = sink_input.volume
  print(volume.values) # list of per-channel values (floats)
  print(volume.value_flat) # average level across channels (float)

  time.sleep(1)

  volume.value_flat = 0.3 # sets all volume.values to 0.3
  pulse.volume_set(sink_input, volume) # applies the change

  time.sleep(1)

  n_channels = len(volume.values)
  new_volume = PulseVolumeInfo(0.5, n_channels) # 0.5 across all n_channels
  # new_volume = PulseVolumeInfo([0.15, 0.25]) # from a list of channel levels (stereo)
  pulse.volume_set(sink_input, new_volume)
  # pulse.sink_input_volume_set(sink_input.index, new_volume) # same as above

In most common cases, doing something like pulse.volume_set_all_chans(sink_input, 0.2) should do the trick though - no need to bother with specific channels in PulseVolumeInfo there.

String values

libpulse explicitly returns utf-8-encoded string values, which are always decoded to "abstract string" type in both python-2 (where it's called "unicode") and python-3 ("str"), for consistency.

It might be wise to avoid mixing these with encoded strings ("bytes") in the code, especially in python-2, where "bytes" is often used as a default string type.

Enumerated/named values (enums)

In place of C integers that correspond to some enum or constant (e.g. -1 for PA_SINK_INVALID_STATE), module returns EnumValue objects, which are comparable to strings ("str" type in py2/py3).

For example:

>>> pulsectl.PulseEventTypeEnum.change == 'change'
True
>>> pulsectl.PulseEventTypeEnum.change
<EnumValue event-type=change>
>>> pulsectl.PulseEventTypeEnum
<Enum event-type [change new remove]>

It might be preferrable to use enums instead of strings in the code so that interpreter can signal error on any typos or unknown values specified, as opposed to always silently failing checks with bogus strings.

Event-handling code, threads

libpulse clients always work as an event loop, though this module kinda hides it, presenting a more old-style blocking interface.

So what happens on any call (e.g. pulse.mute(...)) is:

  • Make a call to libpulse, specifying callback for when operation will be completed.
  • Run libpulse event loop until that callback gets called.
  • Return result passed to that callback call, if any (for various "get" methods).

event_callback_set() and event_listen() calls essentally do raw first and second step here.

Which means that any pulse calls from callback function can't be used when event_listen() (or any other pulse call through this module, for that matter) waits for return value and runs libpulse loop already.

One can raise PulseLoopStop exception there to make event_listen() return, run whatever pulse calls after that, then re-start the event_listen() thing.

This will not miss any events, as all blocking calls do same thing as event_listen() does (second step above), and can cause callable passed to event_callback_set() to be called (when loop is running).

Also, same instance of libpulse eventloop can't be run from different threads, naturally, so if threads are used, client can be initialized with threading_lock=True option (can also accept lock instance instead of True) to create a mutex around step-2 (run event loop) from the list above, so multiple threads won't do it at the same time.

For proper python eventloop integration (think twisted or asyncio), use pulsectl-asyncio module instead.

There are also some tricks mentioned in github #11 to shoehorn this module into async apps, but even with non-asyncio eventloop, starting from pulsectl-asyncio would probably be much easier.

Tests

Test code is packaged/installed with the module and can be useful to run when changing module code, or to check if current python, module and pulseudio versions all work fine together.

Commands to run tests from either checkout directory or installed module:

% python2 -m unittest discover
% python3 -m unittest discover

Note that if "pulsectl" module is available both in current directory (e.g. checkout dir) and user/system python module path, former should always take priority for commands above.

Add e.g. -k test_stream_move for commands above to match and run specific test(s), and when isolating specific failure, it might also be useful to run with PA_DEBUG=1 env-var to get full verbose pulseaudio log, for example:

% PA_DEBUG=1 python -m unittest discover -k test_module_funcs

Test suite runs ad-hoc isolated pulseaudio instance with null-sinks (not touching hardware), custom (non-default) startup script and environment, and interacts only with that instance, terminating it afterwards. Still uses system/user daemon.conf files though, so these can affect the tests.

Any test failures can indicate incompatibilities, bugs in the module code, issues with pulseaudio (or its daemon.conf) and underlying dependencies. There are no "expected" test case failures.

All tests can run for up to 10 seconds currently (v19.9.6), due to some involving playback (using paplay from /dev/urandom) being time-sensitive.

Changelog and versioning scheme

This package uses one-version-per-commit scheme (updated by pre-commit hook) and pretty much one release per git commit, unless more immediate follow-up commits are planned or too lazy to run py setup.py sdist bdist_wheel upload for some trivial README typo fix.

Version scheme: {year}.{month}.{git-commit-count-this-month}
I.e. "16.9.10" is "11th commit on Sep 2016".

There is a CHANGES.rst file with the list of any intentional breaking changes (should be exceptionally rare, if any) and new/added non-trivial functionality.

It can be a bit out of date though, as one has to remember to update it manually.
"Last synced/updated:" line there might give a hint as to by how much.

Installation

It's a regular package for Python (3.x or 2.x).

If a package is available for your distribution, using your package manager is the recommended way to install it.

Otherwise, using pip is the best way:

% pip install pulsectl

(add --user option to install into $HOME for current user only)

Be sure to use python3/python2, pip3/pip2, easy_install-... commands based on which python version you want to install the module for, if you are still using python2 (and likely have python3 on the system as well).

If you don't have "pip" command:

% python -m ensurepip
% python -m pip install --upgrade pip
% python -m pip install pulsectl

(same suggestion wrt "install --user" as above)

On a very old systems, one of these might work:

% curl https://bootstrap.pypa.io/get-pip.py | python
% pip install pulsectl

% easy_install pulsectl

% git clone --depth=1 https://github.com/mk-fg/python-pulse-control
% cd python-pulse-control
% python setup.py install

(all of install-commands here also have --user option)

Current-git version can be installed like this:

% pip install 'git+https://github.com/mk-fg/python-pulse-control#egg=pulsectl'

Note that to install stuff to system-wide PATH and site-packages (without --user), elevated privileges (i.e. root and su/sudo) are often required.

Use "...install --user", ~/.pydistutils.cfg or virtualenv to do unprivileged installs into custom paths.

More info on python packaging can be found at packaging.python.org.

Links

  • pulsemixer - initial source for this project (embedded in the tool).

  • pulsectl-asyncio - similar libpulse wrapper to this one, but for async python code.

  • libpulseaudio - different libpulse bindings module, more low-level, auto-generated from pulseaudio header files.

    Branches there have bindings for different (newer) pulseaudio versions.

  • pypulseaudio - high-level bindings module, rather similar to this one.

  • pulseaudio-mixer-cli - alsamixer-like script built on top of this module.

python-pulse-control's People

Contributors

alex3d avatar avian2 avatar bertwesarg avatar mfeif avatar mhthies avatar mk-fg avatar mraerino avatar mweinelt avatar natesymer avatar sbraz avatar taka-kazu avatar thejohnfreeman avatar tuffnerdstuff avatar tyilo avatar

Stargazers

 avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar

Watchers

 avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar  avatar

python-pulse-control's Issues

c_uint32 type for the 'channels' field in PA_SAMPLE_SPEC

The channels field's type is defined as uint8_t in the sample.h pulselib header.

In the PA_SAMPLE_SPEC Structure defined in _pulsectl.py, it is defined as c_uint32 instead. This works on little endian architectures (most platforms currently) when the compiler has zeroed all the bytes in the structure, but it fails on big endian ones: it is the highest significant byte on a big endian uint32_t that is located at the lowest memory location.

Using Python, one can check that byte 0x01 is the first byte in memory for little endian and the fourth byte for big endian.

>>> import struct
>>> print(struct.pack('<I', 0x04030201))    # integer little endian
b'\x01\x02\x03\x04'
>>> print(struct.pack('>I', 0x04030201))    # integer big endian
b'\x04\x03\x02\x01'

The other Structures that use the 'channels' field define correctly the type as c_uint8.

BTW many thanks for this great project !

Backward-compatibility issue

Is it really necessary to make not backward-compatible changes in API?

AttributeError: module 'pulsectl._pulsectl' has no attribute 'PA_SAMPLE_FLOAT32BE'. Did you mean: 'PA_SAMPLE_FLOAT32NE'?

present in volctl after your last change in package

https://aur.archlinux.org/packages/python-pulsectl#comment-913427

It worked with python 3.11 yesterday already. It is not possible to run it today. Is it possible to ensure backward compatibility?

I'm not sure how much apps can this incompatibility change destroyed. Can you explain the reasons for it and what are migration steps? Thanks.

Trying to set application/client volumes

I'm trying to change the volumes of clients with a simple script:

from pulsectl import Pulse

with Pulse('ALSA plug-in [python2.7]') as pulse:
  print(pulse.sink_input_list())
  for i,cl in enumerate(pulse.sink_input_list()):
    print('-------------------')
    print(cl.name)
    pulse.sink_input_volume_set(index=cl.index,vol=1000.0)

However, I get the error:

Traceback (most recent call last):
  File "ctl.py", line 17, in <module>
    pulse.sink_input_volume_set(index=cl.index,vol=1000.0)
  File "/usr/local/lib/python2.7/dist-packages/pulsectl/pulsectl.py", line 531, in _wrapper
    pulse_args = func(*args, **kws) if func else list()
  File "/usr/local/lib/python2.7/dist-packages/pulsectl/pulsectl.py", line 560, in <lambda>
    c.pa.context_set_sink_input_volume, lambda vol: vol.to_struct() )
AttributeError: 'float' object has no attribute 'to_struct'

According to the docs, "All volume values in this module are float objects in 0-65536 range", so I'm not clear on what sink_input_volume_set is expecting besides an integer. Any insight?

State of sink inputs: Is there sink input?

I am trying to write a script to change the volume of the running application instead of the entire sink.
I tried to figure out which sink input is currently playing sound. Ideally I would get the current sound level for every sink input and if it is greater than zero, it is running.

I found pacmd list-sink-inputs which has state (not perfect since some inputs always seem running).

โฏ pacmd list-sink-inputs |grep -e state: -e index: -e client: # spotify stopped
    index: 1
	state: RUNNING
	client: 7 <TeamSpeak3>
    index: 15
	state: CORKED
	client: 2 <Spotify>
โฏ pacmd list-sink-inputs |grep -e state: -e index: -e client: # spotify playing
    index: 1
	state: RUNNING
	client: 7 <TeamSpeak3>
    index: 15
	state: RUNNING
	client: 2 <Spotify>

I tried to find this state with pulsectl but I failed. Is there any way to get the state of sink-input or its current sound level? It might be the client state I am looking for. I don't know...

error on macbook : "OSError: dlopen(libpulse.so.0, 6): image not found"

I am getting below error on my macbook:

Jans-MBP:tmp jan$ python3
Python 3.7.4 (v3.7.4:e09359112e, Jul  8 2019, 14:54:52) 
[Clang 6.0 (clang-600.0.57)] on darwin
Type "help", "copyright", "credits" or "license" for more information.
>>> import pulsectl
Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/site-packages/pulsectl/__init__.py", line 4, in <module>
    from . import _pulsectl
  File "/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/site-packages/pulsectl/_pulsectl.py", line 669, in <module>
    pa = LibPulse()
  File "/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/site-packages/pulsectl/_pulsectl.py", line 621, in __init__
    p = CDLL('libpulse.so.0')
  File "/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/ctypes/__init__.py", line 364, in __init__
    self._handle = _dlopen(self._name, mode)
OSError: dlopen(libpulse.so.0, 6): image not found
>>> 

Note that pulseaudio is installed on my macbook when trying to import pulsectl

Jans-MBP:tmp jan$ pulseaudio --version
W: [] caps.c: Normally all extra capabilities would be dropped now, but that's impossible because PulseAudio was built without capabilities support.
pulseaudio 13.0
Jans-MBP:tmp jan$ 

Can't list ports of sinks/sources

Hello,

I would like to list all the ports of a sink or a source but I didn't find anything to do this in your wrapper.
Do you have any information about it ? Is it possible ?

Thanks,
Seevoid

PyPy3 errors: TypeError: expected integer, got NoneType object

When I run tests with PyPy3, they pass but I see a lot of these errors. Is PyPy3 officially supported?
For instance:

$ pypy3 -m unittest discover -k test_default_set
TypeError: expected integer, got NoneType object
TypeError: expected integer, got NoneType object
TypeError: expected integer, got NoneType object
TypeError: expected integer, got NoneType object
TypeError: expected integer, got NoneType object
TypeError: expected integer, got NoneType object
TypeError: expected integer, got NoneType object
TypeError: expected integer, got NoneType object
TypeError: expected integer, got NoneType object
TypeError: expected integer, got NoneType object
TypeError: expected integer, got NoneType object
TypeError: expected integer, got NoneType object
TypeError: expected integer, got NoneType object
TypeError: expected integer, got NoneType object
TypeError: expected integer, got NoneType object
TypeError: expected integer, got NoneType object
TypeError: expected integer, got NoneType object
TypeError: expected integer, got NoneType object
TypeError: expected integer, got NoneType object
TypeError: expected integer, got NoneType object
TypeError: expected integer, got NoneType object
TypeError: expected integer, got NoneType object
TypeError: expected integer, got NoneType object
TypeError: expected integer, got NoneType object
TypeError: expected integer, got NoneType object
TypeError: expected integer, got NoneType object
TypeError: expected integer, got NoneType object
.
----------------------------------------------------------------------
Ran 1 test in 0.904s

OK

method connect_to_cli use always using a unix socket

Hi,

The method connect_to_cli of the class Pulse is using a Unix socket even when instantiating with an Ip address:
with Pulse(server='192.168.0.2') as pulse:
pulse.connect_to_cli() #Connect to the local pulse audio server

Took me a little while to understand why my code was not doing what I wanted.

On a side not, I am new to python and pulseaudio, I used connect_to_cli because I needed to change the proplist of a sink-input. How easy would it be to implement this ?

Regards,
Xavier

All unit tests and examples fail

Hello!

The unit tests fail both using python2 and python3. The examples given in the readme fail at pulse.volume_set(sink_input, new_volume) with pulsectl.pulsectl.PulseOperationFailed: 1. Let me know if I can provide you with any other information.

4.14.83-gentoo x86_64 GNU/Linux
pulseaudio 12.2
pulsectl (18.12.5)

python3 -m unittest pulsectl.tests.all

EEE
======================================================================
ERROR: setUpClass (pulsectl.tests.dummy_instance.DummyTests)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/home/asdf/local/src/github.com/mk-fg/python-pulse-control/pulsectl/tests/dummy_instance.py", line 146, in setUpClass
    cls.instance_info = dummy_pulse_init()
  File "/home/asdf/local/src/github.com/mk-fg/python-pulse-control/pulsectl/tests/dummy_instance.py", line 25, in dummy_pulse_init
    try: _dummy_pulse_init(info)
  File "/home/asdf/local/src/github.com/mk-fg/python-pulse-control/pulsectl/tests/dummy_instance.py", line 60, in _dummy_pulse_init
    s.bind((addr, p))
OSError: [Errno 99] Cannot assign requested address

======================================================================
ERROR: test_crash_after_connect (pulsectl.tests.dummy_instance.PulseCrashTests)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/home/asdf/local/src/github.com/mk-fg/python-pulse-control/pulsectl/tests/dummy_instance.py", line 473, in test_crash_after_connect
    info = dummy_pulse_init()
  File "/home/asdf/local/src/github.com/mk-fg/python-pulse-control/pulsectl/tests/dummy_instance.py", line 25, in dummy_pulse_init
    try: _dummy_pulse_init(info)
  File "/home/asdf/local/src/github.com/mk-fg/python-pulse-control/pulsectl/tests/dummy_instance.py", line 60, in _dummy_pulse_init
    s.bind((addr, p))
OSError: [Errno 99] Cannot assign requested address

======================================================================
ERROR: test_reconnect (pulsectl.tests.dummy_instance.PulseCrashTests)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/home/asdf/local/src/github.com/mk-fg/python-pulse-control/pulsectl/tests/dummy_instance.py", line 485, in test_reconnect
    info = dummy_pulse_init()
  File "/home/asdf/local/src/github.com/mk-fg/python-pulse-control/pulsectl/tests/dummy_instance.py", line 25, in dummy_pulse_init
    try: _dummy_pulse_init(info)
  File "/home/asdf/local/src/github.com/mk-fg/python-pulse-control/pulsectl/tests/dummy_instance.py", line 60, in _dummy_pulse_init
    s.bind((addr, p))
OSError: [Errno 99] Cannot assign requested address

----------------------------------------------------------------------
Ran 2 tests in 0.001s

FAILED (errors=3)
Error in atexit._run_exitfuncs:
Traceback (most recent call last):
  File "/home/asdf/local/src/github.com/mk-fg/python-pulse-control/pulsectl/tests/dummy_instance.py", line 151, in tearDownClass
    dummy_pulse_cleanup(cls.instance_info)
AttributeError: type object 'DummyTests' has no attribute 'instance_info'

Thanks

Setting volume on sink linear percentage but on sink source is cuberoot of percentage.

It looks like the volume values are passing through different pathways internally depending on setting the value for a sink input vs a sink. E.g. to get levels set to 50% (.5) for the sink and each input one must do as follows:

  with Pulse('volume-setter') as pulse:
            source = pulse.sink_input_list()[0]
            pulse.volume_set_all_chans(source, np.cbrt(volume_pct))
            # Alternatively
            # vol = source[0].volume
            # vol.value_flat = np.cbrt(volume_pct)
            # pulse.volume_set(source[0], np.cbrt(volume_pct))
            
            pulse.volume_set_all_chans(pulse.sink_list()[0], volume_pct)

get_peak_sample

I am using get_peak_sample() to display the sound level of a playing sample.
But I notice that everytime it runs it floods the pulseaudio events with:

Pulse event: facility=<EnumValue event-facility=source>, index=23, t=<EnumValue event-type=change>
Pulse event: facility=<EnumValue event-facility=source_output>, index=1933, t=<EnumValue event-type=remove>
Pulse event: facility=<EnumValue event-facility=client>, index=403, t=<EnumValue event-type=remove>

When an actual sink/source port is changed it also sends similar event like so:

Pulse event: facility=<EnumValue event-facility=sink>, index=23, t=<EnumValue event-type=change>
Pulse event: facility=<EnumValue event-facility=sink_input>, index=43, t=<EnumValue event-type=remove>
Pulse event: facility=<EnumValue event-facility=client>, index=413, t=<EnumValue event-type=remove>

I need a reliable way to determine when a sink/source port and card profile is updated .

Impossible inheriting Pulse class

Hi, thanks for the project. Now the issue:
Inherited classes cannot use the functions like sink_input_list(), this seems to depend on some decorator. Would be nice if the following print() calls would output the same.

import pulsectl
p=pulsectl.Pulse()
print(p.sink_input_list())
class X(pulsectl.Pulse): pass
x=X()
print(x.sink_input_list())

Feature: Keep pulse() connected

p=pulse() shall not kill the process when pulse disconnects.

It happens from time to time that the user might restart or kill pulse e.g. because he changes the config or so. Everyone who implements some daemon that has a pulse() instance might prefer a Pulse class that reconnects by itself as soon as pulse is available. Is there space somewhere for something like this?

import pulsectl, time, sys
from threading import Thread

class Pulse(pulsectl.Pulse):

    def __init__(self,*args,**xargs):
        super().__init__(*args,connect=False,**xargs)
        self.connect_pulse()

    def connect_pulse(self):
        def connect():
            self.connect()
            self.on_connected()
        def keep_reconnecting():
            while True:
                try: connect()
                except pulsectl.pulsectl.PulseError: time.sleep(3)
                else: break
        print("[%s] Connecting..."%self.__class__.__name__, file=sys.stderr)
        try: connect()
        except pulsectl.pulsectl.PulseError:
            Thread(target=keep_reconnecting,daemon=True).start()
    
    def on_connected(self):
        print("[%s] Connected to Pulseaudio."%self.__class__.__name__, file=sys.stderr)

Of course, using this makes sense with try-catch around all pulse() attribute calls.

The PA_SAMPLE_FLOAT32BE value is incorrect

PA_SAMPLE_FLOAT32BE is set to 6 in the sample.h libpulse header, but it is set to 5 in _pulsectl.py.

In sample.h, PA_SAMPLE_FLOAT32NE (NE stands for native order) is defined as PA_SAMPLE_FLOAT32LE(5) or PA_SAMPLE_FLOAT32BE(6) depending on the endianness of the processor. And PA_SAMPLE_FLOAT32 is defined as PA_SAMPLE_FLOAT32NE. The Python sys module has the sys.byteorder variable to get the native byte order. Using this variable allows to get the same definition for PA_SAMPLE_FLOAT32 as done in sample.h.

FWIW the following Python script uses the preprocessor of gcc to print as Python statements the constants defined by enums in the libpulse header def.h (and also sample.h that is included by def.h):

import re
import subprocess

enums_re = r'typedef\s+enum\s+pa_.*{([^}]+)}\s+(pa_\w+)\s*;'
constant_re = r'(\w+)\s*=\s*(0x[0-9A-Fa-f]+|-?\d+)'

def libpulse_enums(pathname):
    """Parse enums in a libpulse header.

    Print the constants as Python statements.
    """

    proc = subprocess.run(['gcc', '-E', '-P', pathname],
                          capture_output=True, text=True)

    enums = re.findall(enums_re, proc.stdout, flags=re.MULTILINE)
    for enum, name in enums:
        print(f'# Enum {name}.')
        i = 0
        for constant in (x.strip() for x in enum.split(',')):
            if not constant:
                continue
            m = re.match(constant_re, constant)
            if m is not None:
                val = m.group(2)
                print(f'{m.group(1)} = {val}')
                i = eval(val)
            else:
                print(f'{constant} = {i}')
            i += 1
        print()

libpulse_enums('/usr/include/pulse/def.h')

pulse.close() not working

When I run pulse.close() it is not stopping the execution of pulse.play_sample(). The sample still continues to play....

class AudioTestPlay(QObject):

    def __init__(self):
        super().__init__()
        self.pulse = pulsectl.Pulse()

    def play(self):
        self.pulse.play_sample('starwars')

    def stop(self):
        self.pulse.close()

DeprecationWarning: inspect.getargspec() is deprecated since Python 3.0, use inspect.signature() or inspect.getfullargspec()

When pulsectl is imported in a pytest i see the following deprecation warnings:

../../usr/local/lib/python3.9/site-packages/pulsectl/pulsectl.py:611: 20 warnings
  /usr/local/lib/python3.9/site-packages/pulsectl/pulsectl.py:611: DeprecationWarning: inspect.getargspec() is deprecated since Python 3.0, use inspect.signature() or inspect.getfullargspec()
    func_args = list(inspect.getargspec(func or (lambda: None)))

../../usr/local/lib/python3.9/site-packages/pulsectl/pulsectl.py:615: 20 warnings
  /usr/local/lib/python3.9/site-packages/pulsectl/pulsectl.py:615: DeprecationWarning: `formatargspec` is deprecated since Python 3.5. Use `signature` and the `Signature` object directly
    _wrapper.__doc__ = 'Signature: func' + inspect.formatargspec(*func_args)

-- Docs: https://docs.pytest.org/en/stable/warnings.html

test_get_peak_sample sometimes fails with "AssertionError: 0 not greater than 0"

Hi,
If I run the test suite multiple times, I can see this test failing. Could it be because /dev/urandom doesn't contain reliable data?

Here's one trace with pytest:

self = <pulsectl.tests.test_with_dummy_instance.DummyTests testMethod=test_get_peak_sample>                                                                                                                        
                                                                                                                                                                                                                   
    def test_get_peak_sample(self):                                                                                                                                                                                
        # Note: this test takes at least multiple seconds to run                                                                                                                                                   
        with pulsectl.Pulse('t', server=self.sock_unix) as pulse:                                                                                                                                                  
                source_any = max(s.index for s in pulse.source_list())                                                                                                                                             
                source_nx = source_any + 1                                                                                                                                                                         
                                                                                                                                                                                                                   
                time.sleep(0.3) # make sure previous streams die                                                                                                                                                   
                peak = pulse.get_peak_sample(source_any, 0.3)                                                                                                                                                      
                self.assertEqual(peak, 0)                                                                
                                                                                                                                                                                                                   
                stream_started = list()                                                                                                                                                                            
                def stream_ev_cb(ev):                                                                                                                                                                              
                        if ev.t != 'new': return                                                                                                                                                                   
                        stream_started.append(ev.index)                                                                                                                                                            
                        raise pulsectl.PulseLoopStop                                                     
                pulse.event_mask_set('sink_input')                                                       
                pulse.event_callback_set(stream_ev_cb)                                                                                                                                                             
                                                                                                                                                                                                                   
                paplay = subprocess.Popen(                                                                                                                                                                         
                        ['paplay', '--raw', '/dev/urandom'], env=dict(              
                                PATH=os.environ['PATH'], XDG_RUNTIME_DIR=self.tmp_dir ) )                
                try:                                                                                     
                        if not stream_started: pulse.event_listen()                                      
                        stream_idx, = stream_started                                                                                                                                                               
                        si = pulse.sink_input_info(stream_idx)                                           
                        sink = pulse.sink_info(si.sink)                                                                                                                                                            
                        source = pulse.source_info(sink.monitor_source)                                                                                                                                            
                                                                                                                                                                                                                   
                        # First poll can randomly fail if too short, probably due to latency or such                                                                                                               
                        peak = pulse.get_peak_sample(sink.monitor_source, 3)                             
                        self.assertGreater(peak, 0)                                                                                                                                                                
                                                                                                                                                                                                                   
                        peak = pulse.get_peak_sample(source.index, 0.3, si.index)                                                                                                                                  
>                       self.assertGreater(peak, 0)                                                                                                                                                                
E      AssertionError: 0 not greater than 0                                                                                                                                                                        

And another with plain unittest:

======================================================================                                   
FAIL: test_get_peak_sample (pulsectl.tests.test_with_dummy_instance.DummyTests)
----------------------------------------------------------------------
Traceback (most recent call last):                                                                       
  File "/var/tmp/portage/dev-python/pulsectl-21.5.6/work/pulsectl-21.5.6/pulsectl/tests/test_with_dummy_instance.py", line 610, in test_get_peak_sample                                                            
    self.assertGreater(peak, 0)                                                                          
AssertionError: 0 not greater than 0                                                                     
                                                                                                         
----------------------------------------------------------------------       
Ran 16 tests in 9.657s                                                                                   
                                                                                                                                                                                                                   
FAILED (failures=1)                                                                                      

Documentation bug: event example code TypeErrors

~ ยป python3.6 patest.py
Traceback (most recent call last):
  File "patest.py", line 10, in <module>
    print('Event types:', ', '.join(p.event_types))
TypeError: sequence item 0: expected str instance, EnumValue found

File contents:

import pulsectl


def event_cb(evt):
    print(evt)
    print(dir(evt))


with pulsectl.Pulse('test1') as p:
    print('Event types:', ', '.join(p.event_types))
    print('Event facilities:', ', '.join(p.event_facilities))
    print('Event masks:', ', '.join(p.event_masks))

Integrate into gobject.MainLoop?

Hello, great project!

Is it possible to integrate this into a gobject.MainLoop?
I've read about the set_poll_func but I am not quite sure how that fits into gobject.

        import gobject
        mainloop = gobject.MainLoop()
        try:
            mainloop.run()
        except KeyboardInterrupt:
            pass

Furthermore, is that library compatible with any pulseaudio version?

'pa_runtime_path' function not found

When importing pulsectl on Windows 10 I get an AttributeError. Here's the traceback:

Traceback (most recent call last):
  File "main.py", line 2, in <module>
    import multimuter as mm
  File "D:\python\MultiMuter\multimuter.py", line 1, in <module>
    import pulsectl
  File "C:\Users\barte\AppData\Local\Programs\Python\Python38-32\lib\site-packages\pulsectl\__init__.py", line 4, in <module>
    from . import _pulsectl
  File "C:\Users\barte\AppData\Local\Programs\Python\Python38-32\lib\site-packages\pulsectl\_pulsectl.py", line 676, in <module>
    pa = LibPulse()
  File "C:\Users\barte\AppData\Local\Programs\Python\Python38-32\lib\site-packages\pulsectl\_pulsectl.py", line 632, in __init__
    func, args, res_proc = getattr(p, k), None, None
  File "C:\Users\barte\AppData\Local\Programs\Python\Python38-32\lib\ctypes\__init__.py", line 386, in __getattr__
    func = self.__getitem__(name)
  File "C:\Users\barte\AppData\Local\Programs\Python\Python38-32\lib\ctypes\__init__.py", line 391, in __getitem__
    func = self._FuncPtr((name_or_ordinal, self))
AttributeError: function 'pa_runtime_path' not found

How do I use the "Pulse" class?

I installed this library to use pacmd on Ubuntu, basically I just want to avoid having to call subprocess.check_output and then manually parse the output. I'm used to just typing pacmd list-cards or whatever at the command line, but I can see from the examples that to do the equivalent here, I have to create a "Pulse" object, which takes a string in its constructor, which I can't map onto the way I would normally use the CLI at all. How do I find out what that string should be?

some fields not present on sinks/sources

Hi there. Thanks for this library! I've been looking for something like this, and am extremely grateful that someone has put in the hard work to make it possible.

I am interested (it seems) in a feature that isn't present in your wrapper, or perhaps it isn't present in libpulse...

When I use pacmd, and list-sinks, I get output like this:

2 sink(s) available.
    index: 4
    name: <alsa_output.pci-0000_00_1b.0.analog-surround-71>
    driver: <module-alsa-card.c>
    flags: HARDWARE HW_MUTE_CTRL HW_VOLUME_CTRL DECIBEL_VOLUME LATENCY DYNAMIC_LATENCY
    state: RUNNING
    suspend cause: 
    priority: 9959
    volume: front-left: 65536 / 100% / 0.00 dB,   front-right: 65536 / 100% / 0.00 dB,   rear-left: 65536 / 100% / 0.00 dB,   rear-right: 65536 / 100% / 0.00 dB,   front-center: 65536 / 100% / 0.00 dB,   lfe: 65536 / 100% / 0.00 dB,   side-left: 65536 / 100% / 0.00 dB,   side-right: 65536 / 100% / 0.00 dB
            balance 0.00
    base volume: 65536 / 100% / 0.00 dB
    volume steps: 65537
    muted: no
    current latency: 46.21 ms
    max request: 63 KiB
    max rewind: 64 KiB
    monitor source: 7
    sample spec: s32le 8ch 44100Hz
    channel map: front-left,front-right,rear-left,rear-right,front-center,lfe,side-left,side-right
                 Surround 7.1
    used by: 1
    linked by: 1
    configured latency: 46.44 ms; range is 0.50 .. 46.44 ms
    card: 1 <alsa_card.pci-0000_00_1b.0>
    module: 2
    properties:
        alsa.resolution_bits = "32"
        device.api = "alsa"
        device.class = "sound"
        alsa.class = "generic"
        alsa.subclass = "generic-mix"
        alsa.name = "ALC883 Analog"
        alsa.id = "ALC883 Analog"
        alsa.subdevice = "0"
        alsa.subdevice_name = "subdevice #0"
        alsa.device = "0"
        alsa.card = "0"
        alsa.card_name = "HDA Intel"
        alsa.long_card_name = "HDA Intel at 0xfe6f4000 irq 31"
        alsa.driver_name = "snd_hda_intel"
        device.bus_path = "pci-0000:00:1b.0"
        sysfs.path = "/devices/pci0000:00/0000:00:1b.0/sound/card0"
        device.bus = "pci"
        device.vendor.id = "8086"
        device.vendor.name = "Intel Corporation"
        device.product.id = "293e"
        device.product.name = "82801I (ICH9 Family) HD Audio Controller"
        device.form_factor = "internal"
        device.string = "surround71:0"
        device.buffering.buffer_size = "65536"
        device.buffering.fragment_size = "32768"
        device.access_mode = "mmap+timer"
        device.profile.name = "analog-surround-71"
        device.profile.description = "Analog Surround 7.1"
        device.description = "Built-in Audio Analog Surround 7.1"
        alsa.mixer_name = "Realtek ALC883"
        alsa.components = "HDA:10ec0883,1043829e,00100002"
        module-udev-detect.discovered = "1"
        device.icon_name = "audio-card-pci"
    ports:
        analog-output-lineout: Line Out (priority 9900, latency offset 0 usec, available: yes)
            properties:

    active port: <analog-output-lineout>

And doing pulse.sink_list() gives me a python list representation of this (awesome!). But if I inspect one of the sink objects, I get fewer attributes:

s.c_struct_fields      s.index                s.port_active
s.channel_count        s.latency              s.port_list
s.channel_list         s.monitor_source       s.proplist
s.configured_latency   s.monitor_source_name  s.sample_spec
s.description          s.mute                 s.volume
s.driver               s.name                 
s.flags                s.owner_module         

In particular, I'm looking for "state: RUNNING", "suspend cause:"... I'd like to write some code that detects when sound is flowing, or at least when PA thinks it hasn't and has therefore suspended a sink.

Am I missing something?

Thanks again!

Support for device info in ports

Hi,

I'm writing a tool that switches the audio channel when an external display is switched on. Depending on the state of the display, the profile name can change, for example my TV is either HDMI 2 or HDMI 3.

I've figured a way to workaround that but pulsectl seems to be missing that info. When I run pacmd list-sinks, I get something like:

        [...]
	ports:
		hdmi-output-1: HDMI / DisplayPort 2 (priority 5800, latency offset 0 usec, available: yes)
			properties:
				device.icon_name = "video-display"
				device.product.name = "FullHD TV

I would like to access the device.product.name property but it's not exposed by PulsePortInfo, would it be possible to add that?

Thanks!

Exposing flags from sink inputs and source outputs

Hi,
I wrote a script to move all sinks/sources and it fails on some objects because they have the flag DONT_MOVE. I'd like to skip them but I can't see the flags via pulsectl's API.

Could you please expose this info on PulseSinkInputInfo and PulseSourceOutputInfo objects?

Moving a stream to a different sink

I'm trying to make a script that switches the default sink and moves all playing streams to the default sink.

I found the method sink_input_move, but it's completely undocumented. It seems to require 2 integers, but no matter what combination of integers I pass I just get pulsectl.pulsectl.PulseOperationFailed. I've tried inputting the indexes of the stream and the sink, the device number and the card number of the sink.

Is this method broken, or am I doing something wrong?

Can't get get_peak_sample() to ever return any values > 0

I'm sure this is pilot error in some way, but no matter what I do, I can't seem to get back a non-zero return from .get_peak_sample().

I'm hoping to do it to a source/input (it's a USB device, but Linux think's it's a 4-channel microphone). But I can't even get it to work on a sink monitor or anything. I've verified that I can set the volume on the sink, so it's not a permissions issue or something like that.

import pulsectl
p = pulsectl.Pulse('tst')
p.sink_list()
# [<PulseSinkInfo at ae5a2af0 - description='miniDSP 2x4HD Analog Stereo', index=0, mute=0, name='alsa_output.usb-miniDSP_miniDSP_2x4HD-00.analog-stereo', channels=2, volumes=[100% 100%]>]
sink = p.sink_list()[0]

# this works:
p.volume_set_all_chans(sink, 0.5)

# this never does:
p.get_peak_sample(sink, 2)
# 0

And yes, it's playing :-) I've also tried attaching to the monitor source, the real source, etc.

I'm sure I'm missing something fundamental, but I am stuck. I'm on PA 12.2 running on a rPi4. Thanks!

Wrong unit for timeout of mainloop_prepare

The the parameter timeout is passed to pa_mainloop_prepare() in milliseconds instead of micoseconds.
This causes higher CPU load when waiting for an event in case event_listen() is called with a timeout parameter.

delay = max(0, int((ts_deadline - ts) * 1000)) if ts_deadline else -1

Also see https://freedesktop.org/software/pulseaudio/doxygen/mainloop_8h.html#a217ce134af601c1c6ce77c41c99fc0d2
image

Manual download .zip, can't run test_with_dummy_instance.py

Background

My python music player uses pactl to turn up/down volume over 1 second for Play/Pause toggle. The command-line call is inefficient so, a direct call to libpulse0 using pulsectl makes sense.

Steps Taken

Unfortunately I couldn't find an apt list candidate for pulsectl so I manually downloaded the .zip and extracted it to a local directory ~/python/pulsectl. Then chmod a+x *.py to make all python programs executable.

Error

When running test_with_dummy_instance.py the following error occurs:

$ test_with_dummy_instance.py

from: can't read /var/mail/__future__

^C^C^C./test_with_dummy_instance.py: line 12: syntax error near unexpected token `1,'
./test_with_dummy_instance.py: line 12: `	sys.path.insert(1, os.path.join(__file__, *['..']*2))'

Initially only the error from: can't read /var/mail/__future__ appears and program appears to be in infinite loop. So, Ctrl + C must be repeated used to get command prompt back.

A /var/mail directory does exist but, it is empty.

Looking at test_with_dummy_instance.py source around line 12:

try: import pulsectl
except ImportError:
	sys.path.insert(1, os.path.join(__file__, *['..']*2))
	import pulsectl

Unfortunately the os.path.join(__file__, *['..']*2)) is above my pay grade.

Any ideas how to get over this hurtle?


Environment

TL;DR Just in case they're needed, here are some environment details.

$ pulseaudio --version
pulseaudio 8.0

:~/python/pulsectl$ grep version setup.py
	version = '22.1.0',

$ apt list | grep libpulse

WARNING: apt does not have a stable CLI interface. Use with caution in scripts.

libpulse-dev/xenial-updates,xenial-security 1:8.0-0ubuntu3.15 amd64
libpulse-java/xenial,xenial 2.4.7-1 all
libpulse-jni/xenial 2.4.7-1 amd64
libpulse-mainloop-glib0/xenial-updates,xenial-security,now 1:8.0-0ubuntu3.15 amd64 [installed]
libpulse-ocaml/xenial 0.1.2-1build3 amd64
libpulse-ocaml-dev/xenial 0.1.2-1build3 amd64
libpulse0/xenial-updates,xenial-security,now 1:8.0-0ubuntu3.15 amd64 [installed]
libpulsedsp/xenial-updates,xenial-security,now 1:8.0-0ubuntu3.15 amd64 [installed]

$ apt list | grep alsa | grep installed

WARNING: apt does not have a stable CLI interface. Use with caution in scripts.

alsa-base/xenial,xenial,now 1.0.25+dfsg-0ubuntu5 all [installed]
alsa-tools/xenial,now 1.1.0-0ubuntu1 amd64 [installed]
alsa-tools-gui/xenial,now 1.1.0-0ubuntu1 amd64 [installed]
alsa-utils/xenial,now 1.1.0-0ubuntu5 amd64 [installed]
gstreamer1.0-alsa/xenial-updates,xenial-security,now 1.8.3-1ubuntu0.3 amd64 [installed]
python-alsaaudio/xenial,now 0.7-1 amd64 [installed]

Combined sink returns as running / detecting activity

When creating a combined sink, the sink state will return "running" even though no audio is playing back through it. This is my ~/.config/pulse/default.pa:

.include /etc/pulse/default.pa

load-module module-combine-sink sink_name=micspam_output sink_properties=device.description=Micspam
load-module module-remap-source source_name=micspam_input source_properties=device.description=Micspam master=micspam_output.monitor

The media.role of the combined sink is "filter" and its state is "DRAINED". Both appear to be undocumented, as I can't find information on either.

Cannot load modules with `module_load`

Should I expect module_load to work?

I'm just testing in a test script right now but when I try to load a module, I get a large return value but there is no effect in pulseaudio. I'm on a Debian 10 system where pulse is running in system mode. Yes, I know what the pulse maintainers think of system mode. Yes, my use case is in their list of reasons to do it. pulsectl is version 21.5.18 and python3 is version 3.7.3.

#!/usr/bin/env python3

import pulsectl
pulse = pulsectl.Pulse("myclient")

print("sinks:")
for sink in pulse.sink_list():
    print (sink)

print("create unnamed null sink")
print(pulse.module_load("module-null-sink"))

print("sinks:")
for sink in pulse.sink_list():
    print (sink)

print("create named null sink")
print(pulse.module_load("module-null-sink", args='sink_name="default.2.null"'))

print("sinks:")
for sink in pulse.sink_list():
    print (sink)

output:

sinks:
description='Built-in Audio Analog Stereo', index=0, mute=1, name='alsa_output.pci-0000_00_1b.0.analog-stereo', channels=2, volumes=[9% 9%]
description='Null Output', index=1, mute=0, name='default.null', channels=2, volumes=[100% 100%]
create unnamed null sink
4294967295
sinks:
description='Built-in Audio Analog Stereo', index=0, mute=1, name='alsa_output.pci-0000_00_1b.0.analog-stereo', channels=2, volumes=[9% 9%]
description='Null Output', index=1, mute=0, name='default.null', channels=2, volumes=[100% 100%]
create named null sink
4294967295
sinks:
description='Built-in Audio Analog Stereo', index=0, mute=1, name='alsa_output.pci-0000_00_1b.0.analog-stereo', channels=2, volumes=[9% 9%]
description='Null Output', index=1, mute=0, name='default.null', channels=2, volumes=[100% 100%]

Thanks!

Integration with Asyncio

First, thanks! This is a great library. I am writing a asyncio based library and I am integrating pulsectl on it.

For now I just started a thread and let pulsectl run on it, however it would be interesting to have proper asyncio integration with pulsectl with it. I saw the other issue with gobject (#11) and I am interested if you know some tips on how to integrate pulsectl with async.

How to inspect an event further

I've been playing around with pulsectl today. I re-created one of your examples, to see the events that Pulse emits:

import pulsectl

with pulsectl.Pulse('event-printer') as pulse:

    def print_events(ev):
        print('Pulse event:', ev)

    pulse.event_mask_set('all')
    pulse.event_callback_set(print_events)
    pulse.event_listen(timeout=10)

Running the above code and pressing the mic-mute toggle on my keyboard:

$ python experiment4.py 
Pulse event: facility=<EnumValue event-facility=source>, index=51, t=<EnumValue event-type=change>
Pulse event: facility=<EnumValue event-facility=card>, index=45, t=<EnumValue event-type=change>

The events seem very sparse; how can I get more information from the event object? Specifically, I'm looking to identify this event more specifically (as in: what it is, that it's a mute-toggle event, that kind of thing).

Any insights much appreciated + thanks for a great pip package!

set-card-profile feature request

I found your project extremely helpful --thank you!!!
I'd like to request that a set-card-profile feature be implemented that would allow you to change the audio profile on a bluetooth device {headset_head_unit, a2dp_sink, off}

https://github.com/GeorgeFilipkin/pulsemixer/
pulsemixer allows you to change the card-profile within the curses interface.
Since this project is forked from pulsemixer, perhaps it would be possible to port that feature over?

I'd like to use this feature in an AlexaPI fork I'm working on --so that the audio profile can change based on whether Alexa is speaking or listening.

Changelog

Just out of curiosity
are you able to provide a high level changelog.md file?
or could you say what changed between, e.g. 16.9.10 and 16.11.0

thanks a lot
and best regards
flori

Example for updating volume values in sink object

I am developing a small mqtt volume controller for various Linux based satellite speakers in my home. I wanted to use a simple polling mechanism checking with pulse.volume_get_all_chans(sink) for the volume and posting this then via mqtt to my network to show on a dashboard.

Unfortunately, pulse.volume_get_all_chans(sink) always returns the same value (also when I change the volumes with pavucontrol) - I seem to be missing some update calls. I do get events when registering a handler with event_callback_set, however the values in volume_get_all_chans still stay the same. Only when restarting my program, I get once an updated volume value.

Is this the expected behavior? How would I get up-to-date values (like when calling pamixer from the cli).

TypeError in get_source_by_name

When trying to obtain a source by it's name the following exception is thrown:

Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/home/esoares/.local/lib/python3.6/site-packages/pulsectl/pulsectl.py", line 511, in _decorator_or_method
    return _wrapper(func_or_self, index)
  File "/home/esoares/.local/lib/python3.6/site-packages/pulsectl/pulsectl.py", line 498, in _wrapper
    pulse_func(self._ctx, *([index, cb, None] if index is not None else [cb, None]))
  File "/home/esoares/.local/lib/python3.6/site-packages/pulsectl/_pulsectl.py", line 605, in _wrapper
    res = func(*args)
ctypes.ArgumentError: argument 3: <class 'TypeError'>: expected CFunctionType instance instead of CFunctionType

Sample code:

import pulsectl
a = pulsectl.Pulse("my_client")
a.get_source_by_name(a.sink_list()[0].name)

Segmentation fault when getting peak level

I'm writing a small tool to do some very basic normalizing of the sound.
Python script works just fine for a while and then it segfaults.

I've managed to create minimalistic example where segfault happens consistently.
Time before crashing varies, it is sometimes 10-15 seconds, and sometimes a couple of minutes.

Details about my system (Fedora 30):

$ rpm -q pulseaudio
pulseaudio-12.2-9.fc30.x86_64
$ rpm -q python3-pulsectl
python3-pulsectl-19.9.5-1.fc30.noarch
$ rpm -q kernel
kernel-5.2.17-200.fc30.x86_64

Script to reproduce the issue:

#!/usr/bin/python3

import time
import pulsectl

def find_vlc():
	while 1:
		sinks = pulse.sink_input_list()
		for sink in sinks:
			if sink.proplist["application.process.binary"] == "vlc":
				return sink
		print("No VLC found, waiting.")
		time.sleep(5)

pulse = pulsectl.Pulse('segfa')

while 1:
	pa_vlc = find_vlc()
	peak = pulse.get_peak_sample(pulse.sink_info(pa_vlc.sink).monitor_source, 0.2, pa_vlc.index)
	print(round(peak, 3))

Note that finding VLC can be removed from the loop, and it still segfaults.

Please let me know if there is anything else I can provide to further help with resolving.

Audio Playback

I am simply trying to playback a .wav file through my speakers using this API. I want to use PulseAudio to play a wav and not ALSA.

from pulsectl import PulseSimple

ps = PulseSimple()
ps.file_playback('sound.wav')

Traceback (most recent call last):
  File "test.py", line 8, in <module>
    ps = PulseSimple()
  File "/usr/local/lib/python3.8/dist-packages/pulsectl/pulsectl.py", line 993, in __init__
    self.buffer_attr = c.PA_BUFFER_ATTR(c.PA_INVALID, c.PA_INVALID, c.PA_INVALID, c.PA_INVALID, period_size)
TypeError: an integer is required (got type NoneType)

Any ideas what is wrong? Or is there a better way to play audio in python using a sink's name?

PA_VOLUME_MAX and PA_VOLUME_UI_MAX incorrect

The pulseaudio value is:

#define PA_VOLUME_MAX ((pa_volume_t) UINT32_MAX/2)

But pulsectl has:

PA_VOLUME_MAX = 2**32-1 // 2

Which misses parenthesis, I think it should be:

PA_VOLUME_MAX = (2**32-1) // 2

(Note that pulseaudio had a different value before 1.0, but that does not seem relevant anymore: https://gitlab.freedesktop.org/pulseaudio/pulseaudio/-/commit/179b291b18b9a7366955948ce0ddfb2e65a6b66e)

For PA_VOLUME_UI_MAX, pulseaudio has:

#define PA_VOLUME_UI_MAX   (pa_sw_volume_from_dB(+11.0))

pulsectl has a hardcoded value, that seems to have been calculated by a python lambda version of pa_sw_volume_from_dB:

# pa_sw_volume_from_dB = lambda db:\
# min(PA_VOLUME_MAX, int(round(((10.0 ** (db / 20.0)) ** 3) * PA_VOLUME_NORM)))
PA_VOLUME_UI_MAX = 2927386 # pa_sw_volume_from_dB(+11.0)

However, that lambda is wrong, the cbrt function used by pulse is the cube root, not the cube. So the correct version is:

>>> pa_sw_volume_from_dB = lambda db:  min(PA_VOLUME_MAX, int(round(((10.0 ** (db / 20.0)) ** (1/3)) * PA_VOLUME_NORM)))
>>> pa_sw_volume_from_dB(+11.0)
99957

Which matches the pulse version:

matthijs@grubby:~$ cat foo.c
#include <pulse/volume.h>
#include <stdio.h>

int main() {
        printf("%u\n", PA_VOLUME_UI_MAX);
}
matthijs@grubby:~$ gcc foo.c -lpulse && ./a.out 
99957

I wonder, though: Why not just provide a binding for pa_sw_volume_from_dB and calculate the value using that (this is how I originally ended up here, I needed pa_sw_volume_from_dB and while writing my own version, I looked at PA_VOLUME_MAX and noticed it was wrong).

Unloading module raises "IndexError: tuple index out of range"

I load a module on callback like this:

pulse.module_load(name=DSP_MODULE_NAME,
                  args=f'sink_name={DSP_SINK_NAME} master={mojo_sink.name} '
                       f'plugin={DSP_PLUGIN_NAME} label={DSP_PLUGIN_NAME}')

Results in a loaded module (part of sink_list() output):

<PulseSinkInfo at 7f8220767a90 - description='LADSPA Plugin ladspa_dsp on Mojo Analog Stereo', index=6, mute=0, name='dsp', channels=2, volumes=[100% 100%]>

Works awesome.

Then on another callback I want to unload it:

pulse.module_unload(name=DSP_MODULE_NAME)

(the same global variable DSP_MODULE_NAME)

This results in:

Traceback (most recent call last):                         
  File "_ctypes/callbacks.c", line 232, in 'calling callback function'                                                 
  File "/usr/local/lib/python3.7/dist-packages/pulsectl/pulsectl.py", line 406, in _pulse_subscribe_cb                 
    try: self.event_callback(PulseEventInfo(ev_t, ev_fac, idx))                                                        
  File "./ssink_renameme.py", line 70, in print_events     
    _setup_mojo()                                          
  File "./ssink_renameme.py", line 88, in _setup_mojo      
    pulse.module_unload(name=DSP_MODULE_NAME)              
  File "/usr/local/lib/python3.7/dist-packages/pulsectl/pulsectl.py", line 549, in _wrapper                            
    else: index, args = args[0], args[1:]                  
IndexError: tuple index out of range

I want to replace pacmd unload-module module-ladspa-sink with pulse.module_unload(). For now I can supplement with running pacmd for this case, but this isn't the most elegant solution :)

Thank you for creating this library! :) Especially event_callback_set is awesome. My Mojo takes a few seconds to start and I don't have to listen on UDEV and guess how much time I have to sleep before setting up sink for it :)

Please support testing against pipewire

PulseAudio is pretty much dead these days and everyone and their grandmother have already switched over to PipeWire. However, pulsectl's test suite relies on pulseaudio daemon being available, and at least on Gentoo it cannot be installed simultaneously with PipeWire. As a result, I can't test the package without having to replace the media server on my system which is a major hassle.

Could you please add support for testing against PipeWire instead?

Methods for proplist editing

It would be handy to have wrappers around the methods that modify proplists (like pacmd does).

In particular, my use case for this is modifying the descriptions and names (device.description and similar) of sink inputs and source outputs.

add timeout conex

hello I wanted to know how to implement a timeout for the connection
My application is simple but I need to know quickly if it is possible to connect with the remote pulse pulse = pulsectl.Pulse(server = 'ip_remote')

Version 20.2.4

Is it possible to get an pypi release of that? Thanks a lot!

Segmentation fault

I get a segmentation fault when I pulse = pulsectl.Pulse('my-client-name')
Pulse is working OK. I'm running Pulse as a system wide daemon.

Recommend Projects

  • React photo React

    A declarative, efficient, and flexible JavaScript library for building user interfaces.

  • Vue.js photo Vue.js

    ๐Ÿ–– Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.

  • Typescript photo Typescript

    TypeScript is a superset of JavaScript that compiles to clean JavaScript output.

  • TensorFlow photo TensorFlow

    An Open Source Machine Learning Framework for Everyone

  • Django photo Django

    The Web framework for perfectionists with deadlines.

  • D3 photo D3

    Bring data to life with SVG, Canvas and HTML. ๐Ÿ“Š๐Ÿ“ˆ๐ŸŽ‰

Recommend Topics

  • javascript

    JavaScript (JS) is a lightweight interpreted programming language with first-class functions.

  • web

    Some thing interesting about web. New door for the world.

  • server

    A server is a program made to process requests and deliver data to clients.

  • Machine learning

    Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.

  • Game

    Some thing interesting about game, make everyone happy.

Recommend Org

  • Facebook photo Facebook

    We are working to build community through open source technology. NB: members must have two-factor auth.

  • Microsoft photo Microsoft

    Open source projects and samples from Microsoft.

  • Google photo Google

    Google โค๏ธ Open Source for everyone.

  • D3 photo D3

    Data-Driven Documents codes.