|
#!/usr/bin/python |
|
|
|
# Change PulseAudio's default-sink + move all sink-inputs |
|
# Target sink is selected by loosely matching against sinks' `active-port` field |
|
# See `pacmd list-sinks` for `active-port` options |
|
# |
|
# For proper function: |
|
# in: `/etc/pulse/default.pa` |
|
# at the end of the line: `load-module module-stream-restore` |
|
# add: ` restore_device=false` |
|
# then restart PulseAudio (or reboot) |
|
# |
|
# Usage: pa_change_sink <~sink active port> |
|
# Examples: pa_change_sink headphone |
|
# pa_change_sink hdmi |
|
|
|
|
|
from syslog import syslog |
|
import subprocess |
|
import sys |
|
import re |
|
|
|
|
|
# Utility functions for consuming `pacmd list-X` output |
|
def pacmd(*args, **kwargs): |
|
if kwargs.get('log'): syslog('Running pacmd %s' % ' '.join(map(str, list(args)))) |
|
result = subprocess.check_output(['pacmd'] + list(args)) |
|
if kwargs.get('log') and result: syslog('Got back %s' % result) |
|
return result |
|
|
|
def all_group_dicts(pattern, string, flags): |
|
return [match.groupdict() for match in re.finditer(pattern, string, flags)] |
|
|
|
def get_sinks(): |
|
return all_group_dicts(r'(?P<default>\*?) ' |
|
'index: (?P<index>\d+).+?' |
|
'active port: <(?P<active_port>.+?)>', |
|
pacmd('list-sinks'), |
|
re.DOTALL) |
|
|
|
def get_inputs(): |
|
return all_group_dicts(r'index: (?P<index>\d+).+?' |
|
'sink: (?P<sink_index>\d+) <.+?' |
|
'application.process.binary = "(?P<application>.+?)"', |
|
pacmd('list-sink-inputs'), |
|
re.DOTALL) |
|
|
|
|
|
# Script-specific utility functions |
|
def find_default_sink(sinks): |
|
return next(sink for sink in sinks if sink['default'] == '*') |
|
|
|
def find_sink_for_input(sinks, input): |
|
return next(sink for sink in sinks if sink['index'] == input['sink_index']) |
|
|
|
def refresh_input(input): |
|
return next(_input for _input in get_inputs() if _input['index'] == input['index']) |
|
|
|
|
|
# Script |
|
def change_sink(target): |
|
syslog('Changing PulseAudio sink to %s' % target) |
|
|
|
sinks = get_sinks() |
|
inputs = get_inputs() |
|
target_sink = next(sink for sink in sinks if target in sink['active_port']) |
|
|
|
pacmd('set-default-sink', target_sink['index'], log=True) |
|
syslog('Changed default sink from %s to %s' % ( |
|
find_default_sink(sinks)['active_port'], |
|
find_default_sink(get_sinks())['active_port'])) |
|
|
|
for input in inputs: |
|
pacmd('move-sink-input', input['index'], target_sink['index'], log=True) |
|
syslog('Changed sink for %s from %s to %s'% ( |
|
input['application'], |
|
find_sink_for_input(sinks, input)['active_port'], |
|
find_sink_for_input(sinks, refresh_input(input))['active_port'])) |
|
|
|
# Ugly but simple way to notify the user within the most appropriate UI |
|
subprocess.check_output(['xdotool', 'key', 'XF86AudioMute']) |
|
subprocess.check_output(['xdotool', 'key', 'XF86AudioMute']) |
|
|
|
|
|
if __name__ == '__main__': |
|
try: |
|
change_sink(sys.argv[1]) |
|
except Exception as e: |
|
print('ERROR %s: %s' % (type(e), e)) |
|
exit(1) |