python audio player wrapper around ffplay (provided by ffmpeg)

ffplayer.py

import os
import signal
import subprocess
import threading
import time
from typing import Optional, Callable

Callback = Callable[[], None]  # for typing


class FFPlayAudioPlayer:
    """
    A simple media player class that uses `ffplay` (part of ffmpeg) to play audio or video.

    Requirements:
        - ffmpeg must be installed on your system and accessible in the PATH.
        - `ffplay` command should be available.

    Methods:
        - play(media_path: str): Starts playing the media from the specified path.
        - pause(): Pauses the media.
        - resume(): Resumes the media if it was paused.
        - stop(): Stops the media playback.
        - is_media_playing() -> bool: Checks if the media is currently playing.
        - playback_time -> int: Returns the current playback time in milliseconds.
    """

    def __init__(self,
                 on_track_start: Optional[Callback] = None,
                 on_track_end: Optional[Callback] = None,
                 on_pause: Optional[Callback] = None,
                 on_resume: Optional[Callback] = None) -> None:
        """
        Initializes the FFPlayMediaPlayer with default values.
        """
        self.process: Optional[subprocess.Popen] = None
        self._start_ts: float = 0.0
        self._playback_time_accumulator: float = 0.0
        self.is_playing: threading.Event = threading.Event()
        self.end_of_media: threading.Event = threading.Event()
        self.media_path: Optional[str] = None
        self.on_track_start = on_track_start
        self.on_track_end = on_track_end
        self.on_pause = on_pause
        self.on_resume = on_resume

    @property
    def playback_time(self) -> int:
        """
        Returns the current playback time in milliseconds. If the media is playing,
        the time elapsed since the last start is added to the accumulated playback time.

        Returns:
            int: The current playback time in milliseconds.
        """
        delta = 0
        if self._start_ts and self.is_playing.is_set():
            delta = time.time() - self._start_ts
        return int((self._playback_time_accumulator + delta) * 1000)

    @property
    def track_length(self) -> int:
        """
        getting the duration of the audio in milliseconds
        """
        return self.playback_time + 1  # TODO - parse stdout from ffplay

    def play(self, media_path: str) -> None:
        """
        Starts playing the media from the specified file path or URL.

        Args:
            media_path (str): The path or URL to the media file to play.
        """
        if self.process and self.process.poll() is None:
            print("Media is already playing.")
            return

        self.media_path = media_path
        # Redirect stdout and stderr to /dev/null to avoid buffer overflow
        self.process = subprocess.Popen(
            ['ffplay', '-nodisp', '-autoexit', '-vn', media_path],
            stdout=open(os.devnull, 'w'),  # Redirect stdout to /dev/null
            stderr=open(os.devnull, 'w')  # Redirect stderr to /dev/null
        )
        self.is_playing.set()
        self._start_ts = time.time()
        self._playback_time_accumulator = 0
        print(f"Playing: {media_path}")
        if self.on_track_start:
            try:
                self.on_track_start()
            except Exception as e:
                print(f"Error in on_track_start callback: {e}")

        # Monitor the process to detect when it finishes
        self._monitor_playback()

    def _monitor_playback(self) -> None:
        """
        Monitors the playback and calls the end-of-track callback when the process finishes.
        """
        self.end_of_media.clear()

        def check_process():
            while self.process.poll() is None:  # Check if process is still running
                time.sleep(1)  # Sleep a bit before checking again
            print("End of track")
            self.is_playing.clear()
            self.end_of_media.set()
            # Once process ends, call the on_track_end callback
            if self.on_track_end:
                try:
                    self.on_track_end()
                except Exception as e:
                    print(f"Error in on_track_end callback: {e}")
            self.process = None

        # Start a background thread to monitor the playback process
        monitor_thread = threading.Thread(target=check_process, daemon=True)
        monitor_thread.start()

    def wait_for_playback(self):
        self.end_of_media.wait()

    def pause(self) -> None:
        """
        Pauses the media by sending the SIGSTOP signal to the process.
        """
        if self.is_playing.is_set():
            self.process.send_signal(signal.SIGSTOP)  # Send SIGSTOP signal to pause the process
            self.is_playing.clear()
            self._playback_time_accumulator += time.time() - self._start_ts
            self._start_ts = 0
            print("Paused.")
            if self.on_pause:
                try:
                    self.on_pause()
                except Exception as e:
                    print(f"Error in on_pause callback: {e}")

    def resume(self) -> None:
        """
        Resumes the media by sending the SIGCONT signal to the process.
        """
        if not self.is_playing.is_set():
            self.process.send_signal(signal.SIGCONT)  # Send SIGCONT signal to resume the process
            self.is_playing.set()
            self._start_ts = time.time()
            print("Resumed.")
            if self.on_resume:
                try:
                    self.on_resume()
                except Exception as e:
                    print(f"Error in on_resume callback: {e}")

    def stop(self) -> None:
        """
        Stops the media playback and kills the `ffplay` process.
        """
        if self.process:
            self.process.kill()
            self.is_playing.clear()
            self.end_of_media.set()
            self._playback_time_accumulator = self._start_ts = 0
            self.process = None
            print("Stopped.")
            if self.on_track_end:
                try:
                    self.on_track_end()
                except Exception as e:
                    print(f"Error in on_track_end callback: {e}")


if __name__ == "__main__":
    # Example usage: playing an MP3 file from a URL
    url = "https://github.com/OpenVoiceOS/ovos-skill-easter-eggs/raw/refs/heads/dev/sounds/sing/drnimpo-robots.mp3"
    jfk = "/home/miro/PycharmProjects/OVOS/STT/ovos-stt-plugin-fasterwhisper/jfk.wav"

    # Create player instance
    player = FFPlayAudioPlayer()

    # Play media
    player.play(url)
    print(f"Playback time: {player.playback_time} ms")
    time.sleep(5)
    print(f"Playback time: {player.playback_time} ms")

    # Pause media
    player.pause()
    time.sleep(2)
    print(f"Playback time: {player.playback_time} ms")

    # Resume media
    player.resume()
    player.wait_for_playback()