audioplayer.py

from __future__ import annotations
import subprocess as sp
from multiprocessing import Process

COMMAND = "mpg123 -R"

# Commands are only characters in the ASCII set.
STDIN_ENCODING = "ascii"

END_PLAYBACK_SENTINEL = b"@P 0"


def clamp(n: int, min_n: int, max_n: int) -> int:
    return min(max_n, max(min_n, n))


class AudioPlayer:
    def __init__(self):
        """An audio player wrapper over mpg123 that allows for control of playback."""

        self._player_process: sp.Popen = sp.Popen(COMMAND.split(), stdin=sp.PIPE, stdout=sp.PIPE, stderr=sp.DEVNULL)

        # mpg123 does absolute volume setting. To avoid needing to do a lookup to set volume, we're maintaining
        #  and internal volume level. This value is relative to and independent of the system volume.
        self._volume = 100
        # To ensure that the real and "cached" volumes are in sync.
        self.set_volume(self._volume)

    def _send_command(self, command: str) -> None:
        # Apparently, writing directly to the STDIN can cause dead-locks according to the Python documentation.
        # The alternatives though (.communicate) don't work when sending commands without wanting to wait for the
        #  process to terminate.
        self._player_process.stdin.write(command.encode(STDIN_ENCODING) + b"\n")
        self._player_process.stdin.flush()

    def play(self, song_path: str, block_until_finished: bool = True) -> None:
        """Starts playing the given song."""
        self._send_command(f"load {song_path}")

        if block_until_finished:
            while True:
                line = self._player_process.stdout.readline()
                if line.startswith(END_PLAYBACK_SENTINEL) or not line:
                    break

    def stop(self) -> None:
        self._send_command("stop")

    def toggle_pause(self) -> None:
        self._send_command("pause")

    def set_volume(self, new_volume: int) -> None:
        self._volume = clamp(new_volume, 0, 100)
        self._send_command(f"volume {self._volume}")

    def adjust_volume(self, adjust_amount: int) -> None:
        """The provided adjustment should be an integer between -100 and 100 indicating how much to adjust the
        volume by. Negative numbers decrease volume, positive numbers increase."""
        self.set_volume(self._volume + adjust_amount)

    def terminate(self):
        """Should be called either directly or via a context manager to terminate the player process."""
        self._player_process.terminate()

    def __enter__(self) -> AudioPlayer:
        return self

    def __exit__(self, *_) -> None:
        self.terminate()