# import os import threading import vlc # type: ignore # from config import Config from datetime import datetime from helpers import file_is_unreadable from typing import Optional from time import sleep from log import log lock = threading.Lock() class Music: """ Manage the playing of music tracks """ def __init__(self) -> None: self.VLC = vlc.Instance() self.player = None self.max_volume = Config.VOLUME_VLC_DEFAULT def fade(self) -> None: """ Fade the currently playing track. The actual management of fading runs in its own thread so as not to hold up the UI during the fade. """ if not self.player: return if not self.player.get_position() > 0 and self.player.is_playing(): return thread = threading.Thread(target=self._fade) thread.start() def _fade(self) -> None: """ Implementation of fading the current track in a separate thread. """ # Take a copy of current player to allow another track to be # started without interfering here with lock: p = self.player self.player = None # Sanity check if not p: return fade_time = Config.FADE_TIME / 1000 steps = Config.FADE_STEPS sleep_time = fade_time / steps # We reduce volume by one mesure first, then by two measures, # then three, and so on. # The sum of the arithmetic sequence 1, 2, 3, ..n is # (n**2 + n) / 2 total_measures_count = (steps**2 + steps) / 2 measures_to_reduce_by = 0 for i in range(1, steps + 1): measures_to_reduce_by += i volume_factor = 1 - ( measures_to_reduce_by / total_measures_count) p.audio_set_volume(int(self.max_volume * volume_factor)) sleep(sleep_time) with lock: p.stop() log.debug(f"Releasing player {p=}") p.release() def get_playtime(self) -> Optional[int]: """Return elapsed play time""" if not self.player: return None return self.player.get_time() def get_position(self) -> Optional[float]: """Return current position""" if not self.player: return None return self.player.get_position() def play(self, path: str, position: Optional[float] = None) -> Optional[int]: """ Start playing the track at path. Log and return if path not found. """ if file_is_unreadable(path): log.error(f"play({path}): path not readable") return None status = -1 media = self.VLC.media_new_path(path) self.player = media.player_new_from_media() if self.player: self.player.audio_set_volume(self.max_volume) status = self.player.play() if position: self.player.set_position(position) return status # # def set_position(self, ms): # """Set current play time in milliseconds from start""" # # with lock: # return self.player.set_time(ms) def set_volume(self, volume, set_default=True): """Set maximum volume used for player""" if not self.player: return if set_default: self.max_volume = volume self.player.audio_set_volume(volume) def stop(self) -> float: """Immediately stop playing""" with lock: if not self.player: return 0.0 position = self.player.get_position() self.player.stop() self.player.release() # Ensure we don't reference player after release self.player = None return position