126 lines
3.1 KiB
Python
126 lines
3.1 KiB
Python
import threading
|
|
import time
|
|
from enum import Enum
|
|
|
|
import pygame
|
|
from pydantic import BaseModel
|
|
|
|
|
|
class PlayerState(Enum):
|
|
Playing = "Playing"
|
|
Paused = "Paused"
|
|
Stopped = "Stopped"
|
|
|
|
|
|
class Track(BaseModel):
|
|
artist: str
|
|
title: str
|
|
duration: int
|
|
filepath: str
|
|
|
|
|
|
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)
|
|
|
|
|
|
class MusicPlayer:
|
|
def __init__(self):
|
|
pygame.init()
|
|
pygame.mixer.init()
|
|
pygame.mixer.music.set_endevent(pygame.USEREVENT)
|
|
# Threading
|
|
self.lock = threading.Lock()
|
|
self._running = True
|
|
self.thread = threading.Thread(target=self._loop, daemon=True)
|
|
self.thread.start()
|
|
# Music Player
|
|
self.queue: Queue = Queue()
|
|
self._current_track: Track | None = None
|
|
self._state: PlayerState = PlayerState.Stopped
|
|
|
|
def _loop(self):
|
|
while self._running:
|
|
for event in pygame.event.get():
|
|
if event.type == pygame.USEREVENT:
|
|
if self._current_track:
|
|
print(f"Track {self._current_track.title} ended")
|
|
self._play_next_track()
|
|
|
|
time.sleep(0.1)
|
|
|
|
def _load_track(self, track: Track):
|
|
with self.lock:
|
|
self.current_track = track
|
|
pygame.mixer.music.unload()
|
|
pygame.mixer.music.load(self.current_track.filepath)
|
|
|
|
def _play_next_track(self):
|
|
next_track = self.queue.next()
|
|
if next_track:
|
|
self._load_track(next_track)
|
|
pygame.mixer.music.play()
|
|
|
|
def add_to_queue(self, track: Track):
|
|
self.queue.add(track)
|
|
|
|
def play(self):
|
|
import pdb
|
|
|
|
pdb.set_trace()
|
|
if self._current_track:
|
|
with self.lock:
|
|
pygame.mixer.music.play()
|
|
else:
|
|
self._play_next_track()
|
|
self._state = PlayerState.Playing
|
|
|
|
def pause(self):
|
|
with self.lock:
|
|
pygame.mixer.music.pause()
|
|
self._state = PlayerState.Paused
|
|
print("Set player state to Paused")
|
|
|
|
def resume(self):
|
|
with self.lock:
|
|
pygame.mixer.music.unpause()
|
|
self._state = PlayerState.Playing
|
|
print("Set player state to Playing")
|
|
|
|
def stop(self):
|
|
with self.lock:
|
|
pygame.mixer.music.stop()
|
|
self._state = PlayerState.Stopped
|
|
self._current_track = None
|
|
|
|
def shutdown(self):
|
|
self._running = False
|
|
self.thread.join()
|
|
pygame.mixer.quit()
|
|
|
|
def get_queue(self) -> Queue:
|
|
return self.queue
|
|
|
|
def set_volume(self, volume: float):
|
|
with self.lock:
|
|
pygame.mixer.music.set_volume(volume)
|
|
|
|
def get_volume(self):
|
|
return pygame.mixer.music.get_volume()
|
|
|
|
def get_state(self) -> PlayerState:
|
|
return self._state
|
|
|
|
def get_current_track(self) -> Track | None:
|
|
return self._current_track
|