-
-
Save nidefawl/75ca46ba9979062290b066a7c1ee08fd to your computer and use it in GitHub Desktop.
Small tool to configure and maintain PipeWire links between sets of input/output port querie
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/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. | |
# Only the first two channels are linked. And links are created in stereo pairs | |
# | |
# 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) | |
printStatusOnce = False | |
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)) | |
if not len(ports_out) and not printStatusOnce: | |
sys.stderr.write(f"No output ports found for pid {args.pid_out} name regex {args.name_regex_out} application regex {args.application_regex_out} channel {args.channel_out}\n") | |
sys.stderr.flush() | |
if not args.continuous: | |
sys.exit(1) | |
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)) | |
if not len(ports_in) and not printStatusOnce: | |
sys.stderr.write(f"No input ports found for pid {args.pid_in} name regex {args.name_regex_in} application regex {args.application_regex_in} channel {args.channel_in}\n") | |
sys.stderr.flush() | |
if not args.continuous: | |
sys.exit(1) | |
links = pipewire_links(dump=dump) | |
if args.list_only: | |
print("Outputs") | |
print("-------") | |
print(ports_out) | |
print("") | |
print("") | |
print("Inputs") | |
print("------") | |
print(ports_in) | |
if not printStatusOnce: | |
print("Ports out size =", len(ports_out)) | |
print("Ports in size =", len(ports_in)) | |
for i in range(2): | |
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 not port_name_out.endswith(str(i)): | |
continue | |
if not port_name_in.endswith(str(i)): | |
continue | |
if any(link["info"]["output-port-id"] == port_id_out and link["info"]["input-port-id"] == port_id_in for link in links): | |
if not printStatusOnce: | |
print("Skipping already linked ports:", port_name_out, "->", port_name_in) | |
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)) | |
printStatusOnce = True |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment