radio95/RadioPlayer
0
1
Fork 0
mirror of https://github.com/radio95-rnt/RadioPlayer.git synced 2026-02-26 21:53:54 +01:00
Code Issues Activity

major internal changes (might be reverted)

Browse Source
This commit is contained in:
KubaPro010
2025-12-12 21:20:38 +01:00
parent 06b983d3cd
commit 2d31f0a907
8 changed files with 53 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
modules/__init__.py
View File

@@ -7,11 +7,12 @@ from pathlib import Path
@dataclass
class Track:
path: Path
fade_out: bool
fade_in: bool
fade_out: float
fade_in: float
official: bool
args: dict[str, str] | None
offset: float = 0.0
focus_time_offset: float = 0.0 # Offset according to the duration
@dataclass
class Process:
@@ -23,7 +24,7 @@ class Process:
class Skeleton_ProcessManager:
processes: list[Process]
def _get_audio_duration(self, file_path): ...
def play(self, track: Track, fade_in_time: int=0, fade_out_time: int=0) -> Process: ...
def play(self, track: Track, fade_in_time: float=0, fade_out_time: float=0) -> Process: ...
def anything_playing(self) -> bool: ...
def stop_all(self, timeout: float | None = None) -> None: ...
def wait_all(self, timeout: float | None = None) -> None: ...
@@ -72,7 +73,7 @@ class PlayerModule(BaseIMCModule):
"""
Simple passive observer, this allows you to send the current track the your RDS encoder, or to your website
"""
def on_new_playlist(self, playlist: list[Track]) -> None:
def on_new_playlist(self, playlist: list[Track], global_args: dict[str, str]) -> None:
"""This is called every new playlist"""
pass
def on_new_track(self, index: int, track: Track, next_track: Track | None) -> None:
@@ -132,7 +133,7 @@ class ActiveModifier(BaseIMCModule):
The second track object is the next track, which is optional which is also only used for metadata and will not be taken in as data to play
"""
return (track, None), False
def on_new_playlist(self, playlist: list[Track]) -> None:
def on_new_playlist(self, playlist: list[Track], global_args: dict[str, str]) -> None:
"""
Same behaviour as the basic module function
"""

27
modules/active_modifier.py
View File

@@ -18,8 +18,10 @@ class Module(ActiveModifier):
self.can_limit_tracks = False
self.morning_start = self.day_end = 0
self.file_lock = Lock()
def on_new_playlist(self, playlist: list[Track]):
self.crossfade = 5
def on_new_playlist(self, playlist: list[Track], global_args: dict[str, str]):
self.playlist = playlist
self.crossfade = float(global_args.get("crossfade", 5.0))
if not self._imc: return
self.limit_tracks, self.morning_start, self.day_end = self._imc.send(self, "advisor", None) # pyright: ignore[reportGeneralTypeIssues]
@@ -27,8 +29,7 @@ class Module(ActiveModifier):
if self.limit_tracks: logger.info("Skipping tracks if they bleed into other times.")
self.can_limit_tracks = self.limit_tracks
def play(self, index: int, track: Track | None, next_track: Track | None):
if not track:
raise NotImplementedError("This active modifer does not support advisor-less mode")
if not track: raise NotImplementedError("This active modifer does not support advisor-less mode")
if not self.playlist: return (track, next_track), False
@@ -51,16 +52,16 @@ class Module(ActiveModifier):
if len(songs):
song, official = get_song()
if self.last_track: last_track_to_fade_out = self.last_track.fade_out
if self.last_track: last_track_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 (index - 1) >= 0: last_track_fade_out = self.playlist[index - 1].fade_out
else: last_track_fade_out = 0.0
if len(songs) != 0: next_track_to_fade_in = True
if len(songs) != 0: next_track_fade_in = self.crossfade
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 index + 1 < len(self.playlist) and next_track: next_track_fade_in = next_track.fade_in
elif not next_track: next_track_fade_in = 0.0
else: next_track_fade_in = self.crossfade
if not self.originals or self.originals[-1] != track: self.originals.append(track)
@@ -74,10 +75,10 @@ class Module(ActiveModifier):
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, {})
self.last_track = Track(song, self.crossfade if new_official else 0, last_track_fade_out, official, {}, focus_time_offset=-self.crossfade if new_official else 0)
next_track = Track(new_song, self.crossfade if len(songs) else next_track_fade_in, self.crossfade if new_official else 0, new_official, {}, focus_time_offset=-self.crossfade if len(songs) else -next_track_fade_in)
else:
self.last_track = Track(song, next_track_to_fade_in, last_track_to_fade_out, official, {})
self.last_track = Track(song, next_track_fade_in, last_track_fade_out, official, {}, focus_time_offset=-next_track_fade_in)
next_track = track
self.limit_tracks = False
return (self.last_track, next_track), True

7
modules/jingle.py
View File

@@ -20,15 +20,16 @@ class Module(PlaylistModifierModule):
if int(global_args.get("no_jingle", 0)) != 0 or not self.primary: return None
out: list[Track] = []
last_jingiel = True
crossfade = float(global_args.get("crossfade", 5.0))
for track in playlist:
if not last_jingiel and (random.randint(1,3) == 1) and (track.args is None or int(track.args.get("no_jingle", 0)) == 0):
out.append(Track(track.path, True, False, True, track.args))
out.append(Track(track.path, crossfade, 0, True, track.args, focus_time_offset=-crossfade))
jingle = self.primary
if self.secondary and (random.randint(1,3) == 1): jingle = random.choice(self.secondary)
out.append(Track(jingle, False, False, False, {}))
out.append(Track(jingle, 0, 0, False, {}))
last_jingiel = True
continue
out.append(Track(track.path, True, True, True, track.args))
out.append(Track(track.path, crossfade, crossfade, True, track.args,focus_time_offset=-crossfade))
last_jingiel = False
return out

17
modules/progress.py
View File

@@ -1,17 +0,0 @@
from . import PlayerModule, Track, Path
def format_time(seconds) -> str:
hours = int(seconds // 3600)
minutes = int((seconds % 3600) // 60)
secs = int(seconds % 60)
return f"{hours:02d}:{minutes:02d}:{secs:02d}"
class Module(PlayerModule):
def progress(self, index: int, track: Track, elapsed: float, total: float, real_total: float) -> None:
if track.official:
data = f"{track.path.name}: {format_time(elapsed)} / {format_time(total)}\n"
Path("/tmp/radioPlayer_progress").write_text(data)
def shutdown(self):
Path("/tmp/radioPlayer_progress").write_text("")
module = Module()

20
modules/skipper.py Normal file
View File

@@ -0,0 +1,20 @@
from . import PlayerModule, log95, Track
import os
from typing import TextIO
_log_out: TextIO
assert _log_out # pyright: ignore[reportUnboundVariable]
logger = log95.log95("Skipper", output=_log_out)
class Module(PlayerModule):
def __init__(self) -> None: self.playlist = []
def on_new_playlist(self, playlist: list[Track], global_args: dict[str, str]): self.playlist = [str(t.path.absolute()) for t in playlist]
def progress(self, index: int, track: Track, elapsed: float, total: float, real_total: float) -> None:
if os.path.exists("/tmp/radioPlayer_skip"):
self._imc.send(self, "procman", {"op": 2}) # Ask procman to kill every track playing (usually there is one, unless we are in the default 5 seconds of the crossfade)
os.remove("/tmp/radioPlayer_skip")
def on_new_track(self, index: int, track: Track, next_track: Track | None):
if next_track: logger.info("Next up:", next_track.path.name)
module = Module()

5
modules/web.py
View File

@@ -1,4 +1,4 @@
# websocket_module.py
# TODO: focus time in track
import multiprocessing, os
import json
import threading, uuid, time
@@ -165,7 +165,8 @@ class Module(PlayerModule):
if key := message.get("key", None): self.data[key] = out
except Exception: pass
def on_new_playlist(self, playlist: list[Track]) -> None:
def on_new_playlist(self, playlist: list[Track], global_args: dict[str, str]) -> None:
# TODO: add global args to data
api_data = []
for track in playlist:
api_data.append({

31
modules/write_playlists.py
View File

@@ -1,31 +0,0 @@
from . import PlayerModule, log95, Track
import os
from typing import TextIO
_log_out: TextIO
assert _log_out # pyright: ignore[reportUnboundVariable]
logger = log95.log95("PlayView", output=_log_out)
class Module(PlayerModule):
def __init__(self) -> None: self.playlist = []
def on_new_playlist(self, playlist: list[Track]): self.playlist = [str(t.path.absolute()) for t in playlist]
def progress(self, index: int, track: Track, elapsed: float, total: float, real_total: float) -> None:
if os.path.exists("/tmp/radioPlayer_skip"):
self._imc.send(self, "procman", {"op": 2}) # Ask procman to kill every track playing (usually there is one, unless we are in the default 5 seconds of the crossfade)
os.remove("/tmp/radioPlayer_skip")
def on_new_track(self, index: int, track: Track, next_track: Track | None):
if next_track: logger.info("Next up:", next_track.path.name)
if str(track.path) != self.playlist[index]:
# discrepancy, which means that the playing file was modified by the active modifier
# we are playing a file that was not determined in the playlist, that means it was chosen by the active modifier and made up on the fly
lines = self.playlist[:index] + [f"> ({track.path})"] + [self.playlist[index]] + self.playlist[index+1:]
else: lines = self.playlist[:index] + [f"> {self.playlist[index]}"] + self.playlist[index+1:]
with open("/tmp/radioPlayer_playlist", "w") as f:
for line in lines:
try: f.write(line + "\n")
except UnicodeEncodeError:
print(line.encode('utf-8', errors='ignore').decode('utf-8'))
raise
module = Module()

13
radioPlayer.py
View File

@@ -28,7 +28,7 @@ class ProcessManager(Skeleton_ProcessManager):
result = float(result.stdout.strip())
self.duration_cache.saveElement(file_path.as_posix(), result, (60*60*2), False, True)
return result
def play(self, track: Track, fade_in_time: int=0, fade_out_time: int=0) -> Process:
def play(self, track: Track, fade_in_time: float=0, fade_out_time: float=0) -> Process:
assert track.path.exists()
cmd = ['ffplay', '-nodisp', '-hide_banner', '-autoexit', '-loglevel', 'quiet']
@@ -216,13 +216,13 @@ class RadioPlayer:
return
playlist: list[Track] | None = []
[playlist.extend(Track(Path(line).absolute(), True, True, True, args) for line in lns) for (lns, args) in parsed] # i can read this, i think
[playlist.extend(Track(Path(line).absolute(), 0, 0, True, args) for line in lns) for (lns, args) in parsed] # i can read this, i think
[(playlist := module.modify(global_args, playlist) or playlist) for module in self.playlist_modifier_modules if module] # yep
assert len(playlist)
prefetch(playlist[0].path)
[mod.on_new_playlist(playlist) for mod in self.simple_modules + [self.active_modifier] if mod] # one liner'd everything
[mod.on_new_playlist(playlist, global_args) for mod in self.simple_modules + [self.active_modifier] if mod] # one liner'd everything
max_iterator = len(playlist)
else:
@@ -230,7 +230,6 @@ class RadioPlayer:
playlist = None
global_args = {}
return_pending = track = False
cross_fade = int(global_args.get("crossfade", 5)) # TODO: get rid of global_args usage in the core and instead just store fades in track (that would require a pretty much rewrite of the active modifier tho...)
song_i = i = 0
def get_track():
@@ -273,11 +272,9 @@ class RadioPlayer:
self.logger.info(f"Now playing: {track.path.name}")
prefetch(track.path)
pr = self.procman.play(track, cross_fade, cross_fade)
pr = self.procman.play(track, track.fade_in, track.fade_out)
[module.on_new_track(song_i, pr.track, next_track) for module in self.simple_modules if module]
end_time = pr.started_at + pr.duration
if track.fade_out: end_time -= cross_fade
end_time = pr.started_at + pr.duration + pr.track.focus_time_offset
while end_time >= time.monotonic() and pr.process.poll() is None:
start = time.monotonic()