fphammerle / switchbot-mqtt / 3768437106 / 2

Source File

100.0

/switchbot_mqtt/_actors/__init__.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/>.

import logging
import typing

import bluepy.btle
import paho.mqtt.client
import switchbot

from switchbot_mqtt._actors.base import _MQTTCallbackUserdata, _MQTTControlledActor
from switchbot_mqtt._utils import (
    _join_mqtt_topic_levels,
    _MQTTTopicLevel,
    _MQTTTopicPlaceholder,
)

_LOGGER = logging.getLogger(__name__)

_BUTTON_TOPIC_LEVELS_PREFIX = (
    "switch",
    "switchbot",
    _MQTTTopicPlaceholder.MAC_ADDRESS,
)
_CURTAIN_TOPIC_LEVELS_PREFIX = (
    "cover",
    "switchbot-curtain",
    _MQTTTopicPlaceholder.MAC_ADDRESS,
)


class _ButtonAutomator(_MQTTControlledActor):
    # https://www.home-assistant.io/integrations/switch.mqtt/

    MQTT_COMMAND_TOPIC_LEVELS = _BUTTON_TOPIC_LEVELS_PREFIX + ("set",)
    _MQTT_UPDATE_DEVICE_INFO_TOPIC_LEVELS = _BUTTON_TOPIC_LEVELS_PREFIX + (
        "request-device-info",
    )
    MQTT_STATE_TOPIC_LEVELS = _BUTTON_TOPIC_LEVELS_PREFIX + ("state",)
    _MQTT_BATTERY_PERCENTAGE_TOPIC_LEVELS = _BUTTON_TOPIC_LEVELS_PREFIX + (
        "battery-percentage",
    )

    def __init__(
        self, *, mac_address: str, retry_count: int, password: typing.Optional[str]
    ) -> None:
        self.__device = switchbot.Switchbot(
            mac=mac_address, password=password, retry_count=retry_count
        )
        super().__init__(
            mac_address=mac_address, retry_count=retry_count, password=password
        )

    def _get_device(self) -> switchbot.SwitchbotDevice:
        return self.__device

    def execute_command(
        self,
        *,
        mqtt_message_payload: bytes,
        mqtt_client: paho.mqtt.client.Client,
        update_device_info: bool,
        mqtt_topic_prefix: str,
    ) -> None:
        # https://www.home-assistant.io/integrations/switch.mqtt/#payload_on
        if mqtt_message_payload.lower() == b"on":
            if not self.__device.turn_on():
                _LOGGER.error("failed to turn on switchbot %s", self._mac_address)
            else:
                _LOGGER.info("switchbot %s turned on", self._mac_address)
                # https://www.home-assistant.io/integrations/switch.mqtt/#state_on
                self.report_state(
                    mqtt_client=mqtt_client,
                    mqtt_topic_prefix=mqtt_topic_prefix,
                    state=b"ON",
                )
                if update_device_info:
                    self._update_and_report_device_info(mqtt_client, mqtt_topic_prefix)
        # https://www.home-assistant.io/integrations/switch.mqtt/#payload_off
        elif mqtt_message_payload.lower() == b"off":
            if not self.__device.turn_off():
                _LOGGER.error("failed to turn off switchbot %s", self._mac_address)
            else:
                _LOGGER.info("switchbot %s turned off", self._mac_address)
                self.report_state(
                    mqtt_client=mqtt_client,
                    mqtt_topic_prefix=mqtt_topic_prefix,
                    state=b"OFF",
                )
                if update_device_info:
                    self._update_and_report_device_info(mqtt_client, mqtt_topic_prefix)
        else:
            _LOGGER.warning(
                "unexpected payload %r (expected 'ON' or 'OFF')", mqtt_message_payload
            )


