diff --git a/modules/web.py b/modules/web.py index 148beb4..c1ba288 100644 --- a/modules/web.py +++ b/modules/web.py @@ -19,111 +19,149 @@ from . import Track, PlayerModule, Path, BaseIMCModule MAIN_PATH_DIR = Path("/home/user/mixes") -async def ws_handler(websocket: ServerConnection, shared_data: dict, imc_q: multiprocessing.Queue, ws_q: multiprocessing.Queue, writer_q: asyncio.Queue): +# locks: dict[lock_id, websocket | None] — managed inside the async runner's scope +# passed into handlers via a shared dict reference + +async def ws_handler(websocket: ServerConnection, shared_data: dict, imc_q: multiprocessing.Queue, writer_q: asyncio.Queue, locks: dict, clients: set): try: initial = { "track": json.loads(shared_data.get("track", "{}")), "dirs": {"files": [i.name for i in list(MAIN_PATH_DIR.iterdir()) if i.is_file()], "dirs": [i.name for i in list(MAIN_PATH_DIR.iterdir()) if i.is_dir()], "base": str(MAIN_PATH_DIR)}, + "locks": {lid: True for lid, owner in locks.items() if owner is not None}, } - except Exception: initial = {"track": {}, "dirs": {}} + except Exception: initial = {"track": {}, "dirs": {}, "locks": {}} await websocket.send(json.dumps({"event": "state", "data": initial})) await websocket.send(json.dumps({"event": "playlist", "data": json.loads(shared_data.get("playlist", "[]"))})) await websocket.send(json.dumps({"event": "rds", "data": json.loads(shared_data.get("rds", "{}"))})) - async for raw in websocket: - try: msg: dict = json.loads(raw) - except Exception: - await websocket.send(json.dumps({"error": "invalid json"})) - continue + async def get_imc(name, data): + key = str(uuid.uuid4()) + imc_q.put({"name": name, "data": data, "key": key}) + start = time.monotonic() + while time.monotonic() - start < 1: + if key in shared_data: return shared_data.pop(key) + await asyncio.sleep(0.1) + return None - async def get_imc(name, data): - key = str(uuid.uuid4()) - imc_q.put({"name": name, "data": data, "key": key}) - start = time.monotonic() - while time.monotonic() - start < 1: - if key in shared_data: return shared_data.pop(key) - await asyncio.sleep(0.1) - return None + async def broadcast(payload: dict): + msg = json.dumps(payload) + for ws in list(clients): + try: await ws.send(msg) + except Exception: pass - action = msg.get("action") - if action == "skip": - imc_q.put({"name": "procman", "data": {"op": 2}}) - await websocket.send(json.dumps({"event": "skip"})) - elif action == "add_to_toplay": - songs = msg.get("songs") - at_top = msg.get("top", False) - if not isinstance(songs, list): await websocket.send(json.dumps({"error": "songs must be a list"})) - else: - imc_q.put({"name": "activemod", "data": {"action": "add_to_toplay", "songs": songs, "top": at_top}}) + try: + async for raw in websocket: + try: msg: dict = json.loads(raw) + except Exception: + await websocket.send(json.dumps({"error": "invalid json"})) + continue + action = msg.get("action") + if action == "skip": + imc_q.put({"name": "procman", "data": {"op": 2}}) + await websocket.send(json.dumps({"event": "skip"})) + elif action == "add_to_toplay": + songs = msg.get("songs") + at_top = msg.get("top", False) + if not isinstance(songs, list): await websocket.send(json.dumps({"error": "songs must be a list"})) + else: + imc_q.put({"name": "activemod", "data": {"action": "add_to_toplay", "songs": songs, "top": at_top}}) + result = await get_imc("activemod", {"action": "get_toplay"}) + if result is not None: + await broadcast({"data": result, "event": "toplay"}) + elif action == "get_toplay": result = await get_imc("activemod", {"action": "get_toplay"}) - if result is not None: - await asyncio.get_event_loop().run_in_executor(None, ws_q.put, {"data": result, "event": "toplay"}) - elif action == "get_toplay": - result = await get_imc("activemod", {"action": "get_toplay"}) - if result is None: await websocket.send(json.dumps({"error": "timeout", "code": 504})) - else: await websocket.send(json.dumps({"data": result, "event": "toplay"})) - elif action == "clear_toplay": - result = await get_imc("activemod", {"action": "clear_toplay"}) - if result is None: await websocket.send(json.dumps({"error": "timeout", "code": 504})) - else: - await asyncio.get_event_loop().run_in_executor(None, ws_q.put, {"data": result, "event": "toplay"}) - elif action == "skipc" or action == "skipi": - result = await get_imc("activemod", msg) - if result is None: await websocket.send(json.dumps({"error": "timeout", "code": 504})) - else: - await asyncio.get_event_loop().run_in_executor(None, ws_q.put, {"data": result, "event": action}) - elif action == "jingle": - result = await get_imc("jingle", msg.get("top", False)) - if result is None: await websocket.send(json.dumps({"error": "timeout", "code": 504})) - else: - await websocket.send(json.dumps(result)) - result = await get_imc("activemod", {"action": "get_toplay"}) - if result is not None: - await asyncio.get_event_loop().run_in_executor(None, ws_q.put, {"data": result, "event": "toplay"}) - elif action == "request_dir": - what: str = msg.get("what", "") - try: - dir = Path(MAIN_PATH_DIR, what).resolve() - payload = {"files": [i.name for i in list(dir.iterdir()) if i.is_file()], "base": str(dir), "dir": dir.name} - except Exception: payload = {} - await websocket.send(json.dumps({"event": "request_dir", "data": payload})) - elif action == "fsdb_add": - name: str | None = msg.get("name") - try: - if not name: raise Exception("name not defined") - path = Path(MAIN_PATH_DIR, ".playlist", msg.get("playlist", ""), name) - path.touch(exist_ok=True) - await websocket.send(json.dumps({"event": "fsdb_add", "ok": True})) - except Exception as e: await websocket.send(json.dumps({"event": "fsdb_add", "error": str(e)})) - elif action == "fsdb_add_dir": - name: str | None = msg.get("name") - try: - if not name: raise Exception("name not defined") - path = Path(MAIN_PATH_DIR, ".playlist", msg.get("playlist", ""), name) - path.mkdir(parents=True, exist_ok=True) - await websocket.send(json.dumps({"event": "fsdb_add_dir", "ok_dir": True})) - except Exception as e: await websocket.send(json.dumps({"event": "fsdb_add_dir", "error": str(e)})) - elif action == "fsdb_remove": - name: str | None = msg.get("name") - try: - if not name: raise Exception("name not defined") - path = Path(MAIN_PATH_DIR, ".playlist", msg.get("playlist", ""), name) - if path.is_dir(): shutil.rmtree(path) - else: path.unlink(missing_ok=True) - await websocket.send(json.dumps({"event": "fsdb_remove", "ok": True})) - except Exception as e: await websocket.send(json.dumps({"event": "fsdb_remove", "error": str(e)})) - elif action == "fsdb_list": - try: - p = Path(MAIN_PATH_DIR, ".playlist", msg.get("playlist", "")) - payload = { - "files": [i.name for i in p.iterdir() if i.is_file()], - "dirs": [i.name for i in p.iterdir() if i.is_dir()] - } - await websocket.send(json.dumps({"event": "fsdb_list", "data": payload})) - except Exception as e: await websocket.send(json.dumps({"event": "fsdb_list", "data": {}, "error": str(e)})) - elif action == "fm95": await writer_q.put((base64.b64decode(msg.get("data", "")), websocket)) - else: await websocket.send(json.dumps({"event": "error", "error": "unknown action"})) + if result is None: await websocket.send(json.dumps({"error": "timeout", "code": 504})) + else: await websocket.send(json.dumps({"data": result, "event": "toplay"})) + elif action == "clear_toplay": + result = await get_imc("activemod", {"action": "clear_toplay"}) + if result is None: await websocket.send(json.dumps({"error": "timeout", "code": 504})) + else: + await broadcast({"data": result, "event": "toplay"}) + elif action == "skipc" or action == "skipi": + result = await get_imc("activemod", msg) + if result is None: await websocket.send(json.dumps({"error": "timeout", "code": 504})) + else: + await broadcast({"data": result, "event": action}) + elif action == "jingle": + result = await get_imc("jingle", msg.get("top", False)) + if result is None: await websocket.send(json.dumps({"error": "timeout", "code": 504})) + else: + await websocket.send(json.dumps(result)) + result = await get_imc("activemod", {"action": "get_toplay"}) + if result is not None: + await broadcast({"data": result, "event": "toplay"}) + elif action == "request_dir": + what: str = msg.get("what", "") + try: + dir = Path(MAIN_PATH_DIR, what).resolve() + payload = {"files": [i.name for i in list(dir.iterdir()) if i.is_file()], "base": str(dir), "dir": dir.name} + except Exception: payload = {} + await websocket.send(json.dumps({"event": "request_dir", "data": payload})) + elif action == "fsdb_add": + name: str | None = msg.get("name") + try: + if not name: raise Exception("name not defined") + path = Path(MAIN_PATH_DIR, ".playlist", msg.get("playlist", ""), name) + path.touch(exist_ok=True) + await websocket.send(json.dumps({"event": "fsdb_add", "ok": True})) + except Exception as e: await websocket.send(json.dumps({"event": "fsdb_add", "error": str(e)})) + elif action == "fsdb_add_dir": + name: str | None = msg.get("name") + try: + if not name: raise Exception("name not defined") + path = Path(MAIN_PATH_DIR, ".playlist", msg.get("playlist", ""), name) + path.mkdir(parents=True, exist_ok=True) + await websocket.send(json.dumps({"event": "fsdb_add_dir", "ok_dir": True})) + except Exception as e: await websocket.send(json.dumps({"event": "fsdb_add_dir", "error": str(e)})) + elif action == "fsdb_remove": + name: str | None = msg.get("name") + try: + if not name: raise Exception("name not defined") + path = Path(MAIN_PATH_DIR, ".playlist", msg.get("playlist", ""), name) + if path.is_dir(): shutil.rmtree(path) + else: path.unlink(missing_ok=True) + await websocket.send(json.dumps({"event": "fsdb_remove", "ok": True})) + except Exception as e: await websocket.send(json.dumps({"event": "fsdb_remove", "error": str(e)})) + elif action == "fsdb_list": + try: + p = Path(MAIN_PATH_DIR, ".playlist", msg.get("playlist", "")) + payload = { + "files": [i.name for i in p.iterdir() if i.is_file()], + "dirs": [i.name for i in p.iterdir() if i.is_dir()] + } + await websocket.send(json.dumps({"event": "fsdb_list", "data": payload})) + except Exception as e: await websocket.send(json.dumps({"event": "fsdb_list", "data": {}, "error": str(e)})) + elif action == "fm95": await writer_q.put((base64.b64decode(msg.get("data", "")), websocket)) + elif action == "lock": + lid = msg.get("id") + if lid is None: await websocket.send(json.dumps({"event": "error", "error": "lock id required"})) + elif locks.get(lid) not in (None, websocket): + # already held by someone else + await websocket.send(json.dumps({"event": "lock", "id": lid, "data": True, "error": "already locked"})) + else: + locks[lid] = websocket + await broadcast({"event": "lock", "id": lid, "data": True}) + + elif action == "unlock": + lid = msg.get("id") + if lid is None: + await websocket.send(json.dumps({"event": "error", "error": "lock id required"})) + elif locks.get(lid) is not websocket: + # not the owner + await websocket.send(json.dumps({"event": "error", "error": "not lock owner"})) + else: + locks[lid] = None + await broadcast({"event": "lock", "id": lid, "data": False}) + + else: await websocket.send(json.dumps({"event": "error", "error": "unknown action"})) + finally: + # release every lock this client held on disconnect + for lid, owner in list(locks.items()): + if owner is websocket: + locks[lid] = None + await broadcast({"event": "lock", "id": lid, "data": False}) + async def broadcast_worker(ws_q: multiprocessing.Queue, clients: set): loop = asyncio.get_event_loop() @@ -164,7 +202,9 @@ async def socket_handler(socket: asyncio.StreamReader, writer: asyncio.StreamWri def websocket_server_process(shared_data: dict, imc_q: multiprocessing.Queue, ws_q: multiprocessing.Queue): async def runner(): - clients = set() + clients: set[ServerConnection] = set() + locks: dict[int, ServerConnection | None] = {} # lock_id -> owning websocket or None + reader, writer = await asyncio.open_unix_connection("/etc/fm95/ctl.socket") # pyright: ignore[reportAttributeAccessIssue] reader: asyncio.StreamReader writer: asyncio.StreamWriter @@ -173,11 +213,12 @@ def websocket_server_process(shared_data: dict, imc_q: multiprocessing.Queue, ws async def handler_wrapper(websocket: ServerConnection): clients.add(websocket) await asyncio.get_event_loop().run_in_executor(None, ws_q.put, {"event": "users", "data": len(clients)}) - try: await ws_handler(websocket, shared_data, imc_q, ws_q, writer_q) + try: await ws_handler(websocket, shared_data, imc_q, writer_q, locks, clients) finally: await websocket.close(1001, "") clients.discard(websocket) await asyncio.get_event_loop().run_in_executor(None, ws_q.put, {"event": "users", "data": len(clients)}) + async def process_request(websocket: ServerConnection, request: Request): if request.path == "/ws": if not "upgrade" in request.headers.get("Connection", "").lower(): @@ -191,28 +232,13 @@ def websocket_server_process(shared_data: dict, imc_q: multiprocessing.Queue, ws else: if request.path == "/" and (file := Path(__file__, "..", "web", "index.html").resolve()).exists(): data = file.read_bytes() - return Response( - 200, - "OK", - Headers([("Content-Type", "text/html; charset=utf-8"), ("Content-Length", f"{len(data)}")]), - data - ) + return Response(200, "OK", Headers([("Content-Type", "text/html; charset=utf-8"), ("Content-Length", f"{len(data)}")]), data) elif (file := Path(__file__, "..", "web", request.path.removeprefix("/").strip()).resolve()).exists(): data = file.read_bytes() - return Response( - 200, - "OK", - Headers([("Content-Type", get_content_type(file.name)), ("Content-Length", f"{len(data)}")]), - data - ) + return Response(200, "OK", Headers([("Content-Type", get_content_type(file.name)), ("Content-Length", f"{len(data)}")]), data) else: data = b"Not Found\n" - return Response( - 404, - "Not Found", - Headers([("Content-Length", f"{len(data)}")]), - data - ) + return Response(404, "Not Found", Headers([("Content-Length", f"{len(data)}")]), data) server = await websockets.serve(handler_wrapper, "0.0.0.0", 3001, server_header="RadioPlayer ws plugin", process_request=process_request) broadcaster = asyncio.create_task(broadcast_worker(ws_q, clients)) @@ -251,7 +277,6 @@ class Module(PlayerModule): self.ws_process = multiprocessing.Process(target=websocket_server_process, args=(self.data, self.imc_q, self.ws_q), daemon=False) self.ws_process.start() if os.name == "posix": - # We have to manage the process ourselves. Linux sends signals to everyone in the process group, meaning a CTRL+C would not kill the main one, but this one would die try: os.setpgid(self.ws_process.pid, self.ws_process.pid) except Exception: pass @@ -268,15 +293,7 @@ class Module(PlayerModule): def on_new_playlist(self, playlist: list[Track], global_args: dict[str, str]) -> None: api_data = [] for track in playlist: - api_data.append({ - "path": str(track.path), - "fade_out": track.fade_out, - "fade_in": track.fade_in, - "official": track.official, - "args": track.args, - "offset": track.offset, - "focus_time_offset": track.focus_time_offset - }) + api_data.append({"path": str(track.path), "fade_out": track.fade_out, "fade_in": track.fade_in, "official": track.official, "args": track.args, "offset": track.offset, "focus_time_offset": track.focus_time_offset}) output_data = {"playlist": api_data, "global_args": global_args} self.data["playlist"] = json.dumps(output_data) try: self.ws_q.put({"event": "playlist", "data": output_data}) diff --git a/modules/web/web.js b/modules/web/web.js index 747028f..c3cf29a 100644 --- a/modules/web/web.js +++ b/modules/web/web.js @@ -17,6 +17,7 @@ let skippedIndices = []; let lastElapsed = 0; let lastUpdateTime = 0; let currentRealTotal = 1; +let pollLockHeld = false; function toggleSection(id) { document.getElementById(id).classList.toggle("collapsed"); @@ -63,9 +64,6 @@ function connectWs() { ws.addEventListener("open", () => { statusEl.textContent = "connected"; reconnectDelay = 1500; - ws.send(JSON.stringify({ action: "get_toplay" })); - ws.send(JSON.stringify({ action: "skipc" })); - ws.send(JSON.stringify({ action: "skipi" })); }); ws.addEventListener("close", () => { @@ -88,12 +86,29 @@ function wsSend(obj) { ws.send(JSON.stringify(obj)); } +function handleLockState(msg) { + if(msg[0] == true) pollLockHeld = false; + else { + setTimeout(() => { + wsSend({action: "lock", "id": 0}); + wsSend({action: "get_toplay"}); + wsSend({action: "skipc"}); + wsSend({action: "skipi"}); + }, 500 + (Math.random() * 1000)) + } +} + function handleMessage(msg) { switch (msg.event) { case "state": { const d = msg.data || {}; - if (d.dirs) updateDirs(d.dirs); - if (d.track) applyProgressState(d.track); + if(d.dirs) updateDirs(d.dirs); + if(d.track) applyProgressState(d.track); + if(d.locks) handleLockState(d.locks) + break; + } + case "lock": { + if(msg.error) pollLockHeld = false; break; } case "rds": @@ -105,9 +120,11 @@ function handleMessage(msg) { break; case "new_track": applyTrackState(msg.data); - wsSend({ action: "get_toplay" }); - wsSend({ action: "skipc" }); - wsSend({ action: "skipi" }); + if(pollLockHeld) { + wsSend({ action: "get_toplay" }); + wsSend({ action: "skipc" }); + wsSend({ action: "skipi" }); + } break; case "progress": applyProgressState(msg.data);