Additional functionality for Bot and Curtain, reliability improvements. (#30)

* Add_extended_options

* Change get_basic_info to allow reuse for other device notification keys.

* Update __init__.py

* Fix subscribe to notifications flag.

* Add curtain advanced page data.

* Update __init__.py

* Update __init__.py

* Update __init__.py

* Update __init__.py

* Update __init__.py

* Check response from SB after sending command.

-Check response from SB after sending command.
-Allow passive scanning.
-Combine notification and send_command methods.

* Add functions for setting SB mode, cleanup password function.

-Add function to set SB mode (switch/toggle) with strength.
-Add function to set SB long press duration.
-Update get_basic_info to return strength setting.
-Cleanup password function. (_commandkey)

* interface = 0 in bluepy by default. Just matching to save a few seconds.

-interface = 0 in bluepy by default. Just matching to save a few seconds in execution.
-The press, turn_on, turn_off still executes despite bot mode. Just logging to inform user of state.

* Error handling on get_basic_settings

-Error handling on get_basic_settings.
-Add CONNECT_LOCK for discovery.
-Add Inverse to set switchbot mode.

* Update __init__.py

* Add hand up/down functions.

-Add hand up/down functions.
-Log debug messages when bot is in different mode.

* If no notification message return 00 bytes.

All notifications messages return a byte value or multiples of it. If 00 we know the command failed. Set 00 as default if no status returned for consistency.

* Check if bluepy helper is started, wait if device busy.

Check if bluepy helper is started, wait if device busy.
*Logger warning and error is present just for testing*

* Switchbot devices can't immediately accept another connection. Add sleep between commands.

-Try with 0.5 seconds. Might need to increase to 1, 1.5 or 2.

* Update __init__.py

* Update __init__.py

* Update __init__.py

* Shorten and reduce duplicate code.

* Refactor duplicate code to new functions.

* List comprehension for filters on GetSwitchbotDevices

* Call disconnect bluepy helper running.

* bluepy does properly terminate helper. No point in checking.

#Looks like there are some reponse states added to latest bluepy master branch. Could be the reason for connect failures on some conditions. Or the devices take a while to wake.

* Update __init__.py

* Revert "Update __init__.py"

This reverts commit 097710dc44ed33c0d6842368db4e5ba60c6aa1e9.

* No need for .getScanData() as .getValue returns bytes data directly.

* Optimizations and error proofing.

* Update __init__.py

* Found a fix for bluepy disconnect errors.

* Add _stopHelper() when timeout condition.

* Refine _connect method so that the library can handle all error conditions. Bluepy is raises non spesific exceptions on everything and some of them can be corrected.

* Update __init__.py

* Update __init__.py

* Just handle disc as part of connection. (will fail if timeout is reached)

* Cleanup.

* 100 tests later...warning is pointless.

* Sent commands to SB for 1 hour on 10 sec intervals without +99.9% success rate.

* Fix typing errors.

setup.py

@@ -4,7 +4,7 @@ setup(
     name = 'PySwitchbot',
     packages = ['switchbot'],
     install_requires=['bluepy'],
-    version = '0.12.0',
+    version = '0.13.0',
     description = 'A library to communicate with Switchbot',
     author='Daniel Hjelseth Hoyer',
     url='https://github.com/Danielhiversen/pySwitchbot/',

switchbot/__init__.py

@@ -3,7 +3,7 @@ from __future__ import annotations
 
 import binascii
 import logging
-import threading
+from threading import Lock
 import time
 from typing import Any
 
@@ -13,239 +13,246 @@ DEFAULT_RETRY_COUNT = 3
 DEFAULT_RETRY_TIMEOUT = 1
 DEFAULT_SCAN_TIMEOUT = 5
 
-UUID = "cba20d00-224d-11e6-9fb8-0002a5d5c51b"
-HANDLE = "cba20002-224d-11e6-9fb8-0002a5d5c51b"
-NOTIFICATION_HANDLE = "cba20003-224d-11e6-9fb8-0002a5d5c51b"
-
-KEY_PASSWORD_PREFIX = "5711"
-KEY_PASSWORD_NOTIFY_PREFIX = "5712"
+# Keys common to all device types
+DEVICE_GET_BASIC_SETTINGS_KEY = "5702"
+DEVICE_SET_MODE_KEY = "5703"
+DEVICE_SET_EXTENDED_KEY = "570f"
 
+# Bot keys
 PRESS_KEY = "570100"
 ON_KEY = "570101"
 OFF_KEY = "570102"
+DOWN_KEY = "570103"
+UP_KEY = "570104"
 
+# Curtain keys
 OPEN_KEY = "570f450105ff00"  # 570F4501010100
 CLOSE_KEY = "570f450105ff64"  # 570F4501010164
 POSITION_KEY = "570F450105ff"  # +actual_position ex: 570F450105ff32 for 50%
 STOP_KEY = "570F450100ff"
-DEVICE_BASIC_SETTINGS_KEY = "5702"
+CURTAIN_EXT_SUM_KEY = "570f460401"
+CURTAIN_EXT_ADV_KEY = "570f460402"
+CURTAIN_EXT_CHAIN_INFO_KEY = "570f468101"
 
-ON_KEY_SUFFIX = "01"
-OFF_KEY_SUFFIX = "02"
-PRESS_KEY_SUFFIX = "00"
+# Base key when encryption is set
+KEY_PASSWORD_PREFIX = "571"
 
 _LOGGER = logging.getLogger(__name__)
-CONNECT_LOCK = threading.Lock()
+CONNECT_LOCK = Lock()
 
 
-def _process_wohand(data: bytes) -> dict[str, bool | int]:
-    """Process woHand/Bot services data."""
-    _bot_data: dict[str, bool | int] = {}
+def _sb_uuid(comms_type: str = "service") -> bluepy.btle.UUID:
+    """Return Switchbot UUID."""
 
-    # 128 switch or 0 press.
-    _bot_data["switchMode"] = bool(data[1] & 0b10000000)
+    _uuid = {"tx": "002", "rx": "003", "service": "d00"}
 
-    # 64 off or 0 for on, if not inversed in app.
-    if _bot_data["switchMode"]:
-        _bot_data["isOn"] = not bool(data[1] & 0b01000000)
+    if comms_type in _uuid:
+        return bluepy.btle.UUID(f"cba20{_uuid[comms_type]}-224d-11e6-9fb8-0002a5d5c51b")
+
+    return "Incorrect type, choose between: tx, rx or service"
 
-    else:
-        _bot_data["isOn"] = False
 
-    _bot_data["battery"] = data[2] & 0b01111111
+def _process_wohand(data: bytes) -> dict[str, bool | int]:
+    """Process woHand/Bot services data."""
+    _switch_mode = bool(data[1] & 0b10000000)
+
+    _bot_data = {
+        "switchMode": _switch_mode,
+        "isOn": not bool(data[1] & 0b01000000) if _switch_mode else False,
+        "battery": data[2] & 0b01111111,
+    }
 
     return _bot_data
 
 
 def _process_wocurtain(data: bytes, reverse: bool = True) -> dict[str, bool | int]:
     """Process woCurtain/Curtain services data."""
-    _curtain_data: dict[str, bool | int] = {}
 
-    _curtain_data["calibration"] = bool(data[1] & 0b01000000)
-    _curtain_data["battery"] = data[2] & 0b01111111
-    _curtain_data["inMotion"] = bool(data[3] & 0b10000000)
     _position = max(min(data[3] & 0b01111111, 100), 0)
-    _curtain_data["position"] = (100 - _position) if reverse else _position
 
-    # light sensor level (1-10)
-    _curtain_data["lightLevel"] = (data[4] >> 4) & 0b00001111
-    _curtain_data["deviceChain"] = data[4] & 0b00000111
+    _curtain_data = {
+        "calibration": bool(data[1] & 0b01000000),
+        "battery": data[2] & 0b01111111,
+        "inMotion": bool(data[3] & 0b10000000),
+        "position": (100 - _position) if reverse else _position,
+        "lightLevel": (data[4] >> 4) & 0b00001111,
+        "deviceChain": data[4] & 0b00000111,
+    }
 
     return _curtain_data
 
 
-def _process_wosensorth(data: bytes) -> dict[str, Any]:
+def _process_wosensorth(data: bytes) -> dict[str, object]:
     """Process woSensorTH/Temp sensor services data."""
-    _wosensorth_data: dict[str, Any] = {}
 
     _temp_sign = 1 if data[4] & 0b10000000 else -1
     _temp_c = _temp_sign * ((data[4] & 0b01111111) + (data[3] / 10))
     _temp_f = (_temp_c * 9 / 5) + 32
     _temp_f = (_temp_f * 10) / 10
 
-    _wosensorth_data["temp"] = {}
-    _wosensorth_data["temp"]["c"] = _temp_c
-    _wosensorth_data["temp"]["f"] = _temp_f
-
-    _wosensorth_data["fahrenheit"] = bool(data[5] & 0b10000000)
-    _wosensorth_data["humidity"] = data[5] & 0b01111111
-    _wosensorth_data["battery"] = data[2] & 0b01111111
+    _wosensorth_data = {
+        "temp": {"c": _temp_c, "f": _temp_f},
+        "fahrenheit": bool(data[5] & 0b10000000),
+        "humidity": data[5] & 0b01111111,
+        "battery": data[2] & 0b01111111,
+    }
 
     return _wosensorth_data
 
 
+def _process_btle_adv_data(dev: bluepy.btle.ScanEntry) -> dict[str, Any]:
+    """Process bt le adv data."""
+    _adv_data = {"mac_address": dev.addr}
+    _data = dev.getValue(22)[2:]
+
+    supported_types: dict[str, dict[str, Any]] = {
+        "H": {"modelName": "WoHand", "func": _process_wohand},
+        "c": {"modelName": "WoCurtain", "func": _process_wocurtain},
+        "T": {"modelName": "WoSensorTH", "func": _process_wosensorth},
+    }
+
+    _model = chr(_data[0] & 0b01111111)
+    _adv_data["isEncrypted"] = bool(_data[0] & 0b10000000)
+    _adv_data["model"] = _model
+    if _model in supported_types:
+        _adv_data["data"] = supported_types[_model]["func"](_data)
+        _adv_data["data"]["rssi"] = dev.rssi
+        _adv_data["modelName"] = supported_types[_model]["modelName"]
+    else:
+        _adv_data["rawAdvData"] = dev.getValueText(22)
+
+    return _adv_data
+
+
 class GetSwitchbotDevices:
     """Scan for all Switchbot devices and return by type."""
 
-    def __init__(self, interface: int | None = None) -> None:
+    def __init__(self, interface: int = 0) -> None:
         """Get switchbot devices class constructor."""
         self._interface = interface
-        self._all_services_data: dict[str, Any] = {}
+        self._adv_data: dict[str, Any] = {}
 
     def discover(
-        self, retry: int = DEFAULT_RETRY_COUNT, scan_timeout: int = DEFAULT_SCAN_TIMEOUT
-    ) -> dict | None:
+        self,
+        retry: int = DEFAULT_RETRY_COUNT,
+        scan_timeout: int = DEFAULT_SCAN_TIMEOUT,
+        passive: bool = False,
+        mac: str | None = None,
+    ) -> dict[str, Any]:
         """Find switchbot devices and their advertisement data."""
-
         devices = None
 
-        try:
-            devices = bluepy.btle.Scanner(self._interface).scan(scan_timeout)
-
-        except bluepy.btle.BTLEManagementError:
-            _LOGGER.error("Error scanning for switchbot devices", exc_info=True)
+        with CONNECT_LOCK:
+            try:
+                devices = bluepy.btle.Scanner(self._interface).scan(
+                    scan_timeout, passive
+                )
+            except bluepy.btle.BTLEManagementError:
+                _LOGGER.error("Error scanning for switchbot devices", exc_info=True)
 
         if devices is None:
             if retry < 1:
                 _LOGGER.error(
                     "Scanning for Switchbot devices failed. Stop trying", exc_info=True
                 )
-                return None
+                return self._adv_data
 
             _LOGGER.warning(
                 "Error scanning for Switchbot devices. Retrying (remaining: %d)",
                 retry,
             )
             time.sleep(DEFAULT_RETRY_TIMEOUT)
-            return self.discover(retry - 1, scan_timeout)
+            return self.discover(
+                retry=retry - 1,
+                scan_timeout=scan_timeout,
+                passive=passive,
+                mac=mac,
+            )
 
         for dev in devices:
-            if dev.getValueText(7) == UUID:
+            if dev.getValueText(7) == str(_sb_uuid()):
                 dev_id = dev.addr.replace(":", "")
-                self._all_services_data[dev_id] = {}
-                self._all_services_data[dev_id]["mac_address"] = dev.addr
-                for (adtype, desc, value) in dev.getScanData():
-                    if adtype == 22:
-                        _data = bytes.fromhex(value[4:])
-                        _model = chr(_data[0] & 0b01111111)
-                        if _model == "H":
-                            self._all_services_data[dev_id]["data"] = _process_wohand(
-                                _data
-                            )
-                            self._all_services_data[dev_id]["data"]["rssi"] = dev.rssi
-                            self._all_services_data[dev_id]["isEncrypted"] = bool(
-                                _data[0] & 0b10000000
-                            )
-                            self._all_services_data[dev_id]["model"] = _model
-                            self._all_services_data[dev_id]["modelName"] = "WoHand"
-                        elif _model == "c":
-                            self._all_services_data[dev_id][
-                                "data"
-                            ] = _process_wocurtain(_data)
-                            self._all_services_data[dev_id]["data"]["rssi"] = dev.rssi
-                            self._all_services_data[dev_id]["isEncrypted"] = bool(
-                                _data[0] & 0b10000000
-                            )
-                            self._all_services_data[dev_id]["model"] = _model
-                            self._all_services_data[dev_id]["modelName"] = "WoCurtain"
-                        elif _model == "T":
-                            self._all_services_data[dev_id][
-                                "data"
-                            ] = _process_wosensorth(_data)
-                            self._all_services_data[dev_id]["data"]["rssi"] = dev.rssi
-                            self._all_services_data[dev_id]["isEncrypted"] = bool(
-                                _data[0] & 0b10000000
-                            )
-                            self._all_services_data[dev_id]["model"] = _model
-                            self._all_services_data[dev_id]["modelName"] = "WoSensorTH"
-
-                        else:
-                            continue
-                    else:
-                        self._all_services_data[dev_id][desc] = value
-
-        return self._all_services_data
+                if mac:
+                    if dev.addr.lower() == mac.lower():
+                        self._adv_data[dev_id] = _process_btle_adv_data(dev)
+                else:
+                    self._adv_data[dev_id] = _process_btle_adv_data(dev)
+
+        return self._adv_data
 
     def get_curtains(self) -> dict:
         """Return all WoCurtain/Curtains devices with services data."""
-        if not self._all_services_data:
+        if not self._adv_data:
             self.discover()
 
-        _curtain_devices = {}
-
-        for device, data in self._all_services_data.items():
-            if data.get("model") == "c":
-                _curtain_devices[device] = data
+        _curtain_devices = {
+            device: data
+            for device, data in self._adv_data.items()
+            if data.get("model") == "c"
+        }
 
         return _curtain_devices
 
     def get_bots(self) -> dict:
         """Return all WoHand/Bot devices with services data."""
-        if not self._all_services_data:
+        if not self._adv_data:
             self.discover()
 
-        _bot_devices = {}
-
-        for device, data in self._all_services_data.items():
-            if data.get("model") == "H":
-                _bot_devices[device] = data
+        _bot_devices = {
+            device: data
+            for device, data in self._adv_data.items()
+            if data.get("model") == "H"
+        }
 
         return _bot_devices
 
     def get_tempsensors(self) -> dict:
         """Return all WoSensorTH/Temp sensor devices with services data."""
-        if not self._all_services_data:
+        if not self._adv_data:
             self.discover()
 
-        _bot_temp = {}
-
-        for device, data in self._all_services_data.items():
-            if data.get("model") == "T":
-                _bot_temp[device] = data
+        _bot_temp = {
+            device: data
+            for device, data in self._adv_data.items()
+            if data.get("model") == "T"
+        }
 
         return _bot_temp
 
     def get_device_data(self, mac: str) -> dict:
         """Return data for specific device."""
-        if not self._all_services_data:
+        if not self._adv_data:
             self.discover()
 
-        _switchbot_data = {}
-
-        for device in self._all_services_data.values():
-            if device["mac_address"] == mac:
-                _switchbot_data = device
+        _switchbot_data = {
+            device: data
+            for device, data in self._adv_data.items()
+            if data.get("mac_address") == mac
+        }
 
         return _switchbot_data
 
 
-class SwitchbotDevice:
+class SwitchbotDevice(bluepy.btle.Peripheral):
     """Base Representation of a Switchbot Device."""
 
     def __init__(
         self,
         mac: str,
         password: str | None = None,
-        interface: int | None = None,
+        interface: int = 0,
         **kwargs: Any,
     ) -> None:
         """Switchbot base class constructor."""
-        self._interface = interface
-        self._mac = mac
-        self._device = bluepy.btle.Peripheral(
-            deviceAddr=None, addrType=bluepy.btle.ADDR_TYPE_RANDOM, iface=interface
+        bluepy.btle.Peripheral.__init__(
+            self,
+            deviceAddr=None,
+            addrType=bluepy.btle.ADDR_TYPE_RANDOM,
+            iface=interface,
         )
-        self._switchbot_device_data: dict[str, Any] = {}
+        self._interface = interface
+        self._mac = mac.replace("-", ":").lower()
+        self._sb_adv_data: dict[str, Any] = {}
         self._scan_timeout: int = kwargs.pop("scan_timeout", DEFAULT_SCAN_TIMEOUT)
         self._retry_count: int = kwargs.pop("retry_count", DEFAULT_RETRY_COUNT)
         if password is None or password == "":
@@ -255,41 +262,84 @@ class SwitchbotDevice:
                 binascii.crc32(password.encode("ascii")) & 0xFFFFFFFF
             )
 
-    def _connect(self) -> None:
-        try:
-            _LOGGER.debug("Connecting to Switchbot")
-            self._device.connect(
-                self._mac, bluepy.btle.ADDR_TYPE_RANDOM, self._interface
+    # pylint: disable=arguments-differ
+    def _connect(self, retry: int, timeout: int | None = None) -> None:
+        _LOGGER.debug("Connecting to Switchbot")
+
+        if retry < 1:  # failsafe
+            self._stopHelper()
+            return
+
+        if self._helper is None:
+            self._startHelper(self._interface)
+
+        self._writeCmd("conn %s %s\n" % (self._mac, bluepy.btle.ADDR_TYPE_RANDOM))
+
+        rsp = self._getResp(["stat", "err"], timeout)
+
+        while rsp.get("state") and rsp["state"][0] in [
+            "tryconn",
+            "scan",
+            "disc",
+        ]:  # Wait for any operations to finish.
+            rsp = self._getResp(["stat", "err"], timeout)
+
+        if rsp and rsp["rsp"][0] == "err":
+            errcode = rsp["code"][0]
+            _LOGGER.debug(
+                "Error trying to connect to peripheral %s, error: %s",
+                self._mac,
+                errcode,
             )
-            _LOGGER.debug("Connected to Switchbot")
-        except bluepy.btle.BTLEException:
-            _LOGGER.debug("Failed connecting to Switchbot", exc_info=True)
-            raise
 
-    def _disconnect(self) -> None:
-        _LOGGER.debug("Disconnecting")
-        try:
-            self._device.disconnect()
-        except bluepy.btle.BTLEException:
-            _LOGGER.warning("Error disconnecting from Switchbot", exc_info=True)
+            if errcode == "connfail":  # not terminal, can retry connection.
+                return self._connect(retry - 1, timeout)
+
+            if errcode == "nomgmt":
+                raise bluepy.btle.BTLEManagementError(
+                    "Management not available (permissions problem?)", rsp
+                )
+            if errcode == "atterr":
+                raise bluepy.btle.BTLEGattError("Bluetooth command failed", rsp)
+
+            raise bluepy.btle.BTLEException(
+                "Error from bluepy-helper (%s)" % errcode, rsp
+            )
+
+        if not rsp or rsp["state"][0] != "conn":
+            _LOGGER.warning("Bluehelper returned unable to connect state: %s", rsp)
+            self._stopHelper()
+
+            if rsp is None:
+                raise bluepy.btle.BTLEDisconnectError(
+                    "Timed out while trying to connect to peripheral %s" % self._mac,
+                    rsp,
+                )
+
+            raise bluepy.btle.BTLEDisconnectError(
+                "Failed to connect to peripheral %s, rsp: %s" % (self._mac, rsp)
+            )
 
     def _commandkey(self, key: str) -> str:
         if self._password_encoded is None:
             return key
-        key_suffix = PRESS_KEY_SUFFIX
-        if key == ON_KEY:
-            key_suffix = ON_KEY_SUFFIX
-        elif key == OFF_KEY:
-            key_suffix = OFF_KEY_SUFFIX
-        elif key == DEVICE_BASIC_SETTINGS_KEY:
-            return KEY_PASSWORD_NOTIFY_PREFIX + self._password_encoded
-        return KEY_PASSWORD_PREFIX + self._password_encoded + key_suffix
+        key_action = key[3]
+        key_suffix = key[4:]
+        return KEY_PASSWORD_PREFIX + key_action + self._password_encoded + key_suffix
 
     def _writekey(self, key: str) -> Any:
+        if self._helper is None:
+            return
         _LOGGER.debug("Prepare to send")
-        hand = self._device.getCharacteristics(uuid=HANDLE)[0]
-        _LOGGER.debug("Sending command, %s", key)
-        write_result = hand.write(binascii.a2b_hex(key), withResponse=False)
+        try:
+            hand = self.getCharacteristics(uuid=_sb_uuid("tx"))[0]
+            _LOGGER.debug("Sending command, %s", key)
+            write_result = hand.write(bytes.fromhex(key), withResponse=False)
+        except bluepy.btle.BTLEException:
+            _LOGGER.warning(
+                "Error while enabling notifications on Switchbot", exc_info=True
+            )
+            raise
         if not write_result:
             _LOGGER.error(
                 "Sent command but didn't get a response from Switchbot confirming command was sent."
@@ -299,43 +349,71 @@ class SwitchbotDevice:
             _LOGGER.info("Successfully sent command to Switchbot (MAC: %s)", self._mac)
         return write_result
 
-    def _subscribe(self, key: str) -> Any:
+    def _subscribe(self) -> None:
+        if self._helper is None:
+            return
         _LOGGER.debug("Subscribe to notifications")
-        handle = self._device.getCharacteristics(uuid=NOTIFICATION_HANDLE)[0]
-        notify_handle = handle.getHandle() + 1
-        response = self._device.writeCharacteristic(
-            notify_handle, binascii.a2b_hex(key), withResponse=False
-        )
-        return response
+        enable_notify_flag = b"\x01\x00"  # standard gatt flag to enable notification
+        try:
+            handle = self.getCharacteristics(uuid=_sb_uuid("rx"))[0]
+            notify_handle = handle.getHandle() + 1
+            self.writeCharacteristic(
+                notify_handle, enable_notify_flag, withResponse=False
+            )
+        except bluepy.btle.BTLEException:
+            _LOGGER.warning(
+                "Error while enabling notifications on Switchbot", exc_info=True
+            )
+            raise
 
-    def _readkey(self) -> bytes | None:
+    def _readkey(self) -> bytes:
+        if self._helper is None:
+            return b''
         _LOGGER.debug("Prepare to read")
-        receive_handle = self._device.getCharacteristics(uuid=NOTIFICATION_HANDLE)
-        if receive_handle:
-            for char in receive_handle:
-                read_result: bytes = char.read()
-            return read_result
-        return None
-
-    def _sendcommand(self, key: str, retry: int) -> bool:
+        try:
+            receive_handle = self.getCharacteristics(uuid=_sb_uuid("rx"))
+            if receive_handle:
+                for char in receive_handle:
+                    read_result: bytes = char.read()
+                return read_result
+            return b"\x00"
+        except bluepy.btle.BTLEException:
+            _LOGGER.warning(
+                "Error while reading notifications from Switchbot", exc_info=True
+            )
+            raise
+
+    def _sendcommand(self, key: str, retry: int, timeout: int | None = None) -> bytes:
         send_success = False
         command = self._commandkey(key)
+        notify_msg = b"\x00"
         _LOGGER.debug("Sending command to switchbot %s", command)
+
+        if len(self._mac.split(":")) != 6:
+            raise ValueError("Expected MAC address, got %s" % repr(self._mac))
+
         with CONNECT_LOCK:
             try:
-                self._connect()
+                self._connect(retry, timeout)
+                self._subscribe()
                 send_success = self._writekey(command)
+                notify_msg = self._readkey()
             except bluepy.btle.BTLEException:
                 _LOGGER.warning("Error talking to Switchbot", exc_info=True)
             finally:
-                self._disconnect()
+                self.disconnect()
         if send_success:
-            return True
+            if notify_msg == b"\x07":
+                _LOGGER.error("Password required")
+            elif notify_msg == b"\t":
+                _LOGGER.error("Password incorrect")
+
+            return notify_msg
         if retry < 1:
             _LOGGER.error(
                 "Switchbot communication failed. Stopping trying", exc_info=True
             )
-            return False
+            return notify_msg
         _LOGGER.warning("Cannot connect to Switchbot. Retrying (remaining: %d)", retry)
         time.sleep(DEFAULT_RETRY_TIMEOUT)
         return self._sendcommand(key, retry - 1)
@@ -346,112 +424,32 @@ class SwitchbotDevice:
 
     def get_battery_percent(self) -> Any:
         """Return device battery level in percent."""
-        if not self._switchbot_device_data:
+        if not self._sb_adv_data.get("data"):
             return None
-        return self._switchbot_device_data["data"]["battery"]
+        return self._sb_adv_data["data"].get("battery")
 
     def get_device_data(
-        self, retry: int = DEFAULT_RETRY_COUNT, interface: int | None = None
-    ) -> dict | None:
+        self,
+        retry: int = DEFAULT_RETRY_COUNT,
+        interface: int | None = None,
+        passive: bool = False,
+    ) -> dict[str, Any]:
         """Find switchbot devices and their advertisement data."""
-        if interface:
-            _interface: int | None = interface
-        else:
-            _interface = self._interface
-
-        devices = None
-
-        try:
-            devices = bluepy.btle.Scanner(_interface).scan(self._scan_timeout)
+        _interface: int = interface if interface else self._interface
 
-        except bluepy.btle.BTLEManagementError:
-            _LOGGER.error("Error scanning for switchbot devices", exc_info=True)
+        dev_id = self._mac.replace(":", "")
 
-        if devices is None:
-            if retry < 1:
-                _LOGGER.error(
-                    "Scanning for Switchbot devices failed. Stop trying", exc_info=True
-                )
-                return None
-
-            _LOGGER.warning(
-                "Error scanning for Switchbot devices. Retrying (remaining: %d)",
-                retry,
-            )
-            time.sleep(DEFAULT_RETRY_TIMEOUT)
-            return self.get_device_data(retry=retry - 1, interface=_interface)
-
-        for dev in devices:
-            if self._mac.lower() == dev.addr.lower():
-                self._switchbot_device_data["mac_address"] = dev.addr
-                for (adtype, desc, value) in dev.getScanData():
-                    if adtype == 22:
-                        _data = bytes.fromhex(value[4:])
-                        _model = chr(_data[0] & 0b01111111)
-                        if _model == "H":
-                            self._switchbot_device_data["data"] = _process_wohand(_data)
-                            self._switchbot_device_data["data"]["rssi"] = dev.rssi
-                            self._switchbot_device_data["isEncrypted"] = bool(
-                                _data[0] & 0b10000000
-                            )
-                            self._switchbot_device_data["model"] = _model
-                            self._switchbot_device_data["modelName"] = "WoHand"
-                        elif _model == "c":
-                            self._switchbot_device_data["data"] = _process_wocurtain(
-                                _data
-                            )
-                            self._switchbot_device_data["data"]["rssi"] = dev.rssi
-                            self._switchbot_device_data["isEncrypted"] = bool(
-                                _data[0] & 0b10000000
-                            )
-                            self._switchbot_device_data["model"] = _model
-                            self._switchbot_device_data["modelName"] = "WoCurtain"
-                        elif _model == "T":
-                            self._switchbot_device_data["data"] = _process_wosensorth(
-                                _data
-                            )
-                            self._switchbot_device_data["data"]["rssi"] = dev.rssi
-                            self._switchbot_device_data["isEncrypted"] = bool(
-                                _data[0] & 0b10000000
-                            )
-                            self._switchbot_device_data["model"] = _model
-                            self._switchbot_device_data["modelName"] = "WoSensorTH"
-
-                        else:
-                            continue
-                    else:
-                        self._switchbot_device_data[desc] = value
-
-        return self._switchbot_device_data
-
-    def _get_basic_info(self, retry: int = DEFAULT_RETRY_COUNT) -> bytes:
-        """Get device basic settings."""
-        send_success = False
-        command = self._commandkey(DEVICE_BASIC_SETTINGS_KEY)
-        try:
-            self._connect()
-            self._subscribe(command)
-            send_success = self._writekey(command)
-            value = self._readkey()
-        except bluepy.btle.BTLEException:
-            _LOGGER.warning("Error talking to Switchbot", exc_info=True)
-        finally:
-            self._disconnect()
-
-        if send_success and value:
-
-            print("Successfully retrieved data from device " + str(self._mac))
+        _data = GetSwitchbotDevices(interface=_interface).discover(
+            retry=retry,
+            scan_timeout=self._scan_timeout,
+            passive=passive,
+            mac=self._mac,
+        )
 
-            return value
+        if _data.get(dev_id):
+            self._sb_adv_data = _data[dev_id]
 
-        if retry < 1:
-            _LOGGER.error(
-                "Switchbot communication failed. Stopping trying", exc_info=True
-            )
-            return bytes(0)
-        _LOGGER.warning("Cannot connect to Switchbot. Retrying (remaining: %d)", retry)
-        time.sleep(DEFAULT_RETRY_TIMEOUT)
-        return self._get_basic_info(retry - 1)
+        return self._sb_adv_data
 
 
 class Switchbot(SwitchbotDevice):
@@ -463,52 +461,141 @@ class Switchbot(SwitchbotDevice):
         self._inverse: bool = kwargs.pop("inverse_mode", False)
         self._settings: dict[str, Any] = {}
 
-    def update(self, interface: int | None = None) -> None:
+    def update(self, interface: int | None = None, passive: bool = False) -> None:
         """Update mode, battery percent and state of device."""
-        self.get_device_data(retry=self._retry_count, interface=interface)
+        self.get_device_data(
+            retry=self._retry_count, interface=interface, passive=passive
+        )
 
     def turn_on(self) -> bool:
         """Turn device on."""
-        return self._sendcommand(ON_KEY, self._retry_count)
+        result = self._sendcommand(ON_KEY, self._retry_count)
+
+        if result[0] == 1:
+            return True
+
+        if result[0] == 5:
+            _LOGGER.debug("Bot is in press mode and doesn't have on state")
+            return True
+
+        return False
 
     def turn_off(self) -> bool:
         """Turn device off."""
-        return self._sendcommand(OFF_KEY, self._retry_count)
+        result = self._sendcommand(OFF_KEY, self._retry_count)
+        if result[0] == 1:
+            return True
+
+        if result[0] == 5:
+            _LOGGER.debug("Bot is in press mode and doesn't have off state")
+            return True
+
+        return False
+
+    def hand_up(self) -> bool:
+        """Raise device arm."""
+        result = self._sendcommand(UP_KEY, self._retry_count)
+        if result[0] == 1:
+            return True
+
+        if result[0] == 5:
+            _LOGGER.debug("Bot is in press mode")
+            return True
+
+        return False
+
+    def hand_down(self) -> bool:
+        """Lower device arm."""
+        result = self._sendcommand(DOWN_KEY, self._retry_count)
+        if result[0] == 1:
+            return True
+
+        if result[0] == 5:
+            _LOGGER.debug("Bot is in press mode")
+            return True
+
+        return False
 
     def press(self) -> bool:
         """Press command to device."""
-        return self._sendcommand(PRESS_KEY, self._retry_count)
+        result = self._sendcommand(PRESS_KEY, self._retry_count)
+        if result[0] == 1:
+            return True
+
+        if result[0] == 5:
+            _LOGGER.debug("Bot is in switch mode")
+            return True
+
+        return False
+
+    def set_switch_mode(
+        self, switch_mode: bool = False, strength: int = 100, inverse: bool = False
+    ) -> bool:
+        """Change bot mode."""
+        mode_key = format(switch_mode, "b") + format(inverse, "b")
+        strength_key = f"{strength:0{2}x}"  # to hex with padding to double digit
+
+        result = self._sendcommand(
+            DEVICE_SET_MODE_KEY + strength_key + mode_key, self._retry_count
+        )
+
+        if result[0] == 1:
+            return True
+
+        return False
+
+    def set_long_press(self, duration: int = 0) -> bool:
+        """Set bot long press duration."""
+        duration_key = f"{duration:0{2}x}"  # to hex with padding to double digit
+
+        result = self._sendcommand(
+            DEVICE_SET_EXTENDED_KEY + "08" + duration_key, self._retry_count
+        )
+
+        if result[0] == 1:
+            return True
+
+        return False
 
-    def get_basic_info(self) -> dict[str, Any]:
+    def get_basic_info(self) -> dict[str, Any] | None:
         """Get device basic settings."""
-        settings_data = self._get_basic_info()
-        self._settings["battery"] = settings_data[1]
-        self._settings["firmware"] = settings_data[2] / 10.0
+        _data = self._sendcommand(
+            key=DEVICE_GET_BASIC_SETTINGS_KEY, retry=self._retry_count
+        )
+
+        if _data in (b"\x07", b"\x00"):
+            _LOGGER.error("Unsuccessfull, please try again")
+            return None
 
-        self._settings["timers"] = settings_data[8]
-        self._settings["dualStateMode"] = bool(settings_data[9] & 16)
-        self._settings["inverseDirection"] = bool(settings_data[9] & 1)
-        self._settings["holdSeconds"] = settings_data[10]
+        self._settings = {
+            "battery": _data[1],
+            "firmware": _data[2] / 10.0,
+            "strength": _data[3],
+            "timers": _data[8],
+            "switchMode": bool(_data[9] & 16),
+            "inverseDirection": bool(_data[9] & 1),
+            "holdSeconds": _data[10],
+        }
 
         return self._settings
 
     def switch_mode(self) -> Any:
         """Return true or false from cache."""
         # To get actual position call update() first.
-        if not self._switchbot_device_data:
+        if not self._sb_adv_data.get("data"):
             return None
-        return self._switchbot_device_data["data"]["switchMode"]
+        return self._sb_adv_data.get("switchMode")
 
     def is_on(self) -> Any:
         """Return switch state from cache."""
         # To get actual position call update() first.
-        if not self._switchbot_device_data:
+        if not self._sb_adv_data.get("data"):
             return None
 
         if self._inverse:
-            return not self._switchbot_device_data["data"]["isOn"]
+            return not self._sb_adv_data["data"].get("isOn")
 
-        return self._switchbot_device_data["data"]["isOn"]
+        return self._sb_adv_data["data"].get("isOn")
 
 
 class SwitchbotCurtain(SwitchbotDevice):
@@ -528,77 +615,164 @@ class SwitchbotCurtain(SwitchbotDevice):
         super().__init__(*args, **kwargs)
         self._reverse: bool = kwargs.pop("reverse_mode", True)
         self._settings: dict[str, Any] = {}
+        self.ext_info_sum: dict[str, Any] = {}
+        self.ext_info_adv: dict[str, Any] = {}
 
     def open(self) -> bool:
         """Send open command."""
-        return self._sendcommand(OPEN_KEY, self._retry_count)
+        result = self._sendcommand(OPEN_KEY, self._retry_count)
+        if result[0] == 1:
+            return True
+
+        return False
 
     def close(self) -> bool:
         """Send close command."""
-        return self._sendcommand(CLOSE_KEY, self._retry_count)
+        result = self._sendcommand(CLOSE_KEY, self._retry_count)
+        if result[0] == 1:
+            return True
+
+        return False
 
     def stop(self) -> bool:
         """Send stop command to device."""
-        return self._sendcommand(STOP_KEY, self._retry_count)
+        result = self._sendcommand(STOP_KEY, self._retry_count)
+        if result[0] == 1:
+            return True
+
+        return False
 
     def set_position(self, position: int) -> bool:
         """Send position command (0-100) to device."""
         position = (100 - position) if self._reverse else position
         hex_position = "%0.2X" % position
-        return self._sendcommand(POSITION_KEY + hex_position, self._retry_count)
+        result = self._sendcommand(POSITION_KEY + hex_position, self._retry_count)
+        if result[0] == 1:
+            return True
 
-    def update(self, interface: int | None = None) -> None:
+        return False
+
+    def update(self, interface: int | None = None, passive: bool = False) -> None:
         """Update position, battery percent and light level of device."""
-        self.get_device_data(retry=self._retry_count, interface=interface)
+        self.get_device_data(
+            retry=self._retry_count, interface=interface, passive=passive
+        )
 
     def get_position(self) -> Any:
         """Return cached position (0-100) of Curtain."""
         # To get actual position call update() first.
-        if not self._switchbot_device_data:
+        if not self._sb_adv_data.get("data"):
             return None
-        return self._switchbot_device_data["data"]["position"]
+        return self._sb_adv_data["data"].get("position")
 
-    def get_basic_info(self) -> dict[str, Any]:
+    def get_basic_info(self) -> dict[str, Any] | None:
         """Get device basic settings."""
-        settings_data = self._get_basic_info()
-        self._settings["battery"] = settings_data[1]
-        self._settings["firmware"] = settings_data[2] / 10.0
+        _data = self._sendcommand(
+            key=DEVICE_GET_BASIC_SETTINGS_KEY, retry=self._retry_count
+        )
 
-        self._settings["chainLength"] = settings_data[3]
+        if _data in (b"\x07", b"\x00"):
+            _LOGGER.error("Unsuccessfull, please try again")
+            return None
 
-        self._settings["openDirection"] = (
-            "right_to_left" if settings_data[4] & 0b10000000 == 128 else "left_to_right"
-        )
+        _position = max(min(_data[6], 100), 0)
+
+        self._settings = {
+            "battery": _data[1],
+            "firmware": _data[2] / 10.0,
+            "chainLength": _data[3],
+            "openDirection": (
+                "right_to_left" if _data[4] & 0b10000000 == 128 else "left_to_right"
+            ),
+            "touchToOpen": bool(_data[4] & 0b01000000),
+            "light": bool(_data[4] & 0b00100000),
+            "fault": bool(_data[4] & 0b00001000),
+            "solarPanel": bool(_data[5] & 0b00001000),
+            "calibrated": bool(_data[5] & 0b00000100),
+            "inMotion": bool(_data[5] & 0b01000011),
+            "position": (100 - _position) if self._reverse else _position,
+            "timers": _data[7],
+        }
 
-        self._settings["touchToOpen"] = bool(settings_data[4] & 0b01000000)
-        self._settings["light"] = bool(settings_data[4] & 0b00100000)
-        self._settings["fault"] = bool(settings_data[4] & 0b00001000)
+        return self._settings
 
-        self._settings["solarPanel"] = bool(settings_data[5] & 0b00001000)
-        self._settings["calibrated"] = bool(settings_data[5] & 0b00000100)
-        self._settings["inMotion"] = bool(settings_data[5] & 0b01000011)
+    def get_extended_info_summary(self) -> dict[str, Any] | None:
+        """Get basic info for all devices in chain."""
+        _data = self._sendcommand(key=CURTAIN_EXT_SUM_KEY, retry=self._retry_count)
 
-        _position = max(min(settings_data[6], 100), 0)
-        self._settings["position"] = (100 - _position) if self._reverse else _position
+        if _data in (b"\x07", b"\x00"):
+            _LOGGER.error("Unsuccessfull, please try again")
+            return None
 
-        self._settings["timers"] = settings_data[7]
+        self.ext_info_sum["device0"] = {
+            "openDirectionDefault": not bool(_data[1] & 0b10000000),
+            "touchToOpen": bool(_data[1] & 0b01000000),
+            "light": bool(_data[1] & 0b00100000),
+            "openDirection": (
+                "left_to_right" if _data[1] & 0b00010000 == 1 else "right_to_left"
+            ),
+        }
+
+        # if grouped curtain device present.
+        if _data[2] != 0:
+            self.ext_info_sum["device1"] = {
+                "openDirectionDefault": not bool(_data[1] & 0b10000000),
+                "touchToOpen": bool(_data[1] & 0b01000000),
+                "light": bool(_data[1] & 0b00100000),
+                "openDirection": (
+                    "left_to_right" if _data[1] & 0b00010000 else "right_to_left"
+                ),
+            }
+
+        return self.ext_info_sum
+
+    def get_extended_info_adv(self) -> dict[str, Any] | None:
+        """Get advance page info for device chain."""
+        _data = self._sendcommand(key=CURTAIN_EXT_ADV_KEY, retry=self._retry_count)
+
+        if _data in (b"\x07", b"\x00"):
+            _LOGGER.error("Unsuccessfull, please try again")
+            return None
 
-        return self._settings
+        _state_of_charge = [
+            "not_charging",
+            "charging_by_adapter",
+            "charging_by_solar",
+            "fully_charged",
+            "solar_not_charging",
+            "charging_error",
+        ]
+
+        self.ext_info_adv["device0"] = {
+            "battery": _data[1],
+            "firmware": _data[2] / 10.0,
+            "stateOfCharge": _state_of_charge[_data[3]],
+        }
+
+        # If grouped curtain device present.
+        if _data[4]:
+            self.ext_info_adv["device1"] = {
+                "battery": _data[4],
+                "firmware": _data[5] / 10.0,
+                "stateOfCharge": _state_of_charge[_data[6]],
+            }
+
+        return self.ext_info_adv
 
     def get_light_level(self) -> Any:
         """Return cached light level."""
         # To get actual light level call update() first.
-        if not self._switchbot_device_data:
+        if not self._sb_adv_data.get("data"):
             return None
-        return self._switchbot_device_data["data"]["lightLevel"]
+        return self._sb_adv_data["data"].get("lightLevel")
 
     def is_reversed(self) -> bool:
-        """Return True if the curtain open from left to right."""
+        """Return True if curtain position is opposite from SB data."""
         return self._reverse
 
     def is_calibrated(self) -> Any:
         """Return True curtain is calibrated."""
         # To get actual light level call update() first.
-        if not self._switchbot_device_data:
+        if not self._sb_adv_data.get("data"):
             return None
-        return self._switchbot_device_data["data"]["calibration"]
+        return self._sb_adv_data["data"].get("calibration")