186 lines
4.8 KiB
Python
186 lines
4.8 KiB
Python
import threading
|
|
import time
|
|
from enum import Enum
|
|
|
|
import pygame
|
|
from pydantic import BaseModel
|
|
|
|
|
|
class Track(BaseModel):
|
|
artist: str
|
|
title: str
|
|
duration: int
|
|
filepath: str
|
|
|
|
|
|
class PlaybackState(Enum):
|
|
Playing = "Playing"
|
|
Paused = "Paused"
|
|
Stopped = "Stopped"
|
|
|
|
|
|
class PlayerState(BaseModel):
|
|
playback_state: PlaybackState = PlaybackState.Stopped
|
|
track: Track | None = None
|
|
volume: float
|
|
|
|
|
|
class Queue:
|
|
queue: list[Track]
|
|
|
|
def __init__(self) -> None:
|
|
self._queue: list[Track] = []
|
|
|
|
def next(self) -> Track | None:
|
|
if len(self._queue) > 0:
|
|
return self._queue.pop(0)
|
|
else:
|
|
return None
|
|
|
|
def add(self, track: Track) -> None:
|
|
self._queue.append(track)
|
|
|
|
def len(self) -> int:
|
|
return len(self._queue)
|
|
|
|
|
|
class MusicPlayer:
|
|
def __init__(self):
|
|
# Player State
|
|
self._state = PlayerState(volume=1)
|
|
self._state_changed: bool = False
|
|
|
|
# Sound initialization
|
|
pygame.init()
|
|
pygame.mixer.init()
|
|
pygame.mixer.music.set_endevent(pygame.USEREVENT)
|
|
pygame.mixer.music.set_volume(self._state.volume)
|
|
|
|
# Threading
|
|
self.lock = threading.Lock()
|
|
self._running = True
|
|
self._thread = threading.Thread(target=self._loop, daemon=True)
|
|
self._thread.start()
|
|
|
|
# Queue
|
|
self._queue: Queue = Queue()
|
|
self._queue_changed: bool = False
|
|
|
|
def _loop(self):
|
|
while self._running:
|
|
for event in pygame.event.get():
|
|
if event.type == pygame.USEREVENT: # a song just ended
|
|
self._handle_track_finished()
|
|
self._play_next_track()
|
|
|
|
time.sleep(0.1)
|
|
|
|
def _set_playback_state(self, value: PlaybackState):
|
|
self._state.playback_state = value
|
|
self._state_changed = True
|
|
|
|
def _set_track(self, track: Track | None):
|
|
self._state.track = track
|
|
self._state_changed = True
|
|
|
|
def _handle_track_finished(self) -> None:
|
|
print(f"Finished playing {self._state.track}")
|
|
self._set_track(None)
|
|
self._set_playback_state(PlaybackState.Stopped)
|
|
|
|
def _load_track(self, track: Track):
|
|
self._set_track(track)
|
|
pygame.mixer.music.unload()
|
|
pygame.mixer.music.load(track.filepath)
|
|
|
|
def _queue_add(self, track):
|
|
self._queue.add(track)
|
|
self._queue_changed = True
|
|
|
|
def _queue_next(self) -> Track | None:
|
|
next_track = self._queue.next()
|
|
self._queue_changed = True
|
|
return next_track
|
|
|
|
def _start_playback(self):
|
|
pygame.mixer.music.play()
|
|
self._set_playback_state(PlaybackState.Playing)
|
|
|
|
def _pause_playback(self):
|
|
pygame.mixer.music.pause()
|
|
self._set_playback_state(PlaybackState.Paused)
|
|
|
|
def _resume_playback(self):
|
|
pygame.mixer.music.unpause()
|
|
self._set_playback_state(PlaybackState.Playing)
|
|
|
|
def _stop_playback(self):
|
|
pygame.mixer.music.stop()
|
|
self._set_playback_state(PlaybackState.Stopped)
|
|
self._set_track(None)
|
|
|
|
def _play_next_track(self):
|
|
next_track = self._queue_next()
|
|
if next_track:
|
|
self._load_track(next_track)
|
|
self._start_playback()
|
|
|
|
def add_to_queue(self, track: Track):
|
|
with self.lock:
|
|
que_len = self._queue.len()
|
|
self._queue_add(track)
|
|
# If queue is empty and no corrent track, start playing
|
|
if que_len == 0 and not self._state.track:
|
|
self._play_next_track()
|
|
|
|
def play(self):
|
|
with self.lock:
|
|
if self._state.playback_state == PlaybackState.Playing:
|
|
return
|
|
if self._state.track:
|
|
self._start_playback()
|
|
else:
|
|
self._play_next_track()
|
|
|
|
def pause(self):
|
|
with self.lock:
|
|
self._pause_playback()
|
|
|
|
def resume(self):
|
|
with self.lock:
|
|
self._resume_playback()
|
|
|
|
def stop(self):
|
|
with self.lock:
|
|
self._stop_playback()
|
|
|
|
def next(self):
|
|
with self.lock:
|
|
self._play_next_track()
|
|
|
|
def shutdown(self):
|
|
with self.lock:
|
|
self._running = False
|
|
self._thread.join()
|
|
pygame.mixer.quit()
|
|
|
|
def get_queue(self) -> list[Track]:
|
|
return self._queue._queue
|
|
|
|
def _set_volume(self, volume: float):
|
|
self._state.volume = volume
|
|
pygame.mixer.music.set_volume(volume)
|
|
self._state_changed = True
|
|
|
|
def set_volume(self, volume: float) -> None:
|
|
with self.lock:
|
|
self._set_volume(volume)
|
|
|
|
def get_volume(self) -> float:
|
|
return self._state.volume
|
|
|
|
def get_state(self) -> PlayerState:
|
|
return self._state
|
|
|
|
def get_current_track(self) -> Track | None:
|
|
return self._state.track
|