4082719430

Build # 4082719430

Build Type

push

github

Committed by GitHub

Commit Message

build(deps): bump docker/build-push-action from 3.3.0 to 4.0.0 (#134)

Run Details

310 of 310 relevant lines covered (100.0%)

5.0 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

100.0

/switchbot_mqtt/_actors/base.py

# switchbot-mqtt - MQTT client controlling SwitchBot button & curtain automators,
# compatible with home-assistant.io's MQTT Switch & Cover platform
#
# Copyright (C) 2020 Fabian Peter Hammerle <fabian@hammerle.me>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <https://www.gnu.org/licenses/>.

# > Even with __all__ set appropriately, internal interfaces (packages,
# > modules, classes, functions, attributes or other names) should still be
# > prefixed with a single leading underscore. An interface is also considered
# > internal if any containing namespace (package, module or class) is
# > considered internal.
# https://peps.python.org/pep-0008/#public-and-internal-interfaces

from __future__ import annotations  # PEP563 (default in python>=3.10)

import abc
import dataclasses
import logging
import queue
import shlex
import typing

import bluepy.btle
import paho.mqtt.client
import switchbot
from switchbot_mqtt._utils import (
    _join_mqtt_topic_levels,
    _mac_address_valid,
    _MQTTTopicLevel,
    _MQTTTopicPlaceholder,
    _parse_mqtt_topic,
    _QueueLogHandler,
)

_LOGGER = logging.getLogger(__name__)


@dataclasses.dataclass
class _MQTTCallbackUserdata:
    retry_count: int
    device_passwords: typing.Dict[str, str]
    fetch_device_info: bool
    mqtt_topic_prefix: str


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, *, mac_address: str, retry_count: int, password: typing.Optional[str]
    ) -> None:
        # alternative: pySwitchbot >=0.10.0 provides SwitchbotDevice.get_mac()
        self._mac_address = mac_address

    @abc.abstractmethod
    def _get_device(self) -> switchbot.SwitchbotDevice:
        raise NotImplementedError()

    def _update_device_info(self) -> None:
        log_queue: queue.Queue[logging.LogRecord] = queue.Queue(maxsize=0)
        logging.getLogger("switchbot").addHandler(_QueueLogHandler(log_queue))
        try:
            self._get_device().update()
            # pySwitchbot>=v0.10.1 catches bluepy.btle.BTLEManagementError :(
            # https://github.com/Danielhiversen/pySwitchbot/blob/0.10.1/switchbot/__init__.py#L141
            # pySwitchbot<0.11.0 WARNING, >=0.11.0 ERROR
            while not log_queue.empty():
                log_record = log_queue.get()
                if log_record.exc_info:
                    exc: typing.Optional[BaseException] = log_record.exc_info[1]
                    if (
                        isinstance(exc, bluepy.btle.BTLEManagementError)
                        and exc.emsg == "Permission Denied"
                    ):
                        raise exc
        except bluepy.btle.BTLEManagementError as exc:
            if (
                exc.emsg == "Permission Denied"
                and exc.message == "Failed to execute management command 'le on'"
            ):
                raise PermissionError(
                    "bluepy-helper failed to enable low energy mode"
                    " due to insufficient permissions."
                    "\nSee https://github.com/IanHarvey/bluepy/issues/313#issuecomment-428324639"
                    ", https://github.com/fphammerle/switchbot-mqtt/pull/31#issuecomment-846383603"
                    ", and https://github.com/IanHarvey/bluepy/blob/v/1.3.0/bluepy"
                    "/bluepy-helper.c#L1260."
                    "\nInsecure workaround:"
                    "\n1. sudo apt-get install --no-install-recommends libcap2-bin"
                    f"\n2. sudo setcap cap_net_admin+ep {shlex.quote(bluepy.btle.helperExe)}"
                    "\n3. restart switchbot-mqtt"
                    "\nIn docker-based setups, you could use"
                    " `sudo docker run --cap-drop ALL --cap-add NET_ADMIN --user 0 …`"
                    " (seriously insecure)."
                ) from exc
            raise

    def _report_battery_level(
        self, mqtt_client: paho.mqtt.client.Client, mqtt_topic_prefix: str
    ) -> None:
        # > battery: Percentage of battery that is left.
        # https://www.home-assistant.io/integrations/sensor/#device-class
        self._mqtt_publish(
            topic_prefix=mqtt_topic_prefix,
            topic_levels=self._MQTT_BATTERY_PERCENTAGE_TOPIC_LEVELS,
            payload=str(self._get_device().get_battery_percent()).encode(),
            mqtt_client=mqtt_client,
        )

    def _update_and_report_device_info(
        self, mqtt_client: paho.mqtt.client.Client, mqtt_topic_prefix: str
    ) -> None:
        self._update_device_info()
        self._report_battery_level(
            mqtt_client=mqtt_client, mqtt_topic_prefix=mqtt_topic_prefix
        )

    @classmethod
    def _init_from_topic(
        cls,
        topic: str,
        expected_topic_levels: typing.Collection[_MQTTTopicLevel],
        settings: _MQTTCallbackUserdata,
    ) -> typing.Optional[_MQTTControlledActor]:
        try:
            mac_address = _parse_mqtt_topic(
                topic=topic,
                expected_prefix=settings.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
        return cls(
            mac_address=mac_address,
            retry_count=settings.retry_count,
            password=settings.device_passwords.get(mac_address, None),
        )

    @classmethod
    def _mqtt_update_device_info_callback(
        # pylint: disable=duplicate-code; other callbacks with same params
        cls,
        mqtt_client: paho.mqtt.client.Client,
        userdata: _MQTTCallbackUserdata,
        message: paho.mqtt.client.MQTTMessage,
    ) -> None:
        # pylint: disable=unused-argument; callback
        # https://github.com/eclipse/paho.mqtt.python/blob/v1.5.0/src/paho/mqtt/client.py#L469
        _LOGGER.debug("received topic=%s payload=%r", message.topic, message.payload)
        if message.retain:
            _LOGGER.info("ignoring retained message")
            return
        actor = cls._init_from_topic(
            topic=message.topic,
            expected_topic_levels=cls._MQTT_UPDATE_DEVICE_INFO_TOPIC_LEVELS,
            settings=userdata,
        )
        if actor:
            # pylint: disable=protected-access; own instance
            actor._update_and_report_device_info(
                mqtt_client=mqtt_client, mqtt_topic_prefix=userdata.mqtt_topic_prefix
            )

    @abc.abstractmethod
    def execute_command(  # pylint: disable=duplicate-code; implementations
        self,
        *,
        mqtt_message_payload: bytes,
        mqtt_client: paho.mqtt.client.Client,
        update_device_info: bool,
        mqtt_topic_prefix: str,
    ) -> None:
        raise NotImplementedError()

    @classmethod
    def _mqtt_command_callback(
        # pylint: disable=duplicate-code; other callbacks with same params
        cls,
        mqtt_client: paho.mqtt.client.Client,
        userdata: _MQTTCallbackUserdata,
        message: paho.mqtt.client.MQTTMessage,
    ) -> None:
        # pylint: disable=unused-argument; callback
        # https://github.com/eclipse/paho.mqtt.python/blob/v1.5.0/src/paho/mqtt/client.py#L469
        _LOGGER.debug("received topic=%s payload=%r", message.topic, message.payload)
        if message.retain:
            _LOGGER.info("ignoring retained message")
            return
        actor = cls._init_from_topic(
            topic=message.topic,
            expected_topic_levels=cls.MQTT_COMMAND_TOPIC_LEVELS,
            settings=userdata,
        )
        if actor:
            actor.execute_command(
                mqtt_message_payload=message.payload,
                mqtt_client=mqtt_client,
                update_device_info=userdata.fetch_device_info,
                mqtt_topic_prefix=userdata.mqtt_topic_prefix,
            )

    @classmethod
    def _get_mqtt_message_callbacks(
        cls,
        *,
        enable_device_info_update_topic: bool,
    ) -> typing.Dict[typing.Tuple[_MQTTTopicLevel, ...], typing.Callable]:
        # returning dict because `paho.mqtt.client.Client.message_callback_add` overwrites
        # callbacks with same topic pattern
        # https://github.com/eclipse/paho.mqtt.python/blob/v1.6.1/src/paho/mqtt/client.py#L2304
        # https://github.com/eclipse/paho.mqtt.python/blob/v1.6.1/src/paho/mqtt/matcher.py#L19
        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
    def mqtt_subscribe(
        cls, *, mqtt_client: paho.mqtt.client.Client, settings: _MQTTCallbackUserdata
    ) -> None:
        for topic_levels, callback in cls._get_mqtt_message_callbacks(
            enable_device_info_update_topic=settings.fetch_device_info
        ).items():
            topic = _join_mqtt_topic_levels(
                topic_prefix=settings.mqtt_topic_prefix,
                topic_levels=topic_levels,
                mac_address="+",
            )
            _LOGGER.info("subscribing to MQTT topic %r", topic)
            mqtt_client.subscribe(topic)
            mqtt_client.message_callback_add(sub=topic, callback=callback)

    def _mqtt_publish(
        self,
        *,
        topic_prefix: str,
        topic_levels: typing.Iterable[_MQTTTopicLevel],
        payload: bytes,
        mqtt_client: paho.mqtt.client.Client,
    ) -> None:
        topic = _join_mqtt_topic_levels(
            topic_prefix=topic_prefix,
            topic_levels=topic_levels,
            mac_address=self._mac_address,
        )
        # https://pypi.org/project/paho-mqtt/#publishing
        _LOGGER.debug("publishing topic=%s payload=%r", topic, payload)
        message_info: paho.mqtt.client.MQTTMessageInfo = mqtt_client.publish(
            topic=topic, payload=payload, retain=True
        )
        # wait before checking status?
        if message_info.rc != paho.mqtt.client.MQTT_ERR_SUCCESS:
            _LOGGER.error(
                "Failed to publish MQTT message on topic %s (rc=%d)",
                topic,
                message_info.rc,
            )

    def report_state(
        self,
        state: bytes,
        mqtt_client: paho.mqtt.client.Client,
        mqtt_topic_prefix: str,
    ) -> None:
        self._mqtt_publish(
            topic_prefix=mqtt_topic_prefix,
            topic_levels=self.MQTT_STATE_TOPIC_LEVELS,
            payload=state,
            mqtt_client=mqtt_client,
        )

1	# switchbot-mqtt - MQTT client controlling SwitchBot button & curtain automators,
2	# compatible with home-assistant.io's MQTT Switch & Cover platform
3	#
4	# Copyright (C) 2020 Fabian Peter Hammerle <fabian@hammerle.me>
5	#
6	# This program is free software: you can redistribute it and/or modify
7	# it under the terms of the GNU General Public License as published by
8	# the Free Software Foundation, either version 3 of the License, or
9	# any later version.
10	#
11	# This program is distributed in the hope that it will be useful,
12	# but WITHOUT ANY WARRANTY; without even the implied warranty of
13	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14	# GNU General Public License for more details.
15	#
16	# You should have received a copy of the GNU General Public License
17	# along with this program. If not, see <https://www.gnu.org/licenses/>.
18
19	# > Even with __all__ set appropriately, internal interfaces (packages,
20	# > modules, classes, functions, attributes or other names) should still be
21	# > prefixed with a single leading underscore. An interface is also considered
22	# > internal if any containing namespace (package, module or class) is
23	# > considered internal.
24	# https://peps.python.org/pep-0008/#public-and-internal-interfaces
25
26	from __future__ import annotations # PEP563 (default in python>=3.10)	5✔
27
28	import abc	5✔
29	import dataclasses	5✔
30	import logging	5✔
31	import queue	5✔
32	import shlex	5✔
33	import typing	5✔
34
35	import bluepy.btle	5✔
36	import paho.mqtt.client	5✔
37	import switchbot	5✔
38	from switchbot_mqtt._utils import (	5✔
39	_join_mqtt_topic_levels,
40	_mac_address_valid,
41	_MQTTTopicLevel,
42	_MQTTTopicPlaceholder,
43	_parse_mqtt_topic,
44	_QueueLogHandler,
45	)
46
47	_LOGGER = logging.getLogger(__name__)	5✔
48
49
50	@dataclasses.dataclass	5✔
51	class _MQTTCallbackUserdata:	4✔
52	retry_count: int	5✔
53	device_passwords: typing.Dict[str, str]	5✔
54	fetch_device_info: bool	5✔
55	mqtt_topic_prefix: str	5✔
56
57
58	class _MQTTControlledActor(abc.ABC):	5✔
59	MQTT_COMMAND_TOPIC_LEVELS: typing.Tuple[_MQTTTopicLevel, ...] = NotImplemented	5✔
60	_MQTT_UPDATE_DEVICE_INFO_TOPIC_LEVELS: typing.Tuple[	5✔
61	_MQTTTopicLevel, ...
62	] = NotImplemented
63	MQTT_STATE_TOPIC_LEVELS: typing.Tuple[_MQTTTopicLevel, ...] = NotImplemented	5✔
64	_MQTT_BATTERY_PERCENTAGE_TOPIC_LEVELS: typing.Tuple[	5✔
65	_MQTTTopicLevel, ...
66	] = NotImplemented
67
68	@classmethod	5✔
69	def get_mqtt_update_device_info_topic(cls, *, prefix: str, mac_address: str) -> str:	4✔
70	return _join_mqtt_topic_levels(	5✔
71	topic_prefix=prefix,
72	topic_levels=cls._MQTT_UPDATE_DEVICE_INFO_TOPIC_LEVELS,
73	mac_address=mac_address,
74	)
75
76	@classmethod	5✔
77	def get_mqtt_battery_percentage_topic(cls, *, prefix: str, mac_address: str) -> str:	4✔
78	return _join_mqtt_topic_levels(	5✔
79	topic_prefix=prefix,
80	topic_levels=cls._MQTT_BATTERY_PERCENTAGE_TOPIC_LEVELS,
81	mac_address=mac_address,
82	)
83
84	@abc.abstractmethod	5✔
85	def __init__(	4✔
86	self, *, mac_address: str, retry_count: int, password: typing.Optional[str]
87	) -> None:
88	# alternative: pySwitchbot >=0.10.0 provides SwitchbotDevice.get_mac()
89	self._mac_address = mac_address	5✔
90
91	@abc.abstractmethod	5✔
92	def _get_device(self) -> switchbot.SwitchbotDevice:	4✔
93	raise NotImplementedError()	5✔
94
95	def _update_device_info(self) -> None:	5✔
96	log_queue: queue.Queue[logging.LogRecord] = queue.Queue(maxsize=0)	5✔
97	logging.getLogger("switchbot").addHandler(_QueueLogHandler(log_queue))	5✔
98	try:	5✔
99	self._get_device().update()	5✔
100	# pySwitchbot>=v0.10.1 catches bluepy.btle.BTLEManagementError :(
101	# https://github.com/Danielhiversen/pySwitchbot/blob/0.10.1/switchbot/__init__.py#L141
102	# pySwitchbot<0.11.0 WARNING, >=0.11.0 ERROR
103	while not log_queue.empty():	5✔
104	log_record = log_queue.get()	5✔
105	if log_record.exc_info:	5✔
106	exc: typing.Optional[BaseException] = log_record.exc_info[1]	5✔
107	if (	5✔
108	isinstance(exc, bluepy.btle.BTLEManagementError)
109	and exc.emsg == "Permission Denied"
110	):
111	raise exc	5✔
112	except bluepy.btle.BTLEManagementError as exc:	5✔
113	if (	5✔
114	exc.emsg == "Permission Denied"
115	and exc.message == "Failed to execute management command 'le on'"
116	):
117	raise PermissionError(	5✔
118	"bluepy-helper failed to enable low energy mode"
119	" due to insufficient permissions."
120	"\nSee https://github.com/IanHarvey/bluepy/issues/313#issuecomment-428324639"
121	", https://github.com/fphammerle/switchbot-mqtt/pull/31#issuecomment-846383603"
122	", and https://github.com/IanHarvey/bluepy/blob/v/1.3.0/bluepy"
123	"/bluepy-helper.c#L1260."
124	"\nInsecure workaround:"
125	"\n1. sudo apt-get install --no-install-recommends libcap2-bin"
126	f"\n2. sudo setcap cap_net_admin+ep {shlex.quote(bluepy.btle.helperExe)}"
127	"\n3. restart switchbot-mqtt"
128	"\nIn docker-based setups, you could use"
129	" `sudo docker run --cap-drop ALL --cap-add NET_ADMIN --user 0 …`"
130	" (seriously insecure)."
131	) from exc
132	raise	5✔
133
134	def _report_battery_level(	5✔
135	self, mqtt_client: paho.mqtt.client.Client, mqtt_topic_prefix: str
136	) -> None:
137	# > battery: Percentage of battery that is left.
138	# https://www.home-assistant.io/integrations/sensor/#device-class
139	self._mqtt_publish(	5✔
140	topic_prefix=mqtt_topic_prefix,
141	topic_levels=self._MQTT_BATTERY_PERCENTAGE_TOPIC_LEVELS,
142	payload=str(self._get_device().get_battery_percent()).encode(),
143	mqtt_client=mqtt_client,
144	)
145
146	def _update_and_report_device_info(	5✔
147	self, mqtt_client: paho.mqtt.client.Client, mqtt_topic_prefix: str
148	) -> None:
149	self._update_device_info()	5✔
150	self._report_battery_level(	5✔
151	mqtt_client=mqtt_client, mqtt_topic_prefix=mqtt_topic_prefix
152	)
153
154	@classmethod	5✔
155	def _init_from_topic(	4✔
156	cls,
157	topic: str,
158	expected_topic_levels: typing.Collection[_MQTTTopicLevel],
159	settings: _MQTTCallbackUserdata,
160	) -> typing.Optional[_MQTTControlledActor]:
161	try:	5✔
162	mac_address = _parse_mqtt_topic(	5✔
163	topic=topic,
164	expected_prefix=settings.mqtt_topic_prefix,
165	expected_levels=expected_topic_levels,
166	)[_MQTTTopicPlaceholder.MAC_ADDRESS]
167	except ValueError as exc:	5✔
168	_LOGGER.warning(str(exc), exc_info=False)	5✔
169	return None	5✔
170	if not _mac_address_valid(mac_address):	5✔
171	_LOGGER.warning("invalid mac address %s", mac_address)	5✔
172	return None	5✔
173	return cls(	5✔
174	mac_address=mac_address,
175	retry_count=settings.retry_count,
176	password=settings.device_passwords.get(mac_address, None),
177	)
178
179	@classmethod	5✔
180	def _mqtt_update_device_info_callback(	4✔
181	# pylint: disable=duplicate-code; other callbacks with same params
182	cls,
183	mqtt_client: paho.mqtt.client.Client,
184	userdata: _MQTTCallbackUserdata,
185	message: paho.mqtt.client.MQTTMessage,
186	) -> None:
187	# pylint: disable=unused-argument; callback
188	# https://github.com/eclipse/paho.mqtt.python/blob/v1.5.0/src/paho/mqtt/client.py#L469
189	_LOGGER.debug("received topic=%s payload=%r", message.topic, message.payload)	5✔
190	if message.retain:	5✔
191	_LOGGER.info("ignoring retained message")	5✔
192	return	5✔
193	actor = cls._init_from_topic(	5✔
194	topic=message.topic,
195	expected_topic_levels=cls._MQTT_UPDATE_DEVICE_INFO_TOPIC_LEVELS,
196	settings=userdata,
197	)
198	if actor:	5✔
199	# pylint: disable=protected-access; own instance
200	actor._update_and_report_device_info(	5✔
201	mqtt_client=mqtt_client, mqtt_topic_prefix=userdata.mqtt_topic_prefix
202	)
203
204	@abc.abstractmethod	5✔
205	def execute_command( # pylint: disable=duplicate-code; implementations	4✔
206	self,
207	*,
208	mqtt_message_payload: bytes,
209	mqtt_client: paho.mqtt.client.Client,
210	update_device_info: bool,
211	mqtt_topic_prefix: str,
212	) -> None:
213	raise NotImplementedError()	5✔
214
215	@classmethod	5✔
216	def _mqtt_command_callback(	4✔
217	# pylint: disable=duplicate-code; other callbacks with same params
218	cls,
219	mqtt_client: paho.mqtt.client.Client,
220	userdata: _MQTTCallbackUserdata,
221	message: paho.mqtt.client.MQTTMessage,
222	) -> None:
223	# pylint: disable=unused-argument; callback
224	# https://github.com/eclipse/paho.mqtt.python/blob/v1.5.0/src/paho/mqtt/client.py#L469
225	_LOGGER.debug("received topic=%s payload=%r", message.topic, message.payload)	5✔
226	if message.retain:	5✔
227	_LOGGER.info("ignoring retained message")	5✔
228	return	5✔
229	actor = cls._init_from_topic(	5✔
230	topic=message.topic,
231	expected_topic_levels=cls.MQTT_COMMAND_TOPIC_LEVELS,
232	settings=userdata,
233	)
234	if actor:	5✔
235	actor.execute_command(	5✔
236	mqtt_message_payload=message.payload,
237	mqtt_client=mqtt_client,
238	update_device_info=userdata.fetch_device_info,
239	mqtt_topic_prefix=userdata.mqtt_topic_prefix,
240	)
241
242	@classmethod	5✔
243	def _get_mqtt_message_callbacks(	4✔
244	cls,
245	*,
246	enable_device_info_update_topic: bool,
247	) -> typing.Dict[typing.Tuple[_MQTTTopicLevel, ...], typing.Callable]:
248	# returning dict because `paho.mqtt.client.Client.message_callback_add` overwrites
249	# callbacks with same topic pattern
250	# https://github.com/eclipse/paho.mqtt.python/blob/v1.6.1/src/paho/mqtt/client.py#L2304
251	# https://github.com/eclipse/paho.mqtt.python/blob/v1.6.1/src/paho/mqtt/matcher.py#L19
252	callbacks = {cls.MQTT_COMMAND_TOPIC_LEVELS: cls._mqtt_command_callback}	5✔
253	if enable_device_info_update_topic:	5✔
254	callbacks[	5✔
255	cls._MQTT_UPDATE_DEVICE_INFO_TOPIC_LEVELS
256	] = cls._mqtt_update_device_info_callback
257	return callbacks	5✔
258
259	@classmethod	5✔
260	def mqtt_subscribe(	4✔
261	cls, *, mqtt_client: paho.mqtt.client.Client, settings: _MQTTCallbackUserdata
262	) -> None:
263	for topic_levels, callback in cls._get_mqtt_message_callbacks(	5✔
264	enable_device_info_update_topic=settings.fetch_device_info
265	).items():
266	topic = _join_mqtt_topic_levels(	5✔
267	topic_prefix=settings.mqtt_topic_prefix,
268	topic_levels=topic_levels,
269	mac_address="+",
270	)
271	_LOGGER.info("subscribing to MQTT topic %r", topic)	5✔
272	mqtt_client.subscribe(topic)	5✔
273	mqtt_client.message_callback_add(sub=topic, callback=callback)	5✔
274
275	def _mqtt_publish(	5✔
276	self,
277	*,
278	topic_prefix: str,
279	topic_levels: typing.Iterable[_MQTTTopicLevel],
280	payload: bytes,
281	mqtt_client: paho.mqtt.client.Client,
282	) -> None:
283	topic = _join_mqtt_topic_levels(	5✔
284	topic_prefix=topic_prefix,
285	topic_levels=topic_levels,
286	mac_address=self._mac_address,
287	)
288	# https://pypi.org/project/paho-mqtt/#publishing
289	_LOGGER.debug("publishing topic=%s payload=%r", topic, payload)	5✔
290	message_info: paho.mqtt.client.MQTTMessageInfo = mqtt_client.publish(	5✔
291	topic=topic, payload=payload, retain=True
292	)
293	# wait before checking status?
294	if message_info.rc != paho.mqtt.client.MQTT_ERR_SUCCESS:	5✔
295	_LOGGER.error(	5✔
296	"Failed to publish MQTT message on topic %s (rc=%d)",
297	topic,
298	message_info.rc,
299	)
300
301	def report_state(	5✔
302	self,
303	state: bytes,
304	mqtt_client: paho.mqtt.client.Client,
305	mqtt_topic_prefix: str,
306	) -> None:
307	self._mqtt_publish(	5✔
308	topic_prefix=mqtt_topic_prefix,
309	topic_levels=self.MQTT_STATE_TOPIC_LEVELS,
310	payload=state,
311	mqtt_client=mqtt_client,
312	)

fphammerle / switchbot-mqtt / 4082719430

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous