1
0

adv_parser.py 3.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108
  1. """Library to handle connection with Switchbot."""
  2. from __future__ import annotations
  3. from collections.abc import Callable
  4. from typing import TypedDict
  5. from bleak.backends.device import BLEDevice
  6. from bleak.backends.scanner import AdvertisementData
  7. from .adv_parsers.bot import process_wohand
  8. from .adv_parsers.bulb import process_color_bulb
  9. from .adv_parsers.contact import process_wocontact
  10. from .adv_parsers.curtain import process_wocurtain
  11. from .adv_parsers.meter import process_wosensorth
  12. from .adv_parsers.motion import process_wopresence
  13. from .adv_parsers.plug import process_woplugmini
  14. from .const import SwitchbotModel
  15. from .models import SwitchBotAdvertisement
  16. class SwitchbotSupportedType(TypedDict):
  17. """Supported type of Switchbot."""
  18. modelName: SwitchbotModel
  19. modelFriendlyName: str
  20. func: Callable[[bytes, bytes | None], dict[str, bool | int]]
  21. SUPPORTED_TYPES: dict[str, SwitchbotSupportedType] = {
  22. "d": {
  23. "modelName": SwitchbotModel.CONTACT_SENSOR,
  24. "modelFriendlyName": "Contact Sensor",
  25. "func": process_wocontact,
  26. },
  27. "H": {
  28. "modelName": SwitchbotModel.BOT,
  29. "modelFriendlyName": "Bot",
  30. "func": process_wohand,
  31. },
  32. "s": {
  33. "modelName": SwitchbotModel.MOTION_SENSOR,
  34. "modelFriendlyName": "Motion Sensor",
  35. "func": process_wopresence,
  36. },
  37. "c": {
  38. "modelName": SwitchbotModel.CURTAIN,
  39. "modelFriendlyName": "Curtain",
  40. "func": process_wocurtain,
  41. },
  42. "T": {
  43. "modelName": SwitchbotModel.METER,
  44. "modelFriendlyName": "Meter",
  45. "func": process_wosensorth,
  46. },
  47. "i": {
  48. "modelName": SwitchbotModel.METER,
  49. "modelFriendlyName": "Meter Plus",
  50. "func": process_wosensorth,
  51. },
  52. "g": {
  53. "modelName": SwitchbotModel.PLUG_MINI,
  54. "modelFriendlyName": "Plug Mini",
  55. "func": process_woplugmini,
  56. },
  57. "u": {
  58. "modelName": SwitchbotModel.COLOR_BULB,
  59. "modelFriendlyName": "Color Bulb",
  60. "func": process_color_bulb,
  61. },
  62. }
  63. def parse_advertisement_data(
  64. device: BLEDevice, advertisement_data: AdvertisementData
  65. ) -> SwitchBotAdvertisement | None:
  66. """Parse advertisement data."""
  67. _services = list(advertisement_data.service_data.values())
  68. _mgr_datas = list(advertisement_data.manufacturer_data.values())
  69. if not _services:
  70. return None
  71. _service_data = _services[0]
  72. if not _service_data:
  73. return None
  74. _mfr_data = _mgr_datas[0] if _mgr_datas else None
  75. _model = chr(_service_data[0] & 0b01111111)
  76. data = {
  77. "address": device.address, # MacOS uses UUIDs
  78. "rawAdvData": list(advertisement_data.service_data.values())[0],
  79. "data": {},
  80. }
  81. type_data = SUPPORTED_TYPES.get(_model)
  82. if type_data:
  83. data.update(
  84. {
  85. "isEncrypted": bool(_service_data[0] & 0b10000000),
  86. "model": _model,
  87. "modelFriendlyName": type_data["modelFriendlyName"],
  88. "modelName": type_data["modelName"],
  89. "data": type_data["func"](_service_data, _mfr_data),
  90. }
  91. )
  92. data["data"]["rssi"] = device.rssi
  93. return SwitchBotAdvertisement(device.address, data, device)