"""
src/core/ptz_controller.py - PTZ/设备控制管理器 (M1 迁移)
SADP发现 / ISAPI通信 / PTZ控制 / 运动控制
Author: 雅痞张@南方天文
"""
from __future__ import annotations
import contextlib
import json
import platform
import socket
import struct
import subprocess
import threading
import time
import uuid
import xml.etree.ElementTree as ET
from datetime import UTC, datetime
from pathlib import Path
from src.core.net_detector import suggest_target_ip
from src.ptz.isapi.client import ISAPIClient
from src.ptz.isapi.ptz import ISAPIPTZController
try:
import requests # noqa: F401
from requests.auth import HTTPDigestAuth # noqa: F401
except ImportError:
pass
# ============================================================
# Constants (from M1)
# ============================================================
SADP_MULTICAST_ADDR = "239.255.255.250"
SADP_PORT = 37020
SADP_TIMEOUT_MS = 3000
ISAPI_CHANNEL = 1
DEFAULT_PTZ_PRESET = 10
HOME_COORDS = {"pan": 1800, "tilt": 450, "zoom": 10}
PTZ_MAX_SPEED = 100
PTZ_MIN_SPEED = 1
STABILIZATION_SECONDS = 2
STABLE_POINTS_REQUIRED = 20
STABLE_POINT_DEVIATION = 0
SAMPLE_INTERVAL = 0.1
DEFAULT_USERNAME = "admin"
HIKVISION_MAC_OUI = {
"28:57:BE", "4C:BD:8F", "54:C4:15", "C0:56:E3", "E0:50:8B",
}
def _ping_reachable(ip: str, timeout: float = 3) -> bool:
"""通过 PING 检测 IP 可达性。支持浮点超时（如 0.2 = 200ms）。"""
system = platform.system()
timeout_ms = max(1, int(timeout * 1000))
cmd = f"ping -n 1 -w {timeout_ms} {ip}" if system == "Windows" else f"ping -c 1 -W {max(1, int(timeout))} {ip}"
try:
result = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=timeout + 5,
creationflags=subprocess.CREATE_NO_WINDOW if hasattr(subprocess, 'CREATE_NO_WINDOW') else 0)
# Windows: 不通的IP也会返回自身IP的回复(returncode=0)，需要检查输出
if system == "Windows":
output = result.stdout.lower()
# 这些关键字说明 ping 实际不通
unreachable_markers = ['unreachable', '无法访问', 'destination host', 'ttl expired', 'general failure', 'transmit failed']
for marker in unreachable_markers:
if marker in output:
return False
# 有正常回复且有 TTL = 真正可达
return result.returncode == 0 and 'ttl=' in output
return result.returncode == 0
except Exception:
return False
def check_reachable(ip: str, timeout: float = 3, port: int = 80) -> bool:
"""v1.1.25: ICMP 优先 + TCP:port 回退探测（默认 80 = ISAPI 端口）。
生产 VLAN 可能过滤/限速 ICMP、冷启动 ARP 未暖，导致 ping 全败而
HTTP 实际可达——"设备在线却被判离线"误判的根因放大器；TCP 回退
与手动连接（TCP:80 HTTP Digest）同判定面。
"""
if _ping_reachable(ip, timeout):
return True
import socket
try:
with socket.create_connection((ip, port), timeout=max(0.2, min(timeout, 1.0))):
return True
except Exception:
return False
# ============================================================
# Logger — 复用 src/ptz/core/logger.py (PTZLogger)
# ============================================================
from src.ptz.core.logger import LOG # noqa: E402
# ============================================================
# ConfigManager
# ============================================================
def _normalize_mac(mac: str) -> str:
"""统一 MAC 格式：无分隔符小写（委托 device_path.normalize_mac）。"""
from src.advanced.device_path import normalize_mac
return normalize_mac(mac)
def _atomic_write(path: Path, data: dict) -> None:
"""安全写入 JSON 文件，保证原子性（委托 device_path.atomic_write_json）。"""
from src.advanced.device_path import atomic_write_json
atomic_write_json(path, data)
class ConfigManager:
"""本地配置 + 设备配置管理器。
v6.10: 统一使用 data/devices/{mac}/ 目录。
"""
def __init__(self, local_path: Path, ptz_path: Path) -> None:
self.local_path = local_path.resolve()
self.ptz_path = ptz_path.resolve()
from src.config_paths import DEVICES_DIR, REGISTRY_FILE
self._devices_dir = DEVICES_DIR
self._registry_file = REGISTRY_FILE
# v11.20: 设备配置内存缓存（mtime 校验，写时失效），
# 消除预置点/设备列表路径的每次 rglob 全目录扫描
self._config_cache: dict | None = None
self._cache_mtimes: dict[Path, float] = {}
LOG.log("info", f"初始化 ConfigManager: devices_dir={self._devices_dir}")
self.create_defaults()
def create_defaults(self) -> None:
self._devices_dir.mkdir(parents=True, exist_ok=True)
if not self.local_path.exists():
_atomic_write(self.local_path, {
"hostname": "", "cpu_model": "", "ram_gb": 0, "gpu_count": 0,
"vram_gb": 0, "gpu_names": [],
"selected_nic": {"name": "", "ip": "", "netmask": "", "gateway": ""},
})
LOG.log("done", f"创建本地配置文件: {self.local_path}")
def load_local(self) -> dict:
with open(self.local_path, encoding="utf-8") as f:
return json.load(f)
def save_local(self, data: dict) -> None:
_atomic_write(self.local_path, data)
def load_ptz_config(self) -> dict:
"""加载设备配置。v6.10: 只读取 devices/{mac}/info.json。
v11.20: 内存缓存 + mtime 校验，文件未变时直接返回缓存。
"""
if self._config_cache is not None and self._cache_still_valid():
return self._config_cache
devices = {}
mtimes: dict[Path, float] = {}
if self._devices_dir.exists():
for info_file in self._devices_dir.rglob("info.json"):
try:
mtimes[info_file] = info_file.stat().st_mtime
with open(info_file, encoding="utf-8") as f:
info = json.load(f)
mac = info.get("mac", "")
if mac:
norm_mac = _normalize_mac(mac)
devices[norm_mac] = info
except Exception:
continue
self._config_cache = {"devices": devices}
self._cache_mtimes = mtimes
return self._config_cache
def _cache_still_valid(self) -> bool:
"""校验缓存：文件集合与 mtime 均未变化。"""
try:
if not self._devices_dir.exists():
return not self._cache_mtimes
current = {p: p.stat().st_mtime for p in self._devices_dir.rglob("info.json")}
return current == self._cache_mtimes
except OSError:
return False
def _invalidate_config_cache(self) -> None:
self._config_cache = None
self._cache_mtimes = {}
def save_ptz_config(self, data: dict) -> None:
"""保存设备配置。v10.40: 只写入 devices/{mac}/info.json，不再写 registry.json。"""
devices = data.get("devices", {})
for mac, info in devices.items():
norm_mac = _normalize_mac(mac)
device_dir = self._devices_dir / norm_mac
device_dir.mkdir(parents=True, exist_ok=True)
info_path = device_dir / "info.json"
info["mac"] = norm_mac
info["last_updated"] = datetime.now(UTC).replace(tzinfo=None).isoformat()
_atomic_write(info_path, info)
self._invalidate_config_cache()
def get_device_by_mac(self, mac: str) -> dict | None:
"""v6.10: 读取 devices/{mac}/info.json。"""
norm_mac = _normalize_mac(mac)
info_path = self._devices_dir / norm_mac / "info.json"
if info_path.exists():
try:
with open(info_path, encoding="utf-8") as f:
return json.load(f)
except Exception:
pass
return None
def upsert_device(self, mac: str, info: dict) -> None:
"""v10.40: 写入 devices/{mac}/info.json，不再写 registry.json。
registry.json 由 DeviceManager 统一管理。
v1.1.35: 改 existing 为基合并（修复调用方 dict 整体重写冲掉凭据的
生产事故：SADP 发现 upsert/网关 backfill/{"mac":mac} 触碰任一次
都会丢 password）：调用方未传字段保留原值；保护列空值不覆盖。
"""
norm_mac = _normalize_mac(mac)
device_dir = self._devices_dir / norm_mac
device_dir.mkdir(parents=True, exist_ok=True)
info_path = device_dir / "info.json"
# 如果设备已存在，以已有内容为合并基
existing: dict = {}
if info_path.exists():
try:
with open(info_path, encoding="utf-8") as f:
existing = json.load(f)
except Exception:
pass
# v1.1.35: existing 为基 + 调用方覆盖（未传字段不再丢失）
merged = dict(existing)
merged.update(info)
# v10.41/v1.1.35: 保护列空值不覆盖已有非空（扩展到凭据/编码列）
protected = (
"gateway", "subnet_mask", "model", "name", "serial_number",
"firmware_version", "username", "password", "port", "video_codec",
)
for key in protected:
if not merged.get(key) and existing.get(key):
merged[key] = existing[key]
for merge_key in ("capabilities", "limits", "speed", "home"):
if merge_key in existing and merge_key not in info:
merged[merge_key] = existing[merge_key]
elif merge_key in existing and merge_key in info:
merged_dict = dict(existing[merge_key])
if isinstance(info[merge_key], dict):
merged_dict.update(info[merge_key])
else:
merged_dict = info[merge_key]
merged[merge_key] = merged_dict
merged["mac"] = norm_mac
merged["last_updated"] = datetime.now(UTC).replace(tzinfo=None).isoformat()
if not merged.get("first_seen"):
merged["first_seen"] = merged["last_updated"]
_atomic_write(info_path, merged)
self._invalidate_config_cache()
def remove_device(self, mac: str) -> bool:
"""v10.40: 删除 devices/{mac}/ 目录。registry.json 由 DeviceManager 管理。"""
norm_mac = _normalize_mac(mac)
device_dir = self._devices_dir / norm_mac
if device_dir.exists():
import shutil
shutil.rmtree(device_dir)
self._invalidate_config_cache()
return True
# ============================================================
# SADP Discovery (官方 DLL 封装优先，纯 Python 多播为备用方案)
# ============================================================
SADP_DLL_AVAILABLE = False
try:
from src.core.sadp_discovery import SADPManager as SADPManagerDLL
SADP_DLL_AVAILABLE = True
except ImportError as e:
LOG.log("warning", f"Sadp.dll 封装不可用: {e}，使用纯 Python 多播备用")
# v11.20: 统一使用 sadp_discovery 的 SADPDevice（完整版，含端口字段），消除重复定义
from src.core.sadp_discovery import SADPDevice
def _is_hikvision_mac(mac: str) -> bool:
normalized = mac.replace("-", ":").upper()
return normalized[:8] in HIKVISION_MAC_OUI
def _parse_sadp_response(data: bytes) -> SADPDevice | None:
"""解析 SADP 响应 XML。
SADP 响应格式可能是：
1. 纯 XML: ...
2. Binary header + XML: 4或8字节 header + XML
- v2: \x00\x00\x00\x00 或 \xFF\xFF\xFF\xFF + 4字节大端长度 + XML
- v3: "usdp" + 4字节大端长度 + XML
"""
try:
text = None
xml_str = None
# ---- 策略1：尝试直接作为 XML ----
raw_text = data.decode("utf-8", errors="ignore")
if "" in raw_text or "" in raw_text or " 8:
# 检查是否有 known magic bytes
has_magic = False
if data[:4] == b"usdp" or data[:4] in (b"\x00\x00\x00\x00", b"\xff\xff\xff\xff") or data[:4] in (b"\x00\x00\x00\x01",):
has_magic = True
if has_magic:
# 尝试从 binary data 中提取 XML
# 跳过 header (4或8字节)，找 XML
for start_offset in [4, 8, 10, 12]:
if start_offset < len(data):
chunk = data[start_offset:].decode("utf-8", errors="ignore")
if "")
if xml_start < 0:
xml_start = text.find("")
if xml_start < 0:
xml_start = text.find(" str:
elem = root.find(tag)
if elem is not None and elem.text:
return elem.text.strip()
elem = root.find(f".//{{*}}{tag}")
if elem is not None and elem.text:
return elem.text.strip()
for prefix in ("ipv4", "IPv4", "ip", "IP"):
elem = root.find(f".//{{*}}{prefix}{tag}")
if elem is not None and elem.text:
return elem.text.strip()
elem = root.find(f"{prefix}{tag}")
if elem is not None and elem.text:
return elem.text.strip()
return default
def find_text_ns(tag: str, default: str = "") -> str:
"""在子树中递归查找（处理嵌套命名空间）。"""
for elem in root.iter():
local_name = elem.tag.split("}")[-1] if "}" in elem.tag else elem.tag
if local_name == tag and elem.text:
return elem.text.strip()
return default
# 提取字段
device.mac = (
find_text("MACAddress") or find_text_ns("MACAddress")
or find_text("macAddress") or find_text_ns("macAddress") or ""
)
device.ip = (
find_text("IPv4Address") or find_text_ns("IPv4Address")
or find_text("ipv4Address") or find_text_ns("ipv4Address")
or find_text("IPAddress") or find_text_ns("IPAddress") or ""
)
device.subnet_mask = (
find_text("IPv4SubnetMask") or find_text_ns("IPv4SubnetMask")
or find_text("ipv4SubnetMask") or find_text_ns("ipv4SubnetMask") or ""
)
device.gateway = (
find_text("IPv4Gateway") or find_text_ns("IPv4Gateway")
or find_text("ipv4Gateway") or find_text_ns("ipv4Gateway") or ""
)
device.model = (
find_text("deviceType") or find_text_ns("deviceType")
or find_text("model") or find_text_ns("model")
or find_text("DeviceType") or ""
)
device.serial_number = (
find_text("serialNumber") or find_text_ns("serialNumber")
or find_text("SerialNumber") or ""
)
device.device_name = (
find_text("deviceName") or find_text_ns("deviceName")
or find_text("DeviceName") or ""
)
device.firmware_version = (
find_text("firmwareVersion") or find_text_ns("firmwareVersion")
or find_text("FirmwareVersion") or ""
)
activated_str = (find_text("activated") or find_text_ns("activated") or "").lower()
if activated_str in ("true", "1", "yes"):
device.activated = True
elif activated_str in ("false", "0", "no"):
device.activated = False
else:
device.activated = bool(device.ip and device.ip != "0.0.0.0")
device.is_hikvision = _is_hikvision_mac(device.mac)
if device.mac:
LOG.log("info", f"SADP 解析完成: MAC={device.mac} IP={device.ip} 型号={device.model}")
return device
except ET.ParseError as e:
LOG.log("warning", f"SADP XML 解析失败: {e}")
return None
except Exception as e:
import traceback
LOG.log("warning", f"SADP 响应解析异常: {e}")
LOG.log("warning", f"解析堆栈: {traceback.format_exc()}")
return None
def _get_local_ip() -> str:
"""获取本机网卡的真实 IP 地址（用于 SADP 多播）。"""
import socket
try:
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.connect(("8.8.8.8", 80))
local_ip = s.getsockname()[0]
s.close()
return local_ip
except Exception:
return "127.0.0.1"
def scan_for_devices(bind_ip: str = "0.0.0.0") -> list[SADPDevice]:
"""扫描 SADP 设备。
优先级：
1. 官方 Sadp.dll (如果可用)
2. 纯 Python 多播 (备用方案)
"""
if SADP_DLL_AVAILABLE:
LOG.log("info", "使用官方 Sadp.dll 进行设备扫描")
return _scan_via_sadp_dll()
LOG.log("warning", "Sadp.dll 不可用，使用纯 Python 多播备用方案")
return _scan_via_multicast(bind_ip)
def _scan_via_sadp_dll() -> list[SADPDevice]:
"""通过官方 Sadp.dll 进行设备发现."""
try:
mgr = SADPManagerDLL()
devices_raw = mgr.discover_devices(timeout=6)
result = []
for d in devices_raw:
device = SADPDevice()
device.mac = d.get("mac", "").replace("-", ":")
device.ip = d.get("ip", "")
device.subnet_mask = d.get("subnet_mask", "")
device.gateway = d.get("gateway", "")
device.model = d.get("model", "")
device.serial_number = d.get("serial_number", "")
device.device_name = d.get("device_name", "")
device.firmware_version = d.get("firmware_version", "")
device.activated = d.get("activated", False)
device.is_hikvision = d.get("is_hikvision", False)
if device.mac:
result.append(device)
LOG.log("done", f"SADP DLL 发现设备: {device.display_name()} IP={device.ip}")
if result:
LOG.log("info", f"SADP DLL 扫描结束: 发现 {len(result)} 台设备")
else:
LOG.log("warning", "SADP DLL 扫描: 未发现设备")
return result
except Exception as e:
LOG.log("error", f"SADP DLL 扫描异常: {e}")
LOG.log("warning", "回退到纯 Python 多播...")
return _scan_via_multicast(bind_ip="0.0.0.0")
def _scan_via_multicast(bind_ip: str = "0.0.0.0") -> list[SADPDevice]:
"""纯 Python 多播备用方案."""
actual_ip = _get_local_ip() # 获取本机真实 IP 用于多播接口绑定
devices: list[SADPDevice] = []
seen_macs: set[str] = set()
uuid_str = uuid.uuid4().hex
probe = f"""
{uuid_str}inquiry""".encode()
# 关键：单个 socket 绑定到 SADP_PORT，既发送又接收
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
# Windows SO_REUSEPORT
if platform.system() == "Windows":
with contextlib.suppress(Exception):
sock.setsockopt(socket.SOL_SOCKET, 0x18, 1)
try:
sock.bind(("", SADP_PORT))
LOG.log("info", f"Socket 已绑定到 0.0.0.0:{SADP_PORT}")
except OSError as e:
LOG.log("warning", f"绑定 0.0.0.0:{SADP_PORT} 失败: {e}, 尝试备用方案...")
sock.close()
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.bind(("", SADP_PORT))
LOG.log("info", f"备用方案绑定成功: 0.0.0.0:{SADP_PORT}")
# 加入多播组
try:
mreq = struct.pack("4s4s", socket.inet_aton(SADP_MULTICAST_ADDR), socket.inet_aton(actual_ip))
sock.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, mreq)
LOG.log("info", f"已加入多播组 {SADP_MULTICAST_ADDR} via {actual_ip}")
except Exception as e:
LOG.log("warning", f"加入多播组失败: {e}")
try:
# ---- 发送探测报文（关键：绑定到 SADP_PORT 让设备 unicast 回这个端口）----
send_sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
try:
send_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
if platform.system() == "Windows":
with contextlib.suppress(Exception):
send_sock.setsockopt(socket.SOL_SOCKET, 0x18, 1)
send_sock.bind(("", SADP_PORT))
send_sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, 2)
send_sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_IF, socket.inet_aton(actual_ip))
LOG.log("info", f"发送 socket 绑定到 :{SADP_PORT}, 发送接口: {actual_ip}")
for i in range(3):
send_sock.sendto(probe, (SADP_MULTICAST_ADDR, SADP_PORT))
LOG.log("info", f"SADP probe #{i+1} 已发送 to {SADP_MULTICAST_ADDR}:{SADP_PORT}")
time.sleep(1)
except Exception as e:
import traceback
LOG.log("error", f"SADP 发送异常: {e}")
LOG.log("error", f"发送堆栈: {traceback.format_exc()}")
finally:
send_sock.close()
LOG.log("info", "发送 socket 已关闭")
# ---- 接收响应 ----
LOG.log("info", "开始接收响应...")
sock.settimeout(0.5)
timeout_at = time.time() + (SADP_TIMEOUT_MS / 1000.0)
while time.time() < timeout_at:
try:
data, addr = sock.recvfrom(65535)
LOG.log("info", f"收到 [{len(data)}B] from {addr[0]}:{addr[1]}")
try:
raw = data.decode("utf-8", errors="ignore")[:400]
LOG.log("info", f"原始数据: {raw}")
except Exception:
LOG.log("info", f"十六进制: {data[:200].hex()}")
device = _parse_sadp_response(data)
if device:
LOG.log("info", f"解析成功: MAC={device.mac} IP={device.ip} model={device.model}")
if device and device.mac and device.mac not in seen_macs:
seen_macs.add(device.mac)
devices.append(device)
LOG.log("done", f"发现设备: {device.display_name()} IP={device.ip}")
except TimeoutError:
continue
finally:
with contextlib.suppress(Exception):
sock.setsockopt(socket.IPPROTO_IP, socket.IP_DROP_MEMBERSHIP,
struct.pack("4s4s", socket.inet_aton(SADP_MULTICAST_ADDR), socket.inet_aton(actual_ip)))
sock.close()
LOG.log("info", "接收 socket 已关闭")
LOG.log("info", f"SADP 扫描结束: 发现 {len(devices)} 台设备")
for d in devices:
LOG.log("info", f" MAC={d.mac} IP={d.ip} model={d.model} 激活={d.activated}")
hikvision_devices = [d for d in devices if d.is_hikvision]
if hikvision_devices:
return hikvision_devices
if devices:
LOG.log("warning", f"发现 {len(devices)} 台设备但无 Hikvision OUI 匹配，仍返回全部")
return devices
# ============================================================
class PTZDeviceController:
"""PTZ 设备管理器（高层包装）。
整合 SADP 发现、ISAPI 连接、PTZ 控制能力。
"""
# 方向映射: 前端 direction -> (pan, tilt) values
DIRECTION_MAP = {
"up": (0, 1),
"down": (0, -1),
"left": (-1, 0),
"right": (1, 0),
"up-left": (-1, 1),
"up-right": (1, 1),
"down-left": (-1, -1),
"down-right": (1, -1),
}
def __init__(self) -> None:
LOG.log("info", "PTZDeviceController 初始化完成")
from src.config_paths import CONFIG_DIR as _CONFIG_DIR
self.config = ConfigManager(
local_path=_CONFIG_DIR / "local.json",
ptz_path=_CONFIG_DIR / "ptz_config.json",
)
self._clients: dict[str, ISAPIClient] = {}
self._controllers: dict[str, ISAPIPTZController] = {}
# {ip: {"username": ..., "password": ..., "port": ...}}
self._credentials: dict[str, dict[str, str | int]] = {}
# {mac: device_info_dict} - SADP 发现缓存
self._discovered: dict[str, SADPDevice] = {}
self._credentials = {}
# v9.77: 单设备快速缓存 — 消除每次操作的 MAC→IP 查找
self._active_mac: str | None = None # 当前活跃设备的 MAC（无分隔符小写）
self._active_ip: str | None = None # 当前活跃设备的 IP
# 重连保护：单飞锁防止并发重复重连，冷却期避免失败后反复尝试
self._reconnect_lock: threading.Lock = threading.Lock()
self._reconnect_cooldown_until: float = 0.0
# v10.08: SADP 按需触发，不再后台常驻
self._load_saved_credentials()
# v10.04: 位置更新回调（供 DeviceReader 注册）
self._on_position_update = None
# v10.40: DeviceManager 引用（由 main.py 注入，用于 registry.json 统一管理）
self._device_manager = None
# v10.59: move 取消标记 — stop 时设置，让正在执行的 move 跳过 ISAPI 发送
self._move_cancel_events: dict[str, threading.Event] = {}
# v1.1.25: 命令 generation token — stop 递增 gen 使"已入队未发送"
# 的 move 失效；配合 router 层单 FIFO 线程池保证 stop 到达顺序
self._cmd_gen = 0
self._cmd_gen_lock = threading.Lock()
# v1.1.8: PTZ 行程限位缓存 {ip: {pan_min,pan_max,tilt_min,tilt_max}}（capabilities 一次 IO）
self._ptz_limits: dict[str, dict[str, int]] = {}
def set_device_manager(self, device_manager) -> None:
"""注入 DeviceManager 实例（由 main.py 在启动时调用）。"""
self._device_manager = device_manager
def set_position_update_callback(self, callback) -> None:
"""
设置位置更新回调（供 DeviceReader 注册）
Args:
callback: 位置更新回调函数，接收 {"pan": float, "tilt": float, "zoom": float}
"""
self._on_position_update = callback
def get_config(self) -> ConfigManager:
return self.config
# ================================================================
# SADP 发现 & 设备管理
# ================================================================
def discover_devices(self, bind_ip: str = "0.0.0.0") -> list[dict]:
"""返回已发现的设备列表（按需触发，不常驻后台）。
v10.08: 改为按需创建 SADP 实例，扫描完成后释放资源。
"""
if SADP_DLL_AVAILABLE:
try:
# 按需创建 SADP 实例
sadp_mgr = SADPManagerDLL()
sadp_mgr.start_discovery()
sadp_mgr.send_inquiry()
# 等待设备响应（约 2 秒）
import time
time.sleep(2)
sdk_devices = sadp_mgr.discover_devices()
# 释放资源
with contextlib.suppress(Exception):
sadp_mgr.stop_discovery()
if sdk_devices:
for dev_dict in sdk_devices:
dev = SADPDevice()
dev.mac = _normalize_mac(dev_dict.get("mac", ""))
dev.ip = dev_dict.get("ip", "")
dev.subnet_mask = dev_dict.get("subnet_mask", "")
dev.gateway = dev_dict.get("gateway", "")
dev.model = dev_dict.get("model", "")
dev.serial_number = dev_dict.get("serial_number", "")
dev.device_name = dev_dict.get("device_name", "")
dev.firmware_version = dev_dict.get("firmware_version", "")
dev.activated = dev_dict.get("activated", False)
dev.is_hikvision = dev_dict.get("is_hikvision", False)
self._discovered[dev.mac] = dev
# v10.41: SADP 发现后立即写入 info.json（含网关/子网掩码）
self.config.upsert_device(dev.mac, {
"ip": dev.ip,
"model": dev.model,
"name": dev.device_name,
"gateway": dev.gateway,
"subnet_mask": dev.subnet_mask,
"serial_number": dev.serial_number,
"firmware_version": dev.firmware_version,
})
return [self._sadp_to_dict(d) for d in self._discovered.values()]
except Exception as e:
LOG.log("warning", f"SADP DLL 扫描失败: {e}，回退到多播")
# 回退: 纯 Python 多播
LOG.log("warning", "SADP SDK 不可用或失败，使用纯 Python 多播发现...")
devices = scan_for_devices(bind_ip=bind_ip)
result = []
for d in devices:
d.mac = _normalize_mac(d.mac)
self._discovered[d.mac] = d
# v10.41: SADP 发现后立即写入 info.json（含网关/子网掩码）
self.config.upsert_device(d.mac, {
"ip": d.ip,
"model": d.model,
"name": d.device_name,
"gateway": d.gateway,
"subnet_mask": d.subnet_mask,
"serial_number": d.serial_number,
"firmware_version": d.firmware_version,
})
result.append(self._sadp_to_dict(d))
return result
def get_discovered_devices(self) -> list[dict]:
"""返回已缓存的发现设备列表。"""
return [self._sadp_to_dict(d) for d in self._discovered.values()]
def _sadp_to_dict(self, d: SADPDevice) -> dict:
return {
"mac": d.mac,
"ip": d.ip,
"subnet_mask": d.subnet_mask,
"gateway": d.gateway,
"model": d.model,
"serial_number": d.serial_number,
"device_name": d.device_name,
"firmware_version": d.firmware_version,
"activated": d.activated,
"is_hikvision": d.is_hikvision,
}
# ================================================================
# 凭据管理
# ================================================================
def verify_password_once(self, ip: str, username: str, password: str,
port: int = 80) -> dict:
"""v1.1.43: 密码单次校验（防设备锁定）。
仅调用 1 次 ISAPI GET /System/deviceInfo（Digest Auth 库内部 1 次 401
挑战 + 1 次带凭据），不重试、不落盘、不建持久会话。用于 add_device /
set_credentials 入口前置校验，确保错误密码绝不进入 ptz_config.json。
返回:
{"success": True}
{"success": False, "error_type": str, "message": str}
error_type: auth|timeout|network|forbidden|http|unknown
"""
client = ISAPIClient(ip=ip, username=username, password=password, port=port)
try:
return client.verify_credentials()
finally:
# 立即关闭 session，不占用设备 HTTP 连接槽
import contextlib
with contextlib.suppress(Exception):
client.close()
def _clear_saved_password(self, ip: str) -> None:
"""v1.1.43: 清除指定 IP 设备的已存密码（认证失败时调用，防反复 401 锁定）。
保留设备元数据（mac/ip/model/name/gateway 等），仅清空 password 字段，
使设备回到"需重新输入密码"状态。同时清除内存凭据缓存。
注意：upsert_device 对 password 有"空值不覆盖"保护，故走 save_ptz_config
直接写 info.json（只写目标设备，不影响其他设备）。
"""
# 1. 清除内存缓存
self._credentials.pop(ip, None)
# 2. 清除配置文件中的密码字段
try:
config = self.config.load_ptz_config()
devices = config.get("devices", {})
target_mac: str | None = None
target_info: dict | None = None
for mac, dev in devices.items():
if dev.get("ip") == ip:
target_mac = mac
target_info = dev
break
if target_mac and target_info is not None:
cleared = dict(target_info)
cleared["password"] = ""
# 只写目标设备（save_ptz_config 不影响其他设备）
self.config.save_ptz_config({"devices": {target_mac: cleared}})
LOG.log(
"warning",
f"已清除设备密码（认证失败防锁定）: ip={ip} mac={target_mac}",
)
except Exception as e:
LOG.log("error", f"清除设备密码失败: ip={ip} error={e}")
def save_credentials(self, ip: str, username: str, password: str,
port: int = 80, mac: str = "", model: str = "",
name: str = "", gateway: str = "",
subnet_mask: str = "") -> None:
"""保存设备凭据到内存 + 配置文件。
v10.40: 增加 gateway/subnet_mask 参数，修复网关丢失问题。
"""
if not mac or mac == ip:
LOG.log("warning", f"拒绝保存无效 MAC: {mac} (IP: {ip})")
return
self._credentials[ip] = {
"username": username,
"password": password,
"port": port,
}
self.config.upsert_device(mac, {
"ip": ip,
"username": username,
"password": password,
"port": port,
"model": model,
"name": name,
"mac": mac,
"gateway": gateway,
"subnet_mask": subnet_mask,
"connected": False,
})
def get_credentials(self, ip: str) -> dict | None:
"""获取设备凭据（含设备名称）。
v7.105: 优先从内存读取，回退文件，含 device_name 字段。
"""
# 1. 内存优先
creds = self._credentials.get(ip)
if creds:
return creds
# 2. 内存无，从文件回退读取
# v1.1.35: 同 IP 多 MAC 条目时优先有密码者（生产同 IP 双 MAC 条目
# 曾导致快速连接命中无密码/错密码条目→认证失败）
try:
config = self.config.load_ptz_config()
devices = config.get("devices", {})
fallback: dict | None = None
for pass_with_password in (True, False):
for mac, dev in devices.items():
if dev.get("ip") != ip:
continue
username = dev.get("username")
password = dev.get("password")
if not username or password is None:
continue
if pass_with_password and not password:
continue
if not pass_with_password and password:
continue
creds = {
"username": username,
"password": password,
"port": dev.get("port", 80),
"mac": mac,
"model": dev.get("model", ""),
"device_name": dev.get("name", "") or dev.get("device_name", ""),
"video_codec": dev.get("video_codec", "H.265"),
}
if pass_with_password:
self._credentials[ip] = creds
LOG.log(
"debug",
f"get_credentials 文件回退: ip={ip} mac={mac} 含密码",
)
return creds
if fallback is None:
fallback = creds
if pass_with_password and fallback is None:
continue
break
# v1.1.40: 空密码守卫 — fallback 条目无密码时拒绝返回，
# 防止空密码爆破导致海康设备锁定（生产实证：5次错误后锁定30分钟）
if fallback:
LOG.log(
"warning",
f"get_credentials 文件回退: ip={ip} mac={fallback.get('mac')} 无密码条目，拒绝返回（防设备锁定）",
)
return None
except Exception:
pass
return None
def remove_credentials(self, ip: str) -> bool:
"""移除设备凭据。"""
self._credentials.pop(ip, None)
# 从配置中也移除
config = self.config.load_ptz_config()
devices = config.get("devices", {})
# 通过 IP 或 MAC 查找
for key, dev in list(devices.items()):
if dev.get("ip") == ip or key == ip:
return self.config.remove_device(key)
return False
def _load_saved_credentials(self) -> None:
"""启动时从配置文件加载已保存的凭据.
v10.42: 同时扫描旧格式 {mac}.json 文件（v10.40 迁移兼容）。
"""
try:
config = self.config.load_ptz_config()
devices = config.get("devices", {})
for key, dev in devices.items():
ip = dev.get("ip", key)
if ip and "username" in dev and "password" in dev:
self._credentials[ip] = {
"username": dev["username"],
"password": dev["password"],
"port": dev.get("port", 80),
"mac": dev.get("mac", ""),
"model": dev.get("model", ""),
"name": dev.get("name", ""),
"serial_number": dev.get("serial_number", ""),
"firmware_version": dev.get("firmware_version", ""),
"video_codec": dev.get("video_codec", "H.265"),
}
LOG.log("info", f"已加载凭据: {ip} ({dev.get('model', '')})")
# v10.42: 扫描旧格式 {mac}.json 文件（凭据存储位置）
devices_dir = self.config._devices_dir
if devices_dir.exists():
for old_file in devices_dir.glob("*.json"):
if old_file.name == "devices.json":
continue
try:
with open(old_file, encoding="utf-8") as f:
dev = json.load(f)
ip = dev.get("ip", "")
mac = dev.get("mac", "")
username = dev.get("username")
password = dev.get("password")
# 只加载内存中没有的凭据
if ip and username and password is not None and ip not in self._credentials:
self._credentials[ip] = {
"username": username,
"password": password,
"port": dev.get("port", 80),
"mac": mac,
"model": dev.get("model", ""),
"name": dev.get("name", ""),
"serial_number": dev.get("serial_number", ""),
"firmware_version": dev.get("firmware_version", ""),
"video_codec": dev.get("video_codec", "H.265"),
}
LOG.log("info", f"已加载旧格式凭据: {ip} ({mac})")
except Exception:
continue
if self._credentials:
LOG.log("done", f"已加载 {len(self._credentials)} 个设备凭据")
except Exception as e:
LOG.log("warning", f"加载凭据失败: {e}")
def list_stored_devices(self) -> list[dict]:
"""列出所有已保存的设备（手动添加的）。"""
config = self.config.load_ptz_config()
devices = config.get("devices", {})
result = []
for mac, info in devices.items():
ip = info.get("ip", "")
is_connected = ip in self._controllers
sadp_info = self._discovered.get(mac)
entry = {
"mac": mac,
"ip": ip,
"name": info.get("name", ""),
"model": info.get("model", ""),
"gateway": info.get("gateway", ""), # v6.02
"subnet_mask": info.get("subnet_mask", ""), # v6.02
"port": info.get("port", 80),
"connected": is_connected,
"online": is_connected,
}
if sadp_info:
# 合并 SADP 信息，但不覆盖 name
# v1.1.34: 仅非空值覆盖——防 SADP 缓存空值抹掉已存储网关/子网掩码
sadp_dict = self._sadp_to_dict(sadp_info)
sadp_dict.pop("name", None) # 移除可能的 name 字段
for key, value in sadp_dict.items():
if value not in ("", None):
entry[key] = value
result.append(entry)
return result
# ================================================================
# 设备认证连接
# ================================================================
def connect_device(self, ip: str, username: str, password: str,
port: int = 80) -> dict | None:
"""连接设备，返回连接结果 dict（含成功/失败信息）。
v9.77: 单设备架构重构
- 消除重复认证（复用 client 获取设备信息）
- 设置 _active_mac/_active_ip 缓存
返回值:
{"success": True/False, "message": ..., "device_info": ...}
"""
if ip in self._controllers:
# 已连接，更新状态
ptz_cfg = self.config.load_ptz_config()
for _key, dev in ptz_cfg.get("devices", {}).items():
if dev.get("ip") == ip:
dev["connected"] = True
self.config.save_ptz_config(ptz_cfg)
break
return {"success": True, "message": "设备已连接"}
client = ISAPIClient(ip=ip, username=username, password=password, port=port)
verify_result = client.verify_credentials()
if verify_result.get("success"):
self._clients[ip] = client
controller = ISAPIPTZController(client)
self._controllers[ip] = controller
self._credentials[ip] = {
"username": username,
"password": password,
"port": port,
}
# 保存凭据到配置文件
mac = ""
model = ""
serial_number = ""
firmware_version = ""
# 尝试从发现缓存获取
for sadp_dev in self._discovered.values():
if sadp_dev.ip == ip:
mac = sadp_dev.mac
model = sadp_dev.model
serial_number = sadp_dev.serial_number
firmware_version = sadp_dev.firmware_version
break
# v9.77: 复用 client 获取设备信息，不再创建新 Session
device_name = ""
# v10.41: 网关/子网掩码仅由 SADP 发现时写入，连接时不覆盖
try:
resp = client.get("/System/deviceInfo")
if resp.ok():
_root = ET.fromstring(resp.xml)
for _elem in _root.iter():
_tag = _elem.tag.split("}")[-1] if "}" in _elem.tag else _elem.tag
if _tag == "macAddress" and not mac:
mac = (_elem.text or "").strip().replace(":", "-").upper()
elif _tag == "model" and not model:
model = (_elem.text or "").strip()
elif _tag == "deviceName" and not device_name:
device_name = (_elem.text or "").strip()
elif _tag == "serialNumber" and not serial_number:
serial_number = (_elem.text or "").strip()
elif _tag == "firmwareVersion" and not firmware_version:
firmware_version = (_elem.text or "").strip()
except Exception:
pass
# v1.0.8: 检测视频编码类型（H.264/H.265），缓存到本地设备信息，
# 供 OBS GStreamer 媒体源自适应选择 d3d11 解码链（与凭据同时注入管线）
video_codec = "H.265"
try:
_codec_resp = client.get("/Streaming/channels/101")
if _codec_resp.ok():
_codec_root = ET.fromstring(_codec_resp.xml)
video_codec = (
client.get_xml_text(_codec_root, "videoCodecType", "") or "H.265"
)
except Exception:
pass
if ip in self._credentials:
self._credentials[ip]["video_codec"] = video_codec
LOG.log("info", f"检测到视频编码类型: {video_codec} @ {ip}")
# v1.1.34: 单设备独占——连接成功即断开其他存活会话（拨杆互斥服务端保证）
for other_ip in [ip_ for ip_ in list(self._clients) if ip_ != ip]:
LOG.log("info", f"互斥断开旧会话: {other_ip}")
self.disconnect_device(other_ip)
# v10.41: 从 SADP 缓存获取网关（如果有的话）
sadp_gateway = ""
sadp_subnet = ""
for sadp_dev in self._discovered.values():
if sadp_dev.ip == ip:
sadp_gateway = sadp_dev.gateway
sadp_subnet = sadp_dev.subnet_mask
break
# v1.1.21: 网关兜底 — SADP 缓存缺项时改由设备自报（ISAPI 网络配置），
# 模型无关；修复“连接过的设备网关丢失、未连接设备反而正常”的生产回归。
if not sadp_gateway:
sadp_gateway, sadp_subnet = self._fetch_device_network(client)
self.config.upsert_device(mac or ip, {
"ip": ip,
"username": username,
"password": password,
"port": port,
"model": model,
"mac": mac or ip,
"serial_number": serial_number,
"firmware_version": firmware_version,
"name": device_name,
"gateway": sadp_gateway,
"subnet_mask": sadp_subnet,
"video_codec": video_codec,
"connected": True,
})
# v10.40: 通过 DeviceManager 更新 registry.json
norm_mac = _normalize_mac(mac) if mac else ""
if norm_mac and self._device_manager:
self._device_manager.set_active_device(norm_mac)
self._device_manager.register_device(
mac=norm_mac, ip=ip, name=device_name, model=model
)
# v1.1.25: 自动重连路径也写 online（原仅手动连接路由写，
# 导致 DeviceManager.status 对 auto 路径是死字段）
self._device_manager.update_status(norm_mac, "online")
# v9.77: 设置单设备快速缓存
self._active_mac = norm_mac
self._active_ip = ip
LOG.log("done", f"PTZ 设备已连接: {ip}, MAC={norm_mac}")
device_display = device_name or model or ip
return {
"success": True,
"message": f"设备 [{device_display}] 连接成功",
"steps": [
{"step": "ISAPI 认证", "status": "success"},
],
"ip": ip,
"name": device_name,
"model": model,
}
LOG.log("failed", f"PTZ 设备认证失败: {ip}")
_err_type = verify_result.get("error_type", "") if isinstance(verify_result, dict) else ""
_err_msg = verify_result.get("message", "认证失败，请检查凭据") if isinstance(verify_result, dict) else "认证失败，请检查凭据"
# v1.1.43: 认证失败（401）→ 清除已存密码，防 auto_reconnect 反复 401 触发设备锁定
if _err_type == "auth":
self._clear_saved_password(ip)
_err_msg = "密码错误，请重新输入"
return {
"success": False,
"error_type": _err_type,
"message": _err_msg,
}
def disconnect_device(self, ip: str) -> None:
"""v6.30: 断开设备连接。清除 active_device，保留 last_connected 用于下次快速连接。
v9.77: 清理 _active_mac/_active_ip 缓存
"""
self._controllers.pop(ip, None)
client = self._clients.pop(ip, None)
if client:
import contextlib
with contextlib.suppress(Exception):
# v9.78: 统一关闭控制与轮询两个 session
client.close()
# v9.77: 清理单设备快速缓存
if self._active_ip == ip:
self._active_mac = None
self._active_ip = None
LOG.log("info", "单设备快速缓存已清理")
# 更新设备状态
ptz_cfg = self.config.load_ptz_config()
disconnected_mac = None
for key, dev in ptz_cfg.get("devices", {}).items():
if dev.get("ip") == ip:
dev["connected"] = False
disconnected_mac = key
self.config.save_ptz_config(ptz_cfg)
break
# v10.40: 通过 DeviceManager 清除 active_device，保留 last_connected
if disconnected_mac and self._device_manager:
self._device_manager.clear_active_device(disconnected_mac)
self._device_manager.update_status(disconnected_mac, "offline")
def _fetch_device_network(self, client: ISAPIClient) -> tuple[str, str]:
"""v1.1.21: 读取设备自报网络配置（网关/子网掩码）。
ISAPI: GET /System/Network/channels/1/IPv4 → IPv4Gateway/IPv4SubnetMask。
失败返回 ("", "")，由 upsert 空值不覆盖语义保护已有数据。
"""
try:
resp = client.get("/System/Network/channels/1/IPv4", timeout=3.0)
if resp.status_code == 404:
# 部分型号无 channels 子资源，回退设备级端点
resp = client.get("/System/Network/IPv4", timeout=3.0)
if not resp.ok():
return "", ""
root = ET.fromstring(resp.xml)
gateway = ""
subnet = ""
for elem in root.iter():
tag = elem.tag.split("}")[-1] if "}" in elem.tag else elem.tag
if tag == "IPv4Gateway" and not gateway:
gateway = (elem.text or "").strip()
elif tag == "IPv4SubnetMask" and not subnet:
subnet = (elem.text or "").strip()
if gateway:
LOG.log("info", f"设备自报网关: {gateway} @ {client.ip}")
return gateway, subnet
except Exception as e:
LOG.log("warning", f"读取设备自报网关失败: {e}")
return "", ""
def backfill_gateway_via_scan(self) -> bool:
"""v1.1.34: 网关补偿——已连接设备存储网关为空时跑一次 SADP 扫描落盘。
供后台任务调用（手动连接/自动重连成功路径）。扫描自身已带
upsert 落盘（空值不覆盖语义保护已有数据）；ISAPI 自报端点在
部分型号 404（实测），SADP 是网关唯一可靠来源。
返回补偿后网关是否非空。
"""
# v1.1.34: 优先用单设备快速缓存（_active_mac = 当前连接设备）
mac = self._active_mac or ""
if not mac:
device = self.get_connected_device()
mac = _normalize_mac((device or {}).get("mac", ""))
if not mac:
return False
info = self.config.get_device_by_mac(mac)
if info and info.get("gateway"):
return True
LOG.log("info", f"网关缺失，启动补偿扫描: {mac}")
try:
self.discover_devices()
except Exception as e:
LOG.log("warning", f"网关补偿扫描失败: {e}")
return False
info = self.config.get_device_by_mac(mac)
ok = bool(info and info.get("gateway"))
if ok:
LOG.log("done", f"网关补偿扫描命中: {mac} -> {info.get('gateway')}")
return ok
def fast_reconnect(self, ping_attempts: int = 20) -> dict:
"""v10.40: 启动时快速重连上次设备。
v1.1.25: ping_attempts 参数化——后台退避重试场景每轮少试几次
（原固定 20 次×0.5s ≈ 16-22s 同步阻塞启动关键路径）。
流程：
1. 从 registry.json 读取 last_connected MAC + IP
2. ping 验证 IP 可达（200ms 超时）
3. ARP 表获取 MAC，比对
4. MAC 匹配 -> ISAPI 连接（使用已保存凭据）
5. MAC 不匹配 -> 返回 {"success": False, "mac_mismatch": True}
6. ARP 失败 -> 返回 {"success": False, "message": "设备不可达"}
不使用 ISAPI 兜底获取 MAC。
"""
from src.core.net_detector import get_mac_from_arp
# Step 1: 读取上次连接的设备
if not self._device_manager:
return {"success": False, "message": "DeviceManager 未初始化"}
last_mac = self._device_manager.get_last_connected_mac()
if not last_mac:
return {"success": False, "message": "无上次连接设备"}
# 从配置文件获取 IP 和凭据
device_info = self.config.get_device_by_mac(last_mac)
if not device_info:
return {"success": False, "message": f"设备配置不存在: {last_mac}"}
ip = device_info.get("ip", "")
if not ip:
return {"success": False, "message": f"设备无 IP 记录: {last_mac}"}
username = device_info.get("username", "admin")
password = device_info.get("password", "")
port = device_info.get("port", 80)
if not password:
return {"success": False, "message": f"设备无保存密码: {last_mac}"}
LOG.log("info", f"fast_reconnect: 尝试重连 {last_mac} @ {ip}")
# Step 2: ping 验证 IP 可达 — v1.1.21: 重试循环（冷启动网卡/ARP 未就绪时
# 单次 200ms ping 必败 → 误报不可达放弃自动重连）。
# v1.1.25: check_reachable 已含 TCP:80 回退；次数由参数控制。
reachable = False
for _attempt in range(max(1, ping_attempts)):
if check_reachable(ip, timeout=0.3):
reachable = True
break
time.sleep(0.5)
if not reachable:
LOG.log("warning", f"fast_reconnect: 设备 {ip} 不可达（{ping_attempts} 次重试后）")
return {"success": False, "reason": "unreachable", "message": f"设备 {ip} 不可达（{ping_attempts * 0.8:.0f}s 重试后）"}
# Step 3: ARP 表获取 MAC，比对 — v1.1.21: ping 通后 ARP 条目可能仍需
# 短暂时间写入，重试 3 次。
arp_mac = None
for _attempt in range(3):
arp_mac = get_mac_from_arp(ip)
if arp_mac is not None:
break
time.sleep(0.3)
if arp_mac is None:
LOG.log("warning", f"fast_reconnect: ARP 无法获取 {ip} 的 MAC")
return {"success": False, "reason": "arp_failed", "message": f"ARP 无法获取设备 MAC: {ip}"}
norm_last = _normalize_mac(last_mac)
if arp_mac != norm_last:
LOG.log("error", f"fast_reconnect: MAC 不匹配! 期望={norm_last}, 实际={arp_mac}")
return {
"success": False,
"mac_mismatch": True,
"reason": "mac_mismatch",
"expected_mac": norm_last,
"actual_mac": arp_mac,
"ip": ip,
"message": f"设备已更换（期望 MAC={norm_last}，实际 MAC={arp_mac}）",
}
# Step 4: MAC 匹配，ISAPI 连接
LOG.log("info", f"fast_reconnect: MAC 验证通过，开始 ISAPI 连接 {ip}")
result = self.connect_device(ip, username, password, port)
if result and result.get("success"):
LOG.log("done", f"fast_reconnect: 成功连接 {ip} (MAC={norm_last})")
return {"success": True, "ip": ip, "mac": norm_last, "message": "自动重连成功"}
# v1.1.43: auth 失败 = 终局（密码已被 connect_device 清除），不再重试
_err_type = (result or {}).get("error_type", "")
if _err_type == "auth":
LOG.log("failed", f"fast_reconnect: 密码错误，已清除凭据，终止重试 {ip}")
return {
"success": False,
"reason": "auth_failed",
"terminal": True,
"message": "密码错误，请重新输入",
}
LOG.log("failed", f"fast_reconnect: ISAPI 连接失败 {ip}")
return {"success": False, "reason": "isapi_failed", "message": "ISAPI 连接失败"}
def get_active_ip(self, device_id: str | None = None) -> str | None:
"""v9.77: 快速获取设备 IP，消除每次操作的 MAC→IP 查找。
优先级:
1. device_id 是 IP 格式 → 直接返回
2. device_id 是 MAC 且匹配 _active_mac → 返回 _active_ip
3. device_id 为空 → 返回 _active_ip
4. 回退到旧逻辑（遍历查找）
Args:
device_id: MAC 地址或 IP 地址，可为空
Returns:
设备 IP 地址，未找到返回 None
"""
# 1. device_id 是 IP 格式
if device_id and self._is_ip_address(device_id):
return device_id
# 2. device_id 是 MAC 且匹配缓存
if device_id and self._active_mac:
mac_clean = _normalize_mac(device_id)
if mac_clean == self._active_mac:
return self._active_ip
# 3. device_id 为空，返回缓存
if not device_id and self._active_ip:
return self._active_ip
# 4. 回退到旧逻辑
return None
def _is_ip_address(self, s: str) -> bool:
"""判断字符串是否为 IP 地址格式。"""
import re
return bool(re.match(r'^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$', s))
def list_controllers(self) -> list[str]:
return list(self._controllers.keys())
def is_device_connected(self, ip: str) -> bool:
return ip in self._controllers
def get_connected_device(self) -> dict | None:
"""v7.10: 返回 last_connected 设备信息（用于快速连接）。
注意：此方法返回的是"上次连接的设备"，不一定是"当前连接的设备"。
前端应该通过检查 IP 是否在 _controllers 中判断是否实际连接。
v11.20: 改从 DeviceManager 内存读取，不再每次读盘（热点路径：
前端 5s 轮询 + 每个位置事件回调）。
"""
try:
if not self._device_manager:
return None
last_mac = self._device_manager.get_last_connected_mac()
if not last_mac:
return None
dev = self._device_manager.get_device(last_mac)
if dev is None:
return None
meta = dev.metadata or {}
creds = meta.get('credentials') or {}
return {
'ip': dev.ip,
'mac': dev.mac,
'name': dev.name,
'model': dev.model,
'username': creds.get('username') or meta.get('username'),
'password': creds.get('password') or meta.get('password'),
'port': meta.get('port', 80),
'device_name': meta.get('device_name'),
}
except Exception as e:
LOG.log("error", f"get_connected_device error: {e}")
return None
# ================================================================
# 设备信息获取
# ================================================================
def get_device_info(self, ip: str) -> dict:
"""通过 ISAPI 获取设备详细信息。"""
if ip not in self._clients:
# 尝试自动连接
creds = self.get_credentials(ip)
if creds:
client = ISAPIClient(ip=ip, username=creds["username"],
password=creds["password"], port=creds.get("port", 80))
if not client.verify_credentials():
return {"error": "认证失败"}
self._clients[ip] = client
self._controllers[ip] = ISAPIPTZController(client)
else:
return {"error": "设备未连接且无凭据"}
client = self._clients[ip]
info = {}
# 系统信息
result = client.get("/System/deviceInfo")
if result.ok():
try:
root = ET.fromstring(result.xml)
info["device_name"] = client.get_xml_text(root, "deviceName", "")
info["model"] = client.get_xml_text(root, "model", "")
info["serial_number"] = client.get_xml_text(root, "serialNumber", "")
info["firmware_version"] = client.get_xml_text(root, "firmwareVersion", "")
info["mac_address"] = client.get_xml_text(root, "macAddress", "")
except Exception as e:
LOG.log("warning", f"解析设备信息异常: {e}")
# 网络配置
net_result = client.get("/Network/interfaces/1")
if net_result.ok():
try:
root = ET.fromstring(net_result.xml)
info["network"] = {
"ip": client.get_xml_text(root, "ipAddress", ""),
"subnet_mask": client.get_xml_text(root, "subnetMask", ""),
"gateway": client.get_xml_text(root, "defaultGateway", ""),
}
except Exception:
pass
# PTZ 位置
if ip in self._controllers:
info["ptz_position"] = self._controllers[ip].get_position()
return info
# ================================================================
# IP 修改
# ================================================================
def modify_device_network(self, ip: str, new_ip: str, subnet_mask: str,
gateway: str) -> dict:
"""修改设备网络配置。
优先级：
1. ISAPI（已连接设备）- 可靠
2. SADP（发现未连接设备）- 需要密码，可能失败
"""
# 方案1：通过 ISAPI 修改（设备已连接时）
if ip in self._controllers:
client = self._clients[ip]
LOG.log("info", f"通过 ISAPI 修改网络: {ip} -> {new_ip}")
# ISAPI 网络配置 endpoint
xml = f"""
false
{new_ip}
{subnet_mask}
{gateway}
"""
result = client.put("/Network/channels/1/Network", xml)
if result.ok():
time.sleep(3)
if check_reachable(new_ip, timeout=3):
return {"success": True, "message": f"IP 修改成功: {ip} -> {new_ip}",
"new_ip": new_ip, "method": "ISAPI"}
# 方案2：SADP 修改（需要设备密码）
cred = self._credentials.get(ip)
password = cred["password"] if cred else None
sadp_dev = None
for d in self._discovered.values():
if d.ip == ip:
sadp_dev = d
break
if not sadp_dev:
return {"success": False, "message": "设备未发现且未连接"}
if not password:
return {"success": False, "message": "需要设备密码才能修改 IP"}
return self._modify_ip_via_sadp(sadp_dev, new_ip, subnet_mask, gateway, password)
def _modify_ip_via_sadp(self, sadp_dev, new_ip: str, subnet_mask: str,
gateway: str, password: str, username: str = "admin") -> dict:
"""通过 SADP V40 DLL 修改 IP（官方 API，不绕开 DLL）。
流程:
1. 检测 IP 是否被占用 (PING)
2. 调用 SADP V40 DLL SADP_ModifyDeviceNetParam_V40
3. 查询错误码含义
4. 修改后等待 5 秒，验证新 IP 可达
"""
# 判断 IP 是否被占用
if check_reachable(new_ip, timeout=2):
LOG.log("warning", f"目标 IP {new_ip} 已被占用，需更换 IP")
return {
"success": False,
"message": f"目标 IP {new_ip} 已被占用，请更换",
"error_code": "IP_CONFLICT",
}
# 通过 SADP SDK DLL 修改网络参数（官方 API）
if SADP_DLL_AVAILABLE:
LOG.log("info", f"SADP DLL 修改 IP: MAC={sadp_dev.mac} -> {new_ip}")
try:
mgr = SADPManagerDLL()
result = mgr.modify_device_network(
mac=sadp_dev.mac,
password=password,
new_ip=new_ip,
subnet_mask=subnet_mask,
gateway=gateway,
http_port=getattr(sadp_dev, 'http_port', 80),
)
if result.get("success"):
# v11.20: 等待设备应用新 IP（注释与代码对齐，实际 3 秒）
LOG.log("info", "等待 3 秒让设备应用新 IP...")
time.sleep(3)
if check_reachable(new_ip, timeout=3):
sadp_dev.ip = new_ip
sadp_dev.gateway = gateway
sadp_dev.subnet_mask = subnet_mask
# 修复 BUG 5: 同时保存凭据到 info.json 和内存缓存
self.config.upsert_device(sadp_dev.mac, {
"ip": new_ip,
"model": sadp_dev.model,
"mac": sadp_dev.mac,
"username": username,
"password": password,
})
self._credentials[new_ip] = {
"username": username,
"password": password,
"port": 80,
}
LOG.log("done", f"凭据已保存: {new_ip} (user={username})")
return {
"success": True,
"message": f"IP 修改成功: {sadp_dev.ip} -> {new_ip}",
"new_ip": new_ip,
"method": "SADP_DLL",
}
return {"success": False, "message": f"IP 修改成功但 {new_ip} 不可达"}
# DLL 返回错误
error_msg = result.get("message", "修改失败")
LOG.log("error", f"SADP DLL 修改失败: {error_msg}")
return {
"success": False,
"message": error_msg,
"method": "SADP_DLL",
}
except Exception as e:
LOG.log("error", f"SADP DLL 修改 IP 异常: {e}")
LOG.log("warning", "回退到备用方案（不可用）")
# 回退方案: DLL 不可用时返回错误而非使用裸 socket
return {
"success": False,
"message": "SADP DLL 不可用，无法修改 IP。请复制 Sadp.dll 到项目目录。",
}
def modify_reachable_with_retry(
self,
mac: str,
password: str,
original_ip: str,
target_ip: str,
subnet_mask: str = "255.255.255.0",
gateway: str = "",
max_retries: int = 2,
username: str = "admin",
) -> dict:
"""通过 SADP 修改设备 IP，含重试机制。
Args:
mac: 设备 MAC 地址
password: 设备密码
original_ip: 设备当前 IP
target_ip: 目标 IP
subnet_mask: 子网掩码
gateway: 网关
max_retries: 最大重试次数
username: 设备用户名
"""
# 从 SADP 发现缓存中查找设备
sadp_dev = None
norm_mac = _normalize_mac(mac)
for d_mac, d in self._discovered.items():
if _normalize_mac(d_mac) == norm_mac:
sadp_dev = d
break
# 如果缓存中没有，创建一个临时 SADPDevice
if not sadp_dev:
LOG.log("warning", f"SADP 缓存中未找到 {mac}，使用临时设备对象")
sadp_dev = SADPDevice(
mac=mac,
ip=original_ip,
subnet_mask=subnet_mask,
gateway=gateway,
)
for attempt in range(1, max_retries + 1):
LOG.log("info", f"修改 IP 尝试 {attempt}/{max_retries}: {original_ip} -> {target_ip}")
result = self._modify_ip_via_sadp(sadp_dev, target_ip, subnet_mask, gateway, password, username)
if result.get("success"):
return result
if attempt < max_retries:
LOG.log("warning", f"第 {attempt} 次失败，{result.get('message', '')}，重试中...")
time.sleep(3)
return result
def auto_reconnect_known_devices(
self,
bind_ip: str = "0.0.0.0",
target_ip: str | None = None,
) -> dict:
"""已知设备自动重连。
流程:
1. SADP 扫描发现设备
2. 以 MAC 匹配已保存设备
3. Ping 验证 IP 是否可达
4. IP 不可达时用已保存凭据修改 IP
5. 修改成功后 time.sleep(3) -> 重新扫描确认
Args:
bind_ip: SADP 绑定的网卡 IP (默认 "0.0.0.0" = 全部网卡)
target_ip: 指定目标 IP (默认同网段.64)
Returns:
处理结果字典
"""
LOG.log("info", "开始已知设备自动重连流程...")
# Step 1: SADP 扫描
discovered = self.discover_devices(bind_ip=bind_ip)
if not discovered:
return {"success": False, "message": "SADP 扫描未发现任何设备"}
LOG.log("info", f"SADP 扫描发现 {len(discovered)} 台设备")
# Step 2: 加载已保存设备列表
config = self.config.load_ptz_config()
saved_devices = config.get("devices", {})
if not saved_devices:
return {"success": True, "message": "无已保存设备，跳过自动重连", "processed": 0}
# Step 3: 匹配 + 处理
results = []
processed_count = 0
for dev in discovered:
dev_mac = dev.get("mac", "").strip()
if not dev_mac:
continue
# 统一 MAC 格式进行匹配
norm_discovered = _normalize_mac(dev_mac)
# 尝试匹配已保存设备
saved = None
for mac_key, saved_info in saved_devices.items():
norm_saved = _normalize_mac(mac_key)
if norm_discovered == norm_saved:
saved = saved_info
break
if not saved:
continue # 不是已知设备，跳过
device_name = saved.get("name", dev.get("device_name", ""))
device_model = saved.get("model", dev.get("model", ""))
saved_password = saved.get("password", "")
saved_username = saved.get("username", "admin")
if not saved_password:
LOG.log("warning", f"已知设备 {norm_discovered} ({device_name}) 无保存密码，跳过")
results.append({
"mac": norm_discovered,
"model": device_model,
"status": "skipped",
"reason": "no_password",
})
continue
dev_ip = dev.get("ip", "")
LOG.log("info", f"发现已知设备 {norm_discovered} ({device_model}), IP={dev_ip}")
# Ping 验证可达性（200ms 超时）
if dev_ip and check_reachable(dev_ip, timeout=0.2):
LOG.log("done", f"设备 {norm_discovered} IP {dev_ip} 可达，无需修改")
results.append({
"mac": norm_discovered,
"ip": dev_ip,
"model": device_model,
"status": "reachable",
})
processed_count += 1
continue
# IP 不可达，使用已保存凭据修改
LOG.log("info", f"设备 {norm_discovered} IP {dev_ip} 不可达，执行 IP 修改...")
subnet_mask = dev.get("subnet_mask", "255.255.255.0")
gateway = dev.get("gateway", "")
# 获取目标 IP
eff_target_ip = target_ip
if not eff_target_ip:
eff_target_ip = suggest_target_ip()
if not eff_target_ip:
results.append({
"mac": norm_discovered,
"status": "failed",
"reason": "no_target_ip",
})
continue
if check_reachable(eff_target_ip, timeout=0.2):
LOG.log("warning", f"目标 IP {eff_target_ip} 已被占用")
results.append({
"mac": norm_discovered,
"status": "failed",
"reason": "target_ip_occupied",
})
continue
result = self._modify_ip_via_sadp(
SADPDevice(
mac=norm_discovered.replace(":", "-").lower(),
ip=dev_ip,
subnet_mask=subnet_mask,
gateway=gateway,
model=device_model,
),
eff_target_ip, subnet_mask, gateway, saved_password, saved_username
)
if result.get("success"):
# v11.20: 移除重复的 sleep(3)——_modify_ip_via_sadp 内部已等待并验证可达；
# 重新扫描刷新发现缓存，upsert 仅补充 name/activated 元数据（合并写）
self.discover_devices()
# 更新配置
self.config.upsert_device(norm_discovered, {
"ip": eff_target_ip,
"model": device_model,
"mac": norm_discovered,
"name": device_name,
"username": DEFAULT_USERNAME,
"password": saved_password,
"activated": dev.get("activated", False),
})
LOG.log("done", f"设备 {norm_discovered} IP 修改确认成功: {eff_target_ip}")
results.append({
"mac": norm_discovered,
"old_ip": dev_ip,
"new_ip": eff_target_ip,
"model": device_model,
"status": "modified",
})
else:
LOG.log("error", f"设备 {norm_discovered} IP 修改失败: {result.get('message')}")
results.append({
"mac": norm_discovered,
"status": "failed",
"reason": result.get("message", "unknown"),
})
processed_count += 1
return {
"success": True,
"message": f"已知设备自动重连完成，处理 {processed_count} 台设备",
"processed": processed_count,
"details": results,
}
# ================================================================
# PTZ 高级控制方法
# ================================================================
def _get_ptz_limits(self, ip: str) -> dict | None:
"""v1.1.8: 读取并缓存设备 P/T 轴限位信息。
v1.1.50: 优先从 limit.json 读取（包含 pan_has_limit 字段），
回退到 ISAPI capabilities。
返回 {pan_min, pan_max, tilt_min, tilt_max, pan_has_limit, pan_is_infinite}；失败返回 None。
"""
if ip in self._ptz_limits:
return self._ptz_limits[ip]
# 1. 优先从 limit.json 读取（包含 pan_has_limit 字段）
try:
from src.advanced.device_path import get_data_path_read
# 通过 IP 反查 MAC（使用单设备快速缓存）
mac_clean = self._active_mac if self._active_ip == ip else None
if mac_clean:
limit_path = get_data_path_read(None, mac_clean, "limit")
if limit_path and limit_path.exists():
import json
limit_data = json.loads(limit_path.read_text(encoding="utf-8"))
pan_has_limit = limit_data.get("pan_has_limit", True)
is_infinite = limit_data.get("is_infinite", False)
limits = {
"pan_min": limit_data.get("pan_min", 0),
"pan_max": limit_data.get("pan_max", 3600),
"tilt_min": limit_data.get("tilt_min", -200),
"tilt_max": limit_data.get("tilt_max", 900),
"pan_has_limit": pan_has_limit,
"pan_is_infinite": is_infinite or not pan_has_limit,
}
self._ptz_limits[ip] = limits
LOG.log("info", f"PTZ 限位从 limit.json 加载: pan_infinite={limits['pan_is_infinite']}")
return limits
except Exception as e:
LOG.log("warning", f"读取 limit.json 失败: {e}")
# 2. 回退到 ISAPI capabilities
ctrl, err = self._get_controller(ip)
if err or ctrl is None:
return None
resp = ctrl.client.get(f"/PTZCtrl/channels/{ctrl.channel}/capabilities", timeout=3.0)
if not resp.ok():
return None
try:
import xml.etree.ElementTree as ET
root = ET.fromstring(resp.xml)
space = None
for elem in root.iter():
if elem.tag.split("}")[-1] == "AbsolutePanTiltPositionSpace":
space = elem
break
if space is None:
return None
def _rng(name: str, dft_min: int, dft_max: int) -> tuple[int, int]:
node = None
for child in space:
if child.tag.split("}")[-1] == name:
node = child
break
if node is None:
return dft_min, dft_max
vals = {
sub.tag.split("}")[-1]: (sub.text or "").strip()
for sub in node
}
try:
lo = int(vals.get("Min", dft_min))
except ValueError:
lo = dft_min
try:
hi = int(vals.get("Max", dft_max))
except ValueError:
hi = dft_max
return lo, hi
pan_min, pan_max = _rng("XRange", 0, 3600)
tilt_min, tilt_max = _rng("YRange", 0, 900)
except Exception as e:
LOG.log("warning", f"解析 PTZ capabilities 限位失败: {e}")
return None
limits = {
"pan_min": pan_min, "pan_max": pan_max,
"tilt_min": tilt_min, "tilt_max": tilt_max,
"pan_has_limit": True, # 默认有限位
"pan_is_infinite": False,
}
self._ptz_limits[ip] = limits
return limits
def _check_at_limit(self, ip: str, pan: int, tilt: int) -> str:
"""v1.1.8: 用 DeviceReader 缓存位置判定请求方向是否已顶限位。
v1.1.50: P 轴无限位时跳过 P 轴检查 — 无限位设备可穿越 0/360 边界，
不存在"最小值/最大值"概念。
设备顶限位时 continuous 仍回 HTTP 200 + statusCode=1，云台不动，
用户感知为按键失效；此处提前拦截并明确报错（不新增每命令 IO）。
符号关系经真机实测：pan 为正使 azimuth 减小（趋向 pan_min），
tilt 为正使 elevation 增大（趋向 tilt_max）。
"""
if pan == 0 and tilt == 0:
return ""
limits = self._get_ptz_limits(ip)
if not limits:
return ""
try:
from src.core.device_reader import get_device_reader
reader = get_device_reader()
pos = reader.get_position() if reader else None
if not pos:
return ""
cur_pan = float(pos.get("pan", 0))
cur_tilt = float(pos.get("tilt", 0))
except Exception:
return ""
margin = 5 # 0.5° 容差，避免编码器抖动误判
# v1.1.50: P 轴无限位时跳过 P 轴检查
pan_is_infinite = limits.get("pan_is_infinite", False)
if not pan_is_infinite:
if pan > 0 and cur_pan <= limits["pan_min"] + margin:
return "已达 P 轴限位，命令被设备忽略"
if pan < 0 and cur_pan >= limits["pan_max"] - margin:
return "已达 P 轴限位，命令被设备忽略"
# T 轴总是检查（T 轴通常有限位）
if tilt > 0 and cur_tilt >= limits["tilt_max"] - margin:
return "已达 T 轴限位，命令被设备忽略"
if tilt < 0 and cur_tilt <= limits["tilt_min"] + margin:
return "已达 T 轴限位，命令被设备忽略"
return ""
def _adjust_for_limit(self, ip: str, pan: int, tilt: int) -> tuple[int, int, str]:
"""v1.1.25: 限位方向感知——仅清零"朝向限位"的轴分量，放行逃离方向。
v1.1.50: P 轴无限位时跳过 P 轴修正 — 无限位设备可穿越 0/360 边界。
v1.1.8 全向拦截在失控撞限位后把用户"锁死"（方向键全无效、
只有 HOME 预置点召回能救）；改为轴向修正后逃离方向仍可用。
返回: (调整后 pan, 调整后 tilt, 消息)；消息前缀 ADJUSTED: 表示
部分轴被修正（命令仍发送），否则非空=全轴拦截（不发送）。
"""
if pan == 0 and tilt == 0:
return pan, tilt, ""
limits = self._get_ptz_limits(ip)
if not limits:
return pan, tilt, ""
try:
from src.core.device_reader import get_device_reader
reader = get_device_reader()
pos = reader.get_position() if reader else None
if not pos:
return pan, tilt, ""
cur_pan = float(pos.get("pan", 0))
cur_tilt = float(pos.get("tilt", 0))
except Exception:
return pan, tilt, ""
margin = 5
hit: list[str] = []
# v1.1.50: P 轴无限位时跳过 P 轴修正
pan_is_infinite = limits.get("pan_is_infinite", False)
if not pan_is_infinite:
if pan > 0 and cur_pan <= limits["pan_min"] + margin:
pan, hit = 0, hit + ["P 轴最小限位"]
elif pan < 0 and cur_pan >= limits["pan_max"] - margin:
pan, hit = 0, hit + ["P 轴最大限位"]
# T 轴总是检查
if tilt > 0 and cur_tilt >= limits["tilt_max"] - margin:
tilt, hit = 0, hit + ["T 轴最大限位"]
elif tilt < 0 and cur_tilt <= limits["tilt_min"] + margin:
tilt, hit = 0, hit + ["T 轴最小限位"]
if not hit:
return pan, tilt, ""
if pan == 0 and tilt == 0:
return 0, 0, "已达 " + "/".join(hit) + "，命令被设备忽略"
return pan, tilt, "ADJUSTED:已达 " + "/".join(hit) + "，已自动修正为可动方向"
def current_cmd_gen(self) -> int:
"""v1.1.25: 当前命令 generation（move 入队时捕获）。"""
with self._cmd_gen_lock:
return self._cmd_gen
def bump_cmd_gen(self) -> int:
"""v1.1.25: stop 时递增 generation，使在途/在队 move 失效。"""
with self._cmd_gen_lock:
self._cmd_gen += 1
return self._cmd_gen
def _get_controller(self, ip: str) -> tuple[ISAPIPTZController | None, str]:
"""v10.05: 获取 ISAPIPTZController。v10.05: 移除自动重连，仅返回已连接的控制器。
连接只能手动触发（connect_device），已连接的设备保持连接状态。
"""
# 已连接，直接返回
if ip in self._controllers:
return self._controllers[ip], ""
# v10.05: 不再自动重连，直接返回错误
return None, "设备未连接（请先在设备管理页面点击连接）"
def ptz_move(self, ip: str, direction: str, speed: int = 50,
cmd_gen: int | None = None) -> dict:
"""PTZ 移动控制。
Args:
ip: 设备 IP
direction: up|down|left|right|up-left|up-right|down-left|down-right|zoom-in|zoom-out
speed: 1-100
cmd_gen: v1.1.25 generation token——发送前与当前 gen 比对，
落后（期间有 stop）则放弃发送，根治 stop 反超 move。
"""
import time as _time
_t0 = _time.perf_counter()
# v1.1.25: 废除 v10.59 入口无条件 cancel_event.clear()——它会抹掉
# 早到 stop 的取消语义（TOCTOU）；改用 generation token 判定
# v1.1.25: 入口 stale 检查（廉价前置过滤）
if cmd_gen is not None and cmd_gen != self.current_cmd_gen():
LOG.log("info", f"PTZ 移动已跳过（gen 过期，stop 已到达）: device={ip} direction={direction}")
return {"success": False, "message": "命令已过期（stop 先到达）", "stale": True}
_t_get_ctrl = _time.perf_counter()
ctrl, err = self._get_controller(ip)
_get_ctrl_ms = (_time.perf_counter() - _t_get_ctrl) * 1000
if err:
LOG.log("failed", f"PTZ 移动失败: device={ip} reason={err}")
return {"success": False, "message": err}
# Zoom operations - 官方档位映射
# 档位映射: 1→15, 2→30, 3→45, 4→60, 5→75, 6→90, 7→100
if 1 <= speed <= 7:
speed_mapping = {1: 15, 2: 30, 3: 45, 4: 60, 5: 75, 6: 90, 7: 100}
isapi_speed = speed_mapping.get(speed, 60)
else:
isapi_speed = min(max(speed, 0), 100)
if direction == "zoom-in":
success = ctrl.continuous_move(pan=0, tilt=0, zoom=isapi_speed)
# v11.30: 恢复移动后钩子（升频窗口 + 即时查询推送），补偿事件推送间隔
self._after_move_hook(ip, success)
return {"success": success, "message": f"PTZ zoom-in (档位={speed})", "isapi_speed": isapi_speed}
elif direction == "zoom-out":
success = ctrl.continuous_move(pan=0, tilt=0, zoom=-isapi_speed)
self._after_move_hook(ip, success)
return {"success": success, "message": f"PTZ zoom-out (档位={speed})", "isapi_speed": isapi_speed}
elif direction == "focus-near":
success = ctrl.focus_move_continuous('near', speed)
return {"success": success, "message": f"PTZ focus-near (speed={speed})"}
elif direction == "focus-far":
success = ctrl.focus_move_continuous('far', speed)
return {"success": success, "message": f"PTZ focus-far (speed={speed})"}
elif direction == "focus-auto":
success = ctrl.set_focus_mode("auto")
return {"success": success, "message": "PTZ focus-auto"}
if direction not in self.DIRECTION_MAP:
LOG.log("warning", f"PTZ 移动: 未知方向 {direction}")
return {"success": False, "message": f"未知方向: {direction}"}
# Pan/Tilt operations - 官方档位映射
# 官方 web 端: 1-7 档，默认 4 档
# ISAPI 速度值: 档位7=ISAPI 100 (最快)
# 档位映射: 1→15, 2→30, 3→45, 4→60, 5→75, 6→90, 7→100
pan_val, tilt_val = self.DIRECTION_MAP[direction]
if 1 <= speed <= 7:
speed_mapping = {1: 15, 2: 30, 3: 45, 4: 60, 5: 75, 6: 90, 7: 100}
isapi_speed = speed_mapping.get(speed, 60)
else:
isapi_speed = min(max(speed, 0), 100)
pan = int(pan_val * isapi_speed)
tilt = int(tilt_val * isapi_speed)
# v1.1.25: 限位方向感知——仅拦截朝向限位轴分量，放行逃离方向
pan, tilt, limit_msg = self._adjust_for_limit(ip, pan, tilt)
adjusted = limit_msg.startswith("ADJUSTED:")
if limit_msg and not adjusted:
LOG.log("warning", f"PTZ 移动被限位拦截: device={ip} direction={direction} ({limit_msg})")
return {"success": False, "message": limit_msg, "at_limit": True}
if pan == 0 and tilt == 0:
return {"success": True, "message": "无有效移动分量", "at_limit_adjusted": adjusted}
# v1.1.25: 发送前 gen 复核（锁外入队→此处可能已经过 stop）
if cmd_gen is not None and cmd_gen != self.current_cmd_gen():
LOG.log("info", f"PTZ 移动已跳过（gen 过期，stop 已到达）: device={ip} direction={direction}")
return {"success": False, "message": "命令已过期（stop 先到达）", "stale": True}
# v1.1.26: 发送前日志
LOG.log("info", f"PTZ 移动发送: device={ip} pan={pan} tilt={tilt} isapi_speed={isapi_speed}")
success = ctrl.continuous_move(pan=pan, tilt=tilt)
_total_ms = (_time.perf_counter() - _t0) * 1000
if success:
LOG.log("done", f"PTZ 移动成功: device={ip} direction={direction} speed={speed}(档位)→isapi={isapi_speed}")
# v11.30: 恢复移动后钩子（升频窗口 + 即时查询推送）
self._after_move_hook(ip, success)
else:
LOG.log("failed", f"PTZ 移动失败: device={ip} direction={direction}")
# v10.54-diag: 控制器层分段计时
LOG.log("info", f"PTZ_CTRL TIMING: get_controller={_get_ctrl_ms:.1f}ms total={_total_ms:.1f}ms dir={direction}")
try:
from src.diagnostics.latency_tracker import get_latency_tracker
get_latency_tracker().record(
"controller_move",
get_controller_ms=_get_ctrl_ms,
total_ms=_total_ms,
direction=direction,
)
except ImportError:
pass
return {
"success": success,
"message": f"PTZ 移动: {direction} (档位={speed})",
"direction": direction,
"speed": speed,
"isapi_speed": isapi_speed,
"at_limit_adjusted": adjusted,
}
def ptz_home(self, ip: str) -> dict:
ctrl, err = self._get_controller(ip)
if err:
return {"success": False, "message": err}
success = ctrl.goto_preset(10) # 直接调用 goto_preset(10)
self._after_move_hook(ip, success)
return {
"success": success,
"message": "归位成功" if success else "归位失败",
}
def ptz_stop(self, ip: str) -> dict:
import time as _time
_t0 = _time.perf_counter()
# v1.1.25: 递增 generation——使"已入队未发送"的 move 在发送前自废；
# 废除 v10.59 cancel_event.set()（其语义已被 gen 取代且存在 TOCTOU）
self.bump_cmd_gen()
_t_get_ctrl = _time.perf_counter()
ctrl, err = self._get_controller(ip)
_get_ctrl_ms = (_time.perf_counter() - _t_get_ctrl) * 1000
if err:
return {"success": False, "message": err}
success = ctrl.stop_move()
# v1.1.26: stop 失败时二次确认
if not success:
LOG.log("error", f"PTZ stop 首次失败，尝试二次停止: device={ip}")
success = ctrl.continuous_move(pan=0, tilt=0, zoom=0, focus=0)
if success:
LOG.log("info", f"PTZ 二次停止成功: device={ip}")
else:
LOG.log("error", f"PTZ 二次停止也失败: device={ip}")
_total_ms = (_time.perf_counter() - _t0) * 1000
# v11.30: stop 成功后升频轮询，让停止后的最终位置即时推送
self._after_move_hook(ip, success)
# v10.54-diag: 控制器层分段计时
LOG.log("info", f"PTZ_STOP TIMING: get_controller={_get_ctrl_ms:.1f}ms total={_total_ms:.1f}ms")
try:
from src.diagnostics.latency_tracker import get_latency_tracker
get_latency_tracker().record(
"controller_stop",
get_controller_ms=_get_ctrl_ms,
total_ms=_total_ms,
)
except ImportError:
pass
return {
"success": success,
"message": "停止成功" if success else "停止失败",
}
def ptz_preset(self, ip: str, preset_id: int) -> dict:
ctrl, err = self._get_controller(ip)
if err:
return {"success": False, "message": err}
success = ctrl.goto_preset(preset_id)
return {
"success": success,
"message": f"预置位 {preset_id} {'成功' if success else '失败'}",
"preset_id": preset_id,
}
def set_preset(self, ip: str, preset_id: int, name: str = "") -> dict:
"""设置当前位置为预置点。
路由: POST /api/v1/ptz/{device_id}/preset/{preset_id}/set
"""
ctrl, err = self._get_controller(ip)
if err:
return {"success": False, "message": err}
success = ctrl.set_preset(preset_id, name=name)
return {
"success": success,
"message": f"预置位 {preset_id} 设置{'成功' if success else '失败'}",
"preset_id": preset_id,
}
def ptz_list_presets(self, ip: str) -> dict:
"""获取设备预置点列表。
路由: GET /api/v1/ptz/{device_id}/presets
"""
ctrl, err = self._get_controller(ip)
if err:
return {"success": False, "message": err}
presets = ctrl.list_presets()
return {
"success": True,
"presets": presets,
"count": len(presets),
}
def ptz_absolute(self, ip: str, pan: float, tilt: float,
zoom: float | None = None, speed: int = 50) -> dict:
ctrl, err = self._get_controller(ip)
if err:
return {"success": False, "message": err}
success = ctrl.absolute_move(pan=pan, tilt=tilt, zoom=zoom, speed=speed)
self._after_move_hook(ip, success)
return {
"success": success,
"message": "绝对移动成功" if success else "绝对移动失败",
}
def ptz_get_position(self, ip: str) -> dict:
ctrl, err = self._get_controller(ip)
if err:
return {"success": False, "message": err, "data": {}}
pos = ctrl.get_position()
return {
"success": bool(pos),
"message": "获取位置成功" if pos else "获取位置失败",
"data": pos,
}
def _after_move_hook(self, ip: str, success: bool) -> None:
"""v11.30: 移动/停止成功后钩子。
通知 DeviceReader 进入 1s 升频窗口，并立即查询一次位置触发推送，
作为设备事件推送的补偿（事件订阅不可用时仍能保证位置刷新）。
"""
if not success:
return
try:
from src.core.device_reader import get_device_reader
reader = get_device_reader()
if reader:
reader.notify_move_window()
self._query_position_after_move(ip)
except Exception as e:
LOG.log("warning", f"移动后位置推送失败: {e}")
def _query_position_after_move(self, ip: str) -> None:
"""
v10.04: 移动后主动查询位置，通过回调通知 DeviceReader 更新缓存
Args:
ip: 设备 IP
"""
if not self._on_position_update:
return
try:
ctrl, err = self._get_controller(ip)
if err or not ctrl:
return
pos = ctrl.get_position()
if pos:
self._on_position_update(pos)
except Exception as e:
LOG.log("warning", f"移动后位置查询失败: {e}")
# ================================================================
# 录像管理（ISAPI 原生方法）
#
# 架构说明：
# 之前的 /ContentMgmt/Storage/control 端点返回 404，不适用于 iDS-2DF8C832IXS-A 球机。
# 正确使用:
# PUT /ISAPI/ContentMgmt/record/control/manual/start/tracks/{trackID} — 启动录像
# PUT /ISAPI/ContentMgmt/record/control/manual/stop/tracks/{trackID} — 停止录像
# POST /ISAPI/ContentMgmt/search — 搜索设备端录像
# POST /ISAPI/ContentMgmt/download — 下载录像到本地
# Track ID = channel * 100 + 1（101=通道1主码流）
# 设备需要配置存储介质（NAS/SD卡），无内置硬盘时 hddList 为空但仍可录像到 NAS。
# ================================================================
def _get_isapi_client(self, ip: str) -> ISAPIClient | None:
"""获取已连接的 ISAPIClient，或从凭据创建临时实例。"""
if ip in self._clients:
return self._clients[ip]
creds = self.get_credentials(ip)
if not creds:
return None
port = int(creds.get("port", 80))
username = str(creds.get("username", "admin"))
password = str(creds.get("password", ""))
client = ISAPIClient(ip=ip, username=username, password=password, port=port)
if client.verify_credentials():
self._clients[ip] = client
return client
return None
# ================================================================
# 录像管理（ISAPI 官方方法）
#
# 海康官方 ISAPI 录制接口:
# PUT /ISAPI/ContentMgmt/record/control/manual/start/tracks/{trackID} — 启动录像
# PUT /ISAPI/ContentMgmt/record/control/manual/stop/tracks/{trackID} — 停止录像
# Track ID = channel * 100 + 1（101=通道1主码流）
# ================================================================
def start_recording(self, ip: str, channel: int = ISAPI_CHANNEL, target_name: str = "") -> dict:
"""v7.107: 通过 ISAPI 官方方法启动摄像头录像."""
client = self._get_isapi_client(ip)
if not client:
return {"success": False, "message": f"设备 {ip} 未连接或凭据无效"}
track_id = channel * 100 + 1
endpoint = f"/ContentMgmt/record/control/manual/start/tracks/{track_id}"
result = client.put(endpoint, "")
if result.ok():
LOG.log("done", f"ISAPI 录像已启动: {ip} track={track_id}")
return {"success": True, "message": f"录像已启动: {ip}", "data": {"track_id": track_id}}
else:
LOG.log("failed", f"ISAPI 启动录像失败: {ip} status={result.status_code}")
return {"success": False, "message": f"启动录像失败: {result.error_string or result.status_code}"}
def stop_recording(self, ip: str, channel: int = ISAPI_CHANNEL) -> dict:
"""v7.107: 通过 ISAPI 官方方法停止摄像头录像."""
client = self._get_isapi_client(ip)
if not client:
return {"success": False, "message": f"设备 {ip} 未连接或凭据无效"}
track_id = channel * 100 + 1
endpoint = f"/ContentMgmt/record/control/manual/stop/tracks/{track_id}"
result = client.put(endpoint, "")
if result.ok():
LOG.log("done", f"ISAPI 录像已停止: {ip} track={track_id}")
return {"success": True, "message": f"录像已停止: {ip}", "data": {"track_id": track_id}}
else:
LOG.log("failed", f"ISAPI 停止录像失败: {ip} status={result.status_code}")
return {"success": False, "message": f"停止录像失败: {result.error_string or result.status_code}"}
def update_device_capabilities(self, mac: str, capabilities: dict) -> bool:
"""更新设备能力信息到配置文件。
Args:
mac: 设备 MAC 地址（如传入 IP 则当作 MAC key 使用）
capabilities: 能力字典，如 {"gain": {"supported": True, "min_val": 1, ...}, ...}
"""
try:
norm_mac = _normalize_mac(mac)
config = self.config.load_ptz_config()
devices = config.setdefault("devices", {})
if norm_mac not in devices:
LOG.log("warning", f"更新 capabilities: 设备 {norm_mac} 未找到，跳过")
return False
existing_caps = devices[norm_mac].setdefault("capabilities", {})
existing_caps.update(capabilities)
self.config.save_ptz_config(config)
LOG.log("done", f"设备 {norm_mac} capabilities 已更新: {len(capabilities)} 项")
return True
except Exception as e:
LOG.log("error", f"更新 capabilities 失败: {e}")
return False
def update_device_limits(self, mac: str, limits: dict) -> bool:
"""更新设备限位信息到配置文件。
Args:
mac: 设备 MAC 地址
limits: 限位字典，如 {"pan": {"has_limit": False, ...}, "tilt": {"has_flip": True, ...}, ...}
"""
try:
norm_mac = _normalize_mac(mac)
config = self.config.load_ptz_config()
devices = config.setdefault("devices", {})
if norm_mac not in devices:
LOG.log("warning", f"更新 limits: 设备 {norm_mac} 未找到，跳过")
return False
existing_limits = devices[norm_mac].setdefault("limits", {})
existing_limits.update(limits)
self.config.save_ptz_config(config)
LOG.log("done", f"设备 {norm_mac} limits 已更新: {list(limits.keys())}")
return True
except Exception as e:
LOG.log("error", f"更新 limits 失败: {e}")
return False
def update_device_speed(self, mac: str, speed_data: dict) -> bool:
"""更新设备速度测试数据到配置文件。
Args:
mac: 设备 MAC 地址
speed_data: 速度数据字典，如 {"pan_1": {"displacement": 12.3}, ...}
"""
try:
norm_mac = _normalize_mac(mac)
config = self.config.load_ptz_config()
devices = config.setdefault("devices", {})
if norm_mac not in devices:
LOG.log("warning", f"更新 speed: 设备 {norm_mac} 未找到，跳过")
return False
existing_speed = devices[norm_mac].setdefault("speed", {})
existing_speed.update(speed_data)
self.config.save_ptz_config(config)
LOG.log("done", f"设备 {norm_mac} speed 数据已更新: {len(speed_data)} 项")
return True
except Exception as e:
LOG.log("error", f"更新 speed 数据失败: {e}")
return False
def update_device_home(self, mac: str, home_coords: dict, verified: bool = False) -> bool:
"""更新设备 HOME 位坐标到配置文件。
Args:
mac: 设备 MAC 地址
home_coords: HOME 位坐标，如 {"pan": 1800, "tilt": 450, "zoom": 10}
verified: 是否已验证
"""
try:
norm_mac = _normalize_mac(mac)
config = self.config.load_ptz_config()
devices = config.setdefault("devices", {})
if norm_mac not in devices:
LOG.log("warning", f"更新 home: 设备 {norm_mac} 未找到，跳过")
return False
devices[norm_mac]["home"] = {
"pan": home_coords.get("pan"),
"tilt": home_coords.get("tilt"),
"zoom": home_coords.get("zoom"),
"verified": verified,
}
self.config.save_ptz_config(config)
LOG.log("done", f"设备 {norm_mac} home 位已更新: {home_coords}")
return True
except Exception as e:
LOG.log("error", f"更新 home 位失败: {e}")
return False
def auto_write_advanced_results(self, mac: str, func_results: dict | None = None,
limit_results: dict | None = None,
speed_results: dict | None = None,
home_verified: bool = False) -> dict:
"""自动写入高级功能探测/限位/速度测试结果。
在 function/limit/speed 测试完成后调用，一次性写入所有结果到配置文件。
Args:
mac: 设备 MAC 地址
func_results: function.py detect_all() 结果
limit_results: limit.py run_all_tests() 结果
speed_results: speed.py run_all_tests() 结果
home_verified: HOME 位是否已验证
"""
results = {"capabilities_updated": False, "limits_updated": False,
"speed_updated": False, "home_updated": False}
# 从 FunctionDetector 提取 capabilities
if func_results:
capabilities = {}
for item_key, item_result in func_results.items():
endpoint_def = {}
if isinstance(item_result, dict):
endpoint_def = {
"supported": item_result.get("supported", False),
"min_val": item_result.get("min_val", 0),
"max_val": item_result.get("max_val", 0),
"endpoint": item_result.get("endpoint", ""),
"test_key": item_result.get("test_key", ""),
}
capabilities[item_key] = endpoint_def
if capabilities:
results["capabilities_updated"] = self.update_device_capabilities(mac, capabilities)
# 从 LimitTester 提取 limits
if limit_results:
limits = {}
pan_min = limit_results.get("pan_min", 0)
pan_max = limit_results.get("pan_max", 3600)
tilt_min = limit_results.get("tilt_min", -200)
tilt_max = limit_results.get("tilt_max", 900)
has_flip = limit_results.get("has_flip", False)
limits["pan"] = {
"has_limit": False, # Pan 360° 无绝对限位
"observed_min": pan_min,
"observed_max": pan_max,
}
limits["tilt"] = {
"has_flip": has_flip,
"observed_min": tilt_min,
"observed_max": tilt_max,
}
limits["zoom"] = {
"observed_min": 0,
"observed_max": 320,
}
results["limits_updated"] = self.update_device_limits(mac, limits)
# 从 SpeedTester 提取 speed
if speed_results:
speed_data = {}
results_map = speed_results.get("results", {})
for speed_key, speed_val in results_map.items():
if isinstance(speed_val, dict):
speed_data[f"pan_{speed_key}"] = {
"set_speed": speed_val.get("set", 0),
"actual_speed": speed_val.get("actual", 0),
"displacement": speed_val.get("displacement", 0),
"delta": speed_val.get("delta", 0),
}
if speed_data:
results["speed_updated"] = self.update_device_speed(mac, speed_data)
# 更新 HOME 位
results["home_updated"] = self.update_device_home(mac, HOME_COORDS, verified=home_verified)
LOG.log("done", f"设备 {mac} advanced results 自动写入: {results}")
return results