Source code for miio.discovery

import codecs
import inspect
import logging
import time
from functools import partial
from ipaddress import ip_address
from typing import Callable, Dict, Optional, Type, Union  # noqa: F401

import zeroconf

from miio.integrations.airpurifier import (
    AirDogX3,
    AirFresh,
    AirFreshT2017,
    AirPurifier,
    AirPurifierMiot,
)
from miio.integrations.humidifier import (
    AirHumidifier,
    AirHumidifierJsq,
    AirHumidifierJsqs,
    AirHumidifierMjjsq,
)
from miio.integrations.vacuum import DreameVacuum, RoborockVacuum, ViomiVacuum

from . import (
    AirConditionerMiot,
    AirConditioningCompanion,
    AirConditioningCompanionMcn02,
    AirQualityMonitor,
    AqaraCamera,
    Ceil,
    ChuangmiCamera,
    ChuangmiIr,
    ChuangmiPlug,
    Cooker,
    Device,
    Gateway,
    Heater,
    PowerStrip,
    Toiletlid,
    WaterPurifier,
    WaterPurifierYunmi,
    WifiRepeater,
    WifiSpeaker,
)
from .airconditioningcompanion import (
    MODEL_ACPARTNER_V1,
    MODEL_ACPARTNER_V2,
    MODEL_ACPARTNER_V3,
)
from .airconditioningcompanionMCN import MODEL_ACPARTNER_MCN02
from .airqualitymonitor import (
    MODEL_AIRQUALITYMONITOR_B1,
    MODEL_AIRQUALITYMONITOR_S1,
    MODEL_AIRQUALITYMONITOR_V1,
)
from .alarmclock import AlarmClock
from .chuangmi_plug import (
    MODEL_CHUANGMI_PLUG_HMI205,
    MODEL_CHUANGMI_PLUG_HMI206,
    MODEL_CHUANGMI_PLUG_M1,
    MODEL_CHUANGMI_PLUG_M3,
    MODEL_CHUANGMI_PLUG_V1,
    MODEL_CHUANGMI_PLUG_V2,
    MODEL_CHUANGMI_PLUG_V3,
)
from .heater import MODEL_HEATER_MA1, MODEL_HEATER_ZA1
from .integrations.fan import Fan, FanLeshow, FanMiot, FanZA5
from .integrations.light import (
    PhilipsBulb,
    PhilipsEyecare,
    PhilipsMoonlight,
    PhilipsRwread,
    PhilipsWhiteBulb,
    Yeelight,
)
from .powerstrip import MODEL_POWER_STRIP_V1, MODEL_POWER_STRIP_V2
from .toiletlid import MODEL_TOILETLID_V1

_LOGGER = logging.getLogger(__name__)


