Merge pull request #94 from davidrapan/dev_discovery

Add Discovery tools implemented using asyncio

Author: githubDante

utils/solarman_discovery.py

@@ -0,0 +1,86 @@
+import socket
+import asyncio
+
+from typing import Optional, Union
+from argparse import ArgumentParser
+
+DISCOVERY_IP = "255.255.255.255"
+DISCOVERY_PORT = 48899
+DISCOVERY_MESSAGE = ["WIFIKIT-214028-READ".encode(), "HF-A11ASSISTHREAD".encode()]
+DISCOVERY_TIMEOUT = 1
+
+
+class DiscoveryProtocol:
+    def __init__(self, addresses: Union[list[str], str]):
+        self.addresses = addresses
+        self.responses = asyncio.Queue()
+
+    def connection_made(self, transport: asyncio.DatagramTransport):
+        print(f"DiscoveryProtocol: Send to {self.addresses}")
+        for address in (
+            self.addresses if isinstance(self.addresses, list) else [self.addresses]
+        ):
+            for message in DISCOVERY_MESSAGE:
+                transport.sendto(message, (address, DISCOVERY_PORT))
+
+    def datagram_received(self, data: bytes, addr: tuple[str, int]):
+        if len(d := data.decode().split(",")) == 3 and (s := int(d[2])):
+            self.responses.put_nowait((s, {"ip": d[0], "mac": d[1]}))
+            print(f"DiscoveryProtocol: [{d[0]}, {d[1]}, {s}] from {addr}")
+
+    def error_received(self, e: OSError):
+        print(f"DiscoveryProtocol: {e!r}")
+
+    def connection_lost(self, _: Optional[Exception]):
+        print("DiscoveryProtocol: Connection closed")
+
+
+async def main():
+    parser = ArgumentParser(
+        "solarman-discovery", description="Discovery for Solarman Stick Loggers"
+    )
+    parser.add_argument(
+        "--address",
+        default=DISCOVERY_IP,
+        required=False,
+        type=str,
+        help="Network IPv4 address",
+    )
+    parser.add_argument(
+        "--timeout",
+        default=DISCOVERY_TIMEOUT,
+        required=False,
+        type=int,
+        choices=range(10),
+        help="Timeout in seconds, an integer in the range 0..9",
+    )
+    parser.add_argument(
+        "--wait",
+        default=True,
+        required=False,
+        type=bool,
+        help="Wait for multiple responses",
+    )
+    args = parser.parse_args()
+
+    try:
+        transport, protocol = await asyncio.get_running_loop().create_datagram_endpoint(
+            lambda: DiscoveryProtocol(args.address),
+            family=socket.AF_INET,
+            allow_broadcast=True,
+        )
+        while (
+            await asyncio.wait_for(protocol.responses.get(), args.timeout) is None
+            or args.wait
+        ):
+            pass
+    except TimeoutError:
+        pass
+    except Exception as e:
+        print(repr(e))
+    finally:
+        transport.close()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

utils/solarman_discovery_reply.py

@@ -0,0 +1,62 @@
+import socket
+import asyncio
+import netifaces
+
+from typing import Optional
+from argparse import ArgumentParser
+
+DISCOVERY_IP = "0.0.0.0"
+DISCOVERY_PORT = 48899
+DISCOVERY_MESSAGE = ["WIFIKIT-214028-READ".encode(), "HF-A11ASSISTHREAD".encode()]
+DISCOVERY_TIMEOUT = 3600
+
+ifaces = netifaces.ifaddresses(netifaces.gateways()["default"][2][1])
+iface_inet = ifaces[netifaces.AF_INET][0]["addr"]
+iface_link = ifaces[netifaces.AF_LINK][0]["addr"].replace(":", "").upper()
+
+DISCOVERY_MESSAGE_REPLY = f"{iface_inet},{iface_link},1234567890".encode()
+
+
+class DiscoveryProtocol:
+    def connection_made(self, transport: asyncio.DatagramTransport):
+        self.transport = transport
+
+    def datagram_received(self, data: bytes, addr: tuple[str, int]):
+        if data in DISCOVERY_MESSAGE:
+            print(f"DiscoveryProtocol: {data} from {addr}")
+            self.transport.sendto(DISCOVERY_MESSAGE_REPLY, addr)
+
+    def error_received(self, e: OSError):
+        print(f"DiscoveryProtocol: {e!r}")
+
+    def connection_lost(self, _: Optional[Exception]):
+        print("DiscoveryProtocol: Connection closed")
+
+
+async def main():
+    parser = ArgumentParser(
+        "solarman-discovery-reply", description="Discovery for Solarman Stick Loggers"
+    )
+    parser.add_argument(
+        "--timeout",
+        default=DISCOVERY_TIMEOUT,
+        required=False,
+        type=int,
+        choices=range(3600),
+        help="Timeout in seconds, an integer in the range 0..3600",
+    )
+
+    try:
+        transport, _ = await asyncio.get_running_loop().create_datagram_endpoint(
+            DiscoveryProtocol,
+            local_addr=(DISCOVERY_IP, DISCOVERY_PORT),
+            family=socket.AF_INET,
+            allow_broadcast=True,
+        )
+        await asyncio.sleep(parser.parse_args().timeout)
+    finally:
+        transport.close()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())