Simple python script to switch pulseaudio streams between available sinks

sink-switcher.py

#!/usr/bin/python3

from pulsectl import Pulse, pulsectl
from collections import Counter
import logging
from systemd import journal

log = logging.getLogger("sink-switcher")
log.setLevel(logging.DEBUG)

#logging to stdout
cl = logging.StreamHandler()
cl.setFormatter(logging.Formatter('[%(levelname)s]: %(message)s'))
log.addHandler(cl)

#logging to journald
#filter via journalctl --user -t sink-switcher
jl = journal.JournalHandler(logging.WARNING, SYSLOG_IDENTIFIER=log.name)
log.addHandler(jl)

switch_between_names = ["alsa_output.pci-0000_00_1b.0.analog-stereo",
                        "bluez_sink.AC_FD_93_37_95_A2"]

switch_between_indexes = []

with Pulse("sink-switcher") as pulse:

    existing_sinks = pulse.sink_list()

    #get the indexes of sinks defined in switch_between_names
    for target_sink_name in switch_between_names:
        if target_sink_name in map(lambda x: x.name, existing_sinks):
            switch_between_indexes.append(list(filter(lambda x: x.name == target_sink_name, existing_sinks))[0].index)
        else:
            log.warn("sink %s does not exist.", target_sink_name)

    if len(switch_between_indexes) < 2:
        log.error("less than 2 sinks to switch between, aborting")
        exit(1)
    else:
        log.info("sinks to switch between: %s", str(switch_between_indexes))

    #get the indexes of sink inputs currently connected to sinks defined in switch_between_indexes
    sink_inputs = list(filter(lambda x: x.sink in switch_between_indexes, pulse.sink_input_list()))
    log.info("found %d sink inputs currently mapped to the selected sinks", len(sink_inputs))

    #get the index of the sink that has the majority of inputs connected to it
    dominant_sink = Counter(map(lambda x: x.sink, sink_inputs)).most_common(1)[0][0]
    #get the index of the next sink in the list
    next_sink = switch_between_indexes[(switch_between_indexes.index(dominant_sink) + 1) % len(switch_between_indexes)]
    log.info("dominant sink: %d, next sink: %d", dominant_sink, next_sink)

    #try to move all inputs to the next sink
    for sink_input in sink_inputs:
        try:
            log.info("moving input %d to sink %d", sink_input.index, next_sink)
            pulse.sink_input_move(sink_input.index, next_sink)
        except pulsectl.PulseOperationFailed:
            log.warn("unable to move sink input %d to sink %d", sink_input.index, next_sink)
            pass

Thank you, that just improved the quality of my life :)