From 1f4a184418afcfb7056bbf0138b957ce030ec237 Mon Sep 17 00:00:00 2001 From: Kuba <132459354+KubaPro010@users.noreply.github.com> Date: Fri, 31 Oct 2025 18:04:37 +0100 Subject: [PATCH] massive clean ups (i wonder if it starts) --- modules/__init__.py | 26 +++++++++----- modules/active_modifier.py | 19 +++++----- modules/jingle.py | 16 ++++----- modules/modules.txt | 73 +++++++++++++++++++++++++++++++------- modules/rds.py | 8 ++--- modules/shuffler.py | 4 +-- modules/write_playlists.py | 12 +++---- radioPlayer.py | 56 ++++++++++++----------------- 8 files changed, 130 insertions(+), 84 deletions(-) diff --git a/modules/__init__.py b/modules/__init__.py index b30e5b9..8dd779e 100644 --- a/modules/__init__.py +++ b/modules/__init__.py @@ -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 """ diff --git a/modules/active_modifier.py b/modules/active_modifier.py index 25ae18b..a0726d3 100644 --- a/modules/active_modifier.py +++ b/modules/active_modifier.py @@ -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 diff --git a/modules/jingle.py b/modules/jingle.py index 008f98e..ff1aa49 100644 --- a/modules/jingle.py +++ b/modules/jingle.py @@ -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 diff --git a/modules/modules.txt b/modules/modules.txt index ada229e..9cc143f 100644 --- a/modules/modules.txt +++ b/modules/modules.txt @@ -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: diff --git a/modules/rds.py b/modules/rds.py index 1a2330a..b1fc23d 100644 --- a/modules/rds.py +++ b/modules/rds.py @@ -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=}") diff --git a/modules/shuffler.py b/modules/shuffler.py index 04a3d8f..b88f3d4 100644 --- a/modules/shuffler.py +++ b/modules/shuffler.py @@ -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 diff --git a/modules/write_playlists.py b/modules/write_playlists.py index d02c280..ae5e995 100644 --- a/modules/write_playlists.py +++ b/modules/write_playlists.py @@ -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:] diff --git a/radioPlayer.py b/radioPlayer.py index ffeba6a..a43ff40 100644 --- a/radioPlayer.py +++ b/radioPlayer.py @@ -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}")