123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302 |
- from __future__ import annotations
- import abc
- import logging
- import typing
- import aiomqtt
- import bleak
- import bleak.backends.device
- import switchbot
- from switchbot_mqtt._utils import (
- _join_mqtt_topic_levels,
- _mac_address_valid,
- _MQTTTopicLevel,
- _MQTTTopicPlaceholder,
- _parse_mqtt_topic,
- )
- _LOGGER = logging.getLogger(__name__)
- class _MQTTControlledActor(abc.ABC):
- MQTT_COMMAND_TOPIC_LEVELS: typing.Tuple[_MQTTTopicLevel, ...] = NotImplemented
- _MQTT_UPDATE_DEVICE_INFO_TOPIC_LEVELS: typing.Tuple[_MQTTTopicLevel, ...] = (
- NotImplemented
- )
- MQTT_STATE_TOPIC_LEVELS: typing.Tuple[_MQTTTopicLevel, ...] = NotImplemented
- _MQTT_BATTERY_PERCENTAGE_TOPIC_LEVELS: typing.Tuple[_MQTTTopicLevel, ...] = (
- NotImplemented
- )
- @classmethod
- def get_mqtt_update_device_info_topic(cls, *, prefix: str, mac_address: str) -> str:
- return _join_mqtt_topic_levels(
- topic_prefix=prefix,
- topic_levels=cls._MQTT_UPDATE_DEVICE_INFO_TOPIC_LEVELS,
- mac_address=mac_address,
- )
- @classmethod
- def get_mqtt_battery_percentage_topic(cls, *, prefix: str, mac_address: str) -> str:
- return _join_mqtt_topic_levels(
- topic_prefix=prefix,
- topic_levels=cls._MQTT_BATTERY_PERCENTAGE_TOPIC_LEVELS,
- mac_address=mac_address,
- )
- @abc.abstractmethod
- def __init__(
- self,
- *,
- device: bleak.backends.device.BLEDevice,
- retry_count: int,
- password: typing.Optional[str],
- ) -> None:
-
- self._mac_address = device.address
- self._basic_device_info: typing.Optional[typing.Dict[str, typing.Any]] = None
- @abc.abstractmethod
- def _get_device(self) -> switchbot.SwitchbotDevice:
- raise NotImplementedError()
- async def _report_battery_level(
- self, mqtt_client: aiomqtt.Client, mqtt_topic_prefix: str
- ) -> None:
- assert self._basic_device_info is not None
-
-
- await self._mqtt_publish(
- topic_prefix=mqtt_topic_prefix,
- topic_levels=self._MQTT_BATTERY_PERCENTAGE_TOPIC_LEVELS,
- payload=str(self._basic_device_info["battery"]).encode(),
- mqtt_client=mqtt_client,
- )
- async def _update_and_report_device_info(
- self, mqtt_client: aiomqtt.Client, mqtt_topic_prefix: str
- ) -> None:
- self._basic_device_info = await self._get_device().get_basic_info()
- if self._basic_device_info is None:
- _LOGGER.error(
- "failed to retrieve basic device info from %s", self._mac_address
- )
- else:
- await self._report_battery_level(
- mqtt_client=mqtt_client, mqtt_topic_prefix=mqtt_topic_prefix
- )
- @classmethod
- async def _init_from_topic(
- cls,
- *,
- topic: aiomqtt.Topic,
- mqtt_topic_prefix: str,
- expected_topic_levels: typing.Collection[_MQTTTopicLevel],
- retry_count: int,
- device_passwords: typing.Dict[str, str],
- ) -> typing.Optional[_MQTTControlledActor]:
- try:
- mac_address = _parse_mqtt_topic(
- topic=topic.value,
- expected_prefix=mqtt_topic_prefix,
- expected_levels=expected_topic_levels,
- )[_MQTTTopicPlaceholder.MAC_ADDRESS]
- except ValueError as exc:
- _LOGGER.warning(str(exc), exc_info=False)
- return None
- if not _mac_address_valid(mac_address):
- _LOGGER.warning("invalid mac address %s", mac_address)
- return None
-
-
-
-
- device = await bleak.BleakScanner.find_device_by_address(mac_address)
- if device is None:
- _LOGGER.error(
- "failed to find bluetooth low energy device with mac address %s",
- mac_address,
- )
- return None
- return cls(
- device=device,
- retry_count=retry_count,
- password=device_passwords.get(mac_address, None),
- )
- @classmethod
- async def _mqtt_update_device_info_callback(
-
- cls,
- *,
- mqtt_client: aiomqtt.Client,
- message: aiomqtt.Message,
- mqtt_topic_prefix: str,
- retry_count: int,
- device_passwords: typing.Dict[str, str],
- fetch_device_info: bool,
- ) -> None:
-
-
- _LOGGER.debug("received topic=%s payload=%r", message.topic, message.payload)
- if message.retain:
- _LOGGER.info("ignoring retained message")
- return
- actor = await cls._init_from_topic(
- topic=message.topic,
- mqtt_topic_prefix=mqtt_topic_prefix,
- expected_topic_levels=cls._MQTT_UPDATE_DEVICE_INFO_TOPIC_LEVELS,
- retry_count=retry_count,
- device_passwords=device_passwords,
- )
- if actor:
-
- await actor._update_and_report_device_info(
- mqtt_client=mqtt_client, mqtt_topic_prefix=mqtt_topic_prefix
- )
- @abc.abstractmethod
- async def execute_command(
- self,
- *,
- mqtt_message_payload: bytes,
- mqtt_client: aiomqtt.Client,
- update_device_info: bool,
- mqtt_topic_prefix: str,
- ) -> None:
- raise NotImplementedError()
- @classmethod
- async def _mqtt_command_callback(
-
- cls,
- *,
- mqtt_client: aiomqtt.Client,
- message: aiomqtt.Message,
- mqtt_topic_prefix: str,
- retry_count: int,
- device_passwords: typing.Dict[str, str],
- fetch_device_info: bool,
- ) -> None:
-
-
- _LOGGER.debug("received topic=%s payload=%r", message.topic, message.payload)
- if message.retain:
- _LOGGER.info("ignoring retained message")
- return
- actor = await cls._init_from_topic(
- topic=message.topic,
- mqtt_topic_prefix=mqtt_topic_prefix,
- expected_topic_levels=cls.MQTT_COMMAND_TOPIC_LEVELS,
- retry_count=retry_count,
- device_passwords=device_passwords,
- )
- if actor:
- assert isinstance(message.payload, bytes), message.payload
- await actor.execute_command(
- mqtt_message_payload=message.payload,
- mqtt_client=mqtt_client,
- update_device_info=fetch_device_info,
- mqtt_topic_prefix=mqtt_topic_prefix,
- )
- @classmethod
- def _get_mqtt_message_callbacks(
- cls,
- *,
- enable_device_info_update_topic: bool,
- ) -> typing.Dict[typing.Tuple[_MQTTTopicLevel, ...], typing.Callable]:
-
-
-
-
- callbacks = {cls.MQTT_COMMAND_TOPIC_LEVELS: cls._mqtt_command_callback}
- if enable_device_info_update_topic:
- callbacks[cls._MQTT_UPDATE_DEVICE_INFO_TOPIC_LEVELS] = (
- cls._mqtt_update_device_info_callback
- )
- return callbacks
- @classmethod
- async def mqtt_subscribe(
- cls,
- *,
- mqtt_client: aiomqtt.Client,
- mqtt_topic_prefix: str,
- fetch_device_info: bool,
- ) -> typing.AsyncIterator[typing.Tuple[str, typing.Callable]]:
- for topic_levels, callback in cls._get_mqtt_message_callbacks(
- enable_device_info_update_topic=fetch_device_info
- ).items():
- topic = _join_mqtt_topic_levels(
- topic_prefix=mqtt_topic_prefix,
- topic_levels=topic_levels,
- mac_address="+",
- )
- _LOGGER.info("subscribing to MQTT topic %r", topic)
- await mqtt_client.subscribe(topic)
- yield (topic, callback)
- async def _mqtt_publish(
- self,
- *,
- topic_prefix: str,
- topic_levels: typing.Iterable[_MQTTTopicLevel],
- payload: bytes,
- mqtt_client: aiomqtt.Client,
- ) -> None:
- topic = _join_mqtt_topic_levels(
- topic_prefix=topic_prefix,
- topic_levels=topic_levels,
- mac_address=self._mac_address,
- )
-
- _LOGGER.debug("publishing topic=%s payload=%r", topic, payload)
- try:
- await mqtt_client.publish(topic=topic, payload=payload, retain=True)
- except aiomqtt.MqttCodeError as exc:
- _LOGGER.error(
- "Failed to publish MQTT message on topic %s: aiomqtt.MqttCodeError %s",
- topic,
- exc,
- )
- async def report_state(
- self,
- state: bytes,
- mqtt_client: aiomqtt.Client,
- mqtt_topic_prefix: str,
- ) -> None:
- await self._mqtt_publish(
- topic_prefix=mqtt_topic_prefix,
- topic_levels=self.MQTT_STATE_TOPIC_LEVELS,
- payload=state,
- mqtt_client=mqtt_client,
- )
|