DEVICE_MAP: Dict[str, Union[Type[Device], partial]] = {
    "rockrobo-vacuum-v1": RoborockVacuum,
    "roborock-vacuum-s5": RoborockVacuum,
    "roborock-vacuum-m1s": RoborockVacuum,
    "roborock-vacuum-a10": RoborockVacuum,
    "chuangmi-plug-m1": partial(ChuangmiPlug, model=MODEL_CHUANGMI_PLUG_M1),
    "chuangmi-plug-m3": partial(ChuangmiPlug, model=MODEL_CHUANGMI_PLUG_M3),
    "chuangmi-plug-v1": partial(ChuangmiPlug, model=MODEL_CHUANGMI_PLUG_V1),
    "chuangmi-plug-v2": partial(ChuangmiPlug, model=MODEL_CHUANGMI_PLUG_V2),
    "chuangmi-plug-v3": partial(ChuangmiPlug, model=MODEL_CHUANGMI_PLUG_V3),
    "chuangmi-plug-hmi205": partial(ChuangmiPlug, model=MODEL_CHUANGMI_PLUG_HMI205),
    "chuangmi-plug-hmi206": partial(ChuangmiPlug, model=MODEL_CHUANGMI_PLUG_HMI206),
    "chuangmi-plug_": partial(ChuangmiPlug, model=MODEL_CHUANGMI_PLUG_V1),
    "qmi-powerstrip-v1": partial(PowerStrip, model=MODEL_POWER_STRIP_V1),
    "zimi-powerstrip-v2": partial(PowerStrip, model=MODEL_POWER_STRIP_V2),
    "zimi-clock-myk01": AlarmClock,
    "xiaomi.aircondition.mc1": AirConditionerMiot,
    "xiaomi.aircondition.mc2": AirConditionerMiot,
    "xiaomi.aircondition.mc4": AirConditionerMiot,
    "xiaomi.aircondition.mc5": AirConditionerMiot,
    "airdog-airpurifier-x3": AirDogX3,
    "airdog-airpurifier-x5": AirDogX3,
    "airdog-airpurifier-x7sm": AirDogX3,
    "zhimi-airpurifier-m1": AirPurifier,  # mini model
    "zhimi-airpurifier-m2": AirPurifier,  # mini model 2
    "zhimi-airpurifier-ma1": AirPurifier,  # ms model
    "zhimi-airpurifier-ma2": AirPurifier,  # ms model 2
    "zhimi-airpurifier-sa1": AirPurifier,  # super model
    "zhimi-airpurifier-sa2": AirPurifier,  # super model 2
    "zhimi-airpurifier-v1": AirPurifier,  # v1
    "zhimi-airpurifier-v2": AirPurifier,  # v2
    "zhimi-airpurifier-v3": AirPurifier,  # v3
    "zhimi-airpurifier-v5": AirPurifier,  # v5
    "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)
    "zhimi-airpurifier-vb2": AirPurifierMiot,  # vb2 (Pro H)
    "chuangmi-camera-ipc009": ChuangmiCamera,
    "chuangmi-camera-ipc013": ChuangmiCamera,
    "chuangmi-camera-ipc019": ChuangmiCamera,
    "chuangmi-camera-038a2": ChuangmiCamera,
    "chuangmi-ir-v2": ChuangmiIr,
    "chuangmi-remote-h102a03_": ChuangmiIr,
    "zhimi-humidifier-v1": AirHumidifier,
    "zhimi-humidifier-ca1": AirHumidifier,
    "zhimi-humidifier-cb1": AirHumidifier,
    "shuii-humidifier-jsq001": AirHumidifierJsq,
    "deerma-humidifier-mjjsq": AirHumidifierMjjsq,
    "deerma-humidifier-jsq1": AirHumidifierMjjsq,
    "deerma-humidifier-jsqs": AirHumidifierJsqs,
    "yunmi-waterpuri-v2": WaterPurifier,
    "yunmi.waterpuri.lx9": WaterPurifierYunmi,
    "yunmi.waterpuri.lx11": WaterPurifierYunmi,
    "philips-light-bulb": PhilipsBulb,  # cannot be discovered via mdns
    "philips-light-hbulb": PhilipsWhiteBulb,  # cannot be discovered via mdns
    "philips-light-candle": PhilipsBulb,  # cannot be discovered via mdns
    "philips-light-candle2": PhilipsBulb,  # cannot be discovered via mdns
    "philips-light-ceiling": Ceil,
    "philips-light-zyceiling": Ceil,
    "philips-light-sread1": PhilipsEyecare,  # name needs to be checked
    "philips-light-moonlight": PhilipsMoonlight,  # name needs to be checked
    "philips-light-rwread": PhilipsRwread,  # name needs to be checked
    "xiaomi-wifispeaker-v1": WifiSpeaker,  # name needs to be checked
    "xiaomi-repeater-v1": WifiRepeater,  # name needs to be checked
    "xiaomi-repeater-v3": WifiRepeater,  # name needs to be checked
    "chunmi-cooker-press1": Cooker,
    "chunmi-cooker-press2": Cooker,
    "chunmi-cooker-normal1": Cooker,
    "chunmi-cooker-normal2": Cooker,
    "chunmi-cooker-normal3": Cooker,
    "chunmi-cooker-normal4": Cooker,
    "chunmi-cooker-normal5": Cooker,
    "lumi-acpartner-v1": partial(AirConditioningCompanion, model=MODEL_ACPARTNER_V1),
    "lumi-acpartner-v2": partial(AirConditioningCompanion, model=MODEL_ACPARTNER_V2),
    "lumi-acpartner-v3": partial(AirConditioningCompanion, model=MODEL_ACPARTNER_V3),
    "lumi-acpartner-mcn02": partial(
        AirConditioningCompanionMcn02, model=MODEL_ACPARTNER_MCN02
    ),
    "lumi-camera-aq2": AqaraCamera,
    "yeelink-light-": Yeelight,
    "leshow-fan-ss4": FanLeshow,
    "zhimi-fan-v2": Fan,
    "zhimi-fan-v3": Fan,
    "zhimi-fan-sa1": Fan,
    "zhimi-fan-za1": Fan,
    "zhimi-fan-za3": Fan,
    "zhimi-fan-za4": Fan,
    "dmaker-fan-1c": FanMiot,
    "dmaker-fan-p5": Fan,
    "dmaker-fan-p9": FanMiot,
    "dmaker-fan-p10": FanMiot,
    "dmaker-fan-p11": FanMiot,
    "zhimi-fan-za5": FanZA5,
    "tinymu-toiletlid-v1": partial(Toiletlid, model=MODEL_TOILETLID_V1),
    "zhimi-airfresh-va2": AirFresh,
    "zhimi-airfresh-va4": AirFresh,
    "dmaker-airfresh-t2017": AirFreshT2017,
    "zhimi-airmonitor-v1": partial(AirQualityMonitor, model=MODEL_AIRQUALITYMONITOR_V1),
    "cgllc-airmonitor-b1": partial(AirQualityMonitor, model=MODEL_AIRQUALITYMONITOR_B1),
    "cgllc-airmonitor-s1": partial(AirQualityMonitor, model=MODEL_AIRQUALITYMONITOR_S1),
    "lumi-gateway-": Gateway,
    "viomi-vacuum-v7": ViomiVacuum,
    "viomi-vacuum-v8": ViomiVacuum,
    "zhimi.heater.za1": partial(Heater, model=MODEL_HEATER_ZA1),
    "zhimi.elecheater.ma1": partial(Heater, model=MODEL_HEATER_MA1),
    "dreame-vacuum-mc1808": DreameVacuum,
    "dreame-vacuum-p2008": DreameVacuum,
    "dreame-vacuum-p2028": DreameVacuum,
    "dreame-vacuum-p2009": DreameVacuum,
}


