diff --git a/docs/miio.rst b/docs/miio.rst index f966e7f1b..f97a397ca 100644 --- a/docs/miio.rst +++ b/docs/miio.rst @@ -36,6 +36,14 @@ miio\.airpurifier module :show-inheritance: :undoc-members: +miio\.airpurifier_miot module +----------------------------- + +.. automodule:: miio.airpurifier_miot + :members: + :show-inheritance: + :undoc-members: + miio\.airqualitymonitor module ------------------------------ @@ -93,6 +101,14 @@ miio\.device module :show-inheritance: :undoc-members: +miio\.miot_device module +------------------------ + +.. automodule:: miio.miot_device + :members: + :show-inheritance: + :undoc-members: + miio\.discovery module ---------------------- diff --git a/miio/__init__.py b/miio/__init__.py index b3cb19f8b..17bfd870f 100644 --- a/miio/__init__.py +++ b/miio/__init__.py @@ -9,6 +9,7 @@ from miio.airhumidifier import AirHumidifier, AirHumidifierCA1, AirHumidifierCB1 from miio.airhumidifier_mjjsq import AirHumidifierMjjsq from miio.airpurifier import AirPurifier +from miio.airpurifier_miot import AirPurifierMiot from miio.airqualitymonitor import AirQualityMonitor from miio.aqaracamera import AqaraCamera from miio.ceil import Ceil diff --git a/miio/airfilter_util.py b/miio/airfilter_util.py new file mode 100644 index 000000000..c74fc7c5f --- /dev/null +++ b/miio/airfilter_util.py @@ -0,0 +1,47 @@ +import enum +import re +from typing import Optional + + +class FilterType(enum.Enum): + Regular = "regular" + AntiBacterial = "anti-bacterial" + AntiFormaldehyde = "anti-formaldehyde" + Unknown = "unknown" + + +FILTER_TYPE_RE = ( + (re.compile(r"^\d+:\d+:41:30$"), FilterType.AntiBacterial), + (re.compile(r"^\d+:\d+:(30|0|00):31$"), FilterType.AntiFormaldehyde), + (re.compile(r".*"), FilterType.Regular), +) + + +class FilterTypeUtil: + """Utility class for determining xiaomi air filter type.""" + + _filter_type_cache = {} + + def determine_filter_type( + self, rfid_tag: Optional[str], product_id: Optional[str] + ) -> Optional[FilterType]: + """ + Determine Xiaomi air filter type based on its product ID. + + :param rfid_tag: RFID tag value + :param product_id: Product ID such as "0:0:30:33" + """ + if rfid_tag is None: + return None + if rfid_tag == "0:0:0:0:0:0:0": + return FilterType.Unknown + if product_id is None: + return FilterType.Regular + + ft = self._filter_type_cache.get(product_id, None) + if ft is None: + for filter_re, filter_type in FILTER_TYPE_RE: + if filter_re.match(product_id): + ft = self._filter_type_cache[product_id] = filter_type + break + return ft diff --git a/miio/airpurifier.py b/miio/airpurifier.py index ed36992e3..80ccdc72b 100644 --- a/miio/airpurifier.py +++ b/miio/airpurifier.py @@ -1,11 +1,11 @@ import enum import logging -import re from collections import defaultdict from typing import Any, Dict, Optional import click +from .airfilter_util import FilterType, FilterTypeUtil from .click_common import EnumType, command, format_output from .device import Device from .exceptions import DeviceException @@ -42,20 +42,6 @@ class LedBrightness(enum.Enum): Off = 2 -class FilterType(enum.Enum): - Regular = "regular" - AntiBacterial = "anti-bacterial" - AntiFormaldehyde = "anti-formaldehyde" - Unknown = "unknown" - - -FILTER_TYPE_RE = ( - (re.compile(r"^\d+:\d+:41:30$"), FilterType.AntiBacterial), - (re.compile(r"^\d+:\d+:(30|0|00):31$"), FilterType.AntiFormaldehyde), - (re.compile(r".*"), FilterType.Regular), -) - - class AirPurifierStatus: """Container for status reports from the air purifier.""" @@ -102,6 +88,7 @@ def __init__(self, data: Dict[str, Any]) -> None: A request is limited to 16 properties. """ + self.filter_type_util = FilterTypeUtil() self.data = data @property @@ -239,13 +226,9 @@ def filter_rfid_tag(self) -> Optional[str]: @property def filter_type(self) -> Optional[FilterType]: """Type of installed filter.""" - if self.filter_rfid_tag is None: - return None - if self.filter_rfid_tag == "0:0:0:0:0:0:0": - return FilterType.Unknown - if self.filter_rfid_product_id is None: - return FilterType.Regular - return self._get_filter_type(self.filter_rfid_product_id) + return self.filter_type_util.determine_filter_type( + self.filter_rfid_tag, self.filter_rfid_product_id + ) @property def learn_mode(self) -> bool: @@ -284,16 +267,6 @@ def button_pressed(self) -> Optional[str]: """Last pressed button.""" return self.data["button_pressed"] - @classmethod - def _get_filter_type(cls, product_id: str) -> FilterType: - ft = cls._filter_type_cache.get(product_id, None) - if ft is None: - for filter_re, filter_type in FILTER_TYPE_RE: - if filter_re.match(product_id): - ft = cls._filter_type_cache[product_id] = filter_type - break - return ft - def __repr__(self) -> str: s = ( " None: + self.filter_type_util = FilterTypeUtil() + self.data = data + + @property + def is_on(self) -> bool: + """Return True if device is on.""" + return self.data["power"] + + @property + def power(self) -> str: + """Power state.""" + return "on" if self.is_on else "off" + + @property + def aqi(self) -> int: + """Air quality index.""" + return self.data["aqi"] + + @property + def average_aqi(self) -> int: + """Average of the air quality index.""" + return self.data["average_aqi"] + + @property + def humidity(self) -> int: + """Current humidity.""" + return self.data["humidity"] + + @property + def temperature(self) -> Optional[float]: + """Current temperature, if available.""" + if self.data["temperature"] is not None: + return self.data["temperature"] + + return None + + @property + def fan_level(self) -> int: + """Current fan level.""" + return self.data["fan_level"] + + @property + def mode(self) -> OperationMode: + """Current operation mode.""" + return OperationMode(self.data["mode"]) + + @property + def led(self) -> bool: + """Return True if LED is on.""" + return self.data["led"] + + @property + def led_brightness(self) -> Optional[LedBrightness]: + """Brightness of the LED.""" + if self.data["led_brightness"] is not None: + try: + return LedBrightness(self.data["led_brightness"]) + except ValueError: + return None + + return None + + @property + def buzzer(self) -> Optional[bool]: + """Return True if buzzer is on.""" + if self.data["buzzer"] is not None: + return self.data["buzzer"] + + return None + + @property + def buzzer_volume(self) -> Optional[int]: + """Return buzzer volume.""" + if self.data["buzzer_volume"] is not None: + return self.data["buzzer_volume"] + + return None + + @property + def child_lock(self) -> bool: + """Return True if child lock is on.""" + return self.data["child_lock"] + + @property + def favorite_level(self) -> int: + """Return favorite level, which is used if the mode is ``favorite``.""" + # Favorite level used when the mode is `favorite`. + return self.data["favorite_level"] + + @property + def filter_life_remaining(self) -> int: + """Time until the filter should be changed.""" + return self.data["filter_life_remaining"] + + @property + def filter_hours_used(self) -> int: + """How long the filter has been in use.""" + return self.data["filter_hours_used"] + + @property + def use_time(self) -> int: + """How long the device has been active in seconds.""" + return self.data["use_time"] + + @property + def purify_volume(self) -> int: + """The volume of purified air in cubic meter.""" + return self.data["purify_volume"] + + @property + def motor_speed(self) -> int: + """Speed of the motor.""" + return self.data["motor_speed"] + + @property + def filter_rfid_product_id(self) -> Optional[str]: + """RFID product ID of installed filter.""" + return self.data["filter_rfid_product_id"] + + @property + def filter_rfid_tag(self) -> Optional[str]: + """RFID tag ID of installed filter.""" + return self.data["filter_rfid_tag"] + + @property + def filter_type(self) -> Optional[FilterType]: + """Type of installed filter.""" + return self.filter_type_util.determine_filter_type( + self.filter_rfid_tag, self.filter_rfid_product_id + ) + + def __repr__(self) -> str: + s = ( + "" + % ( + self.power, + self.aqi, + self.average_aqi, + self.temperature, + self.humidity, + self.fan_level, + self.mode, + self.led, + self.led_brightness, + self.buzzer, + self.buzzer_volume, + self.child_lock, + self.favorite_level, + self.filter_life_remaining, + self.filter_hours_used, + self.use_time, + self.purify_volume, + self.motor_speed, + self.filter_rfid_product_id, + self.filter_rfid_tag, + self.filter_type, + ) + ) + return s + + def __json__(self): + return self.data + + +class AirPurifierMiot(MiotDevice): + """Main class representing the air purifier which uses MIoT protocol.""" + + def __init__( + self, + ip: str = None, + token: str = None, + start_id: int = 0, + debug: int = 0, + lazy_discover: bool = True, + ) -> None: + super().__init__(_MAPPING, ip, token, start_id, debug, lazy_discover) + + @command( + default_output=format_output( + "", + "Power: {result.power}\n" + "AQI: {result.aqi} μg/m³\n" + "Average AQI: {result.average_aqi} μg/m³\n" + "Humidity: {result.humidity} %\n" + "Temperature: {result.temperature} °C\n" + "Fan Level: {result.fan_level}\n" + "Mode: {result.mode}\n" + "LED: {result.led}\n" + "LED brightness: {result.led_brightness}\n" + "Buzzer: {result.buzzer}\n" + "Buzzer vol.: {result.buzzer_volume}\n" + "Child lock: {result.child_lock}\n" + "Favorite level: {result.favorite_level}\n" + "Filter life remaining: {result.filter_life_remaining} %\n" + "Filter hours used: {result.filter_hours_used}\n" + "Use time: {result.use_time} s\n" + "Purify volume: {result.purify_volume} m³\n" + "Motor speed: {result.motor_speed} rpm\n" + "Filter RFID product id: {result.filter_rfid_product_id}\n" + "Filter RFID tag: {result.filter_rfid_tag}\n" + "Filter type: {result.filter_type}\n", + ) + ) + def status(self) -> AirPurifierMiotStatus: + """Retrieve properties.""" + + return AirPurifierMiotStatus( + { + prop["did"]: prop["value"] if prop["code"] == 0 else None + for prop in self.get_properties() + } + ) + + @command(default_output=format_output("Powering on")) + def on(self): + """Power on.""" + return self.set_property("power", True) + + @command(default_output=format_output("Powering off")) + def off(self): + """Power off.""" + return self.set_property("power", False) + + @command( + click.argument("level", type=int), + default_output=format_output("Setting fan level to '{level}'"), + ) + def set_fan_level(self, level: int): + """Set fan level.""" + if level < 1 or level > 3: + raise AirPurifierMiotException("Invalid fan level: %s" % level) + return self.set_property("fan_level", level) + + @command( + click.argument("rpm", type=int), + default_output=format_output("Setting favorite motor speed '{rpm}' rpm"), + ) + def set_favorite_rpm(self, rpm: int): + """Set favorite motor speed.""" + # Note: documentation says the maximum is 2300, however, the purifier may return an error for rpm over 2200. + if rpm < 300 or rpm > 2300 or rpm % 10 != 0: + raise AirPurifierMiotException( + "Invalid favorite motor speed: %s. Must be between 300 and 2300 and divisible by 10" + % rpm + ) + return self.set_property("favorite_rpm", rpm) + + @command( + click.argument("volume", type=int), + default_output=format_output("Setting sound volume to {volume}"), + ) + def set_volume(self, volume: int): + """Set buzzer volume.""" + if volume < 0 or volume > 100: + raise AirPurifierMiotException( + "Invalid volume: %s. Must be between 0 and 100" % volume + ) + return self.set_property("buzzer_volume", volume) + + @command( + click.argument("mode", type=EnumType(OperationMode, False)), + default_output=format_output("Setting mode to '{mode.value}'"), + ) + def set_mode(self, mode: OperationMode): + """Set mode.""" + return self.set_property("mode", mode.value) + + @command( + click.argument("level", type=int), + default_output=format_output("Setting favorite level to {level}"), + ) + def set_favorite_level(self, level: int): + """Set the favorite level used when the mode is `favorite`, + should be between 0 and 14. + """ + if level < 0 or level > 14: + raise AirPurifierMiotException("Invalid favorite level: %s" % level) + + return self.set_property("favorite_level", level) + + @command( + click.argument("brightness", type=EnumType(LedBrightness, False)), + default_output=format_output("Setting LED brightness to {brightness}"), + ) + def set_led_brightness(self, brightness: LedBrightness): + """Set led brightness.""" + return self.set_property("led_brightness", brightness.value) + + @command( + click.argument("led", type=bool), + default_output=format_output( + lambda led: "Turning on LED" if led else "Turning off LED" + ), + ) + def set_led(self, led: bool): + """Turn led on/off.""" + return self.set_property("led", led) + + @command( + click.argument("buzzer", type=bool), + default_output=format_output( + lambda buzzer: "Turning on buzzer" if buzzer else "Turning off buzzer" + ), + ) + def set_buzzer(self, buzzer: bool): + """Set buzzer on/off.""" + return self.set_property("buzzer", buzzer) + + @command( + click.argument("lock", type=bool), + default_output=format_output( + lambda lock: "Turning on child lock" if lock else "Turning off child lock" + ), + ) + def set_child_lock(self, lock: bool): + """Set child lock on/off.""" + return self.set_property("child_lock", lock) diff --git a/miio/discovery.py b/miio/discovery.py index 22d0f7956..2ca05a74c 100644 --- a/miio/discovery.py +++ b/miio/discovery.py @@ -14,6 +14,7 @@ AirHumidifier, AirHumidifierMjjsq, AirPurifier, + AirPurifierMiot, AirQualityMonitor, AqaraCamera, Ceil, @@ -108,6 +109,8 @@ "zhimi-airpurifier-v6": AirPurifier, # v6 "zhimi-airpurifier-v7": AirPurifier, # v7 "zhimi-airpurifier-mc1": AirPurifier, # mc1 + "zhimi-airpurifier-mb3": AirPurifierMiot, # mb3 (3/3H) + "zhimi-airpurifier-ma4": AirPurifierMiot, # ma4 (3) "chuangmi.camera.ipc009": ChuangmiCamera, "chuangmi-ir-v2": ChuangmiIr, "chuangmi-remote-h102a03_": ChuangmiIr, diff --git a/miio/miot_device.py b/miio/miot_device.py new file mode 100644 index 000000000..9b45e3ee7 --- /dev/null +++ b/miio/miot_device.py @@ -0,0 +1,55 @@ +import logging + +from .device import Device + +_LOGGER = logging.getLogger(__name__) + + +class MiotDevice(Device): + """Main class representing a MIoT device.""" + + def __init__( + self, + mapping: dict, + ip: str = None, + token: str = None, + start_id: int = 0, + debug: int = 0, + lazy_discover: bool = True, + ) -> None: + self.mapping = mapping + super().__init__(ip, token, start_id, debug, lazy_discover) + + def get_properties(self) -> list: + """Retrieve raw properties based on mapping.""" + + # We send property key in "did" because it's sent back via response and we can identify the property. + properties = [{"did": k, **v} for k, v in self.mapping.items()] + + # A single request is limited to 16 properties. Therefore the + # properties are divided into multiple requests + _props = properties.copy() + values = [] + while _props: + values.extend(self.send("get_properties", _props[:15])) + _props[:] = _props[15:] + + properties_count = len(properties) + values_count = len(values) + if properties_count != values_count: + _LOGGER.debug( + "Count (%s) of requested properties does not match the " + "count (%s) of received values.", + properties_count, + values_count, + ) + + return values + + def set_property(self, property_key: str, value): + """Sets property value.""" + + return self.send( + "set_properties", + [{"did": property_key, **self.mapping[property_key], "value": value}], + ) diff --git a/miio/tests/dummies.py b/miio/tests/dummies.py index fe4a013dc..2ca1c6ad2 100644 --- a/miio/tests/dummies.py +++ b/miio/tests/dummies.py @@ -53,3 +53,21 @@ def _set_state(self, var, value): def _get_state(self, props): """Return wanted properties""" return [self.state[x] for x in props if x in self.state] + + +class DummyMiotDevice(DummyDevice): + """Main class representing a MIoT device.""" + + def __init__(self, *args, **kwargs): + # {prop["did"]: prop["value"] for prop in self.miot_client.get_properties()} + self.state = [{"did": k, "value": v, "code": 0} for k, v in self.state.items()] + super().__init__(*args, **kwargs) + + def get_properties(self): + return self.state + + def set_property(self, property_key: str, value): + for prop in self.state: + if prop["did"] == property_key: + prop["value"] = value + return None diff --git a/miio/tests/test_airfilter_util.py b/miio/tests/test_airfilter_util.py new file mode 100644 index 000000000..d8ff62dbf --- /dev/null +++ b/miio/tests/test_airfilter_util.py @@ -0,0 +1,51 @@ +from unittest import TestCase + +import pytest + +from miio.airfilter_util import FilterType, FilterTypeUtil + + +@pytest.fixture(scope="class") +def airfilter_util(request): + request.cls.filter_type_util = FilterTypeUtil() + + +@pytest.mark.usefixtures("airfilter_util") +class TestAirFilterUtil(TestCase): + def test_determine_filter_type__recognises_unknown_filter(self): + assert ( + self.filter_type_util.determine_filter_type("0:0:0:0:0:0:0", None) + is FilterType.Unknown + ) + + def test_determine_filter_type__recognises_antibacterial_filter(self): + assert ( + self.filter_type_util.determine_filter_type( + "80:64:d1:ba:4f:5f:4", "12:34:41:30" + ) + is FilterType.AntiBacterial + ) + + def test_determine_filter_type__recognises_antiformaldehyde_filter(self): + assert ( + self.filter_type_util.determine_filter_type( + "80:64:d1:ba:4f:5f:4", "12:34:00:31" + ) + is FilterType.AntiFormaldehyde + ) + + def test_determine_filter_type__falls_back_to_regular_filter(self): + regular_filters = [ + "12:34:56:78", + "12:34:56:31", + "12:34:56:31:11:11", + "CO:FF:FF:EE", + None, + ] + for product_id in regular_filters: + assert ( + self.filter_type_util.determine_filter_type( + "80:64:d1:ba:4f:5f:4", product_id + ) + is FilterType.Regular + ) diff --git a/miio/tests/test_airpurifier_miot.py b/miio/tests/test_airpurifier_miot.py new file mode 100644 index 000000000..dddd0e120 --- /dev/null +++ b/miio/tests/test_airpurifier_miot.py @@ -0,0 +1,194 @@ +from unittest import TestCase + +import pytest + +from miio import AirPurifierMiot +from miio.airfilter_util import FilterType +from miio.airpurifier_miot import AirPurifierMiotException, LedBrightness, OperationMode + +from .dummies import DummyMiotDevice + +_INITIAL_STATE = { + "power": True, + "aqi": 10, + "average_aqi": 8, + "humidity": 62, + "temperature": 18.6, + "fan_level": 2, + "mode": 0, + "led": True, + "led_brightness": 1, + "buzzer": False, + "buzzer_volume": 0, + "child_lock": False, + "favorite_level": 10, + "filter_life_remaining": 80, + "filter_hours_used": 682, + "use_time": 2457000, + "purify_volume": 25262, + "motor_speed": 354, + "filter_rfid_product_id": "0:0:41:30", + "filter_rfid_tag": "10:20:30:40:50:60:7", + "button_pressed": "power", +} + + +class DummyAirPurifierMiot(DummyMiotDevice, AirPurifierMiot): + def __init__(self, *args, **kwargs): + self.state = _INITIAL_STATE + self.return_values = { + "get_prop": self._get_state, + "set_power": lambda x: self._set_state("power", x), + "set_mode": lambda x: self._set_state("mode", x), + "set_led": lambda x: self._set_state("led", x), + "set_buzzer": lambda x: self._set_state("buzzer", x), + "set_child_lock": lambda x: self._set_state("child_lock", x), + "set_level_favorite": lambda x: self._set_state("favorite_level", x), + "set_led_b": lambda x: self._set_state("led_b", x), + "set_volume": lambda x: self._set_state("volume", x), + "set_act_sleep": lambda x: self._set_state("act_sleep", x), + "reset_filter1": lambda x: ( + self._set_state("f1_hour_used", [0]), + self._set_state("filter1_life", [100]), + ), + "set_act_det": lambda x: self._set_state("act_det", x), + "set_app_extra": lambda x: self._set_state("app_extra", x), + } + super().__init__(*args, **kwargs) + + +@pytest.fixture(scope="function") +def airpurifier(request): + request.cls.device = DummyAirPurifierMiot() + + +@pytest.mark.usefixtures("airpurifier") +class TestAirPurifier(TestCase): + def test_on(self): + self.device.off() # ensure off + assert self.device.status().is_on is False + + self.device.on() + assert self.device.status().is_on is True + + def test_off(self): + self.device.on() # ensure on + assert self.device.status().is_on is True + + self.device.off() + assert self.device.status().is_on is False + + def test_status(self): + status = self.device.status() + assert status.is_on is _INITIAL_STATE["power"] + assert status.aqi == _INITIAL_STATE["aqi"] + assert status.average_aqi == _INITIAL_STATE["average_aqi"] + assert status.humidity == _INITIAL_STATE["humidity"] + assert status.temperature == _INITIAL_STATE["temperature"] + assert status.fan_level == _INITIAL_STATE["fan_level"] + assert status.mode == OperationMode(_INITIAL_STATE["mode"]) + assert status.led == _INITIAL_STATE["led"] + assert status.led_brightness == LedBrightness(_INITIAL_STATE["led_brightness"]) + assert status.buzzer == _INITIAL_STATE["buzzer"] + assert status.child_lock == _INITIAL_STATE["child_lock"] + assert status.favorite_level == _INITIAL_STATE["favorite_level"] + assert status.filter_life_remaining == _INITIAL_STATE["filter_life_remaining"] + assert status.filter_hours_used == _INITIAL_STATE["filter_hours_used"] + assert status.use_time == _INITIAL_STATE["use_time"] + assert status.purify_volume == _INITIAL_STATE["purify_volume"] + assert status.motor_speed == _INITIAL_STATE["motor_speed"] + assert status.filter_rfid_product_id == _INITIAL_STATE["filter_rfid_product_id"] + assert status.filter_type == FilterType.AntiBacterial + + def test_set_fan_level(self): + def fan_level(): + return self.device.status().fan_level + + self.device.set_fan_level(1) + assert fan_level() == 1 + self.device.set_fan_level(2) + assert fan_level() == 2 + self.device.set_fan_level(3) + assert fan_level() == 3 + + with pytest.raises(AirPurifierMiotException): + self.device.set_fan_level(0) + + with pytest.raises(AirPurifierMiotException): + self.device.set_fan_level(4) + + def test_set_mode(self): + def mode(): + return self.device.status().mode + + self.device.set_mode(OperationMode.Auto) + assert mode() == OperationMode.Auto + + self.device.set_mode(OperationMode.Silent) + assert mode() == OperationMode.Silent + + self.device.set_mode(OperationMode.Favorite) + assert mode() == OperationMode.Favorite + + self.device.set_mode(OperationMode.Fan) + assert mode() == OperationMode.Fan + + def test_set_favorite_level(self): + def favorite_level(): + return self.device.status().favorite_level + + self.device.set_favorite_level(0) + assert favorite_level() == 0 + self.device.set_favorite_level(6) + assert favorite_level() == 6 + self.device.set_favorite_level(14) + assert favorite_level() == 14 + + with pytest.raises(AirPurifierMiotException): + self.device.set_favorite_level(-1) + + with pytest.raises(AirPurifierMiotException): + self.device.set_favorite_level(15) + + def test_set_led_brightness(self): + def led_brightness(): + return self.device.status().led_brightness + + self.device.set_led_brightness(LedBrightness.Bright) + assert led_brightness() == LedBrightness.Bright + + self.device.set_led_brightness(LedBrightness.Dim) + assert led_brightness() == LedBrightness.Dim + + self.device.set_led_brightness(LedBrightness.Off) + assert led_brightness() == LedBrightness.Off + + def test_set_led(self): + def led(): + return self.device.status().led + + self.device.set_led(True) + assert led() is True + + self.device.set_led(False) + assert led() is False + + def test_set_buzzer(self): + def buzzer(): + return self.device.status().buzzer + + self.device.set_buzzer(True) + assert buzzer() is True + + self.device.set_buzzer(False) + assert buzzer() is False + + def test_set_child_lock(self): + def child_lock(): + return self.device.status().child_lock + + self.device.set_child_lock(True) + assert child_lock() is True + + self.device.set_child_lock(False) + assert child_lock() is False