nova-chatmix.py

#!/usr/bin/python3

# Licensed under the 0BSD

from signal import SIGINT, SIGTERM, signal
from subprocess import Popen, check_output

from hid import device
from hid import enumerate as hidenumerate

CMD_PACTL = "pactl"
CMD_PWLOOPBACK = "pw-loopback"


class ChatMix:
    # Create virtual pipewire sinks
    def __init__(self, output_sink: str, main_sink: str, chat_sink: str):
        self.main_sink = main_sink
        self.chat_sink = chat_sink
        self.main_sink_process = self._create_virtual_sink(main_sink, output_sink)
        self.chat_sink_process = self._create_virtual_sink(chat_sink, output_sink)

    def set_main_volume(self, volume: int):
        self._set_volume(self.main_sink, volume)

    def set_chat_volume(self, volume: int):
        self._set_volume(self.chat_sink, volume)

    def set_volumes(self, main_volume: int, chat_volume: int):
        self.set_main_volume(main_volume)
        self.set_chat_volume(chat_volume)

    def close(self):
        self.main_sink_process.terminate()
        self.chat_sink_process.terminate()

    def _create_virtual_sink(self, name: str, output_sink: str) -> Popen:
        return Popen(
            [
                CMD_PWLOOPBACK,
                "-P",
                output_sink,
                "--capture-props=media.class=Audio/Sink",
                "-n",
                name,
            ]
        )

    def _set_volume(self, sink: str, volume: int):
        Popen([CMD_PACTL, "set-sink-volume", f"input.{sink}", f"{volume}%"])


