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