Add auto connect to 360 eye

.gitignore

@@ -15,4 +15,5 @@ main/
 .pydevproject
 docs/_build
 README.rst
-.vscode
+.vscode
+venv/

libpurecool/dyson_360_eye.py

@@ -7,7 +7,7 @@
 
 import paho.mqtt.client as mqtt
 
-from .dyson_device import DysonDevice, NetworkDevice, DEFAULT_PORT
+from .dyson_device import DysonDevice
 from .utils import printable_fields
 from .const import PowerMode, Dyson360EyeMode, Dyson360EyeCommand
 
@@ -17,16 +17,22 @@
 class Dyson360Eye(DysonDevice):
     """Dyson 360 Eye device."""
 
-    def connect(self, device_ip, device_port=DEFAULT_PORT):
-        """Try to connect to device.
+    def auto_connect(self, timeout=5, retry=15):
+        """Try to connect to device using mDNS.
 
-        :param device_ip: Device IP address
-        :param device_port: Device Port (default: 1883)
+        :param timeout: Timeout
+        :param retry: Max retry
         :return: True if connected, else False
         """
-        self._network_device = NetworkDevice(self._name, device_ip,
-                                             device_port)
+        return self._auto_connect("_360eye_mqtt._tcp.local.", timeout, retry)
 
+    @staticmethod
+    def _device_serial_from_name(name):
+        """Get device serial from mDNS name."""
+        return (name.split(".")[0]).split("-", 1)[1]
+
+    def _mqtt_connect(self):
+        """Connect to the MQTT broker."""
         self._mqtt = mqtt.Client(userdata=self, protocol=3)
         self._mqtt.username_pw_set(self._serial, self._credentials)
         self._mqtt.on_message = self.on_message

libpurecool/dyson_device.py

@@ -2,14 +2,16 @@
 
 # pylint: disable=too-many-public-methods,too-many-instance-attributes
 
-from queue import Queue
+from queue import Queue, Empty
 import logging
 import json
 import abc
 import time
+import socket
 
 from .utils import printable_fields
 from .utils import decrypt_password
+from .zeroconf import ServiceBrowser, Zeroconf
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -64,6 +66,41 @@ def __repr__(self):
 class DysonDevice:
     """Abstract Dyson device."""
 
+    class DysonDeviceListener:
+        """Message listener."""
+
+        def __init__(self, serial, add_device_function, serial_from_name):
+            """Create a new message listener.
+
+            :param serial: Device serial
+            :param add_device_function: Callback function
+            """
+            self._serial = serial
+            self.add_device_function = add_device_function
+            self.serial_from_name = serial_from_name
+
+        def remove_service(self, zeroconf, device_type, name):
+            # pylint: disable=unused-argument,no-self-use
+            """Remove listener."""
+            _LOGGER.info("Service %s removed", name)
+
+        def add_service(self, zeroconf, device_type, name):
+            """Add device.
+
+            :param zeroconf: MSDNS object
+            :param device_type: Service type
+            :param name: Device name
+            """
+            device_serial = self.serial_from_name(name)
+            if device_serial == self._serial:
+                # Find searched device
+                info = zeroconf.get_service_info(device_type, name)
+                address = socket.inet_ntoa(info.address)
+                network_device = NetworkDevice(device_serial, address,
+                                               info.port)
+                self.add_device_function(network_device)
+                zeroconf.close()
+
     @staticmethod
     def on_connect(client, userdata, flags, return_code):
         # pylint: disable=unused-argument
@@ -109,21 +146,54 @@ def connection_callback(self, connected):
         """Set function called when device is connected."""
         self._connection_queue.put_nowait(connected)
 
-    @abc.abstractmethod
     def connect(self, device_ip, device_port=DEFAULT_PORT):
         """Connect to the device using ip address.
 
         :param device_ip: Device IP address
         :param device_port: Device Port (default: 1883)
         :return: True if connected, else False
         """
-        return
+        self._network_device = NetworkDevice(self._name, device_ip,
+                                             device_port)
+
+        return self._mqtt_connect()
+
+    def _auto_connect(self, type_, timeout=5, retry=15):
+        """Try to connect to device using mDNS."""
+        for i in range(retry):
+            zeroconf = Zeroconf()
+            listener = self.DysonDeviceListener(self._serial,
+                                                self._add_network_device,
+                                                self._device_serial_from_name)
+            ServiceBrowser(zeroconf, type_, listener)
+            try:
+                self._network_device = self._search_device_queue.get(
+                    timeout=timeout)
+            except Empty:
+                # Unable to find device
+                _LOGGER.warning("Unable to find device %s, try %s",
+                                self._serial, i)
+                zeroconf.close()
+            else:
+                break
+        if self._network_device is None:
+            _LOGGER.error("Unable to connect to device %s", self._serial)
+            return False
+        return self._mqtt_connect()
+
+    @abc.abstractmethod
+    def _mqtt_connect(self):
+        """Connect to the MQTT broker."""
+
+    @staticmethod
+    @abc.abstractmethod
+    def _device_serial_from_name(name):
+        """Get device serial from mDNS name."""
 
     @property
     @abc.abstractmethod
     def status_topic(self):
         """MQTT status topic."""
-        return
 
     @property
     def command_topic(self):

libpurecool/dyson_pure_cool_link.py

@@ -5,7 +5,6 @@
 import json
 import logging
 import time
-import socket
 from threading import Thread
 from queue import Queue, Empty
 
@@ -15,53 +14,18 @@
 from .dyson_pure_state_v2 import \
     DysonEnvironmentalSensorV2State, DysonPureCoolV2State, \
     DysonPureHotCoolV2State
-from .dyson_device import DysonDevice, NetworkDevice, DEFAULT_PORT
+from .dyson_device import DysonDevice
 from .utils import printable_fields, support_heating, is_pure_cool_v2, \
     support_heating_v2
 from .dyson_pure_state import DysonPureHotCoolState, DysonPureCoolState, \
     DysonEnvironmentalSensorState
-from .zeroconf import ServiceBrowser, Zeroconf
 
 _LOGGER = logging.getLogger(__name__)
 
 
 class DysonPureCoolLink(DysonDevice):
     """Dyson device (fan)."""
 
-    class DysonDeviceListener:
-        """Message listener."""
-
-        def __init__(self, serial, add_device_function):
-            """Create a new message listener.
-
-            :param serial: Device serial
-            :param add_device_function: Callback function
-            """
-            self._serial = serial
-            self.add_device_function = add_device_function
-
-        def remove_service(self, zeroconf, device_type, name):
-            # pylint: disable=unused-argument,no-self-use
-            """Remove listener."""
-            _LOGGER.info("Service %s removed", name)
-
-        def add_service(self, zeroconf, device_type, name):
-            """Add device.
-
-            :param zeroconf: MSDNS object
-            :param device_type: Service type
-            :param name: Device name
-            """
-            device_serial = (name.split(".")[0]).split("_")[1]
-            if device_serial == self._serial:
-                # Find searched device
-                info = zeroconf.get_service_info(device_type, name)
-                address = socket.inet_ntoa(info.address)
-                network_device = NetworkDevice(device_serial, address,
-                                               info.port)
-                self.add_device_function(network_device)
-                zeroconf.close()
-
     def __init__(self, json_body):
         """Create a new Pure Cool Link device.
 
