linux-audiomixer/main.py

import json
import mido
import subprocess

nanoKontrol2_mapping: dict[int, str] = {
    0: "Slider_1",
    1: "Slider_2",
    2: "Slider_3",
    3: "Slider_4",
    4: "Slider_5",
    5: "Slider_6",
    6: "Slider_7",
    7: "Slider_8",
    16: "Knob_1",
    17: "Knob_2",
    18: "Knob_3",
    19: "Knob_4",
    20: "Knob_5",
    21: "Knob_6",
    22: "Knob_7",
    23: "Knob_8",
    32: "Solo_1",
    33: "Solo_2",
    34: "Solo_3",
    35: "Solo_4",
    36: "Solo_5",
    37: "Solo_6",
    38: "Solo_7",
    39: "Solo_8",
    41: "Play",
    42: "Stop",
    43: "Prev",
    44: "Next",
    45: "Record",
    46: "Cycle",
    48: "Mute_1",
    49: "Mute_2",
    50: "Mute_3",
    51: "Mute_4",
    52: "Mute_5",
    53: "Mute_6",
    54: "Mute_7",
    55: "Mute_8",
    58: "Track_prev",
    59: "Track_next",
    60: "Marker_set",
    61: "Marker_prev",
    62: "Marker_next",
    64: "Record_1",
    65: "Record_2",
    66: "Record_3",
    67: "Record_4",
    68: "Record_5",
    69: "Record_6",
    70: "Record_7",
    71: "Record_8",
}

audiomixer_mapping = {
    "Slider_1": "mic1.output",
    "Solo_1": "mic1.output",
    "Mute_1": "mic1.output",
    "Record_1": "mic1.output",
    "Slider_2": "mic2.output",
    "Solo_2": "mic2.output",
    "Mute_2": "mic2.output",
    "Record_2": "mic2.output",
    "Slider_3": "system.output",
    "Solo_3": "system.output",
    "Mute_3": "system.output",
    "Record_3": "system.output",
    "Slider_4": "discord.output",
    "Solo_4": "discord.output",
    "Mute_4": "discord.output",
    "Record_4": "discord.output",
    "Slider_5": "music.output",
    "Solo_5": "music.output",
    "Mute_5": "music.output",
    "Record_5": "music.output",
    "Slider_6": "music.output",
    "Solo_6": "music.output",
    "Mute_6": "music.output",
    "Record_6": "music.output",
    "Slider_7": "music.output",
    "Solo_7": "music.output",
    "Mute_7": "music.output",
    "Record_7": "music.output",
    "Slider_8": "browser.output",
    "Solo_8": "browser.output",
    "Mute_8": "browser.output",
    "Record_8": "browser.output",
    "Knob_1": "alsa_output.usb-Focusrite_Scarlett_2i2_USB_Y8T4J6E095585D-00.Direct__Direct__sink",
    "Knob_2": "alsa_output.pci-0000_10_00.4.analog-stereo",
    "Knob_3": "obs_system.output",
    "Knob_4": "obs_discord.output",
    "Knob_5": "obs_music.output",
    "Knob_8": "obs_browser.output",
    "Prev": "XF86Previous",
    "Next": "XF86Next",
}


sliders = [
    "Slider_1",
    "Slider_2",
    "Slider_3",
    "Slider_4",
    "Slider_5",
    "Slider_6",
    "Slider_7",
    "Slider_8",
    "Knob_1",
    "Knob_2",
    "Knob_3",
    "Knob_4",
    "Knob_5",
    "Knob_6",
    "Knob_7",
    "Knob_8",
]

buttons = [
    "Solo_1",
    "Mute_1",
    "Record_1",
    "Solo_2",
    "Mute_2",
    "Record_2",
    "Solo_3",
    "Mute_3",
    "Record_3",
    "Solo_4",
    "Mute_4",
    "Record_4",
    "Solo_5",
    "Mute_5",
    "Record_5",
    "Solo_6",
    "Mute_6",
    "Record_6",
    "Solo_7",
    "Mute_7",
    "Record_7",
    "Solo_8",
    "Mute_8",
    "Record_8",
]


pw_nodes = []
pw_json = subprocess.run("pw-dump", capture_output=True, text=True)
pw_data = json.loads(pw_json.stdout)
for node in pw_data:
    if node["type"] == "PipeWire:Interface:Node":
        pw_nodes.append(node)


def find_id(device: str) -> str:
    device_id: str = ""
    for pw_node in pw_nodes:
        if pw_node["info"]["props"]["node.name"] == device:
            device_id = str(pw_node["id"])
        # else:
        #     raise NotImplementedError
    return device_id


def set_mute(toggle: bool, device: str) -> None:
    device_id = find_id(device)

    if toggle:
        subprocess.Popen(["/usr/bin/wpctl", "set-mute", device_id, "1"])
    else:
        subprocess.Popen(["/usr/bin/wpctl", "set-mute", device_id, "0"])


def set_volume(volume: int, device: str) -> None:
    device_id = find_id(device)

    if volume > 0:
        volume = round((volume / 127), 2)

    subprocess.Popen(["/usr/bin/wpctl", "set-volume", device_id, str(volume)])


def main():
    # TODO threading for MIDI values;

    with mido.open_input("nanoKONTROL2:nanoKONTROL2 _ CTRL 28:0") as mixer:
        for update in mixer:
            midi_input = nanoKontrol2_mapping[update.control]
            if midi_input == "Record":
                if update.value == 127:
                    set_mute(toggle=True, device=audiomixer_mapping["Knob_2"])
                else:
                    set_mute(toggle=False, device=audiomixer_mapping["Knob_2"])

            if midi_input in sliders:
                set_volume(
                    volume=update.value,
                    device=audiomixer_mapping[nanoKontrol2_mapping[update.control]],
                )

            if midi_input in buttons:
                if midi_input.startswith("Mute") and update.value == 127:
                    set_mute(toggle=True, device=audiomixer_mapping[midi_input])
                elif midi_input.startswith("Mute") and update.value == 0:
                    set_mute(toggle=False, device=audiomixer_mapping[midi_input])

    # TODO parsing return to dict to split inputs
    # TODO define mixers in config files
    # TODO pipewire integration
    # TODO user config for audio streams
    # TODO running as a daemon
    # TODO systemctl unit file
    # TODO evaluating update time of script. 1ms? 0.1ms?


if __name__ == "__main__":
    main()