This repository has no description
11 kB
370 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.*; phones left on
5# the "off" profile are switched to A2DP). A new source starts on the
6# config's `rooms` of it, else on all speakers. Each speaker
7# also gets a volume (MQTT number), applied to its node with wpctl. Usage:
8# audio-matrix CONFIG.json [--dry-run], --dry-run prints the pw-link and
9# wpctl calls without running them.
10import json
11import os
12import queue
13import subprocess
14import sys
15import threading
16
17import paho.mqtt.client as mqtt
18
19with open(sys.argv[1]) as f:
20 config = json.load(f)
21dry_run = "--dry-run" in sys.argv[2:]
22speakers = config["speakers"]
23
24events = queue.Queue()
25
26# PipeWire objects by id: devices, nodes, ports and links
27graph = {}
28interesting = {
29 "PipeWire:Interface:Device",
30 "PipeWire:Interface:Node",
31 "PipeWire:Interface:Port",
32 "PipeWire:Interface:Link",
33}
34
35
36def read_graph(proc):
37 # every update is a JSON array whose closing bracket starts a line
38 chunk = []
39 try:
40 for line in proc.stdout:
41 chunk.append(line)
42 if line.startswith("]"):
43 events.put(("graph", json.loads("".join(chunk))))
44 chunk = []
45 finally:
46 proc.kill()
47 events.put(("exit", proc.wait()))
48
49
50def apply(update):
51 for o in update:
52 if "type" not in o:
53 graph.pop(o["id"], None)
54 elif o["type"] in interesting:
55 graph[o["id"]] = o
56
57
58def objects(kind):
59 for o in graph.values():
60 if o["type"] == "PipeWire:Interface:" + kind:
61 yield o["id"], o["info"]
62
63
64def nodes_named(name):
65 return [
66 i for i, info in objects("Node")
67 if info["props"].get("node.name") == name
68 ]
69
70
71def sources():
72 found = {
73 key: {
74 "label": s["label"],
75 "kind": key,
76 "nodes": nodes_named(s["node"]),
77 "rooms": s.get("rooms"),
78 }
79 for key, s in config["sources"].items()
80 }
81 for i, info in objects("Node"):
82 props = info["props"]
83 address = props.get("api.bluez5.address")
84 if not props.get("node.name", "").startswith("bluez_input.") or not address:
85 continue
86 key = address.lower().replace(":", "_")
87 src = found.setdefault(
88 key,
89 {
90 "label": props.get("node.description", address),
91 "kind": "bluetooth",
92 "nodes": [],
93 },
94 )
95 src["nodes"].append(i)
96 return found
97
98
99# bluez devices whose profile was switched on, so a phone that turns it off
100# again is left alone
101profiled = set()
102
103
104# WirePlumber leaves a newly paired phone on the "off" profile when its A2DP
105# transport is not up yet at connect time, so it gets no bluez_input node.
106# Switch such devices to the A2DP (audio-gateway) profile once.
107def bluetooth_profiles():
108 for i, info in objects("Device"):
109 if info["props"].get("device.api") != "bluez5" or i in profiled:
110 continue
111 params = info.get("params", {})
112 current = (params.get("Profile") or [{}])[0].get("name")
113 gateway = [
114 p for p in params.get("EnumProfile", [])
115 if p.get("name") == "audio-gateway" and p.get("available") != "no"
116 ]
117 if current == "off" and gateway:
118 run("wpctl", "set-profile", i, gateway[0]["index"])
119 profiled.add(i)
120
121
122# whether both nodes have their ports to link yet
123def ports(out_node, in_node):
124 have = {
125 (int(info["props"]["node.id"]), info["direction"])
126 for _, info in objects("Port")
127 }
128 return (out_node, "output") in have and (in_node, "input") in have
129
130
131def linked(out_node, in_node):
132 return any(
133 info["output-node-id"] == out_node and info["input-node-id"] == in_node
134 for _, info in objects("Link")
135 )
136
137
138def run(*cmd):
139 print(" ".join(map(str, cmd)), flush=True)
140 if not dry_run:
141 subprocess.run(list(map(str, cmd)))
142
143
144def pw_link(*args):
145 run("pw-link", *args)
146
147
148def load(path):
149 try:
150 with open(path) as f:
151 return json.load(f)
152 except FileNotFoundError:
153 return {}
154
155
156def store(path, data):
157 with open(path + ".tmp", "w") as f:
158 json.dump(data, f)
159 os.replace(path + ".tmp", path)
160
161
162# speakers each source plays on, by source key
163routes = load(config["state"])
164# volume in percent by speaker, 100 if never set
165volumes = load(config["volumes"])
166# volume last set on each speaker node, by node id
167applied = {}
168
169
170def save():
171 store(config["state"], routes)
172
173
174client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
175client.username_pw_set(config["mqtt"]["user"], config["mqtt"]["password"])
176
177
178def entity(key, speaker):
179 return f"audio_route_{key}_{speaker}"
180
181
182def topic(key, speaker):
183 return f"gg23/audio-route/{key}/{speaker}"
184
185
186def publish_state(key, speaker):
187 state = "ON" if speaker in routes[key] else "OFF"
188 client.publish(topic(key, speaker) + "/state", state, retain=True)
189
190
191def volume_topic(speaker):
192 return f"gg23/audio-volume/{speaker}"
193
194
195def publish_volumes():
196 for speaker, sp in speakers.items():
197 base = volume_topic(speaker)
198 client.publish(base + "/state", volumes.get(speaker, 100), retain=True)
199 discovery = {
200 "name": sp["label"],
201 "unique_id": f"acheron_audio_volume_{speaker}",
202 "default_entity_id": f"number.audio_volume_{speaker}",
203 "command_topic": base + "/set",
204 "state_topic": base + "/state",
205 "min": 0,
206 "max": 100,
207 "step": 1,
208 "mode": "slider",
209 "unit_of_measurement": "%",
210 "icon": "mdi:volume-high",
211 "device": {
212 "identifiers": ["acheron_audio_volume"],
213 "name": "Room volume",
214 },
215 }
216 client.publish(
217 f"homeassistant/number/audio_volume_{speaker}/config",
218 json.dumps(discovery),
219 retain=True,
220 )
221
222
223def publish(key, src):
224 for speaker, sp in speakers.items():
225 base = topic(key, speaker)
226 attributes = {
227 "source": src["label"],
228 "speaker": sp["label"],
229 "kind": src["kind"],
230 }
231 client.publish(base + "/attributes", json.dumps(attributes), retain=True)
232 publish_state(key, speaker)
233 discovery = {
234 "name": sp["label"],
235 "unique_id": "acheron_" + entity(key, speaker),
236 "default_entity_id": "switch." + entity(key, speaker),
237 "command_topic": base + "/set",
238 "state_topic": base + "/state",
239 "json_attributes_topic": base + "/attributes",
240 "icon": "mdi:speaker",
241 "device": {
242 "identifiers": ["acheron_audio_" + key],
243 "name": src["label"],
244 },
245 }
246 client.publish(
247 f"homeassistant/switch/{entity(key, speaker)}/config",
248 json.dumps(discovery),
249 retain=True,
250 )
251
252
253def unpublish(key):
254 for speaker in speakers:
255 client.publish(
256 f"homeassistant/switch/{entity(key, speaker)}/config", "", retain=True
257 )
258 for t in ["state", "attributes"]:
259 client.publish(topic(key, speaker) + "/" + t, "", retain=True)
260
261
262# label of each source whose switches are published, None if unknown
263published = {}
264
265
266def sync():
267 bluetooth_profiles()
268 found = sources()
269 for key in [k for k in published if k not in found]:
270 unpublish(key)
271 del published[key]
272 for key, src in found.items():
273 if key not in routes:
274 rooms = src.get("rooms")
275 routes[key] = list(speakers if rooms is None else rooms)
276 save()
277 if key not in published or published[key] != src["label"]:
278 publish(key, src)
279 published[key] = src["label"]
280 for speaker, sp in speakers.items():
281 want = speaker in routes[key]
282 for out_node in src["nodes"]:
283 for in_node in nodes_named(sp["node"]):
284 if linked(out_node, in_node):
285 if not want:
286 pw_link("-d", out_node, in_node)
287 elif want and ports(out_node, in_node):
288 pw_link(out_node, in_node)
289 for speaker, sp in speakers.items():
290 volume = volumes.get(speaker, 100)
291 for node in nodes_named(sp["node"]):
292 if applied.get(node) != volume:
293 run("wpctl", "set-volume", node, volume / 100)
294 applied[node] = volume
295
296
297def command(t, payload):
298 _, _, key, speaker, _ = t.split("/")
299 if key not in published or speaker not in speakers:
300 return
301 if payload == "ON" and speaker not in routes[key]:
302 routes[key].append(speaker)
303 elif payload == "OFF" and speaker in routes[key]:
304 routes[key].remove(speaker)
305 save()
306 publish_state(key, speaker)
307
308
309def volume_command(t, payload):
310 speaker = t.split("/")[2]
311 if speaker not in speakers:
312 return
313 try:
314 volumes[speaker] = max(0, min(100, round(float(payload))))
315 except ValueError:
316 return
317 store(config["volumes"], volumes)
318 client.publish(volume_topic(speaker) + "/state", volumes[speaker], retain=True)
319
320
321def on_connect(client, userdata, flags, reason_code, properties):
322 client.subscribe("gg23/audio-route/+/+/set")
323 client.subscribe("gg23/audio-volume/+/set")
324 events.put(("connect",))
325
326
327def on_message(client, userdata, msg):
328 events.put(("command", msg.topic, msg.payload.decode()))
329
330
331client.on_connect = on_connect
332client.on_message = on_message
333client.connect_async(config["mqtt"]["host"])
334client.loop_start()
335pw_dump = subprocess.Popen(
336 ["pw-dump", "--monitor", "--no-colors"],
337 stdout=subprocess.PIPE,
338 text=True,
339)
340threading.Thread(target=read_graph, args=(pw_dump,), daemon=True).start()
341
342graph_ready = False
343
344
345def handle(event):
346 global graph_ready, published
347 if event[0] == "graph":
348 apply(event[1])
349 graph_ready = True
350 elif event[0] == "connect":
351 # republish everything, and clear the switches of sources that
352 # vanished while nobody was connected
353 published = dict.fromkeys(routes)
354 publish_volumes()
355 elif event[0] == "command":
356 if event[1].startswith("gg23/audio-volume/"):
357 volume_command(*event[1:])
358 else:
359 command(*event[1:])
360 elif event[0] == "exit":
361 sys.exit(f"pw-dump exited with {event[1]}")
362
363
364while True:
365 handle(events.get())
366 # sync once per burst, so links just created are seen before it
367 while not events.empty():
368 handle(events.get())
369 if graph_ready:
370 sync()