diff --git a/src/infuse_iot/tools/gateway.py b/src/infuse_iot/tools/gateway.py index 284d56e..3a50dfe 100644 --- a/src/infuse_iot/tools/gateway.py +++ b/src/infuse_iot/tools/gateway.py @@ -126,6 +126,10 @@ def __init__( self.ddb = ddb self.rpc = rpc_server + def notification_broadcast(self, notification: ClientNotification): + if self.server: + self.server.broadcast(notification) + def query_device_key(self, cb_event: threading.Event | None = None): def security_state_done(pkt: PacketReceived, _: int, response: bytes): cloud_key = response[:32] @@ -204,7 +208,7 @@ def _handle_local_tdf(self, pkt: PacketReceived): if_addr = interface.Address.BluetoothLeAddr.from_tdf_struct(reading.data[0].address) infuse_id = self._common.ddb.infuse_id_from_bluetooth(if_addr) if infuse_id: - self._common.server.broadcast(ClientNotificationConnectionDropped(infuse_id)) + self._common.notification_broadcast(ClientNotificationConnectionDropped(infuse_id)) def _handle_serial_frame(self, frame: bytearray): try: @@ -244,10 +248,9 @@ def _handle_serial_frame(self, frame: bytearray): elif pkt.ptype == InfuseType.KEY_IDS: self._common.query_device_key(None) + # Forward to clients notification = ClientNotificationEpacketReceived(pkt) - if self._common.server: - # Forward to clients - self._common.server.broadcast(notification) + self._common.notification_broadcast(notification) except (ValueError, KeyError) as e: print(f"Decode failed ({e})") @@ -322,19 +325,19 @@ def _bt_connect_cb(self, pkt: PacketReceived, rc: int, response: bytes): else: self._connected[infuse_id] = 1 rsp = ClientNotificationConnectionCreated(infuse_id, 244 - ctypes.sizeof(CtypeBtGattFrame) - 16) - self._common.server.broadcast(rsp) + self._common.notification_broadcast(rsp) def _handle_conn_request(self, req: GatewayRequestConnectionRequest): assert self._common.server is not None if req.infuse_id == InfuseID.GATEWAY or req.infuse_id == self._common.ddb.gateway: # Local gateway always connected - self._common.server.broadcast(ClientNotificationConnectionCreated(req.infuse_id, 512)) + self._common.notification_broadcast(ClientNotificationConnectionCreated(req.infuse_id, 512)) return state = self._common.ddb.devices.get(req.infuse_id, None) if state is None or state.bt_addr is None: - self._common.server.broadcast(ClientNotificationConnectionFailed(req.infuse_id)) + self._common.notification_broadcast(ClientNotificationConnectionFailed(req.infuse_id)) return subs = 0 @@ -403,7 +406,7 @@ def _handle_observed_devices(self): if self._common.ddb.gateway == device: info["gateway"] = True observed_devices[device] = info - self._common.server.broadcast(ClientNotificationObservedDevices(observed_devices)) + self._common.notification_broadcast(ClientNotificationObservedDevices(observed_devices)) def _iter(self) -> None: if self._common.server is None: diff --git a/src/infuse_iot/tools/native_bt.py b/src/infuse_iot/tools/native_bt.py index a8b42ac..dad447e 100644 --- a/src/infuse_iot/tools/native_bt.py +++ b/src/infuse_iot/tools/native_bt.py @@ -30,6 +30,7 @@ PacketReceived, ) from infuse_iot.socket_comms import ( + ClientNotification, ClientNotificationConnectionCreated, ClientNotificationConnectionFailed, ClientNotificationEpacketReceived, @@ -64,6 +65,12 @@ def __init__(self, database: DeviceDatabase, server: LocalServer, bleak_mapping: self._queues: dict[int, asyncio.Queue] = {} self._tasks: dict[int, asyncio.Task] = {} + def wrapped_broadcast(self, notifcation: ClientNotification): + try: + self._server.broadcast(notifcation) + except OSError as e: + Console.log_error(f"Failed to broadcast notification: {str(e)}") + def notification_handler(self, _characteristic: BleakGATTCharacteristic, data: bytearray): try: hdr, decr = CtypeBtGattFrame.decrypt(self._db, None, bytes(data)) @@ -89,9 +96,11 @@ def notification_handler(self, _characteristic: BleakGATTCharacteristic, data: b bytes(decr), ) Console.log_rx(pkt.ptype, len(data)) - self._server.broadcast(ClientNotificationEpacketReceived(pkt)) + self.wrapped_broadcast(ClientNotificationEpacketReceived(pkt)) - async def create_connection(self, request: GatewayRequestConnectionRequest, dev: BLEDevice, queue: asyncio.Queue): + async def create_connection_internal( + self, request: GatewayRequestConnectionRequest, dev: BLEDevice, queue: asyncio.Queue + ): Console.log_info(f"{dev}: Initiating connection") async with BleakClient(dev, timeout=request.timeout_ms / 1000) as client: # Modified from bleak example code @@ -116,7 +125,7 @@ async def create_connection(self, request: GatewayRequestConnectionRequest, dev: Console.log_info(f"{dev}: Connected (MTU {client.mtu_size})") - self._server.broadcast( + self.wrapped_broadcast( ClientNotificationConnectionCreated( request.infuse_id, # ATT header uses 3 bytes of the MTU @@ -147,6 +156,12 @@ async def create_connection(self, request: GatewayRequestConnectionRequest, dev: self._tasks.pop(request.infuse_id) Console.log_info(f"{dev}: Terminating connection") + async def create_connection(self, request: GatewayRequestConnectionRequest, dev: BLEDevice, queue: asyncio.Queue): + try: + await self.create_connection_internal(request, dev, queue) + except TimeoutError as e: + Console.log_info(f"Timeout: {str(e)}") + def datagram_received(self, data: bytes, addr: tuple[str | Any, int]): loop = asyncio.get_event_loop() request = GatewayRequest.from_json(json.loads(data.decode("utf-8"))) @@ -166,7 +181,7 @@ def datagram_received(self, data: bytes, addr: tuple[str | Any, int]): ble_dev = self._mapping.get(request.infuse_id, None) if ble_dev is None: - self._server.broadcast(ClientNotificationConnectionFailed(request.infuse_id)) + self.wrapped_broadcast(ClientNotificationConnectionFailed(request.infuse_id)) return # Create queue for further data transfer @@ -243,7 +258,10 @@ def simple_callback(self, device: BLEDevice, data: AdvertisementData): Console.log_rx(hdr.type, len(payload)) pkt = PacketReceived([hop], hdr.type, decr) notification = ClientNotificationEpacketReceived(pkt) - self.server.broadcast(notification) + try: + self.server.broadcast(notification) + except OSError as e: + Console.log_error(f"Failed to broadcast notification: {str(e)}") async def async_bt_receiver(self): loop = asyncio.get_event_loop()