Skip to content

Commit 1451c61

Browse files
bdracobdraco
authored
Improve handling of passive only data when active scans are not available (#153)
1 parent 2d57f63 commit 1451c61

File tree

12 files changed

+179
-27
lines changed

12 files changed

+179
-27
lines changed

‎switchbot/adv_parser.py‎

Lines changed: 71 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@
2929
"0000fd3d-0000-1000-8000-00805f9b34fb",
3030
"00000d00-0000-1000-8000-00805f9b34fb",
3131
)
32+
MFR_DATA_ORDER = (2409, 89)
3233

3334

3435
class SwitchbotSupportedType(TypedDict):
@@ -44,11 +45,15 @@ class SwitchbotSupportedType(TypedDict):
4445
"modelName": SwitchbotModel.CONTACT_SENSOR,
4546
"modelFriendlyName": "Contact Sensor",
4647
"func": process_wocontact,
48+
"manufacturer_id": 2409,
49+
"manufacturer_data_length": 13,
4750
},
4851
"H": {
4952
"modelName": SwitchbotModel.BOT,
5053
"modelFriendlyName": "Bot",
5154
"func": process_wohand,
55+
"service_uuids": {"cba20d00-224d-11e6-9fb8-0002a5d5c51b"},
56+
"manufacturer_id": 89,
5257
},
5358
"s": {
5459
"modelName": SwitchbotModel.MOTION_SENSOR,
@@ -64,6 +69,12 @@ class SwitchbotSupportedType(TypedDict):
6469
"modelName": SwitchbotModel.CURTAIN,
6570
"modelFriendlyName": "Curtain",
6671
"func": process_wocurtain,
72+
"service_uuids": {
73+
"00001800-0000-1000-8000-00805f9b34fb",
74+
"00001801-0000-1000-8000-00805f9b34fb",
75+
"cba20d00-224d-11e6-9fb8-0002a5d5c51b",
76+
},
77+
"manufacturer_id": 89,
6778
},
6879
"T": {
6980
"modelName": SwitchbotModel.METER,
@@ -107,50 +118,98 @@ class SwitchbotSupportedType(TypedDict):
107118
},
108119
}
109120

121+
_SWITCHBOT_MODEL_TO_CHAR = {
122+
model_data["modelName"]: model_chr
123+
for model_chr, model_data in SUPPORTED_TYPES.items()
124+
}
125+
126+
MODELS_BY_MANUFACTURER_DATA: dict[int, list[tuple[str, SwitchbotSupportedType]]] = {
127+
mfr_id: [] for mfr_id in MFR_DATA_ORDER
128+
}
129+
for model_chr, model in SUPPORTED_TYPES.items():
130+
if "manufacturer_id" in model:
131+
mfr_id = model["manufacturer_id"]
132+
MODELS_BY_MANUFACTURER_DATA[mfr_id].append((model_chr, model))
133+
110134

111135
def parse_advertisement_data(
112-
device: BLEDevice, advertisement_data: AdvertisementData
136+
device: BLEDevice,
137+
advertisement_data: AdvertisementData,
138+
model: SwitchbotModel | None = None,
113139
) -> SwitchBotAdvertisement | None:
114140
"""Parse advertisement data."""
115-
_mgr_datas = list(advertisement_data.manufacturer_data.values())
116141
service_data = advertisement_data.service_data
117142

118-
if not service_data:
119-
return None
120-
121143
_service_data = None
122144
for uuid in SERVICE_DATA_ORDER:
123145
if uuid in service_data:
124146
_service_data = service_data[uuid]
125147
break
126148

127-
if not _service_data:
128-
return None
149+
_mfr_data = None
150+
_mfr_id = None
151+
for mfr_id in MFR_DATA_ORDER:
152+
if mfr_id in advertisement_data.manufacturer_data:
153+
_mfr_id = mfr_id
154+
_mfr_data = advertisement_data.manufacturer_data[mfr_id]
155+
break
129156

130-
_mfr_data = _mgr_datas[0] if _mgr_datas else None
157+
if _mfr_data is None and _service_data is None:
158+
return None
131159