[docs]def pretty_token(token): """Return a pretty string presentation for a token.""" return codecs.encode(token, "hex").decode()
[docs]def get_addr_from_info(info): addrs = info.addresses if len(addrs) > 1: _LOGGER.warning( "More than single IP address in the advertisement, using the first one" ) return str(ip_address(addrs[0]))
[docs]def other_package_info(info, desc): """Return information about another package supporting the device.""" return f"Found {info.name} at {get_addr_from_info(info)}, check {desc}"
[docs]def create_device(name: str, addr: str, device_cls: partial) -> Device: """Return a device object for a zeroconf entry.""" _LOGGER.debug( "Found a supported '%s', using '%s' class", name, device_cls.func.__name__ ) dev = device_cls(ip=addr) m = dev.send_handshake() dev.token = m.checksum _LOGGER.info( "Found a supported '%s' at %s - token: %s", device_cls.func.__name__, addr, pretty_token(dev.token), ) return dev
[docs]class Listener(zeroconf.ServiceListener): """mDNS listener creating Device objects based on detected devices.""" def __init__(self): self.found_devices = {} # type: Dict[str, Device]
[docs] def check_and_create_device(self, info, addr) -> Optional[Device]: """Create a corresponding :class:`Device` implementation for a given info and address..""" name = info.name for identifier, v in DEVICE_MAP.items(): if name.startswith(identifier): if inspect.isclass(v): return create_device(name, addr, partial(v)) elif isinstance(v, partial) and inspect.isclass(v.func): return create_device(name, addr, v) elif callable(v): dev = Device(ip=addr) _LOGGER.info( "%s: token: %s", v(info), pretty_token(dev.send_handshake().checksum), ) return None _LOGGER.warning( "Found unsupported device %s at %s, " "please report to developers", name, addr, ) return None
[docs] def add_service(self, zeroconf: "zeroconf.Zeroconf", type_: str, name: str) -> None: """Callback for discovery responses.""" info = zeroconf.get_service_info(type_, name) addr = get_addr_from_info(info) if addr not in self.found_devices: dev = self.check_and_create_device(info, addr) if dev is not None: self.found_devices[addr] = dev
[docs] def update_service(self, zc: "zeroconf.Zeroconf", type_: str, name: str) -> None: """Callback for state updates, which we ignore for now."""
[docs]class Discovery: """mDNS discoverer for miIO based devices (_miio._udp.local). Calling :func:`discover_mdns` will cause this to subscribe for updates on ``_miio._udp.local`` until any key is pressed, after which a dict of detected devices is returned. """
[docs] @staticmethod def discover_mdns(*, timeout=5) -> Dict[str, Device]: """Discover devices with mdns until any keyboard input.""" _LOGGER.info("Discovering devices with mDNS for %s seconds...", timeout) listener = Listener() browser = zeroconf.ServiceBrowser( zeroconf.Zeroconf(), "_miio._udp.local.", listener ) time.sleep(timeout) browser.cancel() return listener.found_devices