minor internal changes

This commit is contained in:
2025-12-12 20:42:38 +01:00
parent 3afb96a441
commit 06b983d3cd
2 changed files with 17 additions and 50 deletions
+8 -34
View File
@@ -118,23 +118,16 @@ def websocket_server_process(shared_data: dict, imc_q: multiprocessing.Queue, ws
# start server # start server
server = await websockets.serve(handler_wrapper, "0.0.0.0", 3001) server = await websockets.serve(handler_wrapper, "0.0.0.0", 3001)
# background task: broadcast worker
broadcaster = asyncio.create_task(broadcast_worker(ws_q, clients)) broadcaster = asyncio.create_task(broadcast_worker(ws_q, clients))
# run forever until server closes
await server.wait_closed() await server.wait_closed()
# ensure broadcaster stops
ws_q.put(None) ws_q.put(None)
await broadcaster await broadcaster
# On SIGINT/SIGTERM, stop gracefully
loop = asyncio.new_event_loop() loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop) asyncio.set_event_loop(loop)
try: try: loop.run_until_complete(runner())
loop.run_until_complete(runner()) except (KeyboardInterrupt, SystemExit): pass
except (KeyboardInterrupt, SystemExit): finally: loop.close()
pass
finally:
loop.close()
# ---------- Module class (drop-in replacement) ---------- # ---------- Module class (drop-in replacement) ----------
@@ -143,29 +136,21 @@ class Module(PlayerModule):
self.manager = multiprocessing.Manager() self.manager = multiprocessing.Manager()
self.data = self.manager.dict() self.data = self.manager.dict()
self.imc_q = self.manager.Queue() self.imc_q = self.manager.Queue()
# queue for sending broadcasts to the websocket process
self.ws_q = self.manager.Queue() self.ws_q = self.manager.Queue()
# initial state
self.data["playlist"] = "[]" self.data["playlist"] = "[]"
self.data["track"] = "{}" self.data["track"] = "{}"
self.data["progress"] = "{}" self.data["progress"] = "{}"
# ipc thread: listens for responses from other modules (same as before)
self.ipc_thread_running = True self.ipc_thread_running = True
self.ipc_thread = threading.Thread(target=self._ipc_worker, daemon=True) self.ipc_thread = threading.Thread(target=self._ipc_worker, daemon=True)
self.ipc_thread.start() self.ipc_thread.start()
# start websocket server process self.ws_process = multiprocessing.Process(target=websocket_server_process, args=(self.data, self.imc_q, self.ws_q), daemon=False)
self.ws_process = multiprocessing.Process(
target=websocket_server_process, args=(self.data, self.imc_q, self.ws_q), daemon=False
)
self.ws_process.start() self.ws_process.start()
if os.name == "posix": if os.name == "posix":
try: try: os.setpgid(self.ws_process.pid, self.ws_process.pid)
os.setpgid(self.ws_process.pid, self.ws_process.pid) except Exception: pass
except Exception:
pass
def _ipc_worker(self): def _ipc_worker(self):
""" """
@@ -176,17 +161,10 @@ class Module(PlayerModule):
try: try:
message: dict | None = self.imc_q.get() message: dict | None = self.imc_q.get()
if message is None: break if message is None: break
# send to upper layer (existing player IPC)
out = self._imc.send(self, message["name"], message["data"]) out = self._imc.send(self, message["name"], message["data"])
# if message had a key, store the response for the requester if key := message.get("key", None): self.data[key] = out
if key := message.get("key", None): except Exception: pass
# store response into shared dict (accessible to ws process)
self.data[key] = out
except Exception:
# swallow errors to avoid killing the ipc thread
pass
# The following functions update the shared_data and also push a broadcast message onto ws_q
def on_new_playlist(self, playlist: list[Track]) -> None: def on_new_playlist(self, playlist: list[Track]) -> None:
api_data = [] api_data = []
for track in playlist: for track in playlist:
@@ -199,7 +177,6 @@ class Module(PlayerModule):
"offset": track.offset "offset": track.offset
}) })
self.data["playlist"] = json.dumps(api_data) self.data["playlist"] = json.dumps(api_data)
# broadcast
try: self.ws_q.put({"event": "playlist", "data": api_data}) try: self.ws_q.put({"event": "playlist", "data": api_data})
except Exception: pass except Exception: pass
@@ -216,18 +193,15 @@ class Module(PlayerModule):
track_data = {"path": str(track.path), "fade_out": track.fade_out, "fade_in": track.fade_in, "official": track.official, "args": track.args, "offset": track.offset} track_data = {"path": str(track.path), "fade_out": track.fade_out, "fade_in": track.fade_in, "official": track.official, "args": track.args, "offset": track.offset}
payload = {"index": index, "track": track_data, "elapsed": elapsed, "total": total, "real_total": real_total} payload = {"index": index, "track": track_data, "elapsed": elapsed, "total": total, "real_total": real_total}
self.data["progress"] = json.dumps(payload) self.data["progress"] = json.dumps(payload)
# For frequent progress updates you might want to rate-limit; this pushes every call
try: self.ws_q.put({"event": "progress", "data": payload}) try: self.ws_q.put({"event": "progress", "data": payload})
except Exception: pass except Exception: pass
def shutdown(self): def shutdown(self):
# stop ipc thread
self.ipc_thread_running = False self.ipc_thread_running = False
try: self.imc_q.put(None) try: self.imc_q.put(None)
except Exception: pass except Exception: pass
self.ipc_thread.join(timeout=2) self.ipc_thread.join(timeout=2)
# shutdown websocket process by putting sentinel into ws_q and then terminating if needed
try: self.ws_q.put(None) try: self.ws_q.put(None)
except Exception: pass except Exception: pass
+9 -16
View File
@@ -1,6 +1,6 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
import os, subprocess, importlib.util, importlib.machinery, types import os, subprocess, importlib.util, importlib.machinery, types
import sys, signal, glob, time, traceback, atexit import sys, signal, glob, time, traceback
import libcache import libcache
from modules import * from modules import *
from threading import Lock from threading import Lock
@@ -117,8 +117,7 @@ class RadioPlayer:
self.playlist_advisor: PlaylistAdvisor | None = None self.playlist_advisor: PlaylistAdvisor | None = None
self.active_modifier: ActiveModifier | None = None self.active_modifier: ActiveModifier | None = None
self.exit_pending = False self.exit_pending = False
self.exit_status_code = 0 self.exit_status_code = self.intr_time = 0
self.intr_time = 0
self.exit_lock = Lock() self.exit_lock = Lock()
self.procman = ProcessManager() self.procman = ProcessManager()
self.modules: list[tuple[importlib.machinery.ModuleSpec, types.ModuleType, str]] = [] self.modules: list[tuple[importlib.machinery.ModuleSpec, types.ModuleType, str]] = []
@@ -132,19 +131,16 @@ class RadioPlayer:
for module in self.simple_modules: for module in self.simple_modules:
if module: if module:
try: module.shutdown() try: module.shutdown()
except Exception: except Exception: traceback.print_exc(file=self.logger.output)
traceback.print_last(file=self.logger.output)
self.logger.error("Exception while shutting down module.")
self.logger.output.close() self.logger.output.close()
def handle_sigint(self, signum: int, frame: types.FrameType | None): def handle_sigint(self, signum: int, frame: types.FrameType | None):
with self.exit_lock: with self.exit_lock:
self.logger.info("Received CTRL+C (SIGINT)") self.logger.info("Received CTRL+C (SIGINT)")
if (time.monotonic() - self.intr_time) > 5: if (now := time.monotonic()) and ((now - self.intr_time) > 5):
self.intr_time = time.monotonic() self.intr_time = now
self.logger.info("Will quit on song end.") self.logger.info("Will quit on song end.")
self.exit_pending = True self.exit_pending, self.exit_status_code = True, 130
self.exit_status_code = 130
else: else:
self.logger.warning("Force-Quit pending") self.logger.warning("Force-Quit pending")
raise SystemExit(130) raise SystemExit(130)
@@ -178,8 +174,7 @@ class RadioPlayer:
try: try:
start = time.perf_counter() start = time.perf_counter()
spec.loader.exec_module(module) spec.loader.exec_module(module)
time_took = time.perf_counter() - start if (time_took := time.perf_counter() - start) > 0.15: self.logger.warning(f"{module_name} took {time_took:.1f}s to start")
if time_took > 0.15: self.logger.warning(f"{module_name} took {time_took:.1f}s to start")
except Exception as e: except Exception as e:
traceback.print_exc(file=self.logger.output) traceback.print_exc(file=self.logger.output)
self.logger.error(f"Failed loading {module_name} due to {e}, continuing") self.logger.error(f"Failed loading {module_name} due to {e}, continuing")
@@ -235,7 +230,7 @@ class RadioPlayer:
playlist = None playlist = None
global_args = {} global_args = {}
return_pending = track = False return_pending = track = False
cross_fade = int(global_args.get("crossfade", 5)) cross_fade = int(global_args.get("crossfade", 5)) # TODO: get rid of global_args usage in the core and instead just store fades in track (that would require a pretty much rewrite of the active modifier tho...)
song_i = i = 0 song_i = i = 0
def get_track(): def get_track():
@@ -287,9 +282,7 @@ class RadioPlayer:
while end_time >= time.monotonic() and pr.process.poll() is None: while end_time >= time.monotonic() and pr.process.poll() is None:
start = time.monotonic() start = time.monotonic()
[module.progress(song_i, track, time.monotonic() - pr.started_at, pr.duration, end_time - pr.started_at) for module in self.simple_modules if module] [module.progress(song_i, track, time.monotonic() - pr.started_at, pr.duration, end_time - pr.started_at) for module in self.simple_modules if module]
elapsed = time.monotonic() - start if (elapsed := time.monotonic() - start) < 1 and (remaining_until_end := end_time - time.monotonic()) > 0: time.sleep(min(1 - elapsed, remaining_until_end))
remaining_until_end = end_time - time.monotonic()
if elapsed < 1 and remaining_until_end > 0: time.sleep(min(1 - elapsed, remaining_until_end))
i += 1 i += 1
if not extend: song_i += 1 if not extend: song_i += 1