This repository has no description
7.6 kB
266 lines
1# Audio routing matrix: one Home Assistant switch (MQTT discovery) per
2# source and speaker. While a switch is on, the source's PipeWire node is
3# linked to the speaker's node. Sources are the fixed nodes from the config
4# plus every connected Bluetooth A2DP source (bluez_input.*). Usage:
5# audio-matrix CONFIG.json [--dry-run], --dry-run prints the pw-link calls
6# without running them.
7import json
8import os
9import queue
10import subprocess
11import sys
12import threading
13
14import paho.mqtt.client as mqtt
15
16with open(sys.argv[1]) as f:
17 config = json.load(f)
18dry_run = "--dry-run" in sys.argv[2:]
19speakers = config["speakers"]
20
21events = queue.Queue()
22
23# PipeWire objects by id: nodes, ports and links
24graph = {}
25interesting = {
26 "PipeWire:Interface:Node",
27 "PipeWire:Interface:Port",
28 "PipeWire:Interface:Link",
29}
30
31
32def read_graph(proc):
33 # every update is a JSON array whose closing bracket starts a line
34 chunk = []
35 try:
36 for line in proc.stdout:
37 chunk.append(line)
38 if line.startswith("]"):
39 events.put(("graph", json.loads("".join(chunk))))
40 chunk = []
41 finally:
42 proc.kill()
43 events.put(("exit", proc.wait()))
44
45
46def apply(update):
47 for o in update:
48 if "type" not in o:
49 graph.pop(o["id"], None)
50 elif o["type"] in interesting:
51 graph[o["id"]] = o
52
53
54def objects(kind):
55 for o in graph.values():
56 if o["type"] == "PipeWire:Interface:" + kind:
57 yield o["id"], o["info"]
58
59
60def nodes_named(name):
61 return [
62 i for i, info in objects("Node")
63 if info["props"].get("node.name") == name
64 ]
65
66
67def sources():
68 found = {
69 key: {"label": s["label"], "kind": key, "nodes": nodes_named(s["node"])}
70 for key, s in config["sources"].items()
71 }
72 for i, info in objects("Node"):
73 props = info["props"]
74 address = props.get("api.bluez5.address")
75 if not props.get("node.name", "").startswith("bluez_input.") or not address:
76 continue
77 key = address.lower().replace(":", "_")
78 src = found.setdefault(
79 key,
80 {
81 "label": props.get("node.description", address),
82 "kind": "bluetooth",
83 "nodes": [],
84 },
85 )
86 src["nodes"].append(i)
87 return found
88
89
90# whether both nodes have their ports to link yet
91def ports(out_node, in_node):
92 have = {
93 (int(info["props"]["node.id"]), info["direction"])
94 for _, info in objects("Port")
95 }
96 return (out_node, "output") in have and (in_node, "input") in have
97
98
99def linked(out_node, in_node):
100 return any(
101 info["output-node-id"] == out_node and info["input-node-id"] == in_node
102 for _, info in objects("Link")
103 )
104
105
106def pw_link(*args):
107 cmd = ["pw-link", *map(str, args)]
108 print(" ".join(cmd), flush=True)
109 if not dry_run:
110 subprocess.run(cmd)
111
112
113# speakers each source plays on, by source key
114try:
115 with open(config["state"]) as f:
116 routes = json.load(f)
117except FileNotFoundError:
118 routes = {}
119
120
121def save():
122 with open(config["state"] + ".tmp", "w") as f:
123 json.dump(routes, f)
124 os.replace(config["state"] + ".tmp", config["state"])
125
126
127client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
128client.username_pw_set(config["mqtt"]["user"], config["mqtt"]["password"])
129
130
131def entity(key, speaker):
132 return f"audio_route_{key}_{speaker}"
133
134
135def topic(key, speaker):
136 return f"gg23/audio-route/{key}/{speaker}"
137
138
139def publish_state(key, speaker):
140 state = "ON" if speaker in routes[key] else "OFF"
141 client.publish(topic(key, speaker) + "/state", state, retain=True)
142
143
144def publish(key, src):
145 for speaker, sp in speakers.items():
146 base = topic(key, speaker)
147 attributes = {
148 "source": src["label"],
149 "speaker": sp["label"],
150 "kind": src["kind"],
151 }
152 client.publish(base + "/attributes", json.dumps(attributes), retain=True)
153 publish_state(key, speaker)
154 discovery = {
155 "name": sp["label"],
156 "unique_id": "acheron_" + entity(key, speaker),
157 "default_entity_id": "switch." + entity(key, speaker),
158 "command_topic": base + "/set",
159 "state_topic": base + "/state",
160 "json_attributes_topic": base + "/attributes",
161 "icon": "mdi:speaker",
162 "device": {
163 "identifiers": ["acheron_audio_" + key],
164 "name": src["label"],
165 },
166 }
167 client.publish(
168 f"homeassistant/switch/{entity(key, speaker)}/config",
169 json.dumps(discovery),
170 retain=True,
171 )
172
173
174def unpublish(key):
175 for speaker in speakers:
176 client.publish(
177 f"homeassistant/switch/{entity(key, speaker)}/config", "", retain=True
178 )
179 for t in ["state", "attributes"]:
180 client.publish(topic(key, speaker) + "/" + t, "", retain=True)
181
182
183# label of each source whose switches are published, None if unknown
184published = {}
185
186
187def sync():
188 found = sources()
189 for key in [k for k in published if k not in found]:
190 unpublish(key)
191 del published[key]
192 for key, src in found.items():
193 if key not in routes:
194 routes[key] = list(speakers)
195 save()
196 if key not in published or published[key] != src["label"]:
197 publish(key, src)
198 published[key] = src["label"]
199 for speaker, sp in speakers.items():
200 want = speaker in routes[key]
201 for out_node in src["nodes"]:
202 for in_node in nodes_named(sp["node"]):
203 if linked(out_node, in_node):
204 if not want:
205 pw_link("-d", out_node, in_node)
206 elif want and ports(out_node, in_node):
207 pw_link(out_node, in_node)
208
209
210def command(t, payload):
211 _, _, key, speaker, _ = t.split("/")
212 if key not in published or speaker not in speakers:
213 return
214 if payload == "ON" and speaker not in routes[key]:
215 routes[key].append(speaker)
216 elif payload == "OFF" and speaker in routes[key]:
217 routes[key].remove(speaker)
218 save()
219 publish_state(key, speaker)
220
221
222def on_connect(client, userdata, flags, reason_code, properties):
223 client.subscribe("gg23/audio-route/+/+/set")
224 events.put(("connect",))
225
226
227def on_message(client, userdata, msg):
228 events.put(("command", msg.topic, msg.payload.decode()))
229
230
231client.on_connect = on_connect
232client.on_message = on_message
233client.connect_async(config["mqtt"]["host"])
234client.loop_start()
235pw_dump = subprocess.Popen(
236 ["pw-dump", "--monitor", "--no-colors"],
237 stdout=subprocess.PIPE,
238 text=True,
239)
240threading.Thread(target=read_graph, args=(pw_dump,), daemon=True).start()
241
242graph_ready = False
243
244
245def handle(event):
246 global graph_ready, published
247 if event[0] == "graph":
248 apply(event[1])
249 graph_ready = True
250 elif event[0] == "connect":
251 # republish everything, and clear the switches of sources that
252 # vanished while nobody was connected
253 published = dict.fromkeys(routes)
254 elif event[0] == "command":
255 command(*event[1:])
256 elif event[0] == "exit":
257 sys.exit(f"pw-dump exited with {event[1]}")
258
259
260while True:
261 handle(events.get())
262 # sync once per burst, so links just created are seen before it
263 while not events.empty():
264 handle(events.get())
265 if graph_ready:
266 sync()