App-aware Node Linker for PipeWire

pwapp-link

#!/usr/bin/env python3

import re
import subprocess
import sys

header_rx = re.compile(r'^\s+id (\d+), type (.*)$')
property_rx = re.compile(r'^\s+([^ =]+)\s?=\s?"([^"]+)"$')

nodes = {}
ports = {}
links = {}

def get_port(alias):
    for p in ports.values():
        if p["port.alias"] == alias:
            return p

    splat = alias.split(':')
    nodename = splat[0]
    portname = splat[1]

    for n in nodes.values():
        if nodename in n.get("node.name", "") or nodename in n.get("application.name", ""):
            for p2 in n.get("ports", []):
                if p2["port.name"] == portname:
                    return p2
    print("Could not find port for: %s" % alias)
    return None

def get_nodes(name):
    for n in nodes.values():
        if name in n.get("node.name", "") or name in n.get("node.description", "") or name in n.get("application.name", ""):
            yield n

def run_command(line):
    line = line.rstrip()
    print("run-command %s" % line)
    splat = line.split()

    if splat[0] == "unlink-all":
        for n in get_nodes(splat[1]):
            for p in n["ports"]:
                for l in p["links"]:
                    print("unlink", l["id"])
                    subprocess.call(["/usr/bin/pw-link", "-d", l["id"]])

    elif splat[0] == "unlink":
        p = get_port(splat[1])
        if p:
            for l in p["links"]:
                print("unlink", l["id"])
                subprocess.call(["/usr/bin/pw-link", "-d", l["id"]])

    elif splat[0] == "link":
        src = get_port(splat[1])
        dst = get_port(splat[2])
        if src and dst:
            print("link", src["id"], dst["id"])
            subprocess.call(["/usr/bin/pw-link", src["id"], dst["id"]])
    else:
        print("idk")
        
def buildgraph():
    outp = subprocess.check_output(["/usr/bin/pw-cli", "ls"]).decode().splitlines()
    obj = {}
    i = 0
    while i < len(outp):
        line = outp[i]
        i += 1
        
        m = header_rx.match(line)
        if m:
            obj = {
                    "id": m.group(1),
                    "type": m.group(2)
                }
            if "PipeWire:Interface:Port" in obj["type"]:
                obj["links"] = []
                ports[obj["id"]] = obj
            elif "PipeWire:Interface:Node" in obj["type"]:
                obj["ports"] = []
                nodes[obj["id"]] = obj
            elif "PipeWire:Interface:Link" in obj["type"]:
                links[obj["id"]] = obj
            # else throw away
            continue
        m = property_rx.match(line)
        if m:
            obj[m.group(1)] = m.group(2)

    for link in links.values():
        for port_id in [link["link.input.port"], link["link.output.port"]]:
           ports[port_id]["links"].append(link)

    for port in ports.values():
        node_id = port["node.id"]
        nodes[node_id]["ports"].append(port)


if __name__ == "__main__":
    buildgraph()
    print("Nodes: "+str(len(nodes)))
    print("Ports: "+str(len(ports)))
    print("Links: "+str(len(links)))
    for line in sys.stdin.readlines():
        run_command(line)