diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index f39757bbf..ded20028b 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -24,7 +24,7 @@ jobs: python-version: "${{ matrix.python-version }}" - name: "Install dependencies" run: | - python -m pip install --upgrade pip poetry + python -m pip install --upgrade pip "poetry==2.0.1" poetry install --extras docs - name: "Run pre-commit hooks" run: | @@ -48,7 +48,7 @@ jobs: python-version: "${{ matrix.python-version }}" - name: "Install dependencies" run: | - python -m pip install --upgrade pip poetry + python -m pip install --upgrade pip "poetry==2.0.1" poetry install --all-extras - name: "Run tests" run: | diff --git a/README.md b/README.md index c5acc8aa3..0e4aab1f3 100644 --- a/README.md +++ b/README.md @@ -292,6 +292,7 @@ integration, this library supports also the following devices: * Qingping Air Monitor Lite (cgllc.airm.cgdn1) * Xiaomi Walkingpad A1 (ksmb.walkingpad.v3) * Xiaomi Smart Pet Water Dispenser (mmgg.pet_waterer.s1, s4, wi11) +* Xiaomi Smart Pet Fountain 2 (xiaomi.pet_waterer.70m2) * Xiaomi Mi Smart Humidifer S (jsqs, jsq5) * Xiaomi Mi Robot Vacuum Mop 2 (Pro+, Ultra) diff --git a/miio/__init__.py b/miio/__init__.py index 6fd9f4e64..6a863e4bd 100644 --- a/miio/__init__.py +++ b/miio/__init__.py @@ -82,6 +82,7 @@ from miio.integrations.viomi.vacuum import ViomiVacuum from miio.integrations.viomi.viomidishwasher import ViomiDishwasher from miio.integrations.xiaomi.aircondition.airconditioner_miot import AirConditionerMiot +from miio.integrations.xiaomi.petfountain import XiaomiPetFountain from miio.integrations.xiaomi.repeater.wifirepeater import WifiRepeater from miio.integrations.xiaomi.wifispeaker.wifispeaker import WifiSpeaker from miio.integrations.yeelight.dual_switch import YeelightDualControlModule diff --git a/miio/integrations/xiaomi/petfountain/__init__.py b/miio/integrations/xiaomi/petfountain/__init__.py new file mode 100644 index 000000000..e3d927fc9 --- /dev/null +++ b/miio/integrations/xiaomi/petfountain/__init__.py @@ -0,0 +1,8 @@ +# flake8: noqa +from .device import XiaomiPetFountain +from .status import ( + ChargingState, + PetFountainMode, + PetFountainStatus, + XiaomiPetFountainStatus, +) diff --git a/miio/integrations/xiaomi/petfountain/device.py b/miio/integrations/xiaomi/petfountain/device.py new file mode 100644 index 000000000..051a53cd7 --- /dev/null +++ b/miio/integrations/xiaomi/petfountain/device.py @@ -0,0 +1,136 @@ +import logging +from typing import Any + +import click + +from miio.click_common import EnumType, command, format_output +from miio.miot_device import MiotDevice + +from .status import PetFountainMode, XiaomiPetFountainStatus, _time_to_seconds + +_LOGGER = logging.getLogger(__name__) + +MODEL_XIAOMI_PET_WATERER_70M2 = "xiaomi.pet_waterer.70m2" + +MIOT_MAPPING = { + MODEL_XIAOMI_PET_WATERER_70M2: { + "fault_code": {"siid": 2, "piid": 1}, + "status": {"siid": 2, "piid": 3}, + "mode": {"siid": 2, "piid": 4}, + "water_shortage": {"siid": 2, "piid": 10}, + "water_interval": {"siid": 2, "piid": 11}, + "filter_life_remaining": {"siid": 3, "piid": 1}, + "filter_left_time": {"siid": 3, "piid": 2}, + "reset_filter_life": {"siid": 3, "aiid": 1}, + "child_lock": {"siid": 4, "piid": 1}, + "battery": {"siid": 5, "piid": 1}, + "charging_state": {"siid": 5, "piid": 2}, + "do_not_disturb": {"siid": 6, "piid": 1}, + "low_battery": {"siid": 9, "piid": 5}, + "usb_power": {"siid": 9, "piid": 6}, + "dnd_start": {"siid": 9, "piid": 10}, + "dnd_end": {"siid": 9, "piid": 11}, + "pump_blocked": {"siid": 9, "piid": 12}, + } +} + + +class XiaomiPetFountain(MiotDevice): + """Main class representing Xiaomi Pet Fountain 2.""" + + _mappings = MIOT_MAPPING + + @command( + default_output=format_output( + "", + "Status: {result.status}\n" + "Mode: {result.mode}\n" + "Water shortage: {result.water_shortage}\n" + "Pump blocked: {result.pump_blocked}\n" + "Filter life remaining: {result.filter_life_remaining}\n" + "Filter time remaining: {result.filter_left_time}\n" + "Battery: {result.battery}\n" + "Charging state: {result.charging_state}\n" + "Do not disturb: {result.do_not_disturb}\n" + "DND start: {result.dnd_start}\n" + "DND end: {result.dnd_end}\n" + "Child lock: {result.child_lock}\n" + "Fault code: {result.fault_code}\n", + ) + ) + def status(self) -> XiaomiPetFountainStatus: + """Retrieve properties.""" + data = { + prop["did"]: prop["value"] if prop["code"] == 0 else None + for prop in self.get_properties_for_mapping() + } + _LOGGER.debug(data) + return XiaomiPetFountainStatus(data) + + @command( + click.argument("mode", type=EnumType(PetFountainMode)), + default_output=format_output('Changing mode to "{mode.name}"'), + ) + def set_mode(self, mode: PetFountainMode) -> list[dict[str, Any]]: + """Set the water dispensing mode.""" + raw_mode = { + PetFountainMode.Auto: 0, + PetFountainMode.Interval: 1, + PetFountainMode.Continuous: 2, + }[mode] + return self.set_property("mode", raw_mode) + + @command( + click.argument("minutes", type=click.IntRange(10, 120)), + default_output=format_output('Changing water interval to "{minutes}" minutes'), + ) + def set_water_interval(self, minutes: int) -> list[dict[str, Any]]: + """Set the interval mode water interval in minutes.""" + if minutes % 5 != 0: + raise ValueError("Water interval must be set in 5 minute increments") + return self.set_property("water_interval", minutes) + + @command( + click.argument("enabled", type=bool), + default_output=format_output( + lambda enabled: "Enabling child lock" if enabled else "Disabling child lock" + ), + ) + def set_child_lock(self, enabled: bool) -> list[dict[str, Any]]: + """Set the child lock.""" + return self.set_property("child_lock", enabled) + + @command( + click.argument("enabled", type=bool), + default_output=format_output( + lambda enabled: "Enabling do not disturb" + if enabled + else "Disabling do not disturb" + ), + ) + def set_do_not_disturb(self, enabled: bool) -> list[dict[str, Any]]: + """Set do not disturb mode.""" + return self.set_property("do_not_disturb", enabled) + + @command(default_output=format_output("Resetting filter life")) + def reset_filter_life(self) -> dict[str, Any]: + """Reset filter life.""" + return self.call_action_from_mapping("reset_filter_life") + + @command( + click.argument("value", type=click.DateTime(formats=["%H:%M:%S", "%H:%M"])), + default_output=format_output('Changing DnD start to "{value}"'), + ) + def set_dnd_start(self, value) -> list[dict[str, Any]]: + """Set the DnD start time.""" + parsed = value.time() if hasattr(value, "time") else value + return self.set_property("dnd_start", _time_to_seconds(parsed)) + + @command( + click.argument("value", type=click.DateTime(formats=["%H:%M:%S", "%H:%M"])), + default_output=format_output('Changing DnD end to "{value}"'), + ) + def set_dnd_end(self, value) -> list[dict[str, Any]]: + """Set the DnD end time.""" + parsed = value.time() if hasattr(value, "time") else value + return self.set_property("dnd_end", _time_to_seconds(parsed)) diff --git a/miio/integrations/xiaomi/petfountain/status.py b/miio/integrations/xiaomi/petfountain/status.py new file mode 100644 index 000000000..f73289367 --- /dev/null +++ b/miio/integrations/xiaomi/petfountain/status.py @@ -0,0 +1,170 @@ +import enum +from datetime import time +from typing import Any, Optional + +from miio.miot_device import DeviceStatus + + +class PetFountainStatus(enum.Enum): + """The fountain operating status.""" + + NoWater = "no_water" + Watering = "watering" + + +class PetFountainMode(enum.Enum): + """The fountain water mode.""" + + Auto = "auto" + Interval = "interval" + Continuous = "continuous" + + +class ChargingState(enum.Enum): + """The fountain charging state.""" + + NotCharging = "not_charging" + Charging = "charging" + Charged = "charged" + + +class XiaomiPetFountainStatus(DeviceStatus): + """Container for status reports from Xiaomi Pet Fountain 2.""" + + def __init__(self, data: dict[str, Any]) -> None: + """Initialize the status container.""" + self.data = data + + @property + def is_on(self) -> bool: + """Return true to keep option entities available.""" + return True + + @property + def fault_code(self) -> Optional[int]: + """Return the raw fault code.""" + raw_fault = self.data.get("fault_code") + if not isinstance(raw_fault, int): + return None + return raw_fault + + @property + def has_fault(self) -> bool: + """Return true when the device reports a fault.""" + code = self.fault_code + return code is not None and code > 0 + + @property + def status(self) -> Optional[PetFountainStatus]: + """Return the fountain operating status.""" + raw_status = self.data.get("status") + if not isinstance(raw_status, int): + return None + return { + 1: PetFountainStatus.NoWater, + 2: PetFountainStatus.Watering, + }.get(raw_status) + + @property + def mode(self) -> Optional[PetFountainMode]: + """Return the configured water mode.""" + raw_mode = self.data.get("mode") + if not isinstance(raw_mode, int): + return None + return { + 0: PetFountainMode.Auto, + 1: PetFountainMode.Interval, + 2: PetFountainMode.Continuous, + }.get(raw_mode) + + @property + def water_interval(self) -> Optional[int]: + """Return the configured water interval in minutes.""" + return self.data.get("water_interval") + + @property + def water_shortage(self) -> Optional[bool]: + """Return true when the fountain is low on water.""" + return self.data.get("water_shortage") + + @property + def filter_life_remaining(self) -> Optional[int]: + """Return the remaining filter life in percent.""" + return self.data.get("filter_life_remaining") + + @property + def filter_left_time(self) -> Optional[float]: + """Return the remaining filter time in days.""" + value = self.data.get("filter_left_time") + if value is None: + return None + return round(value / 24, 2) + + @property + def child_lock(self) -> Optional[bool]: + """Return true when physical controls are locked.""" + return self.data.get("child_lock") + + @property + def battery(self) -> Optional[int]: + """Return battery level percentage.""" + return self.data.get("battery") + + @property + def charging_state(self) -> Optional[ChargingState]: + """Return the charging state.""" + raw_state = self.data.get("charging_state") + if not isinstance(raw_state, int): + return None + return { + 0: ChargingState.NotCharging, + 1: ChargingState.Charging, + 2: ChargingState.Charged, + }.get(raw_state) + + @property + def do_not_disturb(self) -> Optional[bool]: + """Return true when do not disturb is enabled.""" + return self.data.get("do_not_disturb") + + @property + def low_battery(self) -> Optional[bool]: + """Return true when the device reports low battery.""" + return self.data.get("low_battery") + + @property + def usb_power(self) -> Optional[bool]: + """Return true when USB power is connected.""" + return self.data.get("usb_power") + + @property + def dnd_start(self) -> Optional[time]: + """Return the DnD start time.""" + value = self.data.get("dnd_start") + if value is None: + return None + return _seconds_to_time(value) + + @property + def dnd_end(self) -> Optional[time]: + """Return the DnD end time.""" + value = self.data.get("dnd_end") + if value is None: + return None + return _seconds_to_time(value) + + @property + def pump_blocked(self) -> Optional[bool]: + """Return true when the pump is blocked.""" + return self.data.get("pump_blocked") + + +def _seconds_to_time(value: int) -> time: + value %= 24 * 60 * 60 + hours, remainder = divmod(value, 3600) + minutes, seconds = divmod(remainder, 60) + return time(hour=hours, minute=minutes, second=seconds) + + +def _time_to_seconds(value: time) -> int: + return value.hour * 3600 + value.minute * 60 + value.second diff --git a/miio/integrations/xiaomi/petfountain/tests/__init__.py b/miio/integrations/xiaomi/petfountain/tests/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/miio/integrations/xiaomi/petfountain/tests/test_pet_fountain.py b/miio/integrations/xiaomi/petfountain/tests/test_pet_fountain.py new file mode 100644 index 000000000..451ec562f --- /dev/null +++ b/miio/integrations/xiaomi/petfountain/tests/test_pet_fountain.py @@ -0,0 +1,155 @@ +from datetime import time +from unittest import TestCase + +import pytest + +from miio.tests.dummies import DummyMiotDevice + +from ..device import MODEL_XIAOMI_PET_WATERER_70M2, XiaomiPetFountain +from ..status import ChargingState, PetFountainMode, PetFountainStatus + +_INITIAL_STATE = { + "fault_code": 0, + "status": 2, + "mode": 1, + "water_shortage": False, + "water_interval": 25, + "filter_life_remaining": 76, + "filter_left_time": 23, + "child_lock": False, + "battery": 21, + "charging_state": 0, + "do_not_disturb": False, + "low_battery": False, + "usb_power": False, + "dnd_start": 22 * 3600, + "dnd_end": 8 * 3600 + 30 * 60, + "pump_blocked": False, +} + +_UNKNOWN_STATE = { + "fault_code": None, + "status": None, + "mode": None, + "water_shortage": None, + "water_interval": None, + "filter_life_remaining": None, + "filter_left_time": None, + "child_lock": None, + "battery": None, + "charging_state": None, + "do_not_disturb": None, + "low_battery": None, + "usb_power": None, + "dnd_start": None, + "dnd_end": None, + "pump_blocked": None, +} + + +class DummyXiaomiPetFountain(DummyMiotDevice, XiaomiPetFountain): + def __init__(self, *args, **kwargs): + self._model = MODEL_XIAOMI_PET_WATERER_70M2 + self.state = _INITIAL_STATE + self.return_values = { + "action": lambda payload: payload, + } + super().__init__(*args, **kwargs) + + +@pytest.fixture(scope="function") +def petfountain(request): + request.cls.device = DummyXiaomiPetFountain() + + +@pytest.mark.usefixtures("petfountain") +class TestXiaomiPetFountain(TestCase): + def test_status(self): + status = self.device.status() + + assert status.is_on is True + assert status.fault_code == 0 + assert status.has_fault is False + assert status.status == PetFountainStatus.Watering + assert status.mode == PetFountainMode.Interval + assert status.water_shortage is False + assert status.water_interval == 25 + assert status.filter_life_remaining == 76 + assert status.filter_left_time == round(23 / 24, 2) + assert status.child_lock is False + assert status.battery == 21 + assert status.charging_state == ChargingState.NotCharging + assert status.do_not_disturb is False + assert status.low_battery is False + assert status.usb_power is False + assert status.dnd_start == time(22, 0) + assert status.dnd_end == time(8, 30) + assert status.pump_blocked is False + + def test_set_mode(self): + self.device.set_mode(PetFountainMode.Auto) + assert self.device.status().mode == PetFountainMode.Auto + + self.device.set_mode(PetFountainMode.Interval) + assert self.device.status().mode == PetFountainMode.Interval + + self.device.set_mode(PetFountainMode.Continuous) + assert self.device.status().mode == PetFountainMode.Continuous + + def test_set_water_interval(self): + self.device.set_water_interval(45) + assert self.device.status().water_interval == 45 + + with pytest.raises(ValueError): + self.device.set_water_interval(43) + + def test_set_child_lock(self): + self.device.set_child_lock(True) + assert self.device.status().child_lock is True + + self.device.set_child_lock(False) + assert self.device.status().child_lock is False + + def test_set_do_not_disturb(self): + self.device.set_do_not_disturb(True) + assert self.device.status().do_not_disturb is True + + self.device.set_do_not_disturb(False) + assert self.device.status().do_not_disturb is False + + def test_set_dnd_start(self): + self.device.set_dnd_start(time(21, 15)) + assert self.device.status().dnd_start == time(21, 15) + + def test_set_dnd_end(self): + self.device.set_dnd_end(time(7, 45)) + assert self.device.status().dnd_end == time(7, 45) + + def test_reset_filter_life(self): + result = self.device.reset_filter_life() + assert result["did"] == "call-3-1" + + def test_status_handles_missing_values(self): + self.device.state = [ + {"did": k, "value": v, "code": 0} for k, v in _UNKNOWN_STATE.items() + ] + + status = self.device.status() + + assert status.fault_code is None + assert status.has_fault is False + assert status.status is None + assert status.mode is None + assert status.water_shortage is None + assert status.water_interval is None + assert status.filter_life_remaining is None + assert status.filter_left_time is None + assert status.child_lock is None + assert status.battery is None + assert status.charging_state is None + assert status.do_not_disturb is None + assert status.low_battery is None + assert status.usb_power is None + assert status.dnd_start is None + assert status.dnd_end is None + assert status.pump_blocked is None