adv_parser.py 4.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138
  1. """Library to handle connection with Switchbot."""
  2. from __future__ import annotations
  3. import logging
  4. from collections.abc import Callable
  5. from functools import lru_cache
  6. from typing import TypedDict
  7. from bleak.backends.device import BLEDevice
  8. from bleak.backends.scanner import AdvertisementData
  9. from .adv_parsers.bot import process_wohand
  10. from .adv_parsers.bulb import process_color_bulb
  11. from .adv_parsers.ceiling_light import process_woceiling
  12. from .adv_parsers.contact import process_wocontact
  13. from .adv_parsers.curtain import process_wocurtain
  14. from .adv_parsers.humidifier import process_wohumidifier
  15. from .adv_parsers.light_strip import process_wostrip
  16. from .adv_parsers.meter import process_wosensorth
  17. from .adv_parsers.motion import process_wopresence
  18. from .adv_parsers.plug import process_woplugmini
  19. from .const import SwitchbotModel
  20. from .models import SwitchBotAdvertisement
  21. _LOGGER = logging.getLogger(__name__)
  22. class SwitchbotSupportedType(TypedDict):
  23. """Supported type of Switchbot."""
  24. modelName: SwitchbotModel
  25. modelFriendlyName: str
  26. func: Callable[[bytes, bytes | None], dict[str, bool | int]]
  27. SUPPORTED_TYPES: dict[str, SwitchbotSupportedType] = {
  28. "d": {
  29. "modelName": SwitchbotModel.CONTACT_SENSOR,
  30. "modelFriendlyName": "Contact Sensor",
  31. "func": process_wocontact,
  32. },
  33. "H": {
  34. "modelName": SwitchbotModel.BOT,
  35. "modelFriendlyName": "Bot",
  36. "func": process_wohand,
  37. },
  38. "s": {
  39. "modelName": SwitchbotModel.MOTION_SENSOR,
  40. "modelFriendlyName": "Motion Sensor",
  41. "func": process_wopresence,
  42. },
  43. "r": {
  44. "modelName": SwitchbotModel.LIGHT_STRIP,
  45. "modelFriendlyName": "Light Strip",
  46. "func": process_wostrip,
  47. },
  48. "c": {
  49. "modelName": SwitchbotModel.CURTAIN,
  50. "modelFriendlyName": "Curtain",
  51. "func": process_wocurtain,
  52. },
  53. "T": {
  54. "modelName": SwitchbotModel.METER,
  55. "modelFriendlyName": "Meter",
  56. "func": process_wosensorth,
  57. },
  58. "i": {
  59. "modelName": SwitchbotModel.METER,
  60. "modelFriendlyName": "Meter Plus",
  61. "func": process_wosensorth,
  62. },
  63. "g": {
  64. "modelName": SwitchbotModel.PLUG_MINI,
  65. "modelFriendlyName": "Plug Mini",
  66. "func": process_woplugmini,
  67. },
  68. "u": {
  69. "modelName": SwitchbotModel.COLOR_BULB,
  70. "modelFriendlyName": "Color Bulb",
  71. "func": process_color_bulb,
  72. },
  73. "q": {
  74. "modelName": SwitchbotModel.CEILING_LIGHT,
  75. "modelFriendlyName": "Ceiling Light",
  76. "func": process_woceiling,
  77. },
  78. "e": {
  79. "modelName": SwitchbotModel.HUMIDIFIER,
  80. "modelFriendlyName": "Humidifier",
  81. "func": process_wohumidifier,
  82. },
  83. }
  84. def parse_advertisement_data(
  85. device: BLEDevice, advertisement_data: AdvertisementData
  86. ) -> SwitchBotAdvertisement | None:
  87. """Parse advertisement data."""
  88. _services = list(advertisement_data.service_data.values())
  89. _mgr_datas = list(advertisement_data.manufacturer_data.values())
  90. if not _services:
  91. return None
  92. _service_data = _services[0]
  93. if not _service_data:
  94. return None
  95. _mfr_data = _mgr_datas[0] if _mgr_datas else None
  96. data = _parse_data(_service_data, _mfr_data)
  97. return SwitchBotAdvertisement(device.address, data, device)
  98. @lru_cache(maxsize=128)
  99. def _parse_data(
  100. _service_data: bytes, _mfr_data: bytes | None
  101. ) -> SwitchBotAdvertisement | None:
  102. """Parse advertisement data."""
  103. _model = chr(_service_data[0] & 0b01111111)
  104. data = {
  105. "rawAdvData": _service_data,
  106. "data": {},
  107. "model": _model,
  108. "isEncrypted": bool(_service_data[0] & 0b10000000),
  109. }
  110. type_data = SUPPORTED_TYPES.get(_model)
  111. if type_data:
  112. model_data = type_data["func"](_service_data, _mfr_data)
  113. if model_data:
  114. data.update(
  115. {
  116. "modelFriendlyName": type_data["modelFriendlyName"],
  117. "modelName": type_data["modelName"],
  118. "data": model_data,
  119. }
  120. )
  121. return data