alexei.dolgolyov
638dc526f9
Scan the local network for WLED devices advertising _wled._tcp.local. and present them in the Add Device modal for one-click selection. - New discovery.py: async mDNS browse + parallel /json/info enrichment - GET /api/v1/devices/discover endpoint with already_added dedup - Header scan button (magnifying glass icon) in add-device modal - Discovered devices show name, IP, LED count, version; click to fill form - en/ru locale strings for discovery UI Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
114 lines
3.4 KiB
Python
114 lines
3.4 KiB
Python
"""Network discovery for LED devices (mDNS/DNS-SD)."""
|
|
|
|
import asyncio
|
|
from dataclasses import dataclass
|
|
from typing import List, Optional
|
|
|
|
import httpx
|
|
from zeroconf import ServiceStateChange, Zeroconf
|
|
from zeroconf.asyncio import AsyncServiceBrowser, AsyncServiceInfo, AsyncZeroconf
|
|
|
|
from wled_controller.utils import get_logger
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
WLED_MDNS_TYPE = "_wled._tcp.local."
|
|
DEFAULT_SCAN_TIMEOUT = 3.0
|
|
|
|
|
|
@dataclass
|
|
class DiscoveredDevice:
|
|
"""A device found via network discovery."""
|
|
|
|
name: str
|
|
url: str
|
|
device_type: str
|
|
ip: str
|
|
mac: str
|
|
led_count: Optional[int]
|
|
version: Optional[str]
|
|
|
|
|
|
async def _enrich_wled_device(
|
|
url: str, fallback_name: str
|
|
) -> tuple[str, Optional[str], Optional[int], str]:
|
|
"""Probe a WLED device's /json/info to get name, version, LED count, MAC."""
|
|
try:
|
|
async with httpx.AsyncClient(timeout=2) as client:
|
|
resp = await client.get(f"{url}/json/info")
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
return (
|
|
data.get("name", fallback_name),
|
|
data.get("ver"),
|
|
data.get("leds", {}).get("count"),
|
|
data.get("mac", ""),
|
|
)
|
|
except Exception as e:
|
|
logger.debug(f"Could not fetch WLED info from {url}: {e}")
|
|
return fallback_name, None, None, ""
|
|
|
|
|
|
async def discover_wled_devices(
|
|
timeout: float = DEFAULT_SCAN_TIMEOUT,
|
|
) -> List[DiscoveredDevice]:
|
|
"""Scan the local network for WLED devices via mDNS."""
|
|
discovered: dict[str, AsyncServiceInfo] = {}
|
|
|
|
def on_state_change(**kwargs):
|
|
service_type = kwargs.get("service_type", "")
|
|
name = kwargs.get("name", "")
|
|
state_change = kwargs.get("state_change")
|
|
if state_change in (ServiceStateChange.Added, ServiceStateChange.Updated):
|
|
discovered[name] = AsyncServiceInfo(service_type, name)
|
|
|
|
aiozc = AsyncZeroconf()
|
|
browser = AsyncServiceBrowser(
|
|
aiozc.zeroconf, WLED_MDNS_TYPE, handlers=[on_state_change]
|
|
)
|
|
|
|
await asyncio.sleep(timeout)
|
|
|
|
# Resolve all discovered services
|
|
for info in discovered.values():
|
|
await info.async_request(aiozc.zeroconf, timeout=2000)
|
|
|
|
await browser.async_cancel()
|
|
|
|
# Build raw list with IPs, then enrich in parallel
|
|
raw: list[tuple[str, str, str]] = [] # (service_name, ip, url)
|
|
for name, info in discovered.items():
|
|
addrs = info.parsed_addresses()
|
|
if not addrs:
|
|
continue
|
|
ip = addrs[0]
|
|
port = info.port or 80
|
|
url = f"http://{ip}" if port == 80 else f"http://{ip}:{port}"
|
|
service_name = name.replace(f".{WLED_MDNS_TYPE}", "")
|
|
raw.append((service_name, ip, url))
|
|
|
|
# Enrich all devices in parallel
|
|
enrichment = await asyncio.gather(
|
|
*[_enrich_wled_device(url, sname) for sname, _, url in raw]
|
|
)
|
|
|
|
results: List[DiscoveredDevice] = []
|
|
for (service_name, ip, url), (wled_name, version, led_count, mac) in zip(
|
|
raw, enrichment
|
|
):
|
|
results.append(
|
|
DiscoveredDevice(
|
|
name=wled_name,
|
|
url=url,
|
|
device_type="wled",
|
|
ip=ip,
|
|
mac=mac,
|
|
led_count=led_count,
|
|
version=version,
|
|
)
|
|
)
|
|
|
|
await aiozc.async_close()
|
|
logger.info(f"mDNS scan found {len(results)} WLED device(s)")
|
|
return results
|