132160
try:
133-
data = _parse_data(_service_data, _mfr_data)
161+
data = _parse_data(
162+
"".join(sorted(advertisement_data.service_uuids)),
163+
_service_data,
164+
_mfr_data,
165+
_mfr_id,
166+
model,
167+
)
134168
except Exception as err: # pylint: disable=broad-except
135169
_LOGGER.exception(
136170
"Failed to parse advertisement data: %s: %s", advertisement_data, err
137171
)
138172
return None
139173

174+
if not data:
175+
return None
176+
140177
return SwitchBotAdvertisement(device.address, data, device, advertisement_data.rssi)
141178

142179

143180
@lru_cache(maxsize=128)
144181
def _parse_data(
145-
_service_data: bytes, _mfr_data: bytes | None
182+
_service_uuids_str: str,
183+
_service_data: bytes | None,
184+
_mfr_data: bytes | None,
185+
_mfr_id: int | None = None,
186+
_switchbot_model: SwitchbotModel | None = None,
146187
) -> SwitchBotAdvertisement | None:
147188
"""Parse advertisement data."""
148-
_model = chr(_service_data[0] & 0b01111111)
189+
_model = chr(_service_data[0] & 0b01111111) if _service_data else None
190+
191+
if not _model and _mfr_id and _mfr_id in MODELS_BY_MANUFACTURER_DATA:
192+
_service_uuids = set(_service_uuids_str.split(","))
193+
for model_chr, model_data in MODELS_BY_MANUFACTURER_DATA[_mfr_id]:
194+
if model_data.get("manufacturer_data_length") == len(_mfr_data):
195+
_model = model_chr
196+
break
197+
if model_data.get("service_uuids", set()).intersection(_service_uuids):
198+
_model = model_chr
199+
break
200+
201+
if not _model and _switchbot_model and _switchbot_model in _SWITCHBOT_MODEL_TO_CHAR:
202+
_model = _SWITCHBOT_MODEL_TO_CHAR[_switchbot_model]
203+
204+
if not _model:
205+
return None
206+
207+
_isEncrypted = bool(_service_data[0] & 0b10000000) if _service_data else False
149208
data = {
150209
"rawAdvData": _service_data,
151210
"data": {},
152211
"model": _model,
153-
"isEncrypted": bool(_service_data[0] & 0b10000000),
212+
"isEncrypted": _isEncrypted,
154213
}
155214

156215
type_data = SUPPORTED_TYPES.get(_model)

‎switchbot/adv_parsers/bot.py‎

Lines changed: 11 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,8 +2,18 @@
22
from __future__ import annotations
33

44

5-
def process_wohand(data: bytes, mfr_data: bytes | None) -> dict[str, bool | int]:
5+
def process_wohand(data: bytes | None, mfr_data: bytes | None) -> dict[str, bool | int]:
66
"""Process woHand/Bot services data."""
7+
if data is None and mfr_data is None:
8+
return {}
9+
10+
if data is None:
11+
return {
12+
"switchMode": None,
13+
"isOn": None,
14+
"battery": None,
15+
}
16+
717
_switch_mode = bool(data[1] & 0b10000000)
818

