"""Discover Topbox / EtherGold feed hosts on the local LAN for OHD setup.""" from __future__ import annotations import concurrent.futures import ipaddress import json import socket import urllib.error import urllib.request from typing import Any, Dict, List, Optional, Sequence, Tuple DEFAULT_WEB_PORT = 31234 INFO_PATH = "/Topbox/info.json" PROGRESSIVES_PATH = "/EtherGold/progressives.csv" HEALTH_PATH = "/health" SCAN_CONNECT_TIMEOUT_S = 0.55 MAX_SCAN_WORKERS = 64 def _hostname_starts_with_topbox(hostname: str) -> bool: host = (hostname or "").strip().lower() if host.endswith(".local"): host = host[: -len(".local")] return host.startswith("topbox") def is_topbox_master(device: Dict[str, Any]) -> bool: role = str(device.get("machine_role") or "").strip().upper() if role == "M": return True host = str(device.get("hostname") or "").strip().lower() if host.endswith(".local"): host = host[: -len(".local")] return host.startswith("topbox") and host.endswith("m") def sort_topbox_devices(devices: Sequence[Dict[str, Any]]) -> List[Dict[str, Any]]: return sorted( devices, key=lambda d: ( 0 if str(d.get("source_kind") or "") in ("progressives_csv", "ethergold_health") else 1, 0 if is_topbox_master(d) else 1, str(d.get("hostname") or "").lower(), ), ) def filter_topbox_devices(devices: Sequence[Dict[str, Any]]) -> List[Dict[str, Any]]: """Keep feed sources and Topbox-named DF hosts (drop random noise).""" filtered: List[Dict[str, Any]] = [] for d in devices: kind = str(d.get("source_kind") or "") if kind in ("progressives_csv", "ethergold_health"): filtered.append(d) continue host = str(d.get("hostname") or "") mode = str(d.get("display_mode") or "").strip().upper() if _hostname_starts_with_topbox(host) or mode in ("TBX", "OHD", "MINI", "STRIP"): filtered.append(d) return sort_topbox_devices(filtered) def local_ipv4_addresses() -> List[str]: addrs: List[str] = [] try: probe = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) probe.connect(("8.8.8.8", 80)) addrs.append(probe.getsockname()[0]) probe.close() except OSError: pass try: for info in socket.getaddrinfo(socket.gethostname(), None, socket.AF_INET, socket.SOCK_DGRAM): ip = info[4][0] if not ip.startswith("127."): addrs.append(ip) except OSError: pass unique: List[str] = [] for ip in addrs: if ip not in unique: unique.append(ip) return unique def iter_subnet_hosts(ip_str: str, *, prefix: int = 24) -> List[str]: network = ipaddress.ip_network(f"{ip_str}/{prefix}", strict=False) return [str(host) for host in network.hosts()] def _resolve_hostname(ip: str) -> str: try: host, _, _ = socket.gethostbyaddr(ip) except OSError: return ip return host.split(".")[0] def _http_get(url: str, *, timeout: float) -> Optional[bytes]: try: req = urllib.request.Request(url, headers={"Accept": "*/*"}) with urllib.request.urlopen(req, timeout=timeout) as resp: if getattr(resp, "status", 200) >= 400: return None return resp.read(4096) except (urllib.error.URLError, TimeoutError, OSError): return None def _fetch_topbox_info(ip: str, port: int = DEFAULT_WEB_PORT) -> Optional[Dict[str, Any]]: raw = _http_get(f"http://{ip}:{port}{INFO_PATH}", timeout=SCAN_CONNECT_TIMEOUT_S) if not raw: return None try: payload = json.loads(raw.decode("utf-8", errors="replace")) except json.JSONDecodeError: return None if not isinstance(payload, dict): return None hostname = str(payload.get("hostname") or _resolve_hostname(ip)).strip() payload["ip"] = ip payload["port"] = port payload["hostname"] = hostname payload["progressives_url"] = f"http://{ip}:{port}{PROGRESSIVES_PATH}" payload["source_kind"] = "topbox_info" return payload def _fetch_ethergold_feed(ip: str, port: int = DEFAULT_WEB_PORT) -> Optional[Dict[str, Any]]: """Listen / Topbox EtherGold HTTP that serves progressives (no /Topbox/info.json).""" health_raw = _http_get(f"http://{ip}:{port}{HEALTH_PATH}", timeout=SCAN_CONNECT_TIMEOUT_S) health: Dict[str, Any] = {} if health_raw: try: parsed = json.loads(health_raw.decode("utf-8", errors="replace")) if isinstance(parsed, dict): health = parsed except json.JSONDecodeError: health = {} product = str(health.get("product") or "").strip().lower() # Prefer health product=ethergold; else confirm progressives.csv answers. if product and product != "ethergold": return None prog_url = f"http://{ip}:{port}{PROGRESSIVES_PATH}" if product != "ethergold": prog = _http_get(prog_url, timeout=SCAN_CONNECT_TIMEOUT_S) if prog is None: return None hostname = _resolve_hostname(ip) version = str(health.get("version") or health.get("http_build") or "") return { "ip": ip, "port": port, "hostname": hostname, "machine_role": "", "bank_number": 0, "term_number": 0, "asset_number": None, "display_mode": "FEED", "version": version, "progressives_url": prog_url, "source_kind": "ethergold_health" if product == "ethergold" else "progressives_csv", } def probe_feed_host(ip: str, port: int = DEFAULT_WEB_PORT) -> Optional[Dict[str, Any]]: """One host: Topbox info.json first, else EtherGold progressive feed.""" info = _fetch_topbox_info(ip, port) if info is not None: return info return _fetch_ethergold_feed(ip, port) def scan_local_topbox_devices( *, port: int = DEFAULT_WEB_PORT, prefix: int = 24, max_workers: int = MAX_SCAN_WORKERS, ) -> Tuple[List[Dict[str, Any]], List[str]]: """Scan local /24 subnets for progressive feed hosts (and Topbox info).""" scanned_subnets: List[str] = [] candidate_ips: List[str] = [] for local_ip in local_ipv4_addresses(): subnet = str(ipaddress.ip_network(f"{local_ip}/{prefix}", strict=False)) scanned_subnets.append(subnet) for host in iter_subnet_hosts(local_ip, prefix=prefix): if host not in candidate_ips: candidate_ips.append(host) found: List[Dict[str, Any]] = [] workers = min(max_workers, max(1, len(candidate_ips))) with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as pool: futures = {pool.submit(probe_feed_host, ip, port): ip for ip in candidate_ips} for future in concurrent.futures.as_completed(futures): info = future.result() if info is not None: found.append(info) return filter_topbox_devices(found), scanned_subnets