class NovaProWireless:
    # USB IDs
    VID = 0x1038
    PID = 0x12E0

    # bInterfaceNumber
    INTERFACE = 0x4

    # HID Message length
    MSGLEN = 63

    # Message read timeout
    READ_TIMEOUT = 1000

    # First byte controls data direction.
    TX = 0x6  # To base station.
    RX = 0x7  # From base station.

    # Second Byte
    # This is a very limited list of options, you can control way more. I just haven't implemented those options (yet)
    ## As far as I know, this only controls the icon.
    OPT_SONAR_ICON = 0x8D
    ## Enabling this option enables the ability to switch between volume and ChatMix.
    OPT_CHATMIX_ENABLE = 0x49
    ## Volume controls, 1 byte
    OPT_VOLUME = 0x25
    ## ChatMix controls, 2 bytes show and control game and chat volume.
    OPT_CHATMIX = 0x45
    ## EQ controls, 2 bytes show and control which band and what value.
    OPT_EQ = 0x31
    ## EQ preset controls, 1 byte sets and shows enabled preset. Preset 4 is the custom preset required for OPT_EQ.
    OPT_EQ_PRESET = 0x2E

    # PipeWire Names
    ## String used to automatically select output sink
    PW_OUTPUT_SINK_AUTODETECT = "SteelSeries_Arctis_Nova_Pro_Wireless"
    ## Names of virtual sound devices
    PW_GAME_SINK = "NovaGame"
    PW_CHAT_SINK = "NovaChat"

    # Keeps track of enabled features for when close() is called
    CHATMIX_CONTROLS_ENABLED = False
    SONAR_ICON_ENABLED = False

    # Stops processes when program exits
    CLOSE = False

    # Device not found error string
    ERR_NOTFOUND = "Device not found"

    # Selects correct device, and makes sure we can control it
    def __init__(self, output_sink=None):
        # Find HID device path
        devpath = None
        for hiddev in hidenumerate(self.VID, self.PID):
            if hiddev["interface_number"] == self.INTERFACE:
                devpath = hiddev["path"]
                break
        if not devpath:
            raise DeviceNotFoundException

        # Try to automatically detect output sink, this is skipped if output_sink is given
        if not output_sink:
            sinks = (
                check_output([CMD_PACTL, "list", "sinks", "short"]).decode().split("\n")
            )
            for sink in sinks[:-1]:
                sink_name = sink.split("\t")[1]
                if self.PW_OUTPUT_SINK_AUTODETECT in sink_name:
                    output_sink = sink_name

        self.dev = device()
        self.dev.open_path(devpath)
        self.dev.set_nonblocking(True)
        self.output_sink = output_sink

    # Enables/Disables chatmix controls
    def set_chatmix_controls(self, state: bool):
        assert self.dev, self.ERR_NOTFOUND
        self.dev.write(
            self._create_msgdata((self.TX, self.OPT_CHATMIX_ENABLE, int(state))),
        )
        self.CHATMIX_CONTROLS_ENABLED = state

    # Enables/Disables Sonar Icon
    def set_sonar_icon(self, state: bool):
        assert self.dev, self.ERR_NOTFOUND
        self.dev.write(
            self._create_msgdata((self.TX, self.OPT_SONAR_ICON, int(state))),
        )
        self.SONAR_ICON_ENABLED = state

    # Sets Volume
    def set_volume(self, attenuation: int):
        assert self.dev, self.ERR_NOTFOUND
        self.dev.write(
            self._create_msgdata((self.TX, self.OPT_VOLUME, attenuation)),
        )

    # Sets EQ preset
    def set_eq_preset(self, preset: int):
        assert self.dev, self.ERR_NOTFOUND
        self.dev.write(
            self._create_msgdata((self.TX, self.OPT_EQ_PRESET, preset)),
        )

    # ChatMix implementation
    # Continuously read from base station and ignore everything but ChatMix messages (OPT_CHATMIX)
    def chatmix_volume_control(self, chatmix: ChatMix):
        assert self.dev, self.ERR_NOTFOUND
        while not self.CLOSE:
            try:
                msg = self.dev.read(self.MSGLEN, self.READ_TIMEOUT)
                if not msg or msg[1] is not self.OPT_CHATMIX:
                    continue

                # 4th and 5th byte contain ChatMix data
                gamevol = msg[2]
                chatvol = msg[3]

                # Actually change volume. Everytime you turn the dial, both volumes are set to the correct level
                chatmix.set_volumes(gamevol, chatvol)
            except OSError:
                print("Device was probably disconnected, exiting.")
                self.CLOSE = True
        # Remove virtual sinks on exit
        chatmix.close()

    # Prints output from base station. `debug` argument enables raw output.
    def print_output(self, debug: bool = False):
        assert self.dev
        while not self.CLOSE:
            msg = self.dev.read(self.MSGLEN, self.READ_TIMEOUT)
            if debug:
                print(msg)
            match msg[1]:
                case self.OPT_VOLUME:
                    print(f"Volume: -{msg[2]}")
                case self.OPT_CHATMIX:
                    print(f"Game Volume: {msg[2]} - Chat Volume: {msg[3]}")
                case self.OPT_EQ:
                    print(f"EQ: Bar: {msg[2]} - Value: {(msg[3] - 20) / 2}")
                case self.OPT_EQ_PRESET:
                    print(f"EQ Preset: {msg[2]}")
                case _:
                    print("Unknown Message")

    # Terminates processes and disables features
    def close(self, signum, frame):
        self.CLOSE = True
        if self.CHATMIX_CONTROLS_ENABLED:
            self.set_chatmix_controls(False)
        if self.SONAR_ICON_ENABLED:
            self.set_sonar_icon(False)

    # Takes a tuple of ints and turns it into bytes with the correct length padded with zeroes
    def _create_msgdata(self, data: tuple[int, ...]) -> bytes:
        return bytes(data).ljust(self.MSGLEN, b"\0")


class DeviceNotFoundException(Exception):
    pass


# When run directly, just start the ChatMix implementation. (And activate the icon, just for fun)
if __name__ == "__main__":
    try:
        nova = NovaProWireless()
        nova.set_sonar_icon(state=True)
        nova.set_chatmix_controls(state=True)

        signal(SIGINT, nova.close)
        signal(SIGTERM, nova.close)

        assert nova.output_sink, "Output sink not set"
        chatmix = ChatMix(
            output_sink=nova.output_sink,
            main_sink=nova.PW_GAME_SINK,
            chat_sink=nova.PW_CHAT_SINK,
        )

        nova.chatmix_volume_control(chatmix=chatmix)
    except DeviceNotFoundException:
        print("Device not found, exiting.")