pulsectl-asyncio

此库在 pulsectl 库之上提供了一个 Python 3 asyncio 接口，用于监控和控制 PulseAudio 音频服务器。

pulsectl 是 PulseAudio 客户端 C 库 libpulse 的 Python ctypes 包装器，提供了对 PulseAudio 的源/接收器/流处理和音量混音的高级接口。它最初是从 pulsemixer 命令行应用程序的内部代码分叉而来。

虽然 libpulse 提供了基于回调的异步 C API 用于与 PulseAudio 服务器通信，但 pulsectl 只暴露了阻塞的 Python 接口，让 libpulse 的内部事件循环在收到每个请求的响应之前一直旋转。在 pulsectl 的 README 文件 和 问题 #11 中讨论了将库集成到异步 Python 应用程序的不同方法。然而，这些方法中的任何一种都无法提供与 Python 的 asyncio 事件循环框架的无缝集成。

pulsectl-asyncio 使用基于 ctypes 的 Python 实现了 libpulse 的 main_loop_api，以便使用 Python asyncio 事件循环来处理 libpulse 的异步事件。有了这种事件处理，就不需要调用 libpulse 的阻塞调用，因此可以提供一个 pulsectl 的高层 API 的异步版本：由 pulsectl-asyncio 提供的 PulseAsync 类，与 pulsectl 中的 Pulse 类完全相似，不同之处在于所有方法都声明为 async，并异步等待操作的结果。此外，订阅 PulseAudio 服务器事件的 API 已经从基于回调的接口（event_callback_set() 等）更改为使用异步生成器的一个更符合 asyncio 的接口。

pulsectl-asyncio 依赖于 pulsectl，以重用它对 libpulse 的 ctype 包装以及 PulseObject 类，这些类用于将 PulseAudio 操作结果结构建模为 Python 对象。高级 API 类 PulseAsync 已从 pulsectl 复制并修改为异步控制流。因此，其架构及其代码的主要部分仍然类似于 pulsectl 的代码。

有关 API、返回的 PulseObject 对象和其他值类型、音量指定等更多信息，请参阅 pulsectl 的 README 文件。

使用示例

（大量灵感来自 pulsectl 的 README 文件）

简单示例

import asyncio
import pulsectl_asyncio

async def main():
    async with pulsectl_asyncio.PulseAsync('volume-increaser') as pulse:
        for sink in await pulse.sink_list():
            await pulse.volume_change_all_chans(sink, 0.1)

loop = asyncio.get_event_loop()
loop.run_until_complete(main())

监听服务器状态更改事件

import asyncio
import signal
from contextlib import suppress
import pulsectl_asyncio

# import pulsectl
# print('Event types:', pulsectl.PulseEventTypeEnum)
# print('Event facilities:', pulsectl.PulseEventFacilityEnum)
# print('Event masks:', pulsectl.PulseEventMaskEnum)

async def listen():
    async with pulsectl_asyncio.PulseAsync('event-printer') as pulse:
        async for event in pulse.subscribe_events('all'):
            print('Pulse event:', event)

async def main():
    # Run listen() coroutine in task to allow cancelling it
    listen_task = asyncio.create_task(listen())

    # register signal handlers to cancel listener when program is asked to terminate
    for sig in (signal.SIGTERM, signal.SIGHUP, signal.SIGINT):
        loop.add_signal_handler(sig, listen_task.cancel)
    # Alternatively, the PulseAudio event subscription can be ended by breaking/returning from the `async for` loop

    with suppress(asyncio.CancelledError):
        await listen_task

# Run event loop until main_task finishes
loop = asyncio.get_event_loop()
loop.run_until_complete(main())

其他杂项调整

import asyncio
import pulsectl_asyncio

async def main():
    pulse = pulsectl_asyncio.PulseAsync('my-client-name')
    await pulse.connect()

    print(await pulse.sink_list())
    # [<PulseSinkInfo at 7f85cfd053d0 - desc='Built-in Audio', index=0L, mute=0, name='alsa-speakers', channels=2, volumes='44.0%, 44.0%'>]
    print(await pulse.sink_input_list())
    # [<PulseSinkInputInfo at 7fa06562d3d0 - index=181L, mute=0, name='mpv Media Player', channels=2, volumes='25.0%, 25.0%'>]

    print((await pulse.sink_input_list())[0].proplist)  # Note the parentheses around `await` and the method call
    # {'application.icon_name': 'mpv',
    #  'application.language': 'C',
    #  'application.name': 'mpv Media Player',
    #  ...
    #  'native-protocol.version': '30',
    #  'window.x11.display': ':1.0'}

    print(await 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 = (await pulse.sink_list())[0]
    await pulse.volume_change_all_chans(sink, -0.1)
    await pulse.volume_set_all_chans(sink, 0.5)

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

    card = (await pulse.card_list())[0]
    print(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>]

    await pulse.card_profile_set(card, 'output:hdmi-stereo')

    pulse.close()  # No await here!

loop = asyncio.get_event_loop()
loop.run_until_complete(main())

在命令行中监控默认声卡的输出电平

import asyncio
import signal
from contextlib import suppress

import pulsectl_asyncio

async def listen(pulse: pulsectl_asyncio.PulseAsync, source_name: str):
    async for level in pulse.subscribe_peak_sample(source_name, rate=5):
        print('\x1b[2K\x1b[0E', end='')  # return to beginning of line
        num_o = round(level * 80)
        print('O' * num_o + '-' * (80-num_o), end='', flush=True)


async def main():
    async with pulsectl_asyncio.PulseAsync('peak-listener') as pulse:
        # Get name of monitor_source of default sink
        server_info = await pulse.server_info()
        default_sink_info = await pulse.get_sink_by_name(server_info.default_sink_name)
        source_name = default_sink_info.monitor_source_name

        # Start listening/monitoring task
        listen_task = loop.create_task(listen(pulse, source_name))

        # register signal handlers to cancel listener when program is asked to terminate
        # Alternatively, the PulseAudio event subscription can be ended by breaking/returning from the `async for` loop
        for sig in (signal.SIGTERM, signal.SIGHUP, signal.SIGINT):
            loop.add_signal_handler(sig, listen_task.cancel)

        with suppress(asyncio.CancelledError):
            await listen_task
            print()


# Run event loop until main_task finishes
loop = asyncio.get_event_loop()
loop.run_until_complete(main())