Add Xiaomi Smart Pet Fountain 2 support

.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: |

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)
 

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

miio/integrations/xiaomi/petfountain/__init__.py

@@ -0,0 +1,8 @@
+# flake8: noqa
+from .device import XiaomiPetFountain
+from .status import (
+    ChargingState,
+    PetFountainMode,
+    PetFountainStatus,
+    XiaomiPetFountainStatus,
+)

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))

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