Connect to a WebSocket server on the specified port. This method sets up an asynchronous Socket.IO client with reconnection attempts, starts an asyncio thread, connects to the WebSocket server, and sets the connection status. :param port: The port number to connect
(cls, port: int)
| 147 | |
| 148 | @classmethod |
| 149 | def connect_websocket_server(cls, port: int) -> None: |
| 150 | """ |
| 151 | Connect to a WebSocket server on the specified port. |
| 152 | |
| 153 | This method sets up an asynchronous Socket.IO client with |
| 154 | reconnection attempts, starts an asyncio thread, connects to the WebSocket server, and sets the connection status. |
| 155 | |
| 156 | :param port: The port number to connect to the WebSocket server. |
| 157 | """ |
| 158 | global ws_thread, sio |
| 159 | |
| 160 | if tool.Web.get_web_props().is_connected: |
| 161 | print(f"Already connected to websocket server on port: {port}") |
| 162 | return |
| 163 | |
| 164 | sio = socketio.AsyncClient( |
| 165 | reconnection=True, |
| 166 | reconnection_attempts=RECONNECTION_ATTEMPTS, |
| 167 | reconnection_delay=RECONNECTION_DELAY, |
| 168 | logger=True, |
| 169 | ) |
| 170 | |
| 171 | ws_thread = AsyncioThread() |
| 172 | ws_thread.daemon = True |
| 173 | ws_thread.start() |
| 174 | |
| 175 | ws_url = f"ws://localhost:{port}/blender" |
| 176 | ws_thread.run_coro(cls.sio_connect(ws_url)) |
| 177 | cls.set_is_connected(True) |
| 178 | bpy.app.timers.register(cls.check_operator_queue) |
| 179 | |
| 180 | @classmethod |
| 181 | def disconnect_websocket_server(cls) -> None: |
no test coverage detected