radio95/RadioPlayer
0
1
Fork 0
mirror of https://github.com/radio95-rnt/RadioPlayer.git synced 2026-02-26 21:53:54 +01:00
Code Issues Activity

some changes, also change licence to the UNLICENCE!

Browse Source
This commit is contained in:
KubaPro010
2025-12-02 17:42:11 +01:00
parent 75802ccef4
commit 88b44ddf32
9 changed files with 67 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
modules/__init__.py
View File

@@ -140,14 +140,14 @@ class InterModuleCommunication:
def __init__(self, modules: Sequence[BaseIMCModule | None]) -> None:
self.modules = modules
self.names_modules: dict[str, BaseIMCModule] = {}
for module in modules:
for module in modules:
if module: module.imc(self)
def broadcast(self, source: BaseIMCModule, data: object) -> None:
"""
Send data to all modules, other than ourself
"""
source_name = next((k for k, v in self.names_modules.items() if v is source), None)
for module in [f for f in self.modules if f is not source]:
for module in [f for f in self.modules if f is not source]:
if module: module.imc_data(source, source_name, data, True)
def register(self, module: BaseIMCModule, name: str) -> bool:
"""

9
modules/active_modifier.py
View File

@@ -38,7 +38,6 @@ class Module(ActiveModifier):
if song.startswith("!"):
song = song[1:]
official = False
return Path(song).absolute(), official
if len(songs):
@@ -48,7 +47,7 @@ class Module(ActiveModifier):
else:
if (index - 1) >= 0: 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) and next_track: next_track_to_fade_in = next_track.fade_in
@@ -57,7 +56,7 @@ class Module(ActiveModifier):
if not self.originals or self.originals[-1] != track: self.originals.append(track)
with open("/tmp/radioPlayer_toplay", "w") as f:
with open("/tmp/radioPlayer_toplay", "w") as f:
f.write('\n'.join(songs))
f.write("\n")
@@ -73,14 +72,14 @@ class Module(ActiveModifier):
next_track = track
self.limit_tracks = False
return (self.last_track, next_track), True
elif len(self.originals):
elif len(self.originals):
self.last_track = self.originals.pop(0)
if len(self.originals): next_track = self.originals[0]
else: self.last_track = track
self.limit_tracks = self.can_limit_tracks
if self.limit_tracks:
last_track_duration = self._imc.send(self, "procman", {"op": 1, "arg": self.last_track.path})
last_track_duration = self._imc.send(self, "procman", {"op": 1, "arg": self.last_track.path}) # Ask procman for the duration of this file
assert isinstance(last_track_duration, dict)
last_track_duration = last_track_duration.get("arg")
if last_track_duration:

4
modules/advisor.py
View File

@@ -65,7 +65,7 @@ class Module(PlaylistAdvisor):
self.custom_playlist_last_mod = Time.get_playlist_modification_time(self.custom_playlist)
return self.custom_playlist
elif self.custom_playlist: self.custom_playlist = None
current_day, current_hour = (time := datetime.datetime.now()).strftime('%A').lower(), time.hour
morning_playlist = Path(playlist_dir, current_day, "morning").absolute()
@@ -106,7 +106,7 @@ class Module(PlaylistAdvisor):
self.last_playlist = night_playlist
return self.last_playlist
def new_playlist(self) -> bool:
if self.custom_playlist and self.custom_playlist_path.exists():
if self.custom_playlist and self.custom_playlist_path.exists():
if Time.get_playlist_modification_time(self.custom_playlist) > self.custom_playlist_last_mod:
logger.info("Custom playlist changed on disc, reloading...")
self.custom_playlist = None

2
modules/progress.py
View File

@@ -8,7 +8,7 @@ def format_time(seconds) -> str:
class Module(PlayerModule):
def progress(self, index: int, track: Track, elapsed: float, total: float, real_total: float) -> None:
if track.official:
if track.official:
data = f"{track.path.name}: {format_time(elapsed)} / {format_time(total)}\n"
Path("/tmp/radioPlayer_progress").write_text(data)

8
modules/rds.py
View File

@@ -38,7 +38,7 @@ def update_rds(track_name: str):
except KeyError:
has_name = False
name = track_name.rsplit(".", 1)[0]
name = re.sub(r'^\s*\d+\s*[-.]?\s*', '', name)
if " - " in name:
@@ -52,9 +52,9 @@ def update_rds(track_name: str):
artist = rds_default_artist
title = name
if not has_name: logger.warning(f"File does not have a alias in the name table ({track_name})")
# title = re.sub(r'\s*[\(\[][^\(\)\[\]]*[\)\]]', '', title) # there might be junk
prt = rds_base.format(artist, title)
rtp = [4] # type 1
rtp.append(prt.find(artist)) # start 1
@@ -63,7 +63,7 @@ def update_rds(track_name: str):
rtp.append(prt.find(title)) # start 2
rtp.append(len(title) - 1) # len 2
try:
try:
f = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
f.settimeout(1.0)
data = f"TEXT={prt}\r\nRTP={rtp}\r\n".encode()

4
modules/write_playlists.py
View File

@@ -12,7 +12,7 @@ class Module(PlayerModule):
def on_new_playlist(self, playlist: list[Track]): self.playlist = [str(t.path.absolute()) for t in playlist]
def progress(self, index: int, track: Track, elapsed: float, total: float, real_total: float) -> None:
if os.path.exists("/tmp/radioPlayer_skip"):
self._imc.send(self, "procman", {"op": 2})
self._imc.send(self, "procman", {"op": 2}) # Ask procman to kill every track playing (usually there is one, unless we are in the default 5 seconds of the crossfade)
os.remove("/tmp/radioPlayer_skip")
def on_new_track(self, index: int, track: Track, next_track: Track | None):
if next_track: logger.info("Next up:", next_track.path.name)
@@ -22,7 +22,7 @@ class Module(PlayerModule):
lines = self.playlist[:index] + [f"> ({track.path})"] + [self.playlist[index]] + self.playlist[index+1:]
else: lines = self.playlist[:index] + [f"> {self.playlist[index]}"] + self.playlist[index+1:]
with open("/tmp/radioPlayer_playlist", "w") as f:
for line in lines:
for line in lines:
try: f.write(line + "\n")
except UnicodeEncodeError:
print(line.encode('utf-8', errors='ignore').decode('utf-8'))