Cycles PulseAudio audio output for all applications.

cycle-pulse-sink.py

#!/bin/env python
"""
Cycles PulseAudio audio output for all applications.
"""
import subprocess

SPEAKERS = b'alsa_output.usb-Micronas_Micronas_USB_Codec-00.analog-stereo'
HEADPHONES = b'alsa_output.pci-0000_00_1b.0.analog-stereo'
TV = b'alsa_output.pci-0000_00_03.0.hdmi-stereo-extra1'

SINK_NAME_TO_FRIENDLY_NAME = {
    SPEAKERS: 'SPEAKERS',
    HEADPHONES: 'HEADPHONES',
    TV: 'TV'
}

# The only audio outputs on my system that I use.
USED_SINKS = (SPEAKERS, HEADPHONES, TV)
# Cycling will occur only between these.
CYCLED_SINKS = (SPEAKERS, HEADPHONES)

def shell_exec(cmd_str):
    result = subprocess.run(cmd_str, stdout=subprocess.PIPE)
    return result.stdout

def notify_send(message):
    shell_exec(['notify-send', '-t', '2000', ' ', f"<span font='48px'><b>{message}</b></span>"])

def get_sink_to_id():
    sink_to_id = dict()
    list_output = shell_exec('pactl list short sinks'.split())
    for sink_line in list_output.split(b'\n'):
        if sink_line:
            split_line = sink_line.split(b'\t')
            sink_id = int(split_line[0])
            sink_name = split_line[1]
            sink_to_id[sink_name] = sink_id

            sink_status = split_line[-1]
    return sink_to_id

def get_next_sink_in_cycle(active_sink):
    available_sinks = list(sink_to_id.keys())
    candidate_sinks = [sink for sink in available_sinks if sink in CYCLED_SINKS]
    active_sink_index = candidate_sinks.index(active_sink)
    candidate_sinks.remove(active_sink)

    # No other sinks available, script will quit.
    if not candidate_sinks:
        return None
    else:
        # This index now points to the next sink after the
        # active one.
        try:
            next_sink = candidate_sinks[active_sink_index]
        # Wrap around the list.
        except IndexError:
            next_sink = candidate_sinks[0]
        return next_sink

def move_sink_inputs_to_sink_id(new_sink_id):
    cmd_output = shell_exec("pactl list short sink-inputs".split())
    for sink_input_line in cmd_output.split(b'\n'):
        if sink_input_line:
            split = sink_input_line.split(b'\t')
            sink_id = int(split[1])
            # If I an application to play on the TV, it's unlikely I want to
            # play it on another output so leave it be.
            if sink_id != sink_to_id[TV]:
                sink_input_id = split[0]
                shell_exec(["pactl", "move-sink-input", sink_input_id, str(new_sink_id)])

sink_to_id = get_sink_to_id()
active_sink = shell_exec('pactl get-default-sink'.split()).strip()
new_sink = get_next_sink_in_cycle(active_sink)
if not new_sink:
    notify_send('Nothing to cycle')
    print("Do nothing: only 1 sink for cycling is present")
    sys.exit(0)

move_sink_inputs_to_sink_id(sink_to_id[new_sink])
shell_exec(["pactl", "set-default-sink", str(sink_to_id[new_sink])])
notify_send('Audio output: {}'.format(SINK_NAME_TO_FRIENDLY_NAME[new_sink]))