85 lines
2.8 KiB
Python
85 lines
2.8 KiB
Python
#!/usr/bin/env python3
|
|
"""搜索 V1 工厂协议设备（UDP 4942）。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import ipaddress
|
|
import json
|
|
import secrets
|
|
import socket
|
|
import struct
|
|
import time
|
|
from typing import Any
|
|
|
|
PORT = 4942
|
|
REQUEST = 0x5A0EFF00
|
|
RESPONSE = 0x5A0EFF01
|
|
MAX_DATA = 1024 * 1024
|
|
|
|
|
|
def frame(command: int, payload: bytes) -> bytes:
|
|
return struct.pack("!II", command, len(payload)) + payload
|
|
|
|
|
|
def parse_frame(packet: bytes) -> tuple[int, bytes]:
|
|
if len(packet) < 8:
|
|
raise ValueError("packet shorter than V1 header")
|
|
command, length = struct.unpack("!II", packet[:8])
|
|
if length > MAX_DATA or len(packet) != 8 + length:
|
|
raise ValueError("invalid V1 payload length")
|
|
return command, packet[8:]
|
|
|
|
|
|
def discover(timeout: float, interface: str | None) -> int:
|
|
session = secrets.randbelow(0x7FFFFFFF) + 1
|
|
request = frame(REQUEST, json.dumps({"session": session}).encode("utf-8"))
|
|
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
|
sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
|
|
sock.settimeout(min(0.5, timeout))
|
|
try:
|
|
# Bind the documented client port. If another discovery utility owns
|
|
# it, stop and report the conflict instead of silently using another port.
|
|
sock.bind((interface or "0.0.0.0", PORT))
|
|
sock.sendto(request, ("255.255.255.255", PORT))
|
|
deadline = time.monotonic() + timeout
|
|
seen: set[tuple[str, str]] = set()
|
|
while time.monotonic() < deadline:
|
|
try:
|
|
packet, address = sock.recvfrom(MAX_DATA + 8)
|
|
except socket.timeout:
|
|
continue
|
|
try:
|
|
command, data = parse_frame(packet)
|
|
obj: Any = json.loads(data.decode("utf-8"))
|
|
if command != RESPONSE or not isinstance(obj, dict):
|
|
continue
|
|
if obj.get("session") != session:
|
|
continue
|
|
key = (address[0], str(obj.get("sn", "")))
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
print(json.dumps({"source_ip": address[0], "response": obj}, ensure_ascii=False))
|
|
except (ValueError, UnicodeDecodeError, json.JSONDecodeError):
|
|
continue
|
|
return 0
|
|
finally:
|
|
sock.close()
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--timeout", type=float, default=3.0)
|
|
parser.add_argument("--interface", help="optional local IPv4 address to bind")
|
|
args = parser.parse_args()
|
|
if args.timeout <= 0:
|
|
parser.error("--timeout must be positive")
|
|
if args.interface:
|
|
ipaddress.IPv4Address(args.interface)
|
|
return discover(args.timeout, args.interface)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|