radio95/RadioPlayer
0
1
Fork 0
mirror of https://github.com/radio95-rnt/RadioPlayer.git synced 2026-07-29 15:29:14 +02:00
Code Issues Activity

brand new file format

Browse Source
This commit is contained in:
kuba
2025-10-05 12:54:55 +02:00
parent 8850dbb982
commit c78ad8fafa
3 changed files with 82 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files
radioPlayer.py
+62 -34
View File
@@ -1,4 +1,5 @@
#!/usr/bin/env python3
DEBUG = False
import time, datetime
import os, subprocess
import sys, signal, threading, glob
@@ -6,7 +7,7 @@ import re, unidecode
import random
import socket
from dataclasses import dataclass
import log95
import log95, copy
def print_wait(ttw: float, frequency: float, duration: float=-1, prefix: str="", bias: float = 0):
interval = 1.0 / frequency
@@ -42,8 +43,6 @@ DAY_END = 19
LATE_NIGHT_START = 0
LATE_NIGHT_END = 5
CROSSFADE_DURATION = 5
JINGIEL_FILE = "/home/user/Jingiel.mp3"
playlist_dir = "/home/user/playlists"
@@ -55,7 +54,8 @@ rds_default_name = "Program Godzinny"
udp_host = ("127.0.0.1", 5000)
logger = log95.log95("radioPlayer")
logger_level = log95.log95Levels.DEBUG if DEBUG else log95.log95Levels.CRITICAL_ERROR
logger = log95.log95("radioPlayer", logger_level)
exit_pending = False
reload_pending = False
@@ -84,14 +84,14 @@ class ProcessManager:
result = subprocess.run(['ffprobe', '-v', 'quiet', '-show_entries', 'format=duration', '-of', 'default=noprint_wrappers=1:nokey=1', file_path], capture_output=True, text=True)
if result.returncode == 0: return float(result.stdout.strip())
return None
def play(self, track_path: str, fade_in: bool=False, fade_out: bool=False) -> Process:
def play(self, track_path: str, fade_in: bool=False, fade_out: bool=False, fade_time: int = 5) -> Process:
cmd = ['ffplay', '-nodisp', '-hide_banner', '-autoexit', '-loglevel', 'quiet']
duration = self._get_audio_duration(track_path)
if not duration: raise Exception("Failed to get file duration, does it actually exit?", track_path)
filters = []
if fade_in: filters.append(f"afade=t=in:st=0:d={CROSSFADE_DURATION}")
if fade_out: filters.append(f"afade=t=out:st={duration-CROSSFADE_DURATION}:d={CROSSFADE_DURATION}")
if fade_in: filters.append(f"afade=t=in:st=0:d={fade_time}")
if fade_out: filters.append(f"afade=t=out:st={duration-fade_time}:d={fade_time}")
if filters: cmd.extend(['-af', ",".join(filters)])
cmd.append(track_path)
@@ -229,44 +229,77 @@ def check_if_playlist_modifed(playlist_path: str, custom_playlist: bool = False)
logger.info("Time changed to night hours, switching playlist...")
return True
def play_playlist(playlist_path, custom_playlist: bool=False, do_shuffle=True):
last_modified_time = Time.get_playlist_modification_time(playlist_path)
def parse_playlistfile(playlist_path: str):
parser_log = log95.log95("PARSER", logger_level)
parser_log.debug("Reading", playlist_path)
lines = load_filelines(playlist_path)
if not lines:
logger.info(f"No tracks found in {playlist_path}, checking again in 15 seconds...")
time.sleep(15)
return
def check_for_imports(lines: list[str], seen=None) -> list[str]:
nonlocal parser_log
if seen is None: seen = set()
out = []
for line in lines:
if line.startswith("@"):
target = line.removeprefix("@")
if target not in seen:
parser_log.debug("Importing", target)
seen.add(target)
sub_lines = load_filelines(target)
out.extend(check_for_imports(sub_lines, seen))
else: out.append(line)
return out
lines = check_for_imports(lines)
lines = check_for_imports(lines) # First, import everything
glob_lines = []
out = []
global_arguments = {}
for line in lines:
arguments = {}
if line.startswith(";") or not line.strip(): continue
glob_lines.extend([f for f in glob.glob(line) if os.path.isfile(f)])
if "|" in line:
if line.startswith("|"): # No file name, we're defining global arguments
args = line.removeprefix("|").split(";")
for arg in args:
key, val = arg.split("=", 1)
global_arguments[key] = val
else:
line, args = line.split("|", 1)
args = args.split(";")
for arg in args:
key, val = arg.split("=", 1)
arguments[key] = val
parser_log.debug("Line:", line, "| Global Args:", repr(global_arguments), "| Local args:", repr(arguments))
out.append(([f for f in glob.glob(line) if os.path.isfile(f)], arguments))
return global_arguments, out
def play_playlist(playlist_path, custom_playlist: bool=False):
last_modified_time = Time.get_playlist_modification_time(playlist_path)
if do_shuffle: random.shuffle(glob_lines)
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
lines_args = copy.deepcopy(parsed)
lines = []
for (lns, args) in lines_args:
lns: list[str]
args: dict[str, str]
for i in range(int(args.get("multiplier", 1))): lines.extend(lns)
cross_fade = int(global_args.get("crossfade", 5))
if int(global_args.get("no_shuffle", 0)) == 0: random.shuffle(lines)
playlist: list[tuple[str, bool, bool, bool]] = [] # name, fade in, fade out, official
last_jingiel = True
for glob_line in glob_lines:
for line in lines:
if not last_jingiel and random.choice([False, True, False, False]) and JINGIEL_FILE:
playlist.append((glob_line, True, False, True))
playlist.append((line, True, False, True))
playlist.append((JINGIEL_FILE, False, False, False))
last_jingiel = True
else:
playlist.append((glob_line, True, True, True))
playlist.append((line, True, True, True))
last_jingiel = False
del last_jingiel
@@ -307,7 +340,7 @@ def play_playlist(playlist_path, custom_playlist: bool=False, do_shuffle=True):
pr = procman.play(track_path, to_fade_in, to_fade_out)
ttw = pr.duration
if to_fade_out: ttw -= CROSSFADE_DURATION
if to_fade_out: ttw -= cross_fade
if official: print_wait(ttw, 1, pr.duration, f"{track_name}: ")
else: time.sleep(ttw)
@@ -319,7 +352,6 @@ def can_delete_file(filepath):
def parse_arguments():
"""Parse command line arguments and return configuration"""
arg = sys.argv[1] if len(sys.argv) > 1 else None
do_shuffle = True
selected_list = None
if arg:
@@ -331,7 +363,6 @@ def parse_arguments():
print("Arguments:")
print(" list:playlist;options - Play custom playlist with options")
print()
print(f"Crossfade: {CROSSFADE_DURATION}-second crossfade is automatically applied between tracks")
exit(0)
if can_delete_file("/tmp/radioPlayer_arg"):
@@ -342,23 +373,20 @@ def parse_arguments():
if arg.startswith("list:"):
selected_list = arg.removeprefix("list:")
logger.info(f"The list {selected_list.split(';')[0]} will be played instead of the daily section lists.")
for option in selected_list.split(";"):
if option == "s": do_shuffle = False
selected_list = selected_list.split(";")[0]
else: logger.error(f"Invalid argument or file not found: {arg}")
return do_shuffle, selected_list
return selected_list
def main():
try:
while True:
do_shuffle, selected_list = parse_arguments()
selected_list = parse_arguments()
play_loop = True
while play_loop:
if selected_list:
logger.info("Playing custom list")
result = play_playlist(selected_list, True, do_shuffle)
result = play_playlist(selected_list, True)
if result == "reload": play_loop = False
continue
@@ -386,16 +414,16 @@ def main():
if DAY_START <= current_hour < DAY_END:
logger.info(f"Playing {current_day} day playlist...")
result = play_playlist(day_playlist, False, do_shuffle)
result = play_playlist(day_playlist, False)
elif MORNING_START <= current_hour < MORNING_END:
logger.info(f"Playing {current_day} morning playlist...")
result = play_playlist(morning_playlist, False, do_shuffle)
result = play_playlist(morning_playlist, False)
elif LATE_NIGHT_START <= current_hour < LATE_NIGHT_END:
logger.info(f"Playing {current_day} late_night playlist...")
result = play_playlist(late_night_playlist, False, do_shuffle)
result = play_playlist(late_night_playlist, False)
else:
logger.info(f"Playing {current_day} night playlist...")
result = play_playlist(night_playlist, False, do_shuffle)
result = play_playlist(night_playlist, False)
if exit_pending: exit()
elif reload_pending: