Paul Lipscomb
7ecc718f5d
app -> app-desktop-macos, presets -> app-presets, server -> remote-server, daw-config-reaper -> osc-config-daw, plugin-reaper-realearn -> plugin-reaper-relearn. Updated run.py and presets.py path references accordingly. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
109 lines
4.5 KiB
Python
109 lines
4.5 KiB
Python
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")
|