diff --git a/modules/__init__.py b/modules/__init__.py index cd07c0b..c7093b1 100644 --- a/modules/__init__.py +++ b/modules/__init__.py @@ -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) \ No newline at end of file + 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 {}, [] \ No newline at end of file diff --git a/modules/active_modifier.py b/modules/active_modifier.py index 7c04b34..1228791 100644 --- a/modules/active_modifier.py +++ b/modules/active_modifier.py @@ -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() \ No newline at end of file +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 \ No newline at end of file diff --git a/modules/advisor.py b/modules/advisor.py index 4103bb3..b1090df 100644 --- a/modules/advisor.py +++ b/modules/advisor.py @@ -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() \ No newline at end of file +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 \ No newline at end of file diff --git a/modules/playlist_parser.py b/modules/playlist_parser.py new file mode 100644 index 0000000..08c4043 --- /dev/null +++ b/modules/playlist_parser.py @@ -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() \ No newline at end of file diff --git a/modules/rds.py b/modules/rds.py index 727c783..0947ff3 100644 --- a/modules/rds.py +++ b/modules/rds.py @@ -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) diff --git a/modules/shuffler.py b/modules/shuffler.py index 297ca53..68d8f1c 100644 --- a/modules/shuffler.py +++ b/modules/shuffler.py @@ -7,4 +7,29 @@ class Module(PlaylistModifierModule): if int(global_args.get("no_shuffle", 0)) == 0: random.shuffle(playlist) return playlist -playlistmod = (Module(), 0) \ No newline at end of file +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 \ No newline at end of file diff --git a/modules/skipper.py b/modules/skipper.py index 35f2594..7611b31 100644 --- a/modules/skipper.py +++ b/modules/skipper.py @@ -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) diff --git a/radioPlayer.py b/radioPlayer.py index fda537b..e9978ae 100644 --- a/radioPlayer.py +++ b/radioPlayer.py @@ -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.