You've already forked RadioPlayer
mirror of
https://github.com/radio95-rnt/RadioPlayer.git
synced 2026-02-26 13:52:00 +01:00
95 lines
4.1 KiB
Python
95 lines
4.1 KiB
Python
from . import ActiveModifier, log95, Track, Path
|
|
import os, glob, datetime
|
|
|
|
from .advisor import MORNING_START, DAY_END
|
|
|
|
logger = log95.log95("AC-MOD")
|
|
|
|
class Module(ActiveModifier):
|
|
def __init__(self) -> None:
|
|
self.playlist = None
|
|
self.originals = []
|
|
self.last_track = None
|
|
self.limit_tracks = True
|
|
def on_new_playlist(self, playlist: list[Track]):
|
|
self.playlist = playlist
|
|
|
|
if not self._imc: return
|
|
self.limit_tracks = bool(self._imc.send(self, "advisor", None))
|
|
def play(self, index: int, track: Track, next_track: Track | None):
|
|
if not self.playlist: return (track, next_track), False
|
|
if not os.path.exists("/tmp/radioPlayer_toplay"): open("/tmp/radioPlayer_toplay", "a").close()
|
|
with open("/tmp/radioPlayer_toplay", "r") as f: songs = [s.strip() for s in f.readlines() if s.strip()]
|
|
|
|
songs[:] = [f for s in songs for f in glob.glob(s) if os.path.isfile(f)] # expand glob
|
|
|
|
def get_song(pop: bool = True):
|
|
nonlocal songs
|
|
if pop: song = songs.pop(0)
|
|
else: song = songs[0]
|
|
official = True
|
|
if song.startswith("!"):
|
|
song = song[1:]
|
|
official = False # NOT FLOATINGPOINTERROR
|
|
|
|
song = Path(song).absolute()
|
|
return song, official
|
|
|
|
if len(songs):
|
|
song, official = get_song()
|
|
|
|
if self.last_track: last_track_to_fade_out = self.last_track.fade_out
|
|
else:
|
|
if (index - 1) >= 0: last_track_to_fade_out = self.playlist[index - 1].fade_out
|
|
else: last_track_to_fade_out = False
|
|
|
|
if len(songs) != 0: next_track_to_fade_in = True
|
|
else:
|
|
if index + 1 < len(self.playlist) and next_track: next_track_to_fade_in = next_track.fade_in
|
|
elif not next_track: next_track_to_fade_in = False
|
|
else: next_track_to_fade_in = True
|
|
|
|
if not self.originals or self.originals[-1] != track: self.originals.append(track)
|
|
|
|
with open("/tmp/radioPlayer_toplay", "w") as f:
|
|
f.write('\n'.join(songs))
|
|
f.write("\n")
|
|
|
|
logger.info(f"Playing {song.name} instead, as instructed by toplay")
|
|
|
|
|
|
if len(songs):
|
|
# There are more tracks on the temp list
|
|
new_song, new_official = get_song(False)
|
|
self.last_track = Track(song, new_official, last_track_to_fade_out, official, {})
|
|
next_track = Track(new_song, new_official if len(songs) else next_track_to_fade_in, new_official, new_official, {})
|
|
else:
|
|
self.last_track = Track(song, next_track_to_fade_in, last_track_to_fade_out, official, {})
|
|
next_track = track
|
|
return (self.last_track, next_track), True
|
|
elif len(self.originals):
|
|
self.last_track = self.originals.pop(0)
|
|
if len(self.originals): next_track = self.originals[0]
|
|
else: self.last_track = track
|
|
|
|
if self.limit_tracks:
|
|
last_track_duration = self._imc.send(self, "procman", {"op": 1, "arg": self.last_track.path})
|
|
assert isinstance(last_track_duration, dict)
|
|
last_track_duration = last_track_duration.get("arg")
|
|
|
|
if last_track_duration and last_track_duration > 5*60:
|
|
now = datetime.datetime.now()
|
|
future = datetime.datetime.fromtimestamp(now.timestamp() + last_track_duration)
|
|
if now.hour < MORNING_START and future.hour > MORNING_START:
|
|
logger.warning("Skipping track as it bleeds into the morning")
|
|
return (None, None), None
|
|
elif now.hour < DAY_END and future.hour > DAY_END:
|
|
logger.warning("Skipping track as it bleeds into the night")
|
|
return (None, None), None
|
|
elif future.day > now.day: # late night goes mid day, as it starts at midnight
|
|
logger.warning("Skipping track as it the next day")
|
|
return (None, None), None
|
|
|
|
return (self.last_track, next_track), False
|
|
|
|
activemod = Module() |