Repository: gwww/elkm1 (language: python)

Issue:
[Enhancement] Discovery Support
Now that home assistant has `dhcp` discovery (https://developers.home-assistant.io/docs/creating_integration_manifest/#dhcp), we could watch for digiboard with the oui `00:40:9D` and then do discovery if one is found if the ip isn't already configured.

Example discovery handler: https://github.com/home-assistant/core/blob/dev/homeassistant/components/roomba/config_flow.py#L80

https://github.com/home-assistant/core/blob/dev/homeassistant/components/roomba/manifest.json#L8

For elkm1 it would be:
`"dhcp": [{"macaddress":"00409D*"}]`


## A working discovery implementation
Idea borrowed from https://github.com/pschmitt/roombapy/blob/master/roombapy/discovery.py 
```
import logging
import socket

from struct import unpack


class ElkDiscovery:
    udp_bind_address = ""
    udp_address = "<broadcast>"
    udp_port = 2362
    elk_message = b"XEPID"
    amount_of_broadcasted_messages = 5
    server_socket = None
    log = None

    def __init__(self):
        self.server_socket = _get_socket()
        self.log = logging.getLogger(__name__)

    def find(self, ip=None):
        if ip is not None:
            return self.get(ip)
        return self.get_all()

    def get_all(self):
        self._start_server()
        self._broadcast_message(self.amount_of_broadcasted_messages)
        elks = set()
        while True:
            response = self._get_response()
            if response:
                elks.add(response)
            else:
                break
        return elks

    def get(self, ip):
        self._start_server()
        self._send_message(ip)
        return self._get_response(ip)

    def _get_response(self, ip=None):
        try:
            while True:
                raw_response, addr = self.server_socket.recvfrom(1024)
                if ip is not None and addr[0] != ip:
                    continue
                self.log.debug("Received response: %s, address: %s", raw_response, addr)
                if self._is_from_elk(raw_response):
                    return _decode_data(raw_response)

        except socket.timeout:
            self.log.info("Socket timeout")
            return None

    def _is_from_elk(self, raw_response):
        if not raw_response.startswith(b"M1XEP"):
            return False
        return True

    def _broadcast_message(self, amount):
        for i in range(amount):
            self.server_socket.sendto(
                self.elk_message, (self.udp_address, self.udp_port)
            )
            self.log.debug("Broadcast message sent: " + str(i))

    def _send_message(self, udp_address):
        self.server_socket.sendto(self.elk_message, (udp_address, self.udp_port))
        self.log.debug("Message sent")

    def _start_server(self):
        self.server_socket.bind((self.udp_bind_address, self.udp_port))
        self.log.debug("Socket server started, port %s", self.udp_port)


def _decode_data(raw_response):
    remain = raw_response[5:]
    data = remain[:14]
    (
        mac1,
        mac2,
        mac3,
        mac4,
        mac5,
        mac6,
        ipv4_1,
        ipv4_2,
        ipv4_3,
        ipv4_4,
        port,
        _,
    ) = unpack("!6B4BHH", data)
    mac_address = ":".join(
        [format(val, "02x").upper() for val in (mac1, mac2, mac3, mac4, mac5, mac6)]
    )
    ip_address = f"{ipv4_1}.{ipv4_2}.{ipv4_3}.{ipv4_4}"

    return ElkSystem(mac_address, ip_address, port)


class ElkSystem:
    def __init__(self, mac_address, ip_address, port):
        self.mac_address = mac_address
        self.ip_address = ip_address
        self.port = port

    def __hash__(self):
        return hash(self.mac_address)

    def __eq__(self, o: object) -> bool:
        return isinstance(o, ElkSystem) and self.mac_address == o.mac_address

    def __repr__(self):
        return f"<ElkSystem {self.mac_address} {self.ip_address} {self.port}>"


def _get_socket():
    server_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
    server_socket.settimeout(5)
    return server_socket


logging.basicConfig(level=logging.DEBUG)
import pprint

# Discovery
pprint.pprint(ElkDiscovery().get_all())

# Get the mac of an existing known ip
pprint.pprint(ElkDiscovery().get("192.168.209.56"))
```


Discussion / hints:

Fix patch:
diff --git a/elkm1_lib/discovery.py b/elkm1_lib/discovery.py
new file mode 100644
index 0000000..d3c349f
--- /dev/null
+++ b/elkm1_lib/discovery.py
@@ -0,0 +1,180 @@
+import asyncio
+import logging
+import socket
+import time
+from dataclasses import dataclass
+from struct import unpack
+from typing import Callable, Dict, List, Optional, Tuple
+
+_LOGGER = logging.getLogger(__name__)
+
+
+@dataclass
+class ElkSystem:
+    """An ELKM1 system."""
+
+    mac_address: str
+    ip_address: str
+    port: int
+
+
+def create_udp_socket(discovery_port: int) -> socket.socket:
+    """Create a udp socket used for communicating with the device."""
+    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+    sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
+    sock.bind(("", 0))
+    sock.setblocking(False)
+    return sock
+
+
+class ELKDiscovery(asyncio.DatagramProtocol):
+    def __init__(
+        self,
+        destination: Tuple[str, int],
+        on_response: Callable[[bytes, Tuple[str, int]], None],
+    ) -> None:
+        self.transport = None
+        self.destination = destination
+        self.on_response = on_response
+
+    def datagram_received(self, data: bytes, addr: Tuple[str, int]) -> None:
+        """Trigger on_response."""
+        self.on_response(data, addr)
+
+    def error_received(self, ex: Optional[Exception]) -> None:
+        """Handle error."""
+        _LOGGER.error("ELKDiscovery error: %s", ex)
+
+    def connection_lost(self, ex: Optional[Exception]) -> None:
+        """Do nothing on connection lost."""
+
+
+def _decode_data(raw_response):
+    """Decode an ELK discovery response packet."""
+    remain = raw_response[5:]
+    data = remain[:14]
+    (
+        mac1,
+        mac2,
+        mac3,
+        mac4,
+        mac5,
+        mac6,
+        ipv4_1,
+        ipv4_2,
+        ipv4_3,
+        ipv4_4,
+        port,
+        _,
+    ) = unpack("!6B4BHH", data)
+    mac_address = ":".join(
+        [
+            format(val, "02x").upper().zfill(2)
+            for val in (mac1, mac2, mac3, mac4, mac5, mac6)
+        ]
+    )
+    ip_address = f"{ipv4_1}.{ipv4_2}.{ipv4_3}.{ipv4_4}"
+
+    return ElkSystem(mac_address, ip_address, port)
+
+
+class AIOELKDiscovery:
+    """A 30303 discovery scanner."""
+
+    DISCOVERY_PORT = 2362
+    BROADCAST_FREQUENCY = 3
+    DISCOVER_MESSAGE = b"XEPID"
+    BROADCAST_ADDRESS = "<broadcast>"
+
+    def __init__(self) -> None:
+        self.found_devices: List[ElkSystem] = []
+
+    def _destination_from_address(self, address: Optional[str]) -> Tuple[str, int]:
+        if address is None:
+            address = self.BROADCAST_ADDRESS
+        return (address, self.DISCOVERY_PORT)
+
+    def _process_response(
+        self,
+        data: Optional[bytes],
+        from_address: Tuple[str, int],
+        address: Optional[str],
+        response_list: Dict[str, ELKDiscovery],
+    ) -> bool:
+        """Process a response.
+
+        Returns True if processing should stop
+        """
+        if (
+            data is None
+            or data == self.DISCOVER_MESSAGE
+            or not data.startswith(b"M1XEP")
+        ):
+            return
+        try:
+            response_list[from_address] = _decode_data(data)
+        except Exception as ex:  # pylint: disable=broad-except
+            _LOGGER.warning("Failed to decode response from %s: %s", from_address, ex)
+            return False
+        return from_address[0] == address
+
+    async def _async_run_scan(
+        self,
+        transport: asyncio.DatagramTransport,
+        destination: Tuple[str, int],
+        timeout: int,
+        found_all_future: "asyncio.Future[bool]",
+    ) -> None:
+        """Send the scans."""
+        _LOGGER.debug("discover: %s => %s", destination, self.DISCOVER_MESSAGE)
+        transport.sendto(self.DISCOVER_MESSAGE, destination)
+        quit_time = time.monotonic() + timeout
+        remain_time = timeout
+        while True:
+            time_out = min(remain_time, timeout / self.BROADCAST_FREQUENCY)
+            if time_out <= 0:
+                return
+            try:
+                await asyncio.wait_for(
+                    asyncio.shield(found_all_future), timeout=time_out
+                )
+            except asyncio.TimeoutError:
+                if time.monotonic() >= quit_time:
+                    return
+                # No response, send broadcast again in cast it got lost
+                _LOGGER.debug("discover: %s => %s", destination, self.DISCOVER_MESSAGE)
+                transport.sendto(self.DISCOVER_MESSAGE, destination)
+            else:
+                return  # found_all
+            remain_time = quit_time - time.monotonic()
+
+    async def async_scan(
+        self, timeout: int = 10, address: Optional[str] = None
+    ) -> List[ELKDiscovery]:
+        """Discover ELK devices."""
+        sock = create_udp_socket(self.DISCOVERY_PORT)
+        destination = self._destination_from_address(address)
+        found_all_future = asyncio.Future()
+        response_list = {}
+
+        def _on_response(data: bytes, addr: Tuple[str, int]) -> None:
+            _LOGGER.debug("discover: %s <= %s", addr, data)
+            if self._process_response(data, addr, address, response_list):
+                found_all_future.set_result(True)
+
+        transport, _ = await asyncio.get_running_loop().create_datagram_endpoint(
+            lambda: ELKDiscovery(
+                destination=destination,
+                on_response=_on_response,
+            ),
+            sock=sock,
+        )
+        try:
+            await self._async_run_scan(
+                transport, destination, timeout, found_all_future
+            )
+        finally:
+            transport.close()
+
+        self.found_devices = list(response_list.values())
+        return self.found_devices
diff --git a/examples/aioscanner.py b/examples/aioscanner.py
new file mode 100644
index 0000000..846848f
--- /dev/null
+++ b/examples/aioscanner.py
@@ -0,0 +1,15 @@
+import asyncio
+import logging
+import pprint
+
+from elkm1_lib.discovery import AIOELKDiscovery
+
+logging.basicConfig(level=logging.DEBUG)
+
+
+async def go():
+    scanner = AIOELKDiscovery()
+    pprint.pprint(await scanner.async_scan())
+
+
+asyncio.run(go())


Test patch:
diff --git a/test/test_discovery.py b/test/test_discovery.py
new file mode 100644
index 0000000..d5c5a00
--- /dev/null
+++ b/test/test_discovery.py
@@ -0,0 +1,75 @@
+import asyncio
+import contextlib
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from elkm1_lib.discovery import (
+    AIOELKDiscovery,
+    ELKDiscovery,
+    ElkSystem,
+)
+
+
+@pytest.fixture
+async def mock_discovery_aio_protocol():
+    """Fixture to mock an asyncio connection."""
+    loop = asyncio.get_running_loop()
+    future = asyncio.Future()
+
+    async def _wait_for_connection():
+        transport, protocol = await future
+        await asyncio.sleep(0)
+        await asyncio.sleep(0)
+        return transport, protocol
+
+    async def _mock_create_datagram_endpoint(func, sock=None):
+        protocol: ELKDiscovery = func()
+        transport = MagicMock()
+        protocol.connection_made(transport)
+        with contextlib.suppress(asyncio.InvalidStateError):
+            future.set_result((transport, protocol))
+        return transport, protocol
+
+    with patch.object(loop, "create_datagram_endpoint", _mock_create_datagram_endpoint):
+        yield _wait_for_connection
+
+
+@pytest.mark.asyncio
+async def test_async_scanner_specific_address(mock_discovery_aio_protocol):
+    """Test scanner with a specific address."""
+    scanner = AIOELKDiscovery()
+
+    task = asyncio.ensure_future(
+        scanner.async_scan(timeout=10, address="192.168.209.56")
+    )
+    _, protocol = await mock_discovery_aio_protocol()
+    protocol.datagram_received(
+        b"M1XEP\x00@\x9d\xb1\xad\x03\xc0\xa8\xd18\n)\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01\x00\x00\x00\x01",
+        ("192.168.209.56", 2362),
+    )
+    await task
+    assert scanner.found_devices == [
+        ElkSystem(
+            mac_address="00:40:9D:B1:AD:03", ip_address="192.168.209.56", port=2601
+        )
+    ]
+
+
+@pytest.mark.asyncio
+async def test_async_scanner_broadcast(mock_discovery_aio_protocol):
+    """Test scanner with a broadcast."""
+    scanner = AIOELKDiscovery()
+
+    task = asyncio.ensure_future(scanner.async_scan(timeout=0.01))
+    _, protocol = await mock_discovery_aio_protocol()
+    protocol.datagram_received(
+        b"M1XEP\x00@\x9d\xb1\xad\x03\xc0\xa8\xd18\n)\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x01\x00\x00\x00\x01",
+        ("192.168.209.56", 2362),
+    )
+    await task
+    assert scanner.found_devices == [
+        ElkSystem(
+            mac_address="00:40:9D:B1:AD:03", ip_address="192.168.209.56", port=2601
+        )
+    ]