ITQC/pydase
24
0
Fork 0
mirror of https://github.com/tiqi-group/pydase.git synced 2025-04-22 01:00:02 +02:00
Code Issues Packages Projects Releases Wiki Activity

refactors client sio event setup

Browse Source
This commit is contained in:
Mose Müller 2024-04-15 08:16:46 +02:00
parent f65a0e31c3
commit 5eeaefdd63
1 changed files with 23 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

45
src/pydase/client/client.py
View File

@ -79,27 +79,28 @@ class Client:
) )
async def _setup_events(self) -> None: async def _setup_events(self) -> None:
@self._sio.event self._sio.on("connect", self._handle_connect)
async def connect() -> None: self._sio.on("disconnect", self._handle_disconnect)
logger.debug("Connected to '%s:%s' ...", self._hostname, self._port) self._sio.on("notify", self._handle_update)
serialized_object = cast(
SerializedDataService, await self._sio.call("service_serialization")
)
ProxyLoader.update_data_service_proxy(
self.proxy, serialized_object=serialized_object
)
serialized_object["type"] = "DeviceConnection"
self.proxy._notify_changed("", loads(serialized_object))
self.proxy._connected = True
@self._sio.event async def _handle_connect(self) -> None:
async def disconnect() -> None: logger.debug("Connected to '%s:%s' ...", self._hostname, self._port)
logger.debug("Disconnected from '%s:%s' ...", self._hostname, self._port) serialized_object = cast(
self.proxy._connected = False SerializedDataService, await self._sio.call("service_serialization")
)
ProxyLoader.update_data_service_proxy(
self.proxy, serialized_object=serialized_object
)
serialized_object["type"] = "DeviceConnection"
self.proxy._notify_changed("", loads(serialized_object))
self.proxy._connected = True
@self._sio.event async def _handle_disconnect(self) -> None:
async def notify(data: NotifyDict) -> None: logger.debug("Disconnected from '%s:%s' ...", self._hostname, self._port)
self.proxy._notify_changed( self.proxy._connected = False
data["data"]["full_access_path"],
loads(data["data"]["value"]), async def _handle_update(self, data: NotifyDict) -> None:
) self.proxy._notify_changed(
data["data"]["full_access_path"],
loads(data["data"]["value"]),
)