46 lines
1.4 KiB
Python
46 lines
1.4 KiB
Python
import asyncio
|
|
import logging
|
|
from urllib.parse import urlparse
|
|
|
|
from async_upnp_client.search import async_search
|
|
from async_upnp_client.utils import CaseInsensitiveDict
|
|
|
|
|
|
class FlickerstripDiscoveryClient:
|
|
def __init__(self, max_attempts=3):
|
|
self.max_attempts = max_attempts
|
|
|
|
async def try_discover(self) -> list[str]:
|
|
strips: list[str] = []
|
|
|
|
async def on_response(headers: CaseInsensitiveDict) -> None:
|
|
if "Flickerstrip" in headers["SERVER"]:
|
|
location = headers["LOCATION"]
|
|
res = urlparse(location)
|
|
|
|
if res.hostname is None:
|
|
logging.warning(f"Invalid location header: {location}")
|
|
return
|
|
|
|
strips.append(res.hostname)
|
|
|
|
await async_search(async_callback=on_response)
|
|
|
|
logging.debug(f"Discovered {len(strips)} flickerstrip(s).")
|
|
return strips
|
|
|
|
async def discover(self):
|
|
for i in range(self.max_attempts):
|
|
logging.debug(f"Discovering devices... (Attempt {i + 1})")
|
|
devices = await self.try_discover()
|
|
|
|
if len(devices) > 0:
|
|
return devices
|
|
|
|
if i == self.max_attempts - 1:
|
|
logging.warning("No devices found")
|
|
else:
|
|
logging.debug("No devices found, retrying in 5 seconds")
|
|
await asyncio.sleep(5)
|
|
return []
|