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): 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