919
return {

‎switchbot/adv_parsers/bulb.py‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,9 @@
22
from __future__ import annotations
33

44

5-
def process_color_bulb(data: bytes, mfr_data: bytes | None) -> dict[str, bool | int]:
5+
def process_color_bulb(
6+
data: bytes | None, mfr_data: bytes | None
7+
) -> dict[str, bool | int]:
68
"""Process WoBulb services data."""
79
if mfr_data is None:
810
return {}

‎switchbot/adv_parsers/contact.py‎

Lines changed: 9 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -2,11 +2,16 @@
22
from __future__ import annotations
33

44

5-
def process_wocontact(data: bytes, mfr_data: bytes | None) -> dict[str, bool | int]:
5+
def process_wocontact(
6+
data: bytes | None, mfr_data: bytes | None
7+
) -> dict[str, bool | int]:
68
"""Process woContact Sensor services data."""
7-
battery = data[2] & 0b01111111
8-
tested = bool(data[1] & 0b10000000)
9-
contact_timeout = data[3] & 0b00000100 == 0b00000100
9+
if data is None and mfr_data is None:
10+
return {}
11+
12+
battery = data[2] & 0b01111111 if data else None
13+
tested = bool(data[1] & 0b10000000) if data else None
14+
contact_timeout = data[3] & 0b00000100 == 0b00000100 if data else False
1015

1116
if mfr_data and len(mfr_data) >= 13:
1217
motion_detected = bool(mfr_data[7] & 0b10000000)

‎switchbot/adv_parsers/curtain.py‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3,9 +3,11 @@
33

44

55
def process_wocurtain(
6-
data: bytes, mfr_data: bytes | None, reverse: bool = True
6+
data: bytes | None, mfr_data: bytes | None, reverse: bool = True
77
) -> dict[str, bool | int]:
88
"""Process woCurtain/Curtain services data."""
9+
if data is None:
10+
return {}
911

1012
_position = max(min(data[3] & 0b01111111, 100), 0)
1113

‎switchbot/adv_parsers/humidifier.py‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -13,9 +13,11 @@
1313
# Low: 658000c5222b6300
1414
# Med: 658000c5432b6300
1515
# High: 658000c5642b6300
16-
def process_wohumidifier(data: bytes, mfr_data: bytes | None) -> dict[str, bool | int]:
16+
def process_wohumidifier(
17+
data: bytes | None, mfr_data: bytes | None
18+
) -> dict[str, bool | int]:
1719
"""Process WoHumi services data."""
18-
if mfr_data is None:
20+
if data is None:
1921
return {}
2022
_LOGGER.debug("mfr_data: %s", mfr_data.hex())
2123
_LOGGER.debug("data: %s", data.hex())

‎switchbot/adv_parsers/light_strip.py‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,9 @@
22
from __future__ import annotations
33

44

5-
def process_wostrip(data: bytes, mfr_data: bytes | None) -> dict[str, bool | int]:
5+
def process_wostrip(
6+
data: bytes | None, mfr_data: bytes | None
7+
) -> dict[str, bool | int]:
68
"""Process WoStrip services data."""
79
if mfr_data is None:
810
return {}

‎switchbot/adv_parsers/lock.py‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@
88
_LOGGER = logging.getLogger(__name__)
99

1010

11-
def process_wolock(data: bytes, mfr_data: bytes | None) -> dict[str, bool | int]:
11+
def process_wolock(data: bytes | None, mfr_data: bytes | None) -> dict[str, bool | int]:
1212
"""Process woLock services data."""
1313
if mfr_data is None:
1414
return {}
@@ -17,7 +17,7 @@ def process_wolock(data: bytes, mfr_data: bytes | None) -> dict[str, bool | int]
1717
_LOGGER.debug("data: %s", data.hex())
1818

1919
return {
20-
"battery": data[2] & 0b01111111,
20+
"battery": data[2] & 0b01111111 if data else None,
2121
"calibration": bool(mfr_data[7] & 0b10000000),
2222
"status": LockStatus(mfr_data[7] & 0b01110000),
2323
"update_from_secondary_lock": bool(mfr_data[7] & 0b00001000),

‎switchbot/adv_parsers/meter.py‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,8 +4,10 @@
44
from typing import Any
55

66

7-
def process_wosensorth(data: bytes, mfr_data: bytes | None) -> dict[str, Any]:
7+
def process_wosensorth(data: bytes | None, mfr_data: bytes | None) -> dict[str, Any]:
88
"""Process woSensorTH/Temp sensor services data."""
9+
if data is None:
10+
return {}
911

1012
_temp_sign = 1 if data[4] & 0b10000000 else -1
1113
_temp_c = _temp_sign * ((data[4] & 0b01111111) + ((data[3] & 0b00001111) / 10))

‎switchbot/adv_parsers/motion.py‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,8 +2,12 @@
22
from __future__ import annotations
33

44

5-
def process_wopresence(data: bytes, mfr_data: bytes | None) -> dict[str, bool | int]:
5+
def process_wopresence(
6+
data: bytes | None, mfr_data: bytes | None
7+
) -> dict[str, bool | int]:
68
"""Process WoPresence Sensor services data."""
9+
if data is None:
10+
return {}
711
return {
812
"tested": bool(data[1] & 0b10000000),
913
"motion_detected": bool(data[1] & 0b01000000),

0 commit comments

Comments
 (0)