56 lines
1.5 KiB
Python
56 lines
1.5 KiB
Python
import asyncio
|
|
from async_upnp_client.search import async_search
|
|
import json
|
|
from urllib.parse import urlparse
|
|
from .flickerstrip import Flickerstrip
|
|
|
|
|
|
class FlickerstripDiscoveryClient:
|
|
def __init__(self, max_attempts=3):
|
|
self.max_attempts = max_attempts
|
|
|
|
async def try_discover(self):
|
|
strips: list[Flickerstrip] = []
|
|
|
|
async def on_response(headers) -> None:
|
|
if "Flickerstrip" in headers["SERVER"]:
|
|
res = urlparse(headers["LOCATION"])
|
|
|
|
strips.append(Flickerstrip(
|
|
host=res.hostname,
|
|
port=res.port,
|
|
))
|
|
|
|
await async_search(
|
|
async_callback=on_response
|
|
)
|
|
|
|
print(f"Discovered {len(strips)} flickerstrip(s).")
|
|
if len(strips) > 0:
|
|
return strips
|
|
return None
|
|
|
|
async def discover(self):
|
|
for i in range(self.max_attempts):
|
|
print(f"Discovering devices... (Attempt {i + 1})")
|
|
if (devices := await self.try_discover()) is not None:
|
|
return devices
|
|
return []
|
|
|
|
|
|
async def main():
|
|
client = FlickerstripDiscoveryClient(max_attempts=10)
|
|
flickerstrips = await client.discover()
|
|
if len(flickerstrips) > 0:
|
|
await flickerstrips[0].force_update()
|
|
if flickerstrips[0].status is not None:
|
|
print(json.dumps(flickerstrips[0].status.to_json()))
|
|
|
|
if __name__ == "__main__":
|
|
loop = asyncio.get_event_loop()
|
|
|
|
try:
|
|
loop.run_until_complete(main())
|
|
finally:
|
|
loop.close()
|