class _CurtainMotor(_MQTTControlledActor):

    # https://www.home-assistant.io/integrations/cover.mqtt/
    MQTT_COMMAND_TOPIC_LEVELS = _CURTAIN_TOPIC_LEVELS_PREFIX + ("set",)
    _MQTT_SET_POSITION_TOPIC_LEVELS: typing.Tuple[
        _MQTTTopicLevel, ...
    ] = _CURTAIN_TOPIC_LEVELS_PREFIX + (
        "position",
        "set-percent",
    )
    _MQTT_UPDATE_DEVICE_INFO_TOPIC_LEVELS = _CURTAIN_TOPIC_LEVELS_PREFIX + (
        "request-device-info",
    )
    MQTT_STATE_TOPIC_LEVELS = _CURTAIN_TOPIC_LEVELS_PREFIX + ("state",)
    _MQTT_BATTERY_PERCENTAGE_TOPIC_LEVELS = _CURTAIN_TOPIC_LEVELS_PREFIX + (
        "battery-percentage",
    )
    _MQTT_POSITION_TOPIC_LEVELS = _CURTAIN_TOPIC_LEVELS_PREFIX + ("position",)

    @classmethod
    def get_mqtt_position_topic(cls, prefix: str, mac_address: str) -> str:
        return _join_mqtt_topic_levels(
            topic_prefix=prefix,
            topic_levels=cls._MQTT_POSITION_TOPIC_LEVELS,
            mac_address=mac_address,
        )

    def __init__(
        self, *, mac_address: str, retry_count: int, password: typing.Optional[str]
    ) -> None:
        # > The position of the curtain is saved in self._pos with 0 = open and 100 = closed.
        # https://github.com/Danielhiversen/pySwitchbot/blob/0.10.0/switchbot/__init__.py#L150
        self.__device = switchbot.SwitchbotCurtain(
            mac=mac_address,
            password=password,
            retry_count=retry_count,
            reverse_mode=True,
        )
        super().__init__(
            mac_address=mac_address, retry_count=retry_count, password=password
        )

    def _get_device(self) -> switchbot.SwitchbotDevice:
        return self.__device

    def _report_position(
        self,
        mqtt_client: paho.mqtt.client.Client,  # pylint: disable=duplicate-code; similar param list
        mqtt_topic_prefix: str,
    ) -> None:
        # > position_closed integer (Optional, default: 0)
        # > position_open integer (Optional, default: 100)
        # https://www.home-assistant.io/integrations/cover.mqtt/#position_closed
        # SwitchbotCurtain.get_position() returns a cached value within [0, 100].
        # SwitchbotCurtain.open() and .close() update the position optimistically,
        # SwitchbotCurtain.update() fetches the real position via bluetooth.
        # https://github.com/Danielhiversen/pySwitchbot/blob/0.10.0/switchbot/__init__.py#L202
        self._mqtt_publish(
            topic_prefix=mqtt_topic_prefix,
            topic_levels=self._MQTT_POSITION_TOPIC_LEVELS,
            payload=str(int(self.__device.get_position())).encode(),
            mqtt_client=mqtt_client,
        )

    def _update_and_report_device_info(  # pylint: disable=arguments-differ; report_position is optional
        self,
        mqtt_client: paho.mqtt.client.Client,
        mqtt_topic_prefix: str,
        *,
        report_position: bool = True,
    ) -> None:
        super()._update_and_report_device_info(mqtt_client, mqtt_topic_prefix)
        if report_position:
            self._report_position(
                mqtt_client=mqtt_client, mqtt_topic_prefix=mqtt_topic_prefix
            )

    def execute_command(
        self,
        *,
        mqtt_message_payload: bytes,
        mqtt_client: paho.mqtt.client.Client,
        update_device_info: bool,
        mqtt_topic_prefix: str,
    ) -> None:
        # https://www.home-assistant.io/integrations/cover.mqtt/#payload_open
        report_device_info, report_position = False, False
        if mqtt_message_payload.lower() == b"open":
            if not self.__device.open():
                _LOGGER.error("failed to open switchbot curtain %s", self._mac_address)
            else:
                _LOGGER.info("switchbot curtain %s opening", self._mac_address)
                # > state_opening string (Optional, default: opening)
                # https://www.home-assistant.io/integrations/cover.mqtt/#state_opening
                self.report_state(
                    mqtt_client=mqtt_client,
                    mqtt_topic_prefix=mqtt_topic_prefix,
                    state=b"opening",
                )
                report_device_info = update_device_info
        elif mqtt_message_payload.lower() == b"close":
            if not self.__device.close():
                _LOGGER.error("failed to close switchbot curtain %s", self._mac_address)
            else:
                _LOGGER.info("switchbot curtain %s closing", self._mac_address)
                # https://www.home-assistant.io/integrations/cover.mqtt/#state_closing
                self.report_state(
                    mqtt_client=mqtt_client,
                    mqtt_topic_prefix=mqtt_topic_prefix,
                    state=b"closing",
                )
                report_device_info = update_device_info
        elif mqtt_message_payload.lower() == b"stop":
            if not self.__device.stop():
                _LOGGER.error("failed to stop switchbot curtain %s", self._mac_address)
            else:
                _LOGGER.info("switchbot curtain %s stopped", self._mac_address)
                # no "stopped" state mentioned at
                # https://www.home-assistant.io/integrations/cover.mqtt/#configuration-variables
                # https://community.home-assistant.io/t/mqtt-how-to-remove-retained-messages/79029/2
                self.report_state(
                    mqtt_client=mqtt_client,
                    mqtt_topic_prefix=mqtt_topic_prefix,
                    state=b"",
                )
                report_device_info = update_device_info
                report_position = True
        else:
            _LOGGER.warning(
                "unexpected payload %r (expected 'OPEN', 'CLOSE', or 'STOP')",
                mqtt_message_payload,
            )
        if report_device_info:
            self._update_and_report_device_info(
                mqtt_client=mqtt_client,
                mqtt_topic_prefix=mqtt_topic_prefix,
                report_position=report_position,
            )

    @classmethod
    def _mqtt_set_position_callback(
        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.6.1/src/paho/mqtt/client.py#L3556
        _LOGGER.debug("received topic=%s payload=%r", message.topic, message.payload)
        if message.retain:
            _LOGGER.info("ignoring retained message on topic %s", message.topic)
            return
        actor = cls._init_from_topic(
            topic=message.topic,
            expected_topic_levels=cls._MQTT_SET_POSITION_TOPIC_LEVELS,
            settings=userdata,
        )
        if not actor:
            return  # warning in _init_from_topic
        position_percent = int(message.payload.decode(), 10)
        if position_percent < 0 or position_percent > 100:
            _LOGGER.warning("invalid position %u%%, ignoring message", position_percent)
            return
        # pylint: disable=protected-access; own instance
        if actor._get_device().set_position(position_percent):
            _LOGGER.info(
                "set position of switchbot curtain %s to %u%%",
                actor._mac_address,
                position_percent,
            )
        else:
            _LOGGER.error(
                "failed to set position of switchbot curtain %s", actor._mac_address
            )

    @classmethod
    def _get_mqtt_message_callbacks(
        # pylint: disable=duplicate-code; param list in parent class
        cls,
        *,
        enable_device_info_update_topic: bool,
    ) -> typing.Dict[typing.Tuple[_MQTTTopicLevel, ...], typing.Callable]:
        callbacks = super()._get_mqtt_message_callbacks(
            enable_device_info_update_topic=enable_device_info_update_topic
        )
        callbacks[cls._MQTT_SET_POSITION_TOPIC_LEVELS] = cls._mqtt_set_position_callback
        return callbacks

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	import logging	1✔
20	import typing	1✔
21
22	import bluepy.btle	1✔
23	import paho.mqtt.client	1✔
24	import switchbot	1✔
25
26	from switchbot_mqtt._actors.base import _MQTTCallbackUserdata, _MQTTControlledActor	1✔
27	from switchbot_mqtt._utils import (	1✔
28	_join_mqtt_topic_levels,
29	_MQTTTopicLevel,
30	_MQTTTopicPlaceholder,
31	)
32
33	_LOGGER = logging.getLogger(__name__)	1✔
34
35	_BUTTON_TOPIC_LEVELS_PREFIX = (	1✔
36	"switch",
37	"switchbot",
38	_MQTTTopicPlaceholder.MAC_ADDRESS,
39	)
40	_CURTAIN_TOPIC_LEVELS_PREFIX = (	1✔
41	"cover",
42	"switchbot-curtain",
43	_MQTTTopicPlaceholder.MAC_ADDRESS,
44	)
45
46
47	class _ButtonAutomator(_MQTTControlledActor):	1✔
48	# https://www.home-assistant.io/integrations/switch.mqtt/
49
50	MQTT_COMMAND_TOPIC_LEVELS = _BUTTON_TOPIC_LEVELS_PREFIX + ("set",)	1✔
51	_MQTT_UPDATE_DEVICE_INFO_TOPIC_LEVELS = _BUTTON_TOPIC_LEVELS_PREFIX + (	1✔
52	"request-device-info",
53	)
54	MQTT_STATE_TOPIC_LEVELS = _BUTTON_TOPIC_LEVELS_PREFIX + ("state",)	1✔
55	_MQTT_BATTERY_PERCENTAGE_TOPIC_LEVELS = _BUTTON_TOPIC_LEVELS_PREFIX + (	1✔
56	"battery-percentage",
57	)
58
59	def __init__(	1✔
60	self, *, mac_address: str, retry_count: int, password: typing.Optional[str]
61	) -> None:
62	self.__device = switchbot.Switchbot(	1✔
63	mac=mac_address, password=password, retry_count=retry_count
64	)
65	super().__init__(	1✔
66	mac_address=mac_address, retry_count=retry_count, password=password
67	)
68
69	def _get_device(self) -> switchbot.SwitchbotDevice:	1✔
70	return self.__device	1✔
71
72	def execute_command(	1✔
73	self,
74	*,
75	mqtt_message_payload: bytes,
76	mqtt_client: paho.mqtt.client.Client,
77	update_device_info: bool,
78	mqtt_topic_prefix: str,
79	) -> None:
80	# https://www.home-assistant.io/integrations/switch.mqtt/#payload_on
81	if mqtt_message_payload.lower() == b"on":	1✔
82	if not self.__device.turn_on():	1✔
83	_LOGGER.error("failed to turn on switchbot %s", self._mac_address)	1✔
84	else:
85	_LOGGER.info("switchbot %s turned on", self._mac_address)	1✔
86	# https://www.home-assistant.io/integrations/switch.mqtt/#state_on
87	self.report_state(	1✔
88	mqtt_client=mqtt_client,
89	mqtt_topic_prefix=mqtt_topic_prefix,
90	state=b"ON",
91	)
92	if update_device_info:	1✔
93	self._update_and_report_device_info(mqtt_client, mqtt_topic_prefix)	1✔
94	# https://www.home-assistant.io/integrations/switch.mqtt/#payload_off
95	elif mqtt_message_payload.lower() == b"off":	1✔
96	if not self.__device.turn_off():	1✔
97	_LOGGER.error("failed to turn off switchbot %s", self._mac_address)	1✔
98	else:
99	_LOGGER.info("switchbot %s turned off", self._mac_address)	1✔
100	self.report_state(	1✔
101	mqtt_client=mqtt_client,
102	mqtt_topic_prefix=mqtt_topic_prefix,
103	state=b"OFF",
104	)
105	if update_device_info:	1✔
106	self._update_and_report_device_info(mqtt_client, mqtt_topic_prefix)	1✔
107	else:
108	_LOGGER.warning(	1✔
109	"unexpected payload %r (expected 'ON' or 'OFF')", mqtt_message_payload
110	)
111
112
113	class _CurtainMotor(_MQTTControlledActor):	1✔
114
115	# https://www.home-assistant.io/integrations/cover.mqtt/
116	MQTT_COMMAND_TOPIC_LEVELS = _CURTAIN_TOPIC_LEVELS_PREFIX + ("set",)	1✔
117	_MQTT_SET_POSITION_TOPIC_LEVELS: typing.Tuple[	1✔
118	_MQTTTopicLevel, ...
119	] = _CURTAIN_TOPIC_LEVELS_PREFIX + (
120	"position",
121	"set-percent",
122	)
123	_MQTT_UPDATE_DEVICE_INFO_TOPIC_LEVELS = _CURTAIN_TOPIC_LEVELS_PREFIX + (	1✔
124	"request-device-info",
125	)
126	MQTT_STATE_TOPIC_LEVELS = _CURTAIN_TOPIC_LEVELS_PREFIX + ("state",)	1✔
127	_MQTT_BATTERY_PERCENTAGE_TOPIC_LEVELS = _CURTAIN_TOPIC_LEVELS_PREFIX + (	1✔
128	"battery-percentage",
129	)
130	_MQTT_POSITION_TOPIC_LEVELS = _CURTAIN_TOPIC_LEVELS_PREFIX + ("position",)	1✔
131
132	@classmethod	1✔
133	def get_mqtt_position_topic(cls, prefix: str, mac_address: str) -> str:	1✔
134	return _join_mqtt_topic_levels(	1✔
135	topic_prefix=prefix,
136	topic_levels=cls._MQTT_POSITION_TOPIC_LEVELS,
137	mac_address=mac_address,
138	)
139
140	def __init__(	1✔
141	self, *, mac_address: str, retry_count: int, password: typing.Optional[str]
142	) -> None:
143	# > The position of the curtain is saved in self._pos with 0 = open and 100 = closed.
144	# https://github.com/Danielhiversen/pySwitchbot/blob/0.10.0/switchbot/__init__.py#L150
145	self.__device = switchbot.SwitchbotCurtain(	1✔
146	mac=mac_address,
147	password=password,
148	retry_count=retry_count,
149	reverse_mode=True,
150	)
151	super().__init__(	1✔
152	mac_address=mac_address, retry_count=retry_count, password=password
153	)
154
155	def _get_device(self) -> switchbot.SwitchbotDevice:	1✔
156	return self.__device	1✔
157
158	def _report_position(	1✔
159	self,
160	mqtt_client: paho.mqtt.client.Client, # pylint: disable=duplicate-code; similar param list
161	mqtt_topic_prefix: str,
162	) -> None:
163	# > position_closed integer (Optional, default: 0)
164	# > position_open integer (Optional, default: 100)
165	# https://www.home-assistant.io/integrations/cover.mqtt/#position_closed
166	# SwitchbotCurtain.get_position() returns a cached value within [0, 100].
167	# SwitchbotCurtain.open() and .close() update the position optimistically,
168	# SwitchbotCurtain.update() fetches the real position via bluetooth.
169	# https://github.com/Danielhiversen/pySwitchbot/blob/0.10.0/switchbot/__init__.py#L202
170	self._mqtt_publish(	1✔
171	topic_prefix=mqtt_topic_prefix,
172	topic_levels=self._MQTT_POSITION_TOPIC_LEVELS,
173	payload=str(int(self.__device.get_position())).encode(),
174	mqtt_client=mqtt_client,
175	)
176
177	def _update_and_report_device_info( # pylint: disable=arguments-differ; report_position is optional	1✔
178	self,
179	mqtt_client: paho.mqtt.client.Client,
180	mqtt_topic_prefix: str,
181	*,
182	report_position: bool = True,
183	) -> None:
184	super()._update_and_report_device_info(mqtt_client, mqtt_topic_prefix)	1✔
185	if report_position:	1✔
186	self._report_position(	1✔
187	mqtt_client=mqtt_client, mqtt_topic_prefix=mqtt_topic_prefix
188	)
189
190	def execute_command(	1✔
191	self,
192	*,
193	mqtt_message_payload: bytes,
194	mqtt_client: paho.mqtt.client.Client,
195	update_device_info: bool,
196	mqtt_topic_prefix: str,
197	) -> None:
198	# https://www.home-assistant.io/integrations/cover.mqtt/#payload_open
199	report_device_info, report_position = False, False	1✔
200	if mqtt_message_payload.lower() == b"open":	1✔
201	if not self.__device.open():	1✔
202	_LOGGER.error("failed to open switchbot curtain %s", self._mac_address)	1✔
203	else:
204	_LOGGER.info("switchbot curtain %s opening", self._mac_address)	1✔
205	# > state_opening string (Optional, default: opening)
206	# https://www.home-assistant.io/integrations/cover.mqtt/#state_opening
207	self.report_state(	1✔
208	mqtt_client=mqtt_client,
209	mqtt_topic_prefix=mqtt_topic_prefix,
210	state=b"opening",
211	)
212	report_device_info = update_device_info	1✔
213	elif mqtt_message_payload.lower() == b"close":	1✔
214	if not self.__device.close():	1✔
215	_LOGGER.error("failed to close switchbot curtain %s", self._mac_address)	1✔
216	else:
217	_LOGGER.info("switchbot curtain %s closing", self._mac_address)	1✔
218	# https://www.home-assistant.io/integrations/cover.mqtt/#state_closing
219	self.report_state(	1✔
220	mqtt_client=mqtt_client,
221	mqtt_topic_prefix=mqtt_topic_prefix,
222	state=b"closing",
223	)
224	report_device_info = update_device_info	1✔
225	elif mqtt_message_payload.lower() == b"stop":	1✔
226	if not self.__device.stop():	1✔
227	_LOGGER.error("failed to stop switchbot curtain %s", self._mac_address)	1✔
228	else:
229	_LOGGER.info("switchbot curtain %s stopped", self._mac_address)	1✔
230	# no "stopped" state mentioned at
231	# https://www.home-assistant.io/integrations/cover.mqtt/#configuration-variables
232	# https://community.home-assistant.io/t/mqtt-how-to-remove-retained-messages/79029/2
233	self.report_state(	1✔
234	mqtt_client=mqtt_client,
235	mqtt_topic_prefix=mqtt_topic_prefix,
236	state=b"",
237	)
238	report_device_info = update_device_info	1✔
239	report_position = True	1✔
240	else:
241	_LOGGER.warning(	1✔
242	"unexpected payload %r (expected 'OPEN', 'CLOSE', or 'STOP')",
243	mqtt_message_payload,
244	)
245	if report_device_info:	1✔
246	self._update_and_report_device_info(	1✔
247	mqtt_client=mqtt_client,
248	mqtt_topic_prefix=mqtt_topic_prefix,
249	report_position=report_position,
250	)
251
252	@classmethod	1✔
253	def _mqtt_set_position_callback(	1✔
254	cls,
255	mqtt_client: paho.mqtt.client.Client,
256	userdata: _MQTTCallbackUserdata,
257	message: paho.mqtt.client.MQTTMessage,
258	) -> None:
259	# pylint: disable=unused-argument; callback
260	# https://github.com/eclipse/paho.mqtt.python/blob/v1.6.1/src/paho/mqtt/client.py#L3556
261	_LOGGER.debug("received topic=%s payload=%r", message.topic, message.payload)	1✔
262	if message.retain:	1✔
263	_LOGGER.info("ignoring retained message on topic %s", message.topic)	1✔
264	return	1✔
265	actor = cls._init_from_topic(	1✔
266	topic=message.topic,
267	expected_topic_levels=cls._MQTT_SET_POSITION_TOPIC_LEVELS,
268	settings=userdata,
269	)
270	if not actor:	1✔
271	return # warning in _init_from_topic	1✔
272	position_percent = int(message.payload.decode(), 10)	1✔
273	if position_percent < 0 or position_percent > 100:	1✔
274	_LOGGER.warning("invalid position %u%%, ignoring message", position_percent)	1✔
275	return	1✔
276	# pylint: disable=protected-access; own instance
277	if actor._get_device().set_position(position_percent):	1✔
278	_LOGGER.info(	1✔
279	"set position of switchbot curtain %s to %u%%",
280	actor._mac_address,
281	position_percent,
282	)
283	else:
284	_LOGGER.error(	1✔
285	"failed to set position of switchbot curtain %s", actor._mac_address
286	)
287
288	@classmethod	1✔
289	def _get_mqtt_message_callbacks(	1✔
290	# pylint: disable=duplicate-code; param list in parent class
291	cls,
292	*,
293	enable_device_info_update_topic: bool,
294	) -> typing.Dict[typing.Tuple[_MQTTTopicLevel, ...], typing.Callable]:
295	callbacks = super()._get_mqtt_message_callbacks(	1✔
296	enable_device_info_update_topic=enable_device_info_update_topic
297	)
298	callbacks[cls._MQTT_SET_POSITION_TOPIC_LEVELS] = cls._mqtt_set_position_callback	1✔
299	return callbacks	1✔