Main Loop And Registry Discovery

PipeWire registry updates are delivered through a GLib main context. A program that starts PwgRegistry should run a GMainLoop or otherwise iterate the thread-default main context. Live wrappers such as PwgNode, PwgDevice, PwgMetadata, and PwgStream deliver their signals through the same context.

import gi

gi.require_version("GLib", "2.0")
gi.require_version("Pwg", "0.1")
from gi.repository import GLib, Pwg

Pwg.init()

core = Pwg.Core.new()
registry = Pwg.Registry.new(core)
registry.start()

loop = GLib.MainLoop()


def on_global_added(_registry, global_):
    print(global_.get_id(), global_.get_interface_type(), global_.dup_name() or "")


registry.connect("global-added", on_global_added)
GLib.timeout_add_seconds(2, loop.quit)
loop.run()

The registry keeps immutable PwgGlobal descriptors in a GListModel. Snapshot helper methods return new list models so callers can inspect a stable set of objects without holding the registry’s internal model.

For client-specific discovery, wrap a client global in PwgClientInfo:

clients = registry.dup_globals_by_interface("PipeWire:Interface:Client")
if clients.get_n_items() > 0:
    client = Pwg.ClientInfo.new_from_global(clients.get_item(0))
    if client is not None:
        print(
            client.dup_app_name() or client.dup_name() or "",
            client.dup_app_id() or "",
            client.dup_process_binary() or "",
        )

For device-specific discovery, wrap a device global in PwgDeviceInfo:

devices = registry.dup_globals_by_interface("PipeWire:Interface:Device")
if devices.get_n_items() > 0:
    device = Pwg.DeviceInfo.new_from_global(devices.get_item(0))
    if device is not None:
        print(
            device.dup_description() or device.dup_nick() or device.dup_name() or "",
            device.dup_api() or "",
            device.dup_form_factor() or "",
        )

For link-specific discovery, wrap a link global in PwgLinkInfo:

links = registry.dup_globals_by_interface("PipeWire:Interface:Link")
if links.get_n_items() > 0:
    link = Pwg.LinkInfo.new_from_global(links.get_item(0))
    if link is not None:
        print(
            link.dup_output_node_id() or "",
            link.dup_output_port_id() or "",
            "->",
            link.dup_input_node_id() or "",
            link.dup_input_port_id() or "",
        )

For node-specific discovery, wrap a node global in PwgNodeInfo:

nodes = registry.dup_globals_by_interface("PipeWire:Interface:Node")
if nodes.get_n_items() > 0:
    node_info = Pwg.NodeInfo.new_from_global(nodes.get_item(0))
    if node_info is not None:
        print(node_info.dup_name() or "", node_info.dup_media_class() or "")

For node parameter inspection, bind the same global with PwgNode. The live node proxy exposes copied PwgParamInfo descriptors and emits copied PwgParam values when enumeration results arrive:

node = Pwg.Node.new(core, nodes.get_item(0))
if node is not None:
    node.start()

    for index in range(node.get_param_infos().get_n_items()):
        param_info = node.get_param_infos().get_item(index)
        print(param_info.get_id(), param_info.dup_name() or "")

    def on_param(_node, param):
        print(
            param.get_id(),
            param.dup_name() or "",
            param.dup_format_media_type_name() or "",
            param.dup_format_media_subtype_name() or "",
            param.dup_summary(),
        )
        audio_format = param.dup_audio_format()
        if audio_format is not None:
            print(
                audio_format.get_sample_format(),
                audio_format.get_rate(),
                audio_format.get_channels(),
            )

    node.connect("param", on_param)
    node.enum_all_params()

To observe later node parameter changes, subscribe to the parameter ids that the application cares about. The subscription call replaces the active set of ids; an empty au clears it. PipeWire may deliver current values and later changes through the same signal, so applications should treat subscription events as “the parameter changed or was delivered” rather than trying to infer intent from the parameter sequence number.

volume_param_id = 2
node.connect("param", on_param)
node.subscribe_params(GLib.Variant("au", [volume_param_id]))

# Later, clear the active subscription.
node.subscribe_params(GLib.Variant("au", []))

Nodes can also queue copied parameter updates that were built by this library. The return value only means PipeWire accepted the request for dispatch; it is not an applied-state acknowledgment:

param = Pwg.Param.new_props_volume(0.75)
if param is not None:
    node.set_param(param)

For live device parameter inspection, bind a device global with PwgDevice. The API mirrors PwgNode: parameter descriptors are published through PwgParamInfo, enumeration results are copied PwgParam objects, and subscriptions are event-driven.

PipeWire devices often advertise route parameters that describe input or output routes exposed by the device. Wrap copied EnumRoute or Route parameters in PwgRouteInfo to inspect route fields without parsing SPA POD data in application code:

device_proxy = Pwg.Device.new(core, devices.get_item(0))
if device_proxy is not None:
    device_proxy.start()

    route_ids = []
    for index in range(device_proxy.get_param_infos().get_n_items()):
        param_info = device_proxy.get_param_infos().get_item(index)
        if param_info.dup_name() == "Route":
            route_ids.append(param_info.get_id())

    def on_device_param(_device, param):
        route = Pwg.RouteInfo.new_from_param(param)
        if route is not None:
            print(
                route.get_index(),
                route.dup_direction() or "",
                route.dup_name() or "",
                route.dup_description() or "",
                route.dup_availability() or "",
            )

    device_proxy.connect("param", on_device_param)
    for route_id in route_ids:
        device_proxy.enum_params(route_id, 0, 0)
    device_proxy.subscribe_params(GLib.Variant("au", route_ids))

Route parameters are inspection data from PipeWire. pipewire-gobject does not select hardware routes, change defaults, or replace WirePlumber policy. Apps that need to react to route changes should subscribe and update their own UI or state when new copied route params arrive.

For modules that expose named float controls, build a Props parameter from a GLib.Variant dictionary with signature a{sd}:

controls = GLib.Variant(
    "a{sd}",
    {
        "band_l_0:b0": 1.0,
        "band_l_0:b1": -0.25,
    },
)
param = Pwg.Param.new_props_controls(controls)
if param is not None:
    node.set_param(param)

App-owned PipeWire implementation modules can be loaded from PwgCore. The returned PwgImplModule keeps the module alive until it is unloaded or dropped:

module = core.load_module("libpipewire-module-loopback", "node.name=pwg-loopback")
try:
    print(module.get_loaded())
finally:
    module.unload()

For app-owned audio capture, PwgStream is the canonical API. It exposes negotiated format, level, and optional copied audio blocks. The requested format defaults to F32, 48000 Hz, stereo; set it before starting the stream when the application needs a different F32 mono or stereo rate:

stream = Pwg.Stream.new_audio_capture("alsa_output.example", True)
stream.set_requested_format("F32", 48000, 2)
stream.set_deliver_audio_blocks(True)


def on_audio_block(_stream, block):
    print(block.get_n_frames(), block.get_peak())


stream.connect("audio-block", on_audio_block)
stream.start()

For port-specific discovery, wrap a port global in PwgPortInfo:

ports = registry.dup_globals_by_interface("PipeWire:Interface:Port")
if ports.get_n_items() > 0:
    port = Pwg.PortInfo.new_from_global(ports.get_item(0))
    if port is not None:
        print(
            port.dup_direction() or "",
            port.dup_name() or "",
            port.dup_audio_channel() or "",
        )

The registry is a discovery API, not a session manager. It does not make routing or default-device policy decisions.