More stuff towards making Webstudio sync

player.py

@@ -35,7 +35,7 @@ from plan import PlanItem
 # Stop the Pygame Hello message.
 import os
 os.environ['PYGAME_HIDE_SUPPORT_PROMPT'] = "hide"
-from pygame import mixer
+from pygame import mixer, NOEVENT, USEREVENT, event
 from mutagen.mp3 import MP3
 
 from helpers.myradio_api import MyRadioAPI
@@ -43,7 +43,7 @@ from helpers.os_environment import isMacOS
 from helpers.state_manager import StateManager
 from helpers.logging_manager import LoggingManager
 
-
+PLAYBACK_END = USEREVENT + 1
 class Player():
     state = None
     running = False
@@ -140,6 +140,7 @@ class Player():
     def play(self, pos: float = 0):
         try:
             mixer.music.play(0, pos)
+            mixer.music.set_endevent(PLAYBACK_END)
             self.state.update("pos_offset", pos)
         except:
             self.logger.log.exception("Failed to play at pos: " + str(pos))
@@ -490,6 +491,18 @@ class Player():
                         else:
                             self._retMsg(False)
 
+                #try:
+                #callback_event = event.poll()
+                #print(callback_event)
+                #if callback_event.type == PLAYBACK_END:
+                #    if self.out_q:
+                #        print("Playback endded at end of Track.")
+                #        self.out_q.put("STOP") # Tell clients that we've stopped playing.
+                #elif callback_event.type == NOEVENT:
+                #    pass
+                #print("Another message")
+                #except:
+                #    pass
             # Catch the player being killed externally.
             except KeyboardInterrupt:
                 self.logger.log.info("Received KeyboardInterupt")

server.py

@@ -13,7 +13,10 @@
         October, November 2020
 """
 import asyncio
+import copy
 import multiprocessing
+import queue
+import threading
 import time
 import player
 from flask import Flask, render_template, send_from_directory, request, jsonify
@@ -272,22 +275,22 @@ def add_to_plan(channel: int):
 
     return new_item
 
-@app.route("/player/<int:channel>/move/<int:channel_weight>/<int:position>")
+#@app.route("/player/<int:channel>/move/<int:channel_weight>/<int:position>")
 def move_plan(channel: int, channel_weight: int, position: int):
     channel_to_q[channel].put("MOVE:" + json.dumps({"channel_weight": channel_weight, "position": position}))
 
     # TODO Return
     return True
 
-@app.route("/player/<int:channel>/remove/<int:channel_weight>")
+#@app.route("/player/<int:channel>/remove/<int:channel_weight>")
 def remove_plan(channel: int, channel_weight: int):
-    channel_to_q[channel].put("REMOVE:" + channel_weight)
+    channel_to_q[channel].put("REMOVE:" + str(channel_weight))
 
     # TODO Return
     return True
 
 
-@app.route("/player/<int:channel>/clear")
+#@app.route("/player/<int:channel>/clear")
 def clear_channel_plan(channel: int):
     channel_to_q[channel].put("CLEAR")
 
@@ -306,18 +309,22 @@ def channel_json(channel: int):
 
 def status(channel: int):
     channel_to_q[channel].put("STATUS")
+    i = 0
     while True:
-        response = ui_to_q[channel].get()
-        if response.startswith("STATUS:"):
-            print("Got my status message")
-            response = response[7:]
-            response = response[response.index(":")+1:]
-            try:
-                response = json.loads(response)
-            except:
-                pass
+        try:
+            response = ui_to_q[channel].get_nowait()
+            if response.startswith("STATUS:"):
+                response = response[7:]
+                response = response[response.index(":")+1:]
+                try:
+                    response = json.loads(response)
+                except Exception as e:
+                    raise e
+                return response
+
+        except queue.Empty:
+            pass
 
-            return response
         time.sleep(0.1)
 
 
@@ -369,6 +376,9 @@ def send_logs(path):
 
 
 async def startServer():
+    process_title="startServer"
+    threading.current_thread().name = process_title
+
     if isMacOS():
         multiprocessing.set_start_method("spawn", True)
     for channel in range(state.state["num_channels"]):
@@ -381,7 +391,7 @@ async def startServer():
             multiprocessing.Process(
                 target=player.Player,
                 args=(channel, channel_to_q[-1], channel_from_q[-1]),
-                daemon=True
+                #daemon=True
             )
         )
         channel_p[channel].start()

websocket_server.py

@@ -1,12 +1,13 @@
 import asyncio
 import multiprocessing
+import queue
 from typing import List
 import websockets
 import json
 
 baps_clients = set()
 channel_to_q = None
-channel_from_q: List[multiprocessing.Queue]
+webstudio_to_q: List[multiprocessing.Queue]
 server_name = None
 
 
@@ -15,6 +16,8 @@ async def websocket_handler(websocket, path):
     baps_clients.add(websocket)
     await websocket.send(json.dumps({"message": "Hello", "serverName": server_name}))
     print("New Client: {}".format(websocket))
+    for channel in channel_to_q:
+        channel.put("STATUS")
 
     async def handle_from_webstudio():
         try:
@@ -50,6 +53,7 @@ async def websocket_handler(websocket, path):
                             "artist":  data["newItem"]["artist"] if "artist" in data["newItem"].keys() else None,
                             "timeslotItemId": int(data["newItem"]["timeslotItemId"]) if "timeslotItemId" in data["newItem"].keys() and data["newItem"]["timeslotItemId"] != None else None,
                             "trackId": int(data["newItem"]["trackId"]) if "trackId" in data["newItem"].keys() and data["newItem"]["trackId"] != None else None,
+                            "recordId": int(data["newItem"]["trackId"]) if "trackId" in data["newItem"].keys() and data["newItem"]["trackId"] != None else None,
                             "managedId": managed_id
                         }
                         channel_to_q[channel].put("ADD:" + json.dumps(new_item))
@@ -68,37 +72,44 @@ async def websocket_handler(websocket, path):
             baps_clients.remove(websocket)
 
     async def handle_to_webstudio():
-        global channel_from_q
         while True:
-            for channel in range(len(channel_from_q)):
+            for channel in range(len(webstudio_to_q)):
                 try:
-                    message = channel_from_q[channel].get_nowait()
+                    message = webstudio_to_q[channel].get_nowait()
+                    if not message.startswith("STATUS"):
+                        continue # Ignore non state updates for now.
+                    try:
+                        message = message.split("OKAY:")[1]
+                        message = json.loads(message)
+                    except:
+                        pass
                     data = json.dumps({
-                        "message": message,
-                        "channel:": channel
+                        "command": "STATUS",
+                        "data": message,
+                        "channel": channel
                     })
                     await asyncio.wait([conn.send(data) for conn in baps_clients])
-                except:
+                except queue.Empty:
                     pass
             await asyncio.sleep(0.01)
 
     from_webstudio = asyncio.create_task(handle_from_webstudio())
-    #to_webstudio = asyncio.create_task(handle_to_webstudio())
+    to_webstudio = asyncio.create_task(handle_to_webstudio())
 
     try:
-        await asyncio.gather(from_webstudio)#, to_webstudio)
+        await asyncio.gather(from_webstudio, to_webstudio)
     finally:
         from_webstudio.cancel()
-        #to_webstudio.cancel()
+        to_webstudio.cancel()
 
 
 class WebsocketServer:
 
     def __init__(self, in_q, out_q, state):
         global channel_to_q
-        global channel_from_q
+        global webstudio_to_q
         channel_to_q = in_q
-        channel_from_q = out_q
+        webstudio_to_q = out_q
 
         global server_name
         server_name = state.state["server_name"]