83 lines
2.4 KiB
Python
Executable File
83 lines
2.4 KiB
Python
Executable File
#!/usr/bin/python
|
|
import argparse
|
|
import subprocess
|
|
import time
|
|
|
|
mappingMic = {
|
|
# "output\\.rnnoise_source:capture_1": ["mixed-mic:input_FL"],
|
|
# "output\\.rnnoise_source:capture_2": ["mixed-mic:input_FR"]
|
|
"rnnoise_source:capture_1": ["mixed-mic:input_FL", "mixed-mic:input_FR"],
|
|
}
|
|
|
|
mappingLoopback = {
|
|
"loopback:monitor_FL": ["mixed-mic:input_FL"],
|
|
"loopback:monitor_FR": ["mixed-mic:input_FR"]
|
|
}
|
|
|
|
|
|
parser = argparse.ArgumentParser(
|
|
prog="link-virtual-mic",
|
|
description="link pipewire virtual mics.")
|
|
|
|
parser.add_argument("-a", "--all", action="store_true")
|
|
parser.add_argument("-d", "--disconnect", action="store_true")
|
|
|
|
|
|
def main():
|
|
args = parser.parse_args()
|
|
|
|
connectAll = args.all and not args.disconnect
|
|
disconnectAll = args.all and args.disconnect
|
|
|
|
disconnectMic = args.all and args.disconnect
|
|
disconnectLoopback = args.disconnect
|
|
|
|
tries = -1
|
|
while tries < 20:
|
|
tries += 1
|
|
|
|
micSuccess = True
|
|
loopbackSuccess = True
|
|
if disconnectAll or not args.disconnect:
|
|
micSuccess = manageLinks(mappingMic, disconnectMic)
|
|
if connectAll or disconnectLoopback:
|
|
loopbackSuccess = manageLinks(mappingLoopback, disconnectLoopback)
|
|
|
|
if not micSuccess or not loopbackSuccess:
|
|
time.sleep(1)
|
|
continue
|
|
break
|
|
|
|
|
|
def manageLinks(mapping, disconnect=False):
|
|
for iter_out, iter_inputs in mapping.items():
|
|
cmdOpts = {"shell": True, "capture_output": True, "text": True}
|
|
|
|
cmd = "pw-link -o | grep -e '{}'".format(iter_out)
|
|
completedProcess = subprocess.run(cmd, **cmdOpts)
|
|
if completedProcess.returncode == 1:
|
|
print(f"Output `{iter_out}` not found")
|
|
return False
|
|
output = completedProcess.stdout.split('\n')[0]
|
|
|
|
for iter_input in iter_inputs:
|
|
cmd = "pw-link -i | grep -e '{}'".format(iter_input)
|
|
completedProcess = subprocess.run(cmd, **cmdOpts)
|
|
if completedProcess.returncode == 1:
|
|
print(f"Output `{iter_input}` not found")
|
|
return False
|
|
input = completedProcess.stdout.split('\n')[0]
|
|
|
|
if disconnect:
|
|
cmd = ["pw-link", "-d", output, input]
|
|
else:
|
|
cmd = ["pw-link", output, input]
|
|
print(' '.join(cmd))
|
|
subprocess.run(cmd, capture_output=True)
|
|
|
|
return True
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|