diff --git a/README.rst b/README.rst index 8df828292..d5972f779 100644 --- a/README.rst +++ b/README.rst @@ -34,6 +34,7 @@ Supported devices - Xiaomi Smart WiFi Speaker (:class:`miio.wifispeaker`) (incomplete, please `feel free to help improve the support `__) - Xiaomi Mi WiFi Repeater 2 (:class:`miio.wifirepeater`) - Xiaomi Mi Smart Rice Cooker (:class:`miio.cooker`) +- Xiaomi Smartmi Fresh Air System (:class:`miio.airfresh`) - Yeelight light bulbs (:class:`miio.yeelight`) (only a very rudimentary support, use `python-yeelight `__ for a more complete support) *Feel free to create a pull request to add support for new devices as diff --git a/miio/__init__.py b/miio/__init__.py index 07d65e018..dbfc4c1f4 100644 --- a/miio/__init__.py +++ b/miio/__init__.py @@ -1,5 +1,6 @@ # flake8: noqa from miio.airconditioningcompanion import AirConditioningCompanion +from miio.airfresh import AirFresh from miio.airhumidifier import AirHumidifier from miio.airpurifier import AirPurifier from miio.airqualitymonitor import AirQualityMonitor diff --git a/miio/airfresh.py b/miio/airfresh.py new file mode 100644 index 000000000..ce7adf22b --- /dev/null +++ b/miio/airfresh.py @@ -0,0 +1,293 @@ +import enum +import logging +from collections import defaultdict +from typing import Any, Dict, Optional + +import click + +from .click_common import command, format_output, EnumType +from .device import Device, DeviceException + +_LOGGER = logging.getLogger(__name__) + + +class AirFreshException(DeviceException): + pass + + +class OperationMode(enum.Enum): + # Supported modes of the Air Fresh VA2 (zhimi.airfresh.va2) + Auto = 'auto' + Silent = 'silent' + Interval = 'interval' + Low = 'low' + Middle = 'middle' + Strong = 'strong' + + +class LedBrightness(enum.Enum): + Bright = 0 + Dim = 1 + Off = 2 + + +class AirFreshStatus: + """Container for status reports from the air fresh.""" + + def __init__(self, data: Dict[str, Any]) -> None: + self.data = data + + @property + def power(self) -> str: + """Power state.""" + return self.data["power"] + + @property + def is_on(self) -> bool: + """Return True if device is on.""" + return self.power == "on" + + @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 co2(self) -> int: + """Carbon dioxide.""" + return self.data["co2"] + + @property + def humidity(self) -> int: + """Current humidity.""" + return self.data["humidity"] + + @property + def temperature(self) -> Optional[float]: + """Current temperature, if available.""" + if self.data["temp_dec"] is not None: + return self.data["temp_dec"] / 10.0 + + return None + + @property + def mode(self) -> OperationMode: + """Current operation mode.""" + return OperationMode(self.data["mode"]) + + @property + def led_brightness(self) -> Optional[LedBrightness]: + """Brightness of the LED.""" + if self.data["led_level"] is not None: + try: + return LedBrightness(self.data["led_level"]) + except ValueError: + _LOGGER.error("Unsupported LED brightness discarded: %s", self.data["led_level"]) + 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"] == "on" + + return None + + @property + def child_lock(self) -> bool: + """Return True if child lock is on.""" + return self.data["child_lock"] == "on" + + @property + def filter_life_remaining(self) -> int: + """Time until the filter should be changed.""" + return self.data["filter_life"] + + @property + def filter_hours_used(self) -> int: + """How long the filter has been in use.""" + return self.data["f1_hour_used"] + + @property + def use_time(self) -> int: + """How long the device has been active in seconds.""" + return self.data["use_time"] + + @property + def motor_speed(self) -> int: + """Speed of the motor.""" + return self.data["motor1_speed"] + + @property + def extra_features(self) -> Optional[int]: + return self.data["app_extra"] + + def __repr__(self) -> str: + s = "" % \ + (self.power, + self.aqi, + self.average_aqi, + self.temperature, + self.humidity, + self.co2, + self.mode, + self.led_brightness, + self.buzzer, + self.child_lock, + self.filter_life_remaining, + self.filter_hours_used, + self.use_time, + self.motor_speed, + self.extra_features) + return s + + def __json__(self): + return self.data + + +class AirFresh(Device): + """Main class representing the air fresh.""" + + @command( + default_output=format_output( + "", + "Power: {result.power}\n" + "AQI: {result.aqi} μg/m³\n" + "Average AQI: {result.average_aqi} μg/m³\n" + "Temperature: {result.temperature} °C\n" + "Humidity: {result.humidity} %\n" + "CO2: {result.co2} %\n" + "Mode: {result.mode.value}\n" + "LED brightness: {result.led_brightness}\n" + "Buzzer: {result.buzzer}\n" + "Child lock: {result.child_lock}\n" + "Filter life remaining: {result.filter_life_remaining} %\n" + "Filter hours used: {result.filter_hours_used}\n" + "Use time: {result.use_time} s\n" + "Motor speed: {result.motor_speed} rpm\n" + ) + ) + def status(self) -> AirFreshStatus: + """Retrieve properties.""" + + properties = ["power", "temp_dec", "aqi", "average_aqi", "co2", "buzzer", "child_lock", + "humidity", "led_level", "mode", "motor1_speed", "use_time", + "ntcT", "app_extra", "f1_hour_used", "filter_life", "f_hour", + "favorite_level", "led"] + + # 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_prop", _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 AirFreshStatus( + defaultdict(lambda: None, zip(properties, values))) + + @command( + default_output=format_output("Powering on"), + ) + def on(self): + """Power on.""" + return self.send("set_power", ["on"]) + + @command( + default_output=format_output("Powering off"), + ) + def off(self): + """Power off.""" + return self.send("set_power", ["off"]) + + @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.send("set_mode", [mode.value]) + + @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.send("set_led_level", [brightness.value]) + + @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.""" + if buzzer: + return self.send("set_buzzer", ["on"]) + else: + return self.send("set_buzzer", ["off"]) + + @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.""" + if lock: + return self.send("set_child_lock", ["on"]) + else: + return self.send("set_child_lock", ["off"]) + + @command( + click.argument("value", type=int), + default_output=format_output("Setting extra to {value}") + ) + def set_extra_features(self, value: int): + """Storage register to enable extra features at the app.""" + if value < 0: + raise AirFreshException("Invalid app extra value: %s" % value) + + return self.send("set_app_extra", [value]) + + @command( + default_output=format_output("Resetting filter") + ) + def reset_filter(self): + """Resets filter hours used and remaining life.""" + return self.send('reset_filter1') diff --git a/miio/discovery.py b/miio/discovery.py index fa60784f9..32b6c3b68 100644 --- a/miio/discovery.py +++ b/miio/discovery.py @@ -7,7 +7,7 @@ import zeroconf -from . import (Device, Vacuum, ChuangmiPlug, PowerStrip, AirPurifier, Ceil, +from . import (Device, Vacuum, ChuangmiPlug, PowerStrip, AirPurifier, AirFresh, Ceil, PhilipsBulb, PhilipsEyecare, ChuangmiIr, AirHumidifier, WaterPurifier, WifiSpeaker, WifiRepeater, Yeelight, Fan, Cooker, AirConditioningCompanion) @@ -68,6 +68,7 @@ "zhimi-fan-v2": partial(Fan, model=MODEL_FAN_V2), "zhimi-fan-v3": partial(Fan, model=MODEL_FAN_V3), "zhimi-fan-sa1": partial(Fan, model=MODEL_FAN_SA1), + "zhimi-airfresh-va2": AirFresh, "lumi-gateway-": lambda x: other_package_info( x, "https://github.com/Danielhiversen/PyXiaomiGateway") } # type: Dict[str, Union[Callable, Device]] diff --git a/miio/tests/test_airfresh.py b/miio/tests/test_airfresh.py new file mode 100644 index 000000000..aa200f445 --- /dev/null +++ b/miio/tests/test_airfresh.py @@ -0,0 +1,179 @@ +from unittest import TestCase + +import pytest + +from miio import AirFresh +from miio.airfresh import (OperationMode, LedBrightness, AirFreshStatus, AirFreshException) +from .dummies import DummyDevice + + +class DummyAirFresh(DummyDevice, AirFresh): + def __init__(self, *args, **kwargs): + self.state = { + 'power': 'on', + 'temp_dec': 186, + 'aqi': 10, + 'average_aqi': 8, + 'humidity': 62, + 'co2': 350, + 'buzzer': 'off', + 'child_lock': 'off', + 'led_level': 2, + 'mode': 'auto', + 'motor1_speed': 354, + 'use_time': 2457000, + 'ntcT': None, + 'app_extra': 1, + 'f1_hour_used': 682, + 'filter_life': 80, + 'f_hour': 3500, + 'favorite_level': None, + 'led': None, + } + 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_buzzer': lambda x: self._set_state("buzzer", x), + 'set_child_lock': lambda x: self._set_state("child_lock", x), + 'set_led_level': lambda x: self._set_state("led_level", x), + 'reset_filter1': lambda x: ( + self._set_state('f1_hour_used', [0]), + self._set_state('filter_life', [100]) + ), + 'set_app_extra': lambda x: self._set_state("app_extra", x), + } + super().__init__(args, kwargs) + + +@pytest.fixture(scope="class") +def airfresh(request): + request.cls.device = DummyAirFresh() + # TODO add ability to test on a real device + + +@pytest.mark.usefixtures("airfresh") +class TestAirFresh(TestCase): + def is_on(self): + return self.device.status().is_on + + def state(self): + return self.device.status() + + def test_on(self): + self.device.off() # ensure off + assert self.is_on() is False + + self.device.on() + assert self.is_on() is True + + def test_off(self): + self.device.on() # ensure on + assert self.is_on() is True + + self.device.off() + assert self.is_on() is False + + def test_status(self): + self.device._reset_state() + + assert repr(self.state()) == repr(AirFreshStatus(self.device.start_state)) + + assert self.is_on() is True + assert self.state().aqi == self.device.start_state["aqi"] + assert self.state().average_aqi == self.device.start_state["average_aqi"] + assert self.state().temperature == self.device.start_state["temp_dec"] / 10.0 + assert self.state().humidity == self.device.start_state["humidity"] + assert self.state().co2 == self.device.start_state["co2"] + assert self.state().mode == OperationMode(self.device.start_state["mode"]) + assert self.state().filter_life_remaining == self.device.start_state["filter_life"] + assert self.state().filter_hours_used == self.device.start_state["f1_hour_used"] + assert self.state().use_time == self.device.start_state["use_time"] + assert self.state().motor_speed == self.device.start_state["motor1_speed"] + assert self.state().led_brightness == LedBrightness(self.device.start_state["led_level"]) + assert self.state().buzzer == (self.device.start_state["buzzer"] == 'on') + assert self.state().child_lock == (self.device.start_state["child_lock"] == 'on') + assert self.state().extra_features == self.device.start_state["app_extra"] + + 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.Interval) + assert mode() == OperationMode.Interval + + self.device.set_mode(OperationMode.Low) + assert mode() == OperationMode.Low + + self.device.set_mode(OperationMode.Middle) + assert mode() == OperationMode.Middle + + self.device.set_mode(OperationMode.Strong) + assert mode() == OperationMode.Strong + + 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_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 + + def test_set_extra_features(self): + def extra_features(): + return self.device.status().extra_features + + self.device.set_extra_features(0) + assert extra_features() == 0 + self.device.set_extra_features(1) + assert extra_features() == 1 + self.device.set_extra_features(2) + assert extra_features() == 2 + + with pytest.raises(AirFreshException): + self.device.set_extra_features(-1) + + def test_reset_filter(self): + def filter_hours_used(): + return self.device.status().filter_hours_used + + def filter_life_remaining(): + return self.device.status().filter_life_remaining + + self.device._reset_state() + assert filter_hours_used() != 0 + assert filter_life_remaining() != 100 + self.device.reset_filter() + assert filter_hours_used() == 0 + assert filter_life_remaining() == 100