locks

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:
+    async def get_imc(name, data):
-        try: msg: dict = json.loads(raw)
+        key = str(uuid.uuid4())
-        except Exception:
+        imc_q.put({"name": name, "data": data, "key": key})
-            await websocket.send(json.dumps({"error": "invalid json"}))
+        start = time.monotonic()
-            continue
+        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):
+    async def broadcast(payload: dict):
-            key = str(uuid.uuid4())
+        msg = json.dumps(payload)
-            imc_q.put({"name": name, "data": data, "key": key})
+        for ws in list(clients):
-            start = time.monotonic()
+            try: await ws.send(msg)
-            while time.monotonic() - start < 1:
+            except Exception: pass
-                if key in shared_data: return shared_data.pop(key)
-                await asyncio.sleep(0.1)
-            return None
 
-        action = msg.get("action")
+    try:
-        if action == "skip":
+        async for raw in websocket:
-            imc_q.put({"name": "procman", "data": {"op": 2}})
+            try: msg: dict = json.loads(raw)
-            await websocket.send(json.dumps({"event": "skip"}))
+            except Exception:
-        elif action == "add_to_toplay":
+                await websocket.send(json.dumps({"error": "invalid json"}))
-            songs = msg.get("songs")
+                continue
-            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}})
 
+            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:
+                if result is None: await websocket.send(json.dumps({"error": "timeout", "code": 504}))
-                    await asyncio.get_event_loop().run_in_executor(None, ws_q.put, {"data": result, "event": "toplay"})
+                else: await websocket.send(json.dumps({"data": result, "event": "toplay"}))
-        elif action == "get_toplay":
+            elif action == "clear_toplay":
-            result = await get_imc("activemod", {"action": "get_toplay"})
+                result = await get_imc("activemod", {"action": "clear_toplay"})
-            if result is None: await websocket.send(json.dumps({"error": "timeout", "code": 504}))
+                if result is None: await websocket.send(json.dumps({"error": "timeout", "code": 504}))
-            else: await websocket.send(json.dumps({"data": result, "event": "toplay"}))
+                else:
-        elif action == "clear_toplay":
+                    await broadcast({"data": result, "event": "toplay"})
-            result = await get_imc("activemod", {"action": "clear_toplay"})
+            elif action == "skipc" or action == "skipi":
-            if result is None: await websocket.send(json.dumps({"error": "timeout", "code": 504}))
+                result = await get_imc("activemod", msg)
-            else:
+                if result is None: await websocket.send(json.dumps({"error": "timeout", "code": 504}))
-                await asyncio.get_event_loop().run_in_executor(None, ws_q.put, {"data": result, "event": "toplay"})
+                else:
-        elif action == "skipc" or action == "skipi":
+                    await broadcast({"data": result, "event": action})
-            result = await get_imc("activemod", msg)
+            elif action == "jingle":
-            if result is None: await websocket.send(json.dumps({"error": "timeout", "code": 504}))
+                result = await get_imc("jingle", msg.get("top", False))
-            else:
+                if result is None: await websocket.send(json.dumps({"error": "timeout", "code": 504}))
-                await asyncio.get_event_loop().run_in_executor(None, ws_q.put, {"data": result, "event": action})
+                else:
-        elif action == "jingle":
+                    await websocket.send(json.dumps(result))
-            result = await get_imc("jingle", msg.get("top", False))
+                    result = await get_imc("activemod", {"action": "get_toplay"})
-            if result is None: await websocket.send(json.dumps({"error": "timeout", "code": 504}))
+                    if result is not None:
-            else:
+                        await broadcast({"data": result, "event": "toplay"})
-                await websocket.send(json.dumps(result))
+            elif action == "request_dir":
-                result = await get_imc("activemod", {"action": "get_toplay"})
+                what: str = msg.get("what", "")
-                if result is not None:
+                try:
-                    await asyncio.get_event_loop().run_in_executor(None, ws_q.put, {"data": result, "event": "toplay"})
+                    dir = Path(MAIN_PATH_DIR, what).resolve()
-        elif action == "request_dir":
+                    payload = {"files": [i.name for i in list(dir.iterdir()) if i.is_file()], "base": str(dir), "dir": dir.name}
-            what: str = msg.get("what", "")
+                except Exception: payload = {}
-            try:
+                await websocket.send(json.dumps({"event": "request_dir", "data": payload}))
-                dir = Path(MAIN_PATH_DIR, what).resolve()
+            elif action == "fsdb_add":
-                payload = {"files": [i.name for i in list(dir.iterdir()) if i.is_file()], "base": str(dir), "dir": dir.name}
+                name: str | None = msg.get("name")
-            except Exception: payload = {}
+                try:
-            await websocket.send(json.dumps({"event": "request_dir", "data": payload}))
+                    if not name: raise Exception("name not defined")
-        elif action == "fsdb_add":
+                    path = Path(MAIN_PATH_DIR, ".playlist", msg.get("playlist", ""), name)
-            name: str | None = msg.get("name")
+                    path.touch(exist_ok=True)
-            try:
+                    await websocket.send(json.dumps({"event": "fsdb_add", "ok": True}))
-                if not name: raise Exception("name not defined")
+                except Exception as e: await websocket.send(json.dumps({"event": "fsdb_add", "error": str(e)}))
-                path = Path(MAIN_PATH_DIR, ".playlist", msg.get("playlist", ""), name)
+            elif action == "fsdb_add_dir":
-                path.touch(exist_ok=True)
+                name: str | None = msg.get("name")
-                await websocket.send(json.dumps({"event": "fsdb_add", "ok": True}))
+                try:
-            except Exception as e: await websocket.send(json.dumps({"event": "fsdb_add", "error": str(e)}))
+                    if not name: raise Exception("name not defined")
-        elif action == "fsdb_add_dir":
+                    path = Path(MAIN_PATH_DIR, ".playlist", msg.get("playlist", ""), name)
-            name: str | None = msg.get("name")
+                    path.mkdir(parents=True, exist_ok=True)
-            try:
+                    await websocket.send(json.dumps({"event": "fsdb_add_dir", "ok_dir": True}))
-                if not name: raise Exception("name not defined")
+                except Exception as e: await websocket.send(json.dumps({"event": "fsdb_add_dir", "error": str(e)}))
-                path = Path(MAIN_PATH_DIR, ".playlist", msg.get("playlist", ""), name)
+            elif action == "fsdb_remove":
-                path.mkdir(parents=True, exist_ok=True)
+                name: str | None = msg.get("name")
-                await websocket.send(json.dumps({"event": "fsdb_add_dir", "ok_dir": True}))
+                try:
-            except Exception as e: await websocket.send(json.dumps({"event": "fsdb_add_dir", "error": str(e)}))
+                    if not name: raise Exception("name not defined")
-        elif action == "fsdb_remove":
+                    path = Path(MAIN_PATH_DIR, ".playlist", msg.get("playlist", ""), name)
-            name: str | None = msg.get("name")
+                    if path.is_dir(): shutil.rmtree(path)
-            try:
+                    else: path.unlink(missing_ok=True)
-                if not name: raise Exception("name not defined")
+                    await websocket.send(json.dumps({"event": "fsdb_remove", "ok": True}))
-                path = Path(MAIN_PATH_DIR, ".playlist", msg.get("playlist", ""), name)
+                except Exception as e: await websocket.send(json.dumps({"event": "fsdb_remove", "error": str(e)}))
-                if path.is_dir(): shutil.rmtree(path)
+            elif action == "fsdb_list":
-                else: path.unlink(missing_ok=True)
+                try:
-                await websocket.send(json.dumps({"event": "fsdb_remove", "ok": True}))
+                    p = Path(MAIN_PATH_DIR, ".playlist", msg.get("playlist", ""))
-            except Exception as e: await websocket.send(json.dumps({"event": "fsdb_remove", "error": str(e)}))
+                    payload = {
-        elif action == "fsdb_list":
+                        "files": [i.name for i in p.iterdir() if i.is_file()],
-            try:
+                        "dirs": [i.name for i in p.iterdir() if i.is_dir()]
-                p = Path(MAIN_PATH_DIR, ".playlist", msg.get("playlist", ""))
+                    }
-                payload = {
+                    await websocket.send(json.dumps({"event": "fsdb_list", "data": payload}))
-                    "files": [i.name for i in p.iterdir() if i.is_file()],
+                except Exception as e: await websocket.send(json.dumps({"event": "fsdb_list", "data": {}, "error": str(e)}))
-                    "dirs": [i.name for i in p.iterdir() if i.is_dir()]
+            elif action == "fm95": await writer_q.put((base64.b64decode(msg.get("data", "")), websocket))
-                }
+            elif action == "lock":
-                await websocket.send(json.dumps({"event": "fsdb_list", "data": payload}))
+                lid = msg.get("id")
-            except Exception as e: await websocket.send(json.dumps({"event": "fsdb_list", "data": {}, "error": str(e)}))
+                if lid is None: await websocket.send(json.dumps({"event": "error", "error": "lock id required"}))
-        elif action == "fm95": await writer_q.put((base64.b64decode(msg.get("data", "")), websocket))
+                elif locks.get(lid) not in (None, websocket):
-        else: await websocket.send(json.dumps({"event": "error", "error": "unknown action"}))
+                    # 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(
+                    return Response(200, "OK", Headers([("Content-Type", "text/html; charset=utf-8"), ("Content-Length", f"{len(data)}")]), data)
-                        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(
+                    return Response(200, "OK", Headers([("Content-Type", get_content_type(file.name)), ("Content-Length", f"{len(data)}")]), data)
-                        200,
-                        "OK",
-                        Headers([("Content-Type", get_content_type(file.name)), ("Content-Length", f"{len(data)}")]),
-                        data
-                    )
                 else:
                     data = b"Not Found\n"
-                    return Response(
+                    return Response(404, "Not Found", Headers([("Content-Length", f"{len(data)}")]), data)
-                        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({
+            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})
-                "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})

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.dirs) updateDirs(d.dirs);
-            if (d.track) applyProgressState(d.track);
+            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" });
+            if(pollLockHeld) {
-            wsSend({ action: "skipc" });
+                wsSend({ action: "get_toplay" });
-            wsSend({ action: "skipi" });
+                wsSend({ action: "skipc" });
+                wsSend({ action: "skipi" });
+            }
             break;
         case "progress":
             applyProgressState(msg.data);