@@ -119,37 +83,12 @@ def auto_connect(self, timeout=5, retry=15):
         :param retry: Max retry
         :return: True if connected, else False
         """
-        for i in range(retry):
-            zeroconf = Zeroconf()
-            listener = self.DysonDeviceListener(self._serial,
-                                                self._add_network_device)
-            ServiceBrowser(zeroconf, "_dyson_mqtt._tcp.local.", listener)
-            try:
-                self._network_device = self._search_device_queue.get(
-                    timeout=timeout)
-            except Empty:
-                # Unable to find device
-                _LOGGER.warning("Unable to find device %s, try %s",
-                                self._serial, i)
-                zeroconf.close()
-            else:
-                break
-        if self._network_device is None:
-            _LOGGER.error("Unable to connect to device %s", self._serial)
-            return False
-        return self._mqtt_connect()
-
-    def connect(self, device_ip, device_port=DEFAULT_PORT):
-        """Connect to the device using ip address.
+        return self._auto_connect("_dyson_mqtt._tcp.local.", timeout, retry)
 
-        :param device_ip: Device IP address
-        :param device_port: Device Port (default: 1883)
-        :return: True if connected, else False
-        """
-        self._network_device = NetworkDevice(self._name, device_ip,
-                                             device_port)
-
-        return self._mqtt_connect()
+    @staticmethod
+    def _device_serial_from_name(name):
+        """Get device serial from mDNS name."""
+        return (name.split(".")[0]).split("_")[1]
 
     def _mqtt_connect(self):
         """Connect to the MQTT broker."""

tests/test_360_eye.py

@@ -4,9 +4,10 @@
 from unittest.mock import Mock
 import json
 
-from libpurecool.dyson_360_eye import Dyson360Eye, NetworkDevice, \
+from libpurecool.dyson_360_eye import Dyson360Eye, \
     Dyson360EyeState, Dyson360EyeMapGlobal, Dyson360EyeMapData, \
     Dyson360EyeMapGrid, Dyson360EyeTelemetryData, Dyson360Goodbye
+from libpurecool.dyson_device import NetworkDevice
 from libpurecool.const import PowerMode, Dyson360EyeMode
 
 

tests/test_device.py

@@ -0,0 +1,68 @@
+from libpurecool.dyson_360_eye import Dyson360Eye
+import pytest
+import socket
+from libpurecool.zeroconf import ServiceInfo, Zeroconf
+from unittest import mock
+from unittest.mock import MagicMock
+
+from libpurecool.dyson_pure_cool_link import DysonPureCoolLink
+
+IP_ADDRESS = "192.168.1.2"
+SERIAL = "XXX-XX-XXXXXXXX"
+
+
+def _mocked_zeroconf():
+    def _get_service_info(*args):
+        service_info = MagicMock(spec=ServiceInfo)
+        service_info.address = socket.inet_aton(IP_ADDRESS)
+        service_info.port = 1883
+        return service_info
+
+    zeroconf = MagicMock(spec=Zeroconf)
+    zeroconf.get_service_info = MagicMock(side_effect=_get_service_info)
+    return zeroconf
+
+
+def _get_mocked_service_browser(serial):
+    def _mocked_service_browser(zeroconf, type, listener):
+        listener.add_service(zeroconf, type, "{}.{}".format(serial, type))
+    return _mocked_service_browser
+
+
+@pytest.mark.parametrize(
+    "device_class,serial_prefix",
+    [
+        (DysonPureCoolLink, "PURE-COOL-LINK_"),
+        (Dyson360Eye, "360EYE-"),
+    ]
+)
+def test_auto_connect(device_class, serial_prefix):
+    with mock.patch(
+        'libpurecool.dyson_device.ServiceBrowser',
+        side_effect=_get_mocked_service_browser(serial_prefix + SERIAL),
+    ), mock.patch(
+        'libpurecool.dyson_device.Zeroconf',
+        side_effect=_mocked_zeroconf,
+    ), mock.patch('paho.mqtt.client.Client'):
+        device = device_class({
+            "Active": True,
+            "Serial": SERIAL,
+            "Name": "device-1",
+            "ScaleUnit": "SU01",
+            "Version": "21.03.08",
+            "LocalCredentials": "1/aJ5t52WvAfn+z+fjDuef86kQDQPefbQ6/"
+                                "70ZGysII1Ke1i0ZHakFH84DZuxsSQ4KTT2v"
+                                "bCm7uYeTORULKLKQ==",
+            "AutoUpdate": True,
+            "NewVersionAvailable": False,
+            "ProductType": "475"
+        })
+        device.state_data_available()
+        if hasattr(device, "sensor_data_available"):
+            device.sensor_data_available()
+        device.connection_callback(True)
+        connected = device.auto_connect()
+        assert connected is True
+        assert device.state is None
+        if hasattr(device, "disconnect"):
+            device.disconnect()

tests/test_libpurecoollink.py

@@ -103,6 +103,10 @@ def on_add_device(network_device):
     pass
 
 
+def device_serial_from_name(name):
+    return (name.split(".")[0]).split("_")[1]
+
+
 class TestLibPureCoolLink(unittest.TestCase):
     def setUp(self):
         pass
@@ -236,8 +240,11 @@ def test_status_topic(self):
 
     @mock.patch('socket.inet_ntoa', )
     def test_device_dyson_listener(self, mocked_ntoa):
-        listener = DysonPureCoolLink.DysonDeviceListener('serial-1',
-                                                         on_add_device)
+        listener = DysonPureCoolLink.DysonDeviceListener(
+            'serial-1',
+            on_add_device,
+            device_serial_from_name
+        )
         zeroconf = Mock()
         listener.remove_service(zeroconf, "ptype", "serial-1")
         info = Mock()