custom_components/leafeemag/binary_sensor.py
def _connect_and_subscribe(self) -> bool:
        """Connect to Mag and Subscribe the notification service."""

        # Update time
        self._last_connected_at = time.time()

        # Disconnect from device if necessary
        self._disconnect();

        # Initialize BLE adapter
        _LOGGER.debug('Initializing BLE adapter...')

        try:
            self._ble_adapter.start(False)
        except Exception as error:
            _LOGGER.debug('Error occurred during initializing adapter: %s; However ignored.', error)

        # Connect to device
        _LOGGER.debug('Connecting to Mag... %s', self._name)
        self._mag_device = None

        from pygatt import BLEAddressType

        try:
            AUTO_RECONNECT = True
            self._mag_device = self._ble_adapter.connect(self._mac_address, BLE_CONNECT_TIMEOUT_SEC, BLEAddressType.public, AUTO_RECONNECT)
        except Exception as error:
            _LOGGER.error('Error occurred during connecting to %s: %s; Waiting for retry...', self._name, error)
            self._disconnect();
            return False

        # Get latest state
        _LOGGER.debug('Getting latest state... %s', self._name)

        try:
            value = self._mag_device.char_read(SENSOR_CHARACTERISTIC_UUID, BLE_READ_TIMEOUT_SEC)
            self._set_state_by_received_bytearray(value)
        except Exception as error:
            _LOGGER.debug('Error occurred during getting latest state from %s: %s; However ignored.', self._name, error)

        # Get battery level
        _LOGGER.debug('Getting battery level... %s', self._name)

        try:
            value = self._mag_device.char_read(BATTERY_LEVEL_CHARACTERISTIC_UUID, BLE_READ_TIMEOUT_SEC)
            self._battery_level = int(value.hex(), 16)
        except Exception as error:
            _LOGGER.debug('Error occurred during getting latest state from %s: %s; However ignored.', self._name, error)

        # Subscribe to notifications for state changes
        _LOGGER.debug('Subscribing notification... %s', self._name)

        try:
            self._mag_device.subscribe(SENSOR_CHARACTERISTIC_UUID, lambda handle, value: _on_notification_received_from_mag(self, handle, value))
        except Exception as error:
            _LOGGER.error('Error occurred during subscribing with %s: %s; Waiting for retry...', self._name, error)
            self._disconnect()
            return False

        # Done
        _LOGGER.info('Ready for detect changing: %s', self._name)
        return True
Connect to Mag and Subscribe the notification service.
Connect to Mag and Subscribe the notification service.
Connect to Mag and Subscribe the notification service .
def _connect_and_subscribe(self) -> bool:
        self._last_connected_at = time.time()
        self._disconnect();
        _LOGGER.debug('Initializing BLE adapter...')
        try:
            self._ble_adapter.start(False)
        except Exception as error:
            _LOGGER.debug('Error occurred during initializing adapter: %s; However ignored.', error)
        _LOGGER.debug('Connecting to Mag... %s', self._name)
        self._mag_device = None
        from pygatt import BLEAddressType
        try:
            AUTO_RECONNECT = True
            self._mag_device = self._ble_adapter.connect(self._mac_address, BLE_CONNECT_TIMEOUT_SEC, BLEAddressType.public, AUTO_RECONNECT)
        except Exception as error:
            _LOGGER.error('Error occurred during connecting to %s: %s; Waiting for retry...', self._name, error)
            self._disconnect();
            return False
        _LOGGER.debug('Getting latest state... %s', self._name)
        try:
            value = self._mag_device.char_read(SENSOR_CHARACTERISTIC_UUID, BLE_READ_TIMEOUT_SEC)
            self._set_state_by_received_bytearray(value)
        except Exception as error:
            _LOGGER.debug('Error occurred during getting latest state from %s: %s; However ignored.', self._name, error)
        _LOGGER.debug('Getting battery level... %s', self._name)
        try:
            value = self._mag_device.char_read(BATTERY_LEVEL_CHARACTERISTIC_UUID, BLE_READ_TIMEOUT_SEC)
            self._battery_level = int(value.hex(), 16)
        except Exception as error:
            _LOGGER.debug('Error occurred during getting latest state from %s: %s; However ignored.', self._name, error)
        _LOGGER.debug('Subscribing notification... %s', self._name)
        try:
            self._mag_device.subscribe(SENSOR_CHARACTERISTIC_UUID, lambda handle, value: _on_notification_received_from_mag(self, handle, value))
        except Exception as error:
            _LOGGER.error('Error occurred during subscribing with %s: %s; Waiting for retry...', self._name, error)
            self._disconnect()
            return False
        _LOGGER.info('Ready for detect changing: %s', self._name)
        return True
def _connect_and_subscribe ( self ) -> bool : self . _last_connected_at = time . time ( ) self . _disconnect ( ) ; _LOGGER . debug ( 'Initializing BLE adapter...' ) try : self . _ble_adapter . start ( False ) except Exception as error : _LOGGER . debug ( 'Error occurred during initializing adapter: %s; However ignored.' , error ) _LOGGER . debug ( 'Connecting to Mag... %s' , self . _name ) self . _mag_device = None from pygatt import BLEAddressType try : AUTO_RECONNECT = True self . _mag_device = self . _ble_adapter . connect ( self . _mac_address , BLE_CONNECT_TIMEOUT_SEC , BLEAddressType . public , AUTO_RECONNECT ) except Exception as error : _LOGGER . error ( 'Error occurred during connecting to %s: %s; Waiting for retry...' , self . _name , error ) self . _disconnect ( ) ; return False _LOGGER . debug ( 'Getting latest state... %s' , self . _name ) try : value = self . _mag_device . char_read ( SENSOR_CHARACTERISTIC_UUID , BLE_READ_TIMEOUT_SEC ) self . _set_state_by_received_bytearray ( value ) except Exception as error : _LOGGER . debug ( 'Error occurred during getting latest state from %s: %s; However ignored.' , self . _name , error ) _LOGGER . debug ( 'Getting battery level... %s' , self . _name ) try : value = self . _mag_device . char_read ( BATTERY_LEVEL_CHARACTERISTIC_UUID , BLE_READ_TIMEOUT_SEC ) self . _battery_level = int ( value . hex ( ) , 16 ) except Exception as error : _LOGGER . debug ( 'Error occurred during getting latest state from %s: %s; However ignored.' , self . _name , error ) _LOGGER . debug ( 'Subscribing notification... %s' , self . _name ) try : self . _mag_device . subscribe ( SENSOR_CHARACTERISTIC_UUID , lambda handle , value : _on_notification_received_from_mag ( self , handle , value ) ) except Exception as error : _LOGGER . error ( 'Error occurred during subscribing with %s: %s; Waiting for retry...' , self . _name , error ) self . _disconnect ( ) return False _LOGGER . info ( 'Ready for detect changing: %s' , self . _name ) return True
Connect to Mag and Subscribe the notification service.
Connect to Mag and Subscribe the notification service .
"""Connect to Mag and Subscribe the notification service.""" # Update time # Disconnect from device if necessary # Initialize BLE adapter # Connect to device # Get latest state # Get battery level # Subscribe to notifications for state changes # Done
self