virtual-controller/server/ws_server.py

import json
from PyQt6.QtCore import QObject, pyqtSignal
from PyQt6.QtWebSockets import QWebSocketServer
from PyQt6.QtNetwork import QHostAddress


class WSServer(QObject):
    control_received = pyqtSignal(str, int)
    control_f_received = pyqtSignal(str, float)
    layout_saved = pyqtSignal(str, object)
    widget_visibility_received = pyqtSignal(str, int)
    client_connected = pyqtSignal()
    log_signal = pyqtSignal(str)
    raw_in_signal = pyqtSignal(str)

    def __init__(self, port=8765, parent=None):
        super().__init__(parent)
        self._clients = []
        self._port = port
        self._server = QWebSocketServer("VC", QWebSocketServer.SslMode.NonSecureMode, self)
        if self._server.listen(QHostAddress.SpecialAddress.AnyIPv4, port):
            self.log_signal.emit(f"WS listening on :{port}")
        else:
            self.log_signal.emit(f"WS failed to bind :{port}")
        self._server.newConnection.connect(self._on_new_connection)

    def _on_new_connection(self):
        client = self._server.nextPendingConnection()
        peer = client.peerAddress().toString()
        self._clients.append(client)
        client.textMessageReceived.connect(self._on_message)
        client.disconnected.connect(lambda c=client, p=peer: self._on_disconnect(c, p))
        self.log_signal.emit(f"client connected  {peer}  ({len(self._clients)} total)")
        self.client_connected.emit()

    def _on_disconnect(self, client, peer):
        if client in self._clients:
            self._clients.remove(client)
        client.deleteLater()
        self.log_signal.emit(f"client disconnected  {peer}  ({len(self._clients)} remaining)")

    def _on_message(self, message):
        self.raw_in_signal.emit(message)
        try:
            data = json.loads(message)
            ev = data.get("event")
            if ev == "control":
                uid = data["uid"]
                val = int(data["value"])
                self.log_signal.emit(f"control  uid={uid[:8]}…  val={val}")
                self.control_received.emit(uid, val)
            elif ev == "control_f":
                uid = data["uid"]
                val = float(data["value"])
                self.log_signal.emit(f"control_f  uid={uid[:8]}…  val={val:.4f}")
                self.control_f_received.emit(uid, val)
            elif ev == "save_layout":
                n = len(data.get("layout", {}))
                self.log_signal.emit(f"layout saved  preset={data['preset_uuid'][:8]}…  {n} widgets")
                self.layout_saved.emit(data["preset_uuid"], data["layout"])
            elif ev == "widget_visibility":
                uid = data["uid"]
                dest_id = int(data.get("dest_id", 0))
                self.log_signal.emit(f"widget visibility  uid={uid[:8]}…  dest_id={dest_id}")
                self.widget_visibility_received.emit(uid, dest_id)
            else:
                self.log_signal.emit(f"unknown event: {ev}")
        except Exception as e:
            self.log_signal.emit(f"bad message: {e}")

    @property
    def has_clients(self) -> bool:
        return len(self._clients) > 0

    def broadcast(self, data: dict):
        if not self._clients:
            return
        msg = json.dumps(data)
        for client in list(self._clients):
            client.sendTextMessage(msg)

    def broadcast_preset(self, snapshot: dict, layout: dict):
        name = snapshot.get("preset_name", "")
        self.log_signal.emit(f"broadcast preset  '{name}'  → {len(self._clients)} client(s)")
        # layout omitted — owned and persisted client-side, keyed by preset_uuid
        self.broadcast({"event": "preset", "data": snapshot})

    def broadcast_widget_update(self, uid: str, value: int):
        self.broadcast({"event": "widget_update", "uid": uid, "value": value})

    def broadcast_widget_update_f(self, uid: str, value: float):
        self.broadcast({"event": "widget_update_f", "uid": uid, "value": value})

    def broadcast_widget_label(self, uid: str, label: str):
        self.broadcast({"event": "widget_label", "uid": uid, "label": label})

    def broadcast_widget_feedback(self, uid: str, text: str):
        self.broadcast({"event": "widget_feedback", "uid": uid, "text": text})

    def broadcast_daw_state(self, **kwargs):
        self.broadcast({"event": "daw_state", **kwargs})

    def stop(self):
        self._server.close()
        for client in list(self._clients):
            client.close()
        self._clients.clear()
        self.log_signal.emit("WS server stopped")