From 7769d95f0c1b7628f9b24c650ac5960570232651 Mon Sep 17 00:00:00 2001 From: Kuba <132459354+KubaPro010@users.noreply.github.com> Date: Sun, 2 Nov 2025 10:05:59 +0100 Subject: [PATCH] some technical improvements --- modules/__init__.py | 81 ++++++++++++++++++++++--------------------- modules/advisor.py | 2 +- modules/modules.txt | 84 ++++++++++++++++++++++++--------------------- radioPlayer.py | 2 +- 4 files changed, 88 insertions(+), 81 deletions(-) diff --git a/modules/__init__.py b/modules/__init__.py index 640de69..bd003c3 100644 --- a/modules/__init__.py +++ b/modules/__init__.py @@ -10,26 +10,34 @@ class Track: args: dict[str, str] | None offset: float = 0.0 -class PlayerModule: +class BaseIMCModule: """ - Simple passive observer, this allows you to send the current track the your RDS encoder, or to your website + This is not a module to be used but rather a placeholder IMC api to be used in other modules """ - def on_new_playlist(self, playlist: list[Track]): - """This is called every new playlist""" - pass - 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'): + def imc(self, imc: 'InterModuleCommunication') -> None: """ Receive an IMC object """ pass - def imc_data(self, source: 'PlayerModule | ActiveModifier | PlaylistAdvisor', data: object, broadcast: bool) -> object: + def imc_data(self, source: 'BaseIMCModule', source_name: str | None, data: object, broadcast: bool) -> object: + """ + React to IMC data + """ return None - def progress(self, index: int, track: Track, elapsed: float, total: float, real_total: float): + +class PlayerModule(BaseIMCModule): + """ + 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[Track]) -> None: + """This is called every new playlist""" + pass + def on_new_track(self, index: int, track: Track) -> None: + """ + 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 progress(self, index: int, track: Track, elapsed: float, total: float, real_total: float) -> None: """ Real total and total differ in that, total is how much the track lasts, but real_total will be for how long we will play it for """ @@ -38,13 +46,13 @@ 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[Track]): + def modify(self, global_args: dict, playlist: list[Track]) -> list[Track] | None: """ global_args are playlist global args (see radioPlayer_playlist_file.txt) """ return playlist # No IMC, as we only run on new playlists -class PlaylistAdvisor: +class PlaylistAdvisor(BaseIMCModule): """ 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 """ @@ -58,18 +66,11 @@ class PlaylistAdvisor: Whether to play a new playlist, if this is True, then the player will refresh and fetch a new playlist, calling advise """ return False - def imc(self, imc: 'InterModuleCommunication'): - """ - Receive an IMC object - """ - pass - def imc_data(self, source: 'PlayerModule | ActiveModifier | PlaylistAdvisor', data: object, broadcast: bool) -> object: - return None -class ActiveModifier: +class ActiveModifier(BaseIMCModule): """ 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) -> None: """ Called at start up with the program arguments """ @@ -80,35 +81,35 @@ class ActiveModifier: 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[Track]): + def on_new_playlist(self, playlist: list[Track]) -> None: """ 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, broadcast: bool) -> object: - return None 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: + self.names_modules: dict[str, BaseIMCModule] = {} + def broadcast(self, source: BaseIMCModule, data: object) -> None: """ Send data to all modules, other than ourself """ - if source is not self.advisor: self.advisor.imc_data(source, data, True) - if self.active_modifier and source is not self.active_modifier: self.active_modifier.imc_data(source, data, True) - for module in [f for f in self.simple_modules if f is not source]: module.imc_data(source, data, True) - def register(self, module: PlaylistAdvisor | ActiveModifier | PlayerModule, name: str): + source_name = next((k for k, v in self.names_modules.items() if v is source), None) + if source is not self.advisor: self.advisor.imc_data(source, source_name, data, True) + if self.active_modifier and source is not self.active_modifier: self.active_modifier.imc_data(source, source_name, data, True) + for module in [f for f in self.simple_modules if f is not source]: module.imc_data(source, source_name, data, True) + def register(self, module: BaseIMCModule, name: str) -> bool: + """ + Register our module with a name, so we can be sent data via the send function + """ 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: + def send(self, source: BaseIMCModule, name: str, data: object) -> object: + """ + Sends the data to a named module, and return its response + """ if not name in self.names_modules.keys(): raise Exception - return self.names_modules[name].imc_data(source, 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) \ No newline at end of file diff --git a/modules/advisor.py b/modules/advisor.py index 2be752e..e02d0c0 100644 --- a/modules/advisor.py +++ b/modules/advisor.py @@ -110,7 +110,7 @@ class Module(PlaylistAdvisor): def imc(self, imc: InterModuleCommunication): self.class_imc = imc imc.register(self, "advisor") - def imc_data(self, source: PlayerModule | ActiveModifier | PlaylistAdvisor, data: object, broadcast: bool): + def imc_data(self, source: PlayerModule | ActiveModifier | PlaylistAdvisor, source_name: str | None, data: object, broadcast: bool): return self.custom_playlist advisor = Module() \ No newline at end of file diff --git a/modules/modules.txt b/modules/modules.txt index 1469bd5..11f600a 100644 --- a/modules/modules.txt +++ b/modules/modules.txt @@ -16,36 +16,49 @@ class Track: args: dict[str, str] | None offset: float = 0.0 -class PlayerModule: +class BaseIMCModule: """ - Simple passive observer, this allows you to send the current track the your RDS encoder, or to your website + This is not a module to be used but rather a placeholder IMC api to be used in other modules """ - def on_new_playlist(self, playlist: list[Track]): - """This is called every new playlist""" - pass - 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'): + def imc(self, imc: 'InterModuleCommunication') -> None: """ Receive an IMC object """ pass - def imc_data(self, source: 'PlayerModule | ActiveModifier | PlaylistAdvisor', data: object, broadcast: bool) -> object: + def imc_data(self, source: 'BaseIMCModule', source_name: str | None, data: object, broadcast: bool) -> object: + """ + React to IMC data + """ return None + +class PlayerModule(BaseIMCModule): + """ + 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[Track]) -> None: + """This is called every new playlist""" + pass + def on_new_track(self, index: int, track: Track) -> None: + """ + 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 progress(self, index: int, track: Track, elapsed: float, total: float, real_total: float) -> None: + """ + Real total and total differ in that, total is how much the track lasts, but real_total will be for how long we will play it for + """ + 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[Track]): + def modify(self, global_args: dict, playlist: list[Track]) -> list[Track] | None: """ global_args are playlist global args (see radioPlayer_playlist_file.txt) """ return playlist # No IMC, as we only run on new playlists -class PlaylistAdvisor: +class PlaylistAdvisor(BaseIMCModule): """ 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 """ @@ -59,18 +72,11 @@ class PlaylistAdvisor: Whether to play a new playlist, if this is True, then the player will refresh and fetch a new playlist, calling advise """ return False - def imc(self, imc: 'InterModuleCommunication'): - """ - Receive an IMC object - """ - pass - def imc_data(self, source: 'PlayerModule | ActiveModifier | PlaylistAdvisor', data: object, broadcast: bool) -> object: - return None -class ActiveModifier: +class ActiveModifier(BaseIMCModule): """ 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) -> None: """ Called at start up with the program arguments """ @@ -81,38 +87,38 @@ class ActiveModifier: 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[Track]): + def on_new_playlist(self, playlist: list[Track]) -> None: """ 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, broadcast: bool) -> object: - return None 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: + self.names_modules: dict[str, BaseIMCModule] = {} + def broadcast(self, source: BaseIMCModule, data: object) -> None: """ Send data to all modules, other than ourself """ - if source is not self.advisor: self.advisor.imc_data(source, data, True) - if self.active_modifier and source is not self.active_modifier: self.active_modifier.imc_data(source, data, True) - for module in [f for f in self.simple_modules if f is not source]: module.imc_data(source, data, True) - def register(self, module: PlaylistAdvisor | ActiveModifier | PlayerModule, name: str): + source_name = next((k for k, v in self.names_modules.items() if v is source), None) + if source is not self.advisor: self.advisor.imc_data(source, source_name, data, True) + if self.active_modifier and source is not self.active_modifier: self.active_modifier.imc_data(source, source_name, data, True) + for module in [f for f in self.simple_modules if f is not source]: module.imc_data(source, source_name, data, True) + def register(self, module: BaseIMCModule, name: str) -> bool: + """ + Register our module with a name, so we can be sent data via the send function + """ 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: + def send(self, source: BaseIMCModule, name: str, data: object) -> object: + """ + Sends the data to a named module, and return its response + """ if not name in self.names_modules.keys(): raise Exception - return self.names_modules[name].imc_data(source, 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) ``` 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/radioPlayer.py b/radioPlayer.py index 429d846..863d4f9 100644 --- a/radioPlayer.py +++ b/radioPlayer.py @@ -161,7 +161,7 @@ def play_playlist(playlist_path, starting_index: int = 0): for (lns, args) in parsed: playlist.extend([Track(line, True, True, True, args) for line in lns]) - for module in playlist_modifier_modules: playlist = module.modify(global_args, playlist) + for module in playlist_modifier_modules: playlist = module.modify(global_args, playlist) or playlist for module in simple_modules: module.on_new_playlist(playlist) if active_modifier: active_modifier.on_new_playlist(playlist)