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

massive clean ups (i wonder if it starts)

Browse Source
This commit is contained in:
Kuba
2025-10-31 18:04:37 +01:00
parent 2247e7996c
commit 1f4a184418
8 changed files with 130 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
modules/__init__.py
View File

@@ -1,14 +1,22 @@
import log95
from dataclasses import dataclass
@dataclass
class Track:
path: str
fade_out: bool
fade_in: bool
official: bool
args: dict[str, str] | None
class PlayerModule:
"""
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[tuple[str, bool, bool, bool, dict[str, str]]]):
"""Tuple consists of the track path, to fade out, fade in, official, and args
This is called every new playlist"""
def on_new_playlist(self, playlist: list[Track]):
"""This is called every new playlist"""
pass
def on_new_track(self, index: int, track: str, to_fade_in: bool, to_fade_out: bool, official: bool):
def on_new_track(self, index: int, track: Track):
"""
Called on every track including the ones added by the active modifier, you can check for that comparing the playlists[index] and the track
"""
@@ -24,7 +32,7 @@ class PlaylistModifierModule:
"""
Playlist modifier, this type of module allows you to shuffle, or put jingles into your playlist
"""
def modify(self, global_args: dict, playlist: list[tuple[str, bool, bool, bool, dict[str, str]]]):
def modify(self, global_args: dict, playlist: list[Track]):
"""
global_args are playlist global args (see radioPlayer_playlist_file.txt)
"""
@@ -34,7 +42,7 @@ class PlaylistAdvisor:
"""
Only one of a playlist advisor can be loaded. This module picks the playlist file to play, this can be a scheduler or just a static file
"""
def advise(self, arguments: str | None) -> str:
def advise(self, arguments: str | None) -> str:
"""
Arguments are the arguments passed to the program on startup
"""
@@ -55,18 +63,18 @@ class ActiveModifier:
"""
This changes the next song to be played live, which means that this picks the next song, not the playlist, but this is affected by the playlist
"""
def arguments(self, arguments: str | None):
def arguments(self, arguments: str | None):
"""
Called at start up with the program arguments
"""
pass
def play(self, index:int, track: tuple[str, bool, bool, bool, dict[str, str]]) -> tuple[tuple[str, bool, bool, bool, dict[str, str]], bool] | tuple[None, None]:
def play(self, index:int, track: Track) -> tuple[Track, bool] | tuple[None, None]:
"""
Returns a tuple, in the first case where a is the track and b is a bool, b corresponds to whether to extend the playlist, set to true when adding content instead of replacing it
When None, None is returned then that is treated as a skip, meaning the core will skip this song
"""
return track, False
def on_new_playlist(self, playlist: list[tuple[str, bool, bool, bool, dict[str, str]]]):
def on_new_playlist(self, playlist: list[Track]):
"""
Same behaviour as the basic module function
"""

19
modules/active_modifier.py
View File

@@ -1,5 +1,5 @@
from modules import InterModuleCommunication
from . import ActiveModifier, log95
from . import ActiveModifier, log95, Track
import os
import subprocess
import datetime
@@ -20,31 +20,30 @@ class Module(ActiveModifier):
self.last_track = None
self.limit_tracks = True
self.imc_class = None
def on_new_playlist(self, playlist: list[tuple[str, bool, bool, bool, dict[str, str]]]):
def on_new_playlist(self, playlist: list[Track]):
self.playlist = playlist
if not self.imc_class: return
self.limit_tracks = bool(self.imc_class.send(self, "advisor", None))
def play(self, index: int, track: tuple[str, bool, bool, bool, dict[str, str]]):
def play(self, index: int, track: Track):
if not self.playlist: return track
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()]
with open("/tmp/radioPlayer_toplay", "r") as f: songs = [s.strip() for s in f.readlines() if s.strip()]
if len(songs):
song = songs.pop(0)
if self.last_track:
_, last_track_to_fade_out, _, _, _ = 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]
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):
_, _, next_track_to_fade_in, _, _ = self.playlist[index + 1]
next_track_to_fade_in = self.playlist[index + 1].fade_in
else:
next_track_to_fade_in = True
@@ -56,13 +55,13 @@ class Module(ActiveModifier):
logger.info(f"Playing {song} instead, as instructed by toplay")
self.last_track = (song, next_track_to_fade_in, last_track_to_fade_out, True, {})
self.last_track = Track(song, next_track_to_fade_in, last_track_to_fade_out, True, {})
return self.last_track, True
elif len(self.originals): self.last_track = self.originals.pop(0)
else: self.last_track = track
if self.limit_tracks:
last_track_duration = get_audio_duration(self.last_track[0])
last_track_duration = get_audio_duration(self.last_track.path)
if last_track_duration and last_track_duration > 5*60:
now = datetime.datetime.now()
timestamp = now.timestamp() + last_track_duration

16
modules/jingle.py
View File

@@ -8,22 +8,22 @@ Reacts to the 'no_jingle' argument, for global usage it does not add jingles to
import random
from . import PlaylistModifierModule
from . import PlaylistModifierModule, Track
class Module(PlaylistModifierModule):
def __init__(self, file: str) -> None:
self.file = file
def modify(self, global_args: dict, playlist: list[tuple[str, bool, bool, bool, dict]]):
def modify(self, global_args: dict, playlist: list[Track]):
if int(global_args.get("no_jingle", 0)): return playlist
out: list[tuple[str, bool, bool, bool, dict]] = []
out: list[Track] = []
last_jingiel = True
for (track, _, _, _, args) in playlist:
if not last_jingiel and random.choice([False, True, False, False]) and self.file and int(args.get("no_jingle", 0)) == 0:
out.append((track, True, False, True, args))
out.append((self.file, False, False, False, {}))
for track in playlist:
if not last_jingiel and random.choice([False, True, False, False]) and self.file and (track.args and int(track.args.get("no_jingle", 0)) == 0):
out.append(Track(track, True, False, True, track.args))
out.append(Track(self.file, False, False, False, {}))
last_jingiel = True
else:
out.append((track, True, True, True, args))
out.append(Track(track, True, True, True, track.args))
last_jingiel = False
del last_jingiel
return out

73
modules/modules.txt
View File

@@ -7,62 +7,111 @@ First of all, ther are in total only 4 modules:
- Active modifier (ActiveModifier), this module is optional, but there can still be only one. This module can replace the track while playing, allowing you to skip tracks or play tracks on demand, it can also extend the playlist
```python
@dataclass
class Track:
path: str
fade_out: bool
fade_in: bool
official: bool
args: dict[str, str] | None
class PlayerModule:
"""
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[tuple[str, bool, bool, bool, dict[str, str]]]):
"""Tuple consists of the track path, to fade out, fade in, official, and args
This is called every new playlist"""
def on_new_playlist(self, playlist: list[Track]):
"""This is called every new playlist"""
pass
def on_new_track(self, index: int, track: str, to_fade_in: bool, to_fade_out: bool, official: bool):
def on_new_track(self, index: int, track: Track):
"""
Called on every track including the ones added by the active modifier, you can check for that comparing the playlists[index] and the track
"""
pass
def imc(self, imc: 'InterModuleCommunication'):
"""
Receive an IMC object
"""
pass
def imc_data(self, source: 'PlayerModule | ActiveModifier | PlaylistAdvisor', data: object) -> object:
pass
class PlaylistModifierModule:
"""
Playlist modifier, this type of module allows you to shuffle, or put jingles into your playlist
"""
def modify(self, global_args: dict, playlist: list[tuple[str, bool, bool, bool, dict[str, str]]]):
def modify(self, global_args: dict, playlist: list[Track]):
"""
global_args are playlist global args (see radioPlayer_playlist_file.txt)
"""
return playlist
# No IMC, as we only run on new playlists
class PlaylistAdvisor:
"""
Only one of a playlist advisor can be loaded. This module picks the playlist file to play, this can be a scheduler or just a static file
"""
def advise(self, arguments: str | None) -> str:
def advise(self, arguments: str | None) -> str:
"""
Arguments are the arguments passed to the program on startup
"""
return "/path/to/playlist.txt"
def new_playlist(self) -> int:
def new_playlist(self) -> bool:
"""
Whether to play a new playlist, if this is 1, then the player will refresh, if this is two then the player will refresh quietly
Whether to play a new playlist, if this is True, then the player will refresh and fetch a new playlist, calling advise
"""
return 0
return False
def imc(self, imc: 'InterModuleCommunication'):
"""
Receive an IMC object
"""
pass
def imc_data(self, source: 'PlayerModule | ActiveModifier | PlaylistAdvisor', data: object) -> object:
pass
class ActiveModifier:
"""
This changes the next song to be played live, which means that this picks the next song, not the playlist, but this is affected by the playlist
"""
def arguments(self, arguments: str | None):
def arguments(self, arguments: str | None):
"""
Called at start up with the program arguments
"""
pass
def play(self, index:int, track: tuple[str, bool, bool, bool, dict[str, str]]) -> tuple[tuple[str, bool, bool, bool, dict[str, str]], bool] | tuple[None, None]:
def play(self, index:int, track: Track) -> tuple[Track, bool] | tuple[None, None]:
"""
Returns a tuple, in the first case where a is the track and b is a bool, b corresponds to whether to extend the playlist, set to true when adding content instead of replacing it
When None, None is returned then that is treated as a skip, meaning the core will skip this song
"""
return track, False
def on_new_playlist(self, playlist: list[tuple[str, bool, bool, bool, dict[str, str]]]):
def on_new_playlist(self, playlist: list[Track]):
"""
Same behaviour as the basic module function
"""
pass
def imc(self, imc: 'InterModuleCommunication'):
"""
Receive an IMC object
"""
pass
def imc_data(self, source: 'PlayerModule | ActiveModifier | PlaylistAdvisor', data: object) -> object:
pass
class InterModuleCommunication:
def __init__(self, advisor: PlaylistAdvisor, active_modifier: ActiveModifier | None, simple_modules: list[PlayerModule]) -> None:
self.advisor = advisor
self.active_modifier = active_modifier
self.simple_modules = simple_modules
self.names_modules: dict[str, PlaylistAdvisor | ActiveModifier | PlayerModule] = {}
def broadcast(self, source: PlaylistAdvisor | ActiveModifier | PlayerModule, data: object) -> None:
"""
Send data to all modules
"""
self.advisor.imc_data(source, data)
if self.active_modifier: self.active_modifier.imc_data(source, data)
for module in self.simple_modules: module.imc_data(source, data)
def register(self, module: PlaylistAdvisor | ActiveModifier | PlayerModule, name: str):
if name in self.names_modules.keys(): return False
self.names_modules[name] = module
return True
def send(self, source: PlaylistAdvisor | ActiveModifier | PlayerModule, name: str, data: object) -> object:
if not name in self.names_modules.keys(): raise Exception
return self.names_modules[name].imc_data(source, data)
```
Each module shall have a python script in the modules directory. Each of the modules need to define one or more global variables in order to be seen by the core:

8
modules/rds.py
View File

@@ -1,4 +1,4 @@
from . import PlayerModule, log95
from . import PlayerModule, log95, Track
import socket, re, os
DEBUG = False
@@ -72,9 +72,9 @@ def update_rds(track_name: str):
return prt, ','.join(list(map(str, rtp)))
class Module(PlayerModule):
def on_new_track(self, index: int, track: str, to_fade_in: bool, to_fade_out: bool, official: bool):
if official:
rds_rt, rds_rtp = update_rds(os.path.basename(track))
def on_new_track(self, index: int, track: Track):
if track.official:
rds_rt, rds_rtp = update_rds(os.path.basename(track.path))
logger.info(f"RT set to '{rds_rt}'")
logger.debug(f"{rds_rtp=}")

4
modules/shuffler.py
View File

@@ -1,9 +1,9 @@
import random
from . import PlaylistModifierModule
from . import PlaylistModifierModule, Track
class Module(PlaylistModifierModule):
def modify(self, global_args: dict, playlist: list[tuple[str, bool, bool, bool, dict]]):
def modify(self, global_args: dict, playlist: list[Track]):
if int(global_args.get("no_shuffle", 0)) == 0:
random.shuffle(playlist)
return playlist

12
modules/write_playlists.py
View File

@@ -1,17 +1,17 @@
from . import PlayerModule, log95
from . import PlayerModule, log95, Track
logger = log95.log95("PlayView")
class Module(PlayerModule):
def __init__(self) -> None:
self.playlist = []
def on_new_playlist(self, playlist: list[tuple[str, bool, bool, bool, dict]]):
self.playlist = [t[0] for t in playlist]
def on_new_track(self, index: int, track: str, to_fade_in: bool, to_fade_out: bool, official: bool):
if track != self.playlist[index]:
def on_new_playlist(self, playlist: list[Track]):
self.playlist = [t.path for t in playlist]
def on_new_track(self, index: int, track: Track):
if 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})"] + [self.playlist[index]] + self.playlist[index+1:]
lines = self.playlist[:index] + [f"> ({track.path})"] + [self.playlist[index]] + self.playlist[index+1:]
logger.info("Next up:", self.playlist[index])
else:
lines = self.playlist[:index] + [f"> {self.playlist[index]}"] + self.playlist[index+1:]

56
radioPlayer.py
View File

@@ -1,7 +1,7 @@
#!/usr/bin/env python3
DEBUG = False
import time
import os, subprocess, importlib.util
import os, subprocess, importlib.util, types
import sys, signal, threading, glob
import unidecode
from dataclasses import dataclass
@@ -14,10 +14,8 @@ playlist_modifier_modules: list[PlaylistModifierModule] = []
playlist_advisor: PlaylistAdvisor | None = None
active_modifier: ActiveModifier | None = None
SCRIPT_DIR = Path(__file__).resolve().parent
MODULES_PACKAGE = "modules"
MODULES_DIR = SCRIPT_DIR / MODULES_PACKAGE
MODULES_DIR = MODULES_DIR.resolve()
MODULES_DIR = (Path(__file__).resolve().parent / MODULES_PACKAGE).resolve()
def print_wait(ttw: float, frequency: float, duration: float=-1, prefix: str="", bias: float = 0):
interval = 1.0 / frequency
@@ -101,11 +99,10 @@ procman = ProcessManager()
def handle_sigint(signum, frame):
global exit_pending, intr_time
logger.info("Received SIGINT")
if (time.time() - intr_time) > 10:
if (time.time() - intr_time) > 5:
intr_time = time.time()
logger.info("Will quit on song end.")
exit_pending = True
return
else:
logger.warning("Force-Quit pending")
procman.stop_all()
@@ -120,7 +117,7 @@ def load_filelines(path):
logger.error(f"Playlist not found: {path}")
return []
def parse_playlistfile(playlist_path: str):
def parse_playlistfile(playlist_path: str) -> tuple[dict[str, str], list[tuple[list[str], dict[str, str]]]]:
parser_log = log95.log95("PARSER", logger_level)
parser_log.debug("Reading", playlist_path)
@@ -165,18 +162,15 @@ def parse_playlistfile(playlist_path: str):
def play_playlist(playlist_path):
if not playlist_advisor: raise Exception("No playlist advisor")
try:
global_args, parsed = parse_playlistfile(playlist_path)
try: global_args, parsed = parse_playlistfile(playlist_path)
except Exception:
logger.info(f"Exception while parsing playlist, retrying in 15 seconds...")
time.sleep(15)
return
playlist: list[tuple[str, bool, bool, bool, dict]] = [] # name, fade in, fade out, official, args
playlist: list[Track] = []
for (lns, args) in parsed:
lns: list[str]
args: dict[str, str]
for line in lns: playlist.append((line, True, True, True, args)) # simple entry, just to convert to a format taken by the modules
for line in lns: playlist.append(Track(line, True, True, True, args)) # simple entry, just to convert to a format taken by the modules
for module in playlist_modifier_modules: playlist = module.modify(global_args, playlist)
for module in simple_modules: module.on_new_playlist(playlist)
@@ -187,8 +181,7 @@ def play_playlist(playlist_path):
cross_fade = int(global_args.get("crossfade", 5))
max_iterator = len(playlist)
i = 0
song_i = 0
song_i = i = 0
while i < max_iterator:
if exit_pending:
@@ -199,37 +192,36 @@ def play_playlist(playlist_path):
logger.info("Return reached, next song will reload the playlist.")
procman.wait_all()
return
if playlist_advisor.new_playlist():
logger.info("Reloading now...")
return_pending = True
continue
old_track_tuple = playlist[song_i % len(playlist)]
old_track = playlist[song_i % len(playlist)]
if active_modifier:
track_tuple, extend = active_modifier.play(song_i, old_track_tuple)
if track_tuple is None:
track, extend = active_modifier.play(song_i, old_track)
if track is None:
song_i += 1
continue
if extend: max_iterator += 1
else:
extend = False
track_tuple = old_track_tuple
track, to_fade_in, to_fade_out, official, args = track_tuple
track = old_track
track_path = os.path.abspath(os.path.expanduser(track))
track_path = os.path.abspath(os.path.expanduser(track.path))
track_name = os.path.basename(track_path)
for module in simple_modules: module.on_new_track(song_i, track_path, to_fade_in, to_fade_out, official)
logger.info(f"Now playing: {track_name}")
pr = procman.play(track_path, to_fade_in, to_fade_out, cross_fade)
for module in simple_modules: module.on_new_track(song_i, track)
pr = procman.play(track_path, track.fade_in, track.fade_out, cross_fade)
ttw = pr.duration
if to_fade_out: ttw -= cross_fade
if track.fade_out: ttw -= cross_fade
if official: print_wait(ttw, 1, pr.duration, f"{track_name}: ")
if track.official: print_wait(ttw, 1, pr.duration, f"{track_name}: ")
else: time.sleep(ttw)
i += 1
@@ -250,7 +242,6 @@ def main():
sys.modules[full_module_name] = module
if MODULES_PACKAGE not in sys.modules:
import types
parent = types.ModuleType(MODULES_PACKAGE)
parent.__path__ = [str(MODULES_DIR)]
parent.__package__ = MODULES_PACKAGE
@@ -281,9 +272,8 @@ def main():
if not playlist_advisor:
logger.critical_error("Playlist advisor was not found")
exit(1)
imc = InterModuleCommunication(playlist_advisor, active_modifier, simple_modules)
imc = InterModuleCommunication(playlist_advisor, active_modifier, simple_modules)
playlist_advisor.imc(imc)
if active_modifier: active_modifier.imc(imc)
for module in simple_modules: module.imc(imc)
@@ -292,9 +282,9 @@ def main():
arg = " ".join(sys.argv[1:]) if len(sys.argv) > 1 else None
if active_modifier: active_modifier.arguments(arg)
while True:
playlist = playlist_advisor.advise(arg)
logger.info(f"Advisor picked '{playlist}' to play")
play_playlist(playlist)
if playlist := playlist_advisor.advise(arg):
logger.info(f"Advisor picked '{playlist}' to play")
play_playlist(playlist)
if exit_pending: exit()
except Exception as e:
logger.error(f"Unexpected error: {e}")