auracaster/bumble-auracast
3
0
Fork 0
Code Issues Pull Requests 2 Actions Packages Projects Releases Wiki Activity

refactor: add device cache refresh and rename PipeWire input listing functions

Browse Source
This commit is contained in:
pstruebi
2025-10-06 13:00:09 +02:00
parent 430fb1009d
commit 37984b028b
1 changed files with 13 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
src/auracast/multicast_script.py
View File

@@ -44,7 +44,11 @@ import time
from dotenv import load_dotenv from dotenv import load_dotenv
from auracast import multicast from auracast import multicast
from auracast import auracast_config from auracast import auracast_config
from auracast.utils.sounddevice_utils import list_usb_pw_inputs, list_network_pw_inputs from auracast.utils.sounddevice_utils import (
get_usb_pw_inputs,
get_network_pw_inputs,
refresh_pw_cache,
)
if __name__ == "__main__": if __name__ == "__main__":
@@ -59,12 +63,14 @@ if __name__ == "__main__":
os.environ.setdefault("PULSE_LATENCY_MSEC", "3") os.environ.setdefault("PULSE_LATENCY_MSEC", "3")
usb_inputs = list_usb_pw_inputs() # Refresh device cache and list inputs
refresh_pw_cache()
usb_inputs = get_usb_pw_inputs()
logging.info("USB pw inputs:") logging.info("USB pw inputs:")
for i, d in usb_inputs: for i, d in usb_inputs:
logging.info(f"{i}: {d['name']} in={d['max_input_channels']}") logging.info(f"{i}: {d['name']} in={d['max_input_channels']}")
aes67_inputs = list_network_pw_inputs() aes67_inputs = get_network_pw_inputs()
logging.info("AES67 pw inputs:") logging.info("AES67 pw inputs:")
for i, d in aes67_inputs: for i, d in aes67_inputs:
logging.info(f"{i}: {d['name']} in={d['max_input_channels']}") logging.info(f"{i}: {d['name']} in={d['max_input_channels']}")
@@ -85,7 +91,8 @@ if __name__ == "__main__":
if iface_substr: if iface_substr:
# Loop until a matching AES67 input becomes available # Loop until a matching AES67 input becomes available
while True: while True:
current = list_network_pw_inputs() refresh_pw_cache()
current = get_network_pw_inputs()
sel = next(((i, d) for i, d in current if iface_substr in (d.get('name','').lower())), None) sel = next(((i, d) for i, d in current if iface_substr in (d.get('name','').lower())), None)
if sel: if sel:
input_sel = sel[0] input_sel = sel[0]
@@ -100,7 +107,8 @@ if __name__ == "__main__":
else: else:
# Loop until a USB input becomes available (mirror AES67 retry behavior) # Loop until a USB input becomes available (mirror AES67 retry behavior)
while True: while True:
current = list_usb_pw_inputs() refresh_pw_cache()
current = get_usb_pw_inputs()
if current: if current:
input_sel, selected_dev = current[0] input_sel, selected_dev = current[0]
logging.info(f"Selected first USB input: index={input_sel}, device={selected_dev['name']}") logging.info(f"Selected first USB input: index={input_sel}, device={selected_dev['name']}")