rick/LEDMatrix
1
0
Fork 0
mirror of https://github.com/ChuckBuilds/LEDMatrix.git synced 2026-04-10 21:03:01 +00:00
Code Issues Packages Projects Releases Wiki Activity

updated socketio namespace

Browse Source
This commit is contained in:
ChuckBuilds
2025-05-22 17:51:14 -05:00
parent 5a4e82ab6f
commit 3a73d7cb22
2 changed files with 10 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
src/ytm_client.py
View File

@@ -21,26 +21,26 @@ class YTMClient:
self._data_lock = threading.Lock()
self._connection_event = threading.Event()
@self.sio.event
@self.sio.event(namespace='/api/v1/realtime')
def connect():
logging.info(f"Successfully connected to YTM Companion Socket.IO server at {self.base_url}")
logging.info(f"Successfully connected to YTM Companion Socket.IO server at {self.base_url} on namespace /api/v1/realtime")
self.is_connected = True
self._connection_event.set()
@self.sio.event
@self.sio.event(namespace='/api/v1/realtime')
def connect_error(data):
logging.error(f"YTM Companion Socket.IO connection failed: {data}")
logging.error(f"YTM Companion Socket.IO connection failed for namespace /api/v1/realtime: {data}")
self.is_connected = False
self._connection_event.set()
@self.sio.event
@self.sio.event(namespace='/api/v1/realtime')
def disconnect():
logging.info(f"Disconnected from YTM Companion Socket.IO server at {self.base_url}")
logging.info(f"Disconnected from YTM Companion Socket.IO server at {self.base_url} on namespace /api/v1/realtime")
self.is_connected = False
@self.sio.on('ytm_track_update')
@self.sio.on('ytm_track_update', namespace='/api/v1/realtime')
def on_track_update(data):
logging.debug(f"Received track update from YTM Companion: {data}")
logging.debug(f"Received track update from YTM Companion on /api/v1/realtime: {data}")
with self._data_lock:
self.last_known_track_data = data
@@ -74,9 +74,9 @@ class YTMClient:
def _ensure_connected(self, timeout=5):
if not self.is_connected:
logging.info(f"Attempting to connect to YTM Socket.IO server: {self.base_url}")
logging.info(f"Attempting to connect to YTM Socket.IO server: {self.base_url} on namespace /api/v1/realtime")
try:
self.sio.connect(self.base_url, transports=['websocket'], wait_timeout=timeout)
self.sio.connect(self.base_url, transports=['websocket'], wait_timeout=timeout, namespaces=['/api/v1/realtime'])
self._connection_event.clear()
if not self._connection_event.wait(timeout=timeout):
logging.warning(f"YTM Socket.IO connection attempt timed out after {timeout}s.")