Add coroutine for handling mDNS query responses.

We can now discover our device. Maybe. I'm really not sure if I'm doing
the protocol correctly, but it looks pretty good. *shrug*.
This commit is contained in:
Eli Ribble 2023-05-12 14:31:45 -07:00
parent bc79e842e7
commit 9f3de33337
2 changed files with 39 additions and 0 deletions

37
pnpdevice/discovery.py Normal file
View File

@ -0,0 +1,37 @@
import logging
import socket
from zeroconf import IPVersion
from zeroconf.asyncio import AsyncServiceInfo, AsyncZeroconf
LOGGER = logging.getLogger(__name__)
# From https://stackoverflow.com/questions/166506/finding-local-ip-addresses-using-pythons-stdlib/
def get_ip() -> str:
"Get the primary IP"
LOGGER.info("Determining IP address")
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.settimeout(0)
try:
# doesn't even have to be reachable
s.connect(('10.254.254.254', 1))
ip = s.getsockname()[0]
except Exception:
ip = '127.0.0.1'
finally:
s.close()
LOGGER.info("IP address seems to be %s", ip)
return ip
async def handle():
"Handle requests for discovery"
ip = get_ip()
info = AsyncServiceInfo(
"_http._tcp.local.",
"pnpdevice._http._tcp.local.",
addresses=[socket.inet_aton("127.0.0.1")],
port=13344,
server=ip,
)
aiozc = AsyncZeroconf(ip_version=IPVersion.V4Only)
await aiozc.async_register_service(info)

View File

@ -2,10 +2,12 @@ import argparse
import asyncio
import logging
import pnpdevice.discovery
LOGGER = logging.getLogger(__name__)
async def run():
asyncio.ensure_future(pnpdevice.discovery.handle())
while True:
await asyncio.sleep(1)
LOGGER.info("Tick.")