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

modularize playlist parser

Browse Source
This commit is contained in:
KubaPro010
2026-02-11 21:14:19 +01:00
parent 8175b2ad6a
commit cf0573c4a4
8 changed files with 173 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
modules/__init__.py
View File

@@ -5,6 +5,8 @@ from dataclasses import dataclass
from pathlib import Path
import tinytag
_log_out: log95.TextIO
@dataclass
class Track:
path: Path
@@ -164,4 +166,17 @@ class InterModuleCommunication:
Sends the data to a named module, and return its response
"""
if not name in self.names_modules.keys(): raise ModuleNotFoundError("No such module")
return self.names_modules[name].imc_data(source, next((k for k, v in self.names_modules.items() if v is source), None), data, False)
return self.names_modules[name].imc_data(source, next((k for k, v in self.names_modules.items() if v is source), None), data, False)
class PlaylistParser:
def __init__(self) -> None:
pass
def parse(self, playlist_path: Path) -> tuple[dict[str, str], list[tuple[list[str], dict[str, str]]]]:
"""
This should return the following information:
global arguments,
list of entries:
a entry is just a tuple of a list of strings (file paths)
and a dictionary of str:str consistent of the arguments which affect the files given
"""
return {}, []

27
modules/active_modifier.py
View File

@@ -167,4 +167,29 @@ class Module(ActiveModifier):
if data.get("set", True): self.skip_next = not self.skip_next
return {"status": "ok", "data": self.skip_next}
activemod = Module()
activemod = Module()
# This is free and unencumbered software released into the public domain.
# Anyone is free to copy, modify, publish, use, compile, sell, or
# distribute this software, either in source code form or as a compiled
# binary, for any purpose, commercial or non-commercial, and by any
# means.
# In jurisdictions that recognize copyright laws, the author or authors
# of this software dedicate any and all copyright interest in the
# software to the public domain. We make this dedication for the benefit
# of the public at large and to the detriment of our heirs and
# successors. We intend this dedication to be an overt act of
# relinquishment in perpetuity of all present and future rights to this
# software under copyright law.
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
# IN NO EVENT SHALL THE AUTHORS BE LIABLE FOR ANY CLAIM, DAMAGES OR
# OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE,
# ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
# OTHER DEALINGS IN THE SOFTWARE.
# For more information, please refer to <https://unlicense.org>

27
modules/advisor.py
View File

@@ -127,4 +127,29 @@ class Module(PlaylistAdvisor):
def imc_data(self, source: BaseIMCModule, source_name: str | None, data: object, broadcast: bool):
return (self.custom_playlist, MORNING_START, DAY_END)
advisor = Module()
advisor = Module()
# This is free and unencumbered software released into the public domain.
# Anyone is free to copy, modify, publish, use, compile, sell, or
# distribute this software, either in source code form or as a compiled
# binary, for any purpose, commercial or non-commercial, and by any
# means.
# In jurisdictions that recognize copyright laws, the author or authors
# of this software dedicate any and all copyright interest in the
# software to the public domain. We make this dedication for the benefit
# of the public at large and to the detriment of our heirs and
# successors. We intend this dedication to be an overt act of
# relinquishment in perpetuity of all present and future rights to this
# software under copyright law.
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
# IN NO EVENT SHALL THE AUTHORS BE LIABLE FOR ANY CLAIM, DAMAGES OR
# OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE,
# ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
# OTHER DEALINGS IN THE SOFTWARE.
# For more information, please refer to <https://unlicense.org>

56
modules/playlist_parser.py Normal file
View File

@@ -0,0 +1,56 @@
import glob
from . import log95, _log_out, Path
class PlaintextParser:
def __init__(self): self.logger = log95.log95("PARSER", output=_log_out)
def _check_for_imports(self, path: Path, seen=None) -> list[str]:
if seen is None: seen = set()
if not path.exists():
self.logger.error(f"Playlist not found: {path.name}")
raise Exception("Playlist doesn't exist")
lines = [line.strip() for line in path.read_text().splitlines() if line.strip()]
out = []
for line in lines:
if line.startswith("@"):
target = Path(line.removeprefix("@"))
if target not in seen:
if not target.exists():
self.logger.error(f"Target {target.name} of {path.name} does not exist")
continue
seen.add(target)
out.extend(self._check_for_imports(target, seen))
else: out.append(line)
return out
def parse(self, playlist_path: Path) -> tuple[dict[str, str], list[tuple[list[str], dict[str, str]]]]:
lines = self._check_for_imports(playlist_path)
out = []
global_arguments = {}
for line in lines:
arguments = {}
line = line.strip()
if not line or line.startswith(";") or line.startswith("#"): continue
if "|" in line:
if line.startswith("|"): # No file name, we're defining global arguments
args = line.removeprefix("|").split(";")
for arg in args:
if "=" in arg:
key, val = arg.split("=", 1)
arguments[key] = val
else:
arguments[arg] = True
else:
line, args = line.split("|", 1)
args = args.split(";")
for arg in args:
if "=" in arg:
key, val = arg.split("=", 1)
arguments[key] = val
else:
arguments[arg] = True
out.append(([f for f in glob.glob(line) if Path(f).is_file()], arguments))
return global_arguments, out
parser = PlaintextParser()

7
modules/rds.py
View File

@@ -1,5 +1,5 @@
from . import PlayerModule, log95, Track
import socket, re
from . import PlayerModule, _log_out, log95, Track
import socket
DEBUG = False
@@ -10,9 +10,6 @@ rds_default_artist = "radio95"
udp_host = ("127.0.0.1", 5000)
from typing import TextIO
_log_out: TextIO
logger_level = log95.log95Levels.DEBUG if DEBUG else log95.log95Levels.CRITICAL_ERROR
assert _log_out # pyright: ignore[reportUnboundVariable]
logger = log95.log95("RDS-MODULE", logger_level, output=_log_out)

27
modules/shuffler.py
View File

@@ -7,4 +7,29 @@ class Module(PlaylistModifierModule):
if int(global_args.get("no_shuffle", 0)) == 0: random.shuffle(playlist)
return playlist
playlistmod = (Module(), 0)
playlistmod = (Module(), 0)
# This is free and unencumbered software released into the public domain.
# Anyone is free to copy, modify, publish, use, compile, sell, or
# distribute this software, either in source code form or as a compiled
# binary, for any purpose, commercial or non-commercial, and by any
# means.
# In jurisdictions that recognize copyright laws, the author or authors
# of this software dedicate any and all copyright interest in the
# software to the public domain. We make this dedication for the benefit
# of the public at large and to the detriment of our heirs and
# successors. We intend this dedication to be an overt act of
# relinquishment in perpetuity of all present and future rights to this
# software under copyright law.
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
# IN NO EVENT SHALL THE AUTHORS BE LIABLE FOR ANY CLAIM, DAMAGES OR
# OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE,
# ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
# OTHER DEALINGS IN THE SOFTWARE.
# For more information, please refer to <https://unlicense.org>

5
modules/skipper.py
View File

@@ -1,9 +1,6 @@
from . import PlayerModule, log95, Track
from . import PlayerModule, log95, Track, _log_out
import os
from typing import TextIO
_log_out: TextIO
assert _log_out # pyright: ignore[reportUnboundVariable]
logger = log95.log95("Skipper", output=_log_out)

79
radioPlayer.py
View File

@@ -1,6 +1,6 @@
#!/usr/bin/env python3
import os, importlib.util, importlib.machinery, types
import sys, signal, glob, time, traceback, io
import sys, signal, time, traceback, io
import concurrent.futures
from modules import *
from threading import Lock
@@ -16,58 +16,6 @@ def prefetch(path):
MODULES_PACKAGE = "modules"
MODULES_DIR = Path(__file__, "..", MODULES_PACKAGE).resolve()
class PlaylistParser:
def __init__(self, output: log95.TextIO): self.logger = log95.log95("PARSER", output=output)
def _check_for_imports(self, path: Path, seen=None) -> list[str]:
if seen is None: seen = set()
if not path.exists():
self.logger.error(f"Playlist not found: {path.name}")
raise Exception("Playlist doesn't exist")
lines = [line.strip() for line in path.read_text().splitlines() if line.strip()]
out = []
for line in lines:
if line.startswith("@"):
target = Path(line.removeprefix("@"))
if target not in seen:
if not target.exists():
self.logger.error(f"Target {target.name} of {path.name} does not exist")
continue
seen.add(target)
out.extend(self._check_for_imports(target, seen))
else: out.append(line)
return out
def parse(self, playlist_path: Path) -> tuple[dict[str, str], list[tuple[list[str], dict[str, str]]]]:
lines = self._check_for_imports(playlist_path)
out = []
global_arguments = {}
for line in lines:
arguments = {}
line = line.strip()
if not line or line.startswith(";") or line.startswith("#"): continue
if "|" in line:
if line.startswith("|"): # No file name, we're defining global arguments
args = line.removeprefix("|").split(";")
for arg in args:
if "=" in arg:
key, val = arg.split("=", 1)
arguments[key] = val
else:
arguments[arg] = True
else:
line, args = line.split("|", 1)
args = args.split(";")
for arg in args:
if "=" in arg:
key, val = arg.split("=", 1)
arguments[key] = val
else:
arguments[arg] = True
out.append(([f for f in glob.glob(line) if Path(f).is_file()], arguments))
return global_arguments, out
class ModuleManager:
def __init__(self, output: log95.TextIO) -> None:
self.simple_modules: list[PlayerModule] = []
@@ -105,6 +53,7 @@ class ModuleManager:
self.modules.append((spec, module, module_name))
def start_modules(self, arg):
procman = None
parser = None
"""Executes the module by the python interpreter"""
def timed_loader(spec: importlib.machinery.ModuleSpec, module: types.ModuleType):
assert spec.loader
@@ -139,23 +88,35 @@ class ModuleManager:
else: self.playlist_modifier_modules.append(md)
if md := getattr(module, "advisor", None):
if self.playlist_advisor: raise Exception("Multiple playlist advisors")
if not isinstance(md, PlaylistAdvisor):
self.logger.error("Advisor does not inhirit from PlaylistAdvisor.")
continue
self.playlist_advisor = md
if md := getattr(module, "activemod", None):
if self.active_modifier: raise Exception("Multiple active modifiers")
if not isinstance(md, ActiveModifier):
self.logger.error("Active modifier does not inhirit from ActiveModifier.")
continue
self.active_modifier = md
if md := getattr(module, "procman", None):
if procman: raise Exception("Multiple procmans")
if not isinstance(md, ABC_ProcessManager):
self.logger.error("Modular process manager does not inherit from ABC_ProcessManager.")
self.logger.error("Process manager does not inherit from ABC_ProcessManager.")
continue
procman = md
if md := getattr(module, "parser", None):
if parser: raise Exception("Multiple parsers")
if not isinstance(md, PlaylistParser):
self.logger.error("Parser does not inhirit from PlaylistParser.")
continue
parser = md
if self.active_modifier: self.active_modifier.arguments(arg)
if not self.playlist_advisor: self.logger.warning("Playlist advisor was not found. Beta mode of advisor-less is running (playlist modifiers will not work)")
if not procman:
self.logger.critical_error("Missing process mananger.")
raise SystemExit("Missing process mananger.")
InterModuleCommunication(self.simple_modules + [self.playlist_advisor, ProcmanCommunicator(procman), self.active_modifier])
return procman
return procman, parser
def advisor_advise(self, arguments: str | None):
if not self.playlist_advisor: return None
return self.playlist_advisor.advise(arguments)
@@ -165,7 +126,7 @@ class RadioPlayer:
self.exit_pending = False
self.exit_status_code = self.intr_time = 0
self.exit_lock = Lock()
self.parser = PlaylistParser(output)
self.parser: PlaylistParser | None = None
self.procman: ABC_ProcessManager | None = None
self.arg = arg
self.logger = log95.log95("CORE", output=output)
@@ -191,11 +152,11 @@ class RadioPlayer:
"""Single functon for starting the core, returns but might exit raising an SystemExit"""
self.logger.info("Core starting, loading modules")
self.modman.load_modules()
self.procman = self.modman.start_modules(self.arg)
self.procman, self.parser = self.modman.start_modules(self.arg)
def play_once(self):
"""Plays a single playlist"""
if not (playlist_path := self.modman.advisor_advise(self.arg)):
if not (playlist_path := self.modman.advisor_advise(self.arg)) or not self.parser:
max_iterator = 1
playlist = None
else:
@@ -312,7 +273,7 @@ def main():
core.loop()
except SystemExit:
try: core.shutdown()
except BaseException: traceback.print_exc()
except BaseException: traceback.print_exc(file=f)
raise
# This is free and unencumbered software released into the public domain.