Small tool to configure and maintain PipeWire links between sets of input/output port querie

pw-easylink

#!/usr/bin/python
# coding=utf-8

"""Configures and maintain PipeWire links between sets of queries for input/output ports"""

# The following example configures links between output ports whose application
# name starts with "Firefox" and channel is "Front-Right", and input ports whose
# application name is "ardour" and port name starts with "Track name" and ends
# with "2". Links are automatically created whenver a new port appears for
# either of these queries.
#
#  pw-easylink -ao 'Firefox.*' -co FR -ai 'ardour' -ni 'Track name.*2' --continuous

import re
import sys
import time
import json
import subprocess

def pipewire_dump():
    return json.loads(subprocess.check_output("pw-dump"))


def pipewire_link(port_id_out, port_id_in):
    subprocess.check_call(["pw-link", str(port_id_out), str(port_id_in)])


def pipewire_ports(dump=None):
    dump = dump if dump is not None else pipewire_dump()
    ports = [data for data in dump if data["type"] == "PipeWire:Interface:Port"]
    return ports


def pipewire_links(dump=None):
    dump = dump if dump is not None else pipewire_dump()
    links = [data for data in dump if data["type"] == "PipeWire:Interface:Link"]
    return links


def pipewire_nodes(dump=None):
    dump = dump if dump is not None else pipewire_dump()
    ports = [data for data in dump if data["type"] == "PipeWire:Interface:Node"]
    return ports


def pipewire_find_node_by_id(id, dump=None):
    for node in pipewire_nodes(dump=dump):
        if node["id"] == id:
            return node


def pipewire_find_port(pid=None, name_regex=None, application_regex=None, channel=None, direction=None, dump=None):

    if application_regex is not None:
        application_regex = re.compile(application_regex)
    if name_regex is not None:
        name_regex = re.compile(name_regex)

    for port in pipewire_ports(dump=dump):
        try:
            props = port["info"]["props"]
            if any(v is not None for v in (pid, application_regex)):
                nodeid = props.get("node.id")
                node = pipewire_find_node_by_id(nodeid, dump=dump)
                nprops = node["info"]["props"]
                application_name = nprops.get("application.name")
                application_name = application_name or nprops.get("node.name")
                application_name = application_name or nprops.get("client.name")
                if pid is not None and nprops.get("application.process.id") != pid:
                    continue
                if application_regex is not None and not application_regex.match(application_name):
                    continue
            if name_regex is not None:
                if name_regex.match(props.get("object.path", "")):
                    pass
                elif name_regex.match(props.get("port.alias", "")):
                    pass
                elif name_regex.match(props.get("port.name", "")):
                    pass
                else:
                    continue
            if channel is not None and props.get("audio.channel") != channel:
                continue
            if direction is not None and port["info"]["direction"] != direction:
                continue
        except KeyError:
            pass
        except ValueError:
            pass
        yield port


if __name__ == '__main__':
    import argparse
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--pid-out", "-po", help="Client pid for output part of the link")
    parser.add_argument("--pid-in", "-pi", help="Client pid for input part of the link")
    parser.add_argument("--name-regex-out", "-no", help="Name regex for output part of the link")
    parser.add_argument("--name-regex-in", "-ni", help="Name regex for input part of the link")
    parser.add_argument("--application-regex-out", "-ao", help="Application name regex for output part of the link")
    parser.add_argument("--application-regex-in", "-ai", help="Application name regex for input part of the link")
    parser.add_argument("--channel-out", "-co", help="Channel for output part of the link")
    parser.add_argument("--channel-in", "-ci", help="Channel for input part of the link")
    parser.add_argument("--list-only", "-l", help="Only list ports without connecting", action="store_true")
    parser.add_argument("--continuous", "-c", help="Run continuously", action="store_true")
    parser.add_argument("--interval", "-i", help="Interval between continuous checks (in s)", type=int, default=1)
    args = parser.parse_args()

    if args.pid_out is None and args.name_regex_out is None and args.application_regex_out is None and args.channel_out is None:
        sys.stderr.write(f"Please specify at least one output filter\n")
        sys.stderr.flush()
        sys.exit(1)

    if args.pid_in is None and args.name_regex_in is None and args.application_regex_in is None and args.channel_in is None:
        sys.stderr.write(f"Please specify at least one input filter\n")
        sys.stderr.flush()
        sys.exit(1)

    while True:
        dump = pipewire_dump()
        ports_out = list(pipewire_find_port(pid=args.pid_out, name_regex=args.name_regex_out, application_regex=args.application_regex_out, channel=args.channel_out, direction="output", dump=dump))
        ports_in = list(pipewire_find_port(pid=args.pid_in, name_regex=args.name_regex_in, application_regex=args.application_regex_in, channel=args.channel_in, direction="input", dump=dump))
        links = pipewire_links(dump=dump)
        if args.list_only:
            print("Outputs")
            print("-------")
            print(ports_out)
            print("")
            print("")
            print("Inputs")
            print("------")
            print(ports_in)
        for port_out in ports_out:
            for port_in in ports_in:
                port_id_out = port_out["id"]
                port_id_in = port_in["id"]
                port_name_out = port_out["info"]["props"].get("object.path") or port_out["info"]["props"].get("port.name")
                port_name_in = port_in["info"]["props"].get("object.path") or port_in["info"]["props"].get("port.name")
                if any(link["info"]["output-port-id"] == port_id_out and link["info"]["input-port-id"] == port_id_in for link in links):
                    continue
                print(f"Creating new link between ports {port_id_out}:{port_name_out} -> {port_id_in}:{port_name_in}")
                if not args.list_only:
                    pipewire_link(port_id_out, port_id_in)
        if not args.continuous:
            break
        time.sleep(max(0, args.interval))