"""v9.75: 点击指向 (Click Point) 核心模块 功能: - 像素→角度转换 (TAN/gnomonic 投影) - 1~5点递进式校准建模 - 状态管理 + 数据持久化 数据路径: data/devices/{mac}/CLICK_POINT.JSON """ import json import math import time from src.advanced.device_path import get_click_point_path, get_ptz_models_path, normalize_mac # 校准点间最小间距 (度) MIN_CALIBRATION_DISTANCE_DEG = 45.0 # ============================================================ # 数学: TAN/gnomonic 投影 — 像素→角度 # ============================================================ def pixel_to_angle( px: float, py: float, img_w: int, img_h: int, fov_h: float, fov_v: float ) -> tuple[float, float]: """像素坐标 → 角度偏移 (相对于画面中心)。 使用 TAN/gnomonic 投影 (针孔模型)，禁止线性比例。 Args: px, py: 像素坐标 img_w, img_h: 图片原始分辨率 fov_h, fov_v: 水平/垂直视场角 (度) Returns: (delta_az, delta_el): 方位角/仰角偏移量 (度) """ # 归一化到 [-1, 1]，中心为 (0, 0) nx = (px - img_w / 2) / (img_w / 2) ny = (py - img_h / 2) / (img_h / 2) # TAN 投影: angle = atan(normalized * tan(fov/2)) half_h = math.radians(fov_h / 2) half_v = math.radians(fov_v / 2) delta_az = math.degrees(math.atan(nx * math.tan(half_h))) # 屏幕 Y 向下为正 → 物理 elevation 向上为正，取反 delta_el = -math.degrees(math.atan(ny * math.tan(half_v))) return delta_az, delta_el # ============================================================ # 校准建模 # ============================================================ class ClickPointModel: """点击指向校准模型。 递进式: - 0点: 无校准 (粗模型) - 1点: 偏置校正 - 2点: 偏置 + 各向同性缩放 - 3~5点: 仿射最小二乘 """ def __init__(self): self.points: list[dict] = [] # [{px, py, pan, tilt, img_w, img_h}, ...] self.model_type: str = "none" # none / offset / scale / affine self.params: dict = {} # 模型参数 self.updated_at: float = 0 self._pending_click: dict | None = None # 待应用的点击数据 def add_point(self, px: float, py: float, pan: float, tilt: float, img_w: int, img_h: int) -> int: """添加校准点，返回当前点数。最多5个。""" if len(self.points) >= 5: return len(self.points) self.points.append({ "px": px, "py": py, "pan": pan, "tilt": tilt, "img_w": img_w, "img_h": img_h, }) self.rebuild() return len(self.points) def rebuild(self): """根据当前校准点重建模型。""" n = len(self.points) if n == 0: self.model_type = "none" self.params = {} elif n == 1: self._build_offset() elif n == 2: self._build_scale() else: self._build_affine() self.updated_at = time.time() def _build_offset(self): """1点校准：无法学习缩放，仅记录 scale=1（1点校准意义有限）。""" # v1.1.50: 1点校准无法区分"偏差"和"缩放"，保持 scale=1 # 预测时使用 current_pan + d_az（与无校准相同） self.model_type = "offset" self.params = { "scale_az": 1.0, "scale_el": 1.0, } def _build_scale(self): """2点校准：学习各向同性缩放因子。 v1.1.50: 核心逻辑修正 — 校准模型学习的是"像素→相对角度偏移"的缩放， 而不是"像素→绝对PTZ位置"的映射。 缩放因子 = 实际PTZ变化 / 理论像素角度变化 预测时：target = current_pan + d_az * scale_az """ p0, p1 = self.points[0], self.points[1] fov_h, fov_v = _get_fov(p0["img_w"], p0["img_h"]) d_az0, d_el0 = pixel_to_angle(p0["px"], p0["py"], p0["img_w"], p0["img_h"], fov_h, fov_v) d_az1, d_el1 = pixel_to_angle(p1["px"], p1["py"], p1["img_w"], p1["img_h"], fov_h, fov_v) # 像素差 → 角度差（理论值） pixel_daz = d_az1 - d_az0 pixel_del = d_el1 - d_el0 # PTZ 实际差（测量值） ptz_daz = p1["pan"] - p0["pan"] ptz_del = p1["tilt"] - p0["tilt"] # 缩放因子 = 实际PTZ变化 / 理论像素角度变化 scale_az = ptz_daz / pixel_daz if abs(pixel_daz) > 0.001 else 1.0 scale_el = ptz_del / pixel_del if abs(pixel_del) > 0.001 else 1.0 self.model_type = "scale" self.params = { "scale_az": scale_az, "scale_el": scale_el, } def _build_affine(self): """3~5点校准：仿射最小二乘学习缩放因子。 v1.1.50: 核心逻辑修正 — 学习"像素→相对角度偏移"的缩放， 预测时基于 current_pan/tilt。 """ n = len(self.points) fov_h, fov_v = _get_fov(self.points[0]["img_w"], self.points[0]["img_h"]) # 收集所有点的 (d_az, d_el) 和 (pan, tilt) d_az_list = [] d_el_list = [] pan_list = [] tilt_list = [] for p in self.points: d_az, d_el = pixel_to_angle(p["px"], p["py"], p["img_w"], p["img_h"], fov_h, fov_v) d_az_list.append(d_az) d_el_list.append(d_el) pan_list.append(p["pan"]) tilt_list.append(p["tilt"]) # 计算相对变化（以第一个点为基准） # scale_az = sum((pan_i - pan_0) * (d_az_i - d_az_0)) / sum((d_az_i - d_az_0)^2) base_d_az = d_az_list[0] base_d_el = d_el_list[0] base_pan = pan_list[0] base_tilt = tilt_list[0] sum_daz_diff2 = 0.0 sum_pan_daz = 0.0 sum_del_diff2 = 0.0 sum_tilt_del = 0.0 for i in range(1, n): daz_diff = d_az_list[i] - base_d_az del_diff = d_el_list[i] - base_d_el pan_diff = pan_list[i] - base_pan tilt_diff = tilt_list[i] - base_tilt sum_daz_diff2 += daz_diff * daz_diff sum_pan_daz += pan_diff * daz_diff sum_del_diff2 += del_diff * del_diff sum_tilt_del += tilt_diff * del_diff scale_az = sum_pan_daz / sum_daz_diff2 if sum_daz_diff2 > 0.001 else 1.0 scale_el = sum_tilt_del / sum_del_diff2 if sum_del_diff2 > 0.001 else 1.0 self.model_type = "affine" self.params = { "scale_az": scale_az, "scale_el": scale_el, } def predict(self, px: float, py: float, img_w: int, img_h: int, current_pan: float = 0, current_tilt: float = 0) -> tuple[float, float]: """预测: 像素坐标 → PTZ 目标位置。 v1.1.50: 核心逻辑修正 — 校准模型学习的是"像素→相对角度偏移"的缩放， 预测时基于当前 PTZ 位置加上（可能缩放的）角度偏移。 Args: px, py: 像素坐标 img_w, img_h: 图片分辨率 current_pan, current_tilt: 当前PTZ位置 (度) Returns: (target_pan, target_tilt) 度 """ fov_h, fov_v = _get_fov(img_w, img_h) d_az, d_el = pixel_to_angle(px, py, img_w, img_h, fov_h, fov_v) if self.model_type == "none": # 无校准：直接使用理论偏移 return current_pan + d_az, current_tilt + d_el # 有校准：使用学习的缩放因子 scale_az = self.params.get("scale_az", 1.0) scale_el = self.params.get("scale_el", 1.0) return ( current_pan + d_az * scale_az, current_tilt + d_el * scale_el, ) def reset(self): """重置所有校准数据。""" self.points = [] self.model_type = "none" self.params = {} self.updated_at = 0 # ============================================================ # v11.12: 从 router.py 迁入的业务逻辑 # ============================================================ def execute_move( self, client, ctrl, device_id: str, x: float, y: float, pos: dict | None = None, ) -> dict: """点击屏幕点 → PTZ 绝对值移动（同步方法，由调用方 asyncio.to_thread 包装）。 流程: ISAPI截图 → 比例坐标→像素 → 校准预测 → 限位保护 → absoluteEx 移动。 Args: client: ISAPIClient 实例 ctrl: ISAPIPTZController 实例 device_id: 设备标识 (MAC 或 IP) x, y: 比例坐标 (0~1) pos: v11.30: DeviceReader 缓存位置（设计约束：位置只从后端缓存读）； 为 None 时回退实时 ISAPI 查询 Returns: dict: 移动结果 """ import cv2 import numpy as np from src.advanced.device_path import get_data_path_read # 1. ISAPI 截图 jpg = client.capture_picture() if not jpg: return {"success": False, "message": "ISAPI截图失败"} bgr = cv2.imdecode(np.frombuffer(jpg, np.uint8), cv2.IMREAD_COLOR) if bgr is None: return {"success": False, "message": "图像解码失败"} img_h, img_w = bgr.shape[:2] # 2. 比例坐标 → 像素坐标 px = x * img_w py = y * img_h # 3. 当前 PTZ 位置（v11.30: 优先用缓存，减少一次 ISAPI 占槽） if not pos: pos = ctrl.get_position() if not pos: return {"success": False, "message": "获取PTZ位置失败"} current_pan = pos.get("pan", 0) # 单位: 0.1度 current_tilt = pos.get("tilt", 0) # 4. 校准模型预测 target_pan, target_tilt = self.predict( px, py, img_w, img_h, current_pan=current_pan / 10, current_tilt=current_tilt / 10, ) # 5. 限位保护: 读取 limit.json 裁剪目标范围 mac_clean = normalize_mac(device_id) limit_path = get_data_path_read(None, mac_clean, "limit") if limit_path and limit_path.exists(): try: limit_data = json.loads(limit_path.read_text(encoding="utf-8")) tilt_min_deg = limit_data.get("tilt_min", -200) / 10.0 tilt_max_deg = limit_data.get("tilt_max", 900) / 10.0 target_tilt = max(tilt_min_deg, min(tilt_max_deg, target_tilt)) if limit_data.get("pan_has_limit", False): pan_min_deg = limit_data.get("pan_min", 0) / 10.0 pan_max_deg = limit_data.get("pan_max", 3600) / 10.0 target_pan = max(pan_min_deg, min(pan_max_deg, target_pan)) except Exception: pass # 6. absolute_ex 移动 success = ctrl.absolute_ex(pan=target_pan, tilt=target_tilt) # 保存待应用数据（供 apply 使用） self._pending_click = { "px": px, "py": py, "img_w": img_w, "img_h": img_h, "current_pan": current_pan / 10, "current_tilt": current_tilt / 10, "target_pan": target_pan, "target_tilt": target_tilt, } return { "success": success, "message": "点击指向完成" if success else "指向失败", "data": { "pixel_x": round(px, 1), "pixel_y": round(py, 1), "img_w": img_w, "img_h": img_h, "target_pan": round(target_pan, 4), "target_tilt": round(target_tilt, 4), "model_type": self.model_type, }, } def execute_apply( self, ctrl, point_index: int, pos: dict | None = None, ) -> dict: """应用校准点 — 记录当前 PTZ 位置 + 重建模型（同步方法）。 Args: ctrl: ISAPIPTZController 实例 point_index: 校准点索引 (0~4) pos: v11.30: DeviceReader 缓存位置（位置只从后端缓存读）； 为 None 时回退实时 ISAPI 查询 Returns: dict: 应用结果 """ # 获取待应用的点击数据 pending = self._pending_click if not pending: return {"success": False, "message": "无待应用的点击数据,请先点击画面"} # 当前 PTZ 位置 (用户已手动对准；v11.30: 优先用缓存) if not pos: pos = ctrl.get_position() if not pos: return {"success": False, "message": "获取PTZ位置失败"} actual_pan = pos.get("pan", 0) / 10 # 0.1度 → 度 actual_tilt = pos.get("tilt", 0) / 10 # 后端距离校验 (>=45度) if self.points: last_pt = self.points[-1] dist_deg = ((actual_pan - last_pt["pan"]) ** 2 + (actual_tilt - last_pt["tilt"]) ** 2) ** 0.5 if dist_deg < MIN_CALIBRATION_DISTANCE_DEG: return { "success": False, "message": f"距离不足: {dist_deg:.1f}° < {MIN_CALIBRATION_DISTANCE_DEG:.0f}°，请选择更远位置", } # 添加校准点 point_count = self.add_point( px=pending["px"], py=pending["py"], pan=actual_pan, tilt=actual_tilt, img_w=pending["img_w"], img_h=pending["img_h"], ) # 清除待应用数据 self._pending_click = None # 计算预期值 (用于前端显示) current_zoom = pos.get("zoom", 1) fov_h, fov_v = _get_fov(pending["img_w"], pending["img_h"], zoom=int(current_zoom)) expected_pan, expected_tilt = self.predict( pending["px"], pending["py"], pending["img_w"], pending["img_h"], current_pan=actual_pan, current_tilt=actual_tilt, ) # 计算误差 (欧氏距离) error_deg = ((actual_pan - expected_pan) ** 2 + (actual_tilt - expected_tilt) ** 2) ** 0.5 # v1.1.12: 预测误差超过 10° 拒绝应用该点并回滚 if error_deg > 10.0: self.points.pop() self.rebuild() return { "success": False, "message": f"校准误差 {error_deg:.2f}° 超过 10° 阈值，已拒绝应用该校准点", } return { "success": True, "message": f"校准点{point_index + 1}已应用", "data": { "point_index": point_index, "point_count": point_count, "actual_pan": round(actual_pan, 4), "actual_tilt": round(actual_tilt, 4), "expected_pan": round(expected_pan, 4), "expected_tilt": round(expected_tilt, 4), "error_deg": round(error_deg, 2), "model_type": self.model_type, }, } # v11.30: FOV 进程内缓存 — ptz_models.json 极少变化，按 mtime 失效， # 避免 /status 高频调用每次读盘 _fov_cache: dict[tuple, tuple[float, float]] = {} _fov_cache_mtime: float = 0.0 def _get_fov(img_w: int, img_h: int, zoom: int = 1, model_name: str = "") -> tuple[float, float]: """获取当前分辨率和变焦下的 FOV。 v9.75: 从 PTZ_models.json 读取真实参数，支持多zoom点插值。 v11.30: mtime 缓存，读盘仅在文件变化时发生。 Args: img_w, img_h: 当前分辨率 zoom: 当前变焦倍数 model_name: 设备型号名称 Returns: (fov_h, fov_v): 水平/垂直视场角（度） """ global _fov_cache_mtime ptz_models_path = get_ptz_models_path() mtime = 0.0 if ptz_models_path.exists(): try: mtime = ptz_models_path.stat().st_mtime except OSError: mtime = 0.0 if mtime != _fov_cache_mtime: _fov_cache.clear() _fov_cache_mtime = mtime cache_key = (img_w, img_h, zoom, model_name) if cache_key in _fov_cache: return _fov_cache[cache_key] result = _compute_fov(ptz_models_path, img_w, img_h, zoom, model_name) _fov_cache[cache_key] = result return result def _compute_fov(ptz_models_path, img_w: int, img_h: int, zoom: int, model_name: str) -> tuple[float, float]: """实际从 PTZ_models.json 计算 FOV（由 _get_fov 缓存包装）。""" # 尝试从 PTZ_models.json 读取 if ptz_models_path.exists(): try: data = json.loads(ptz_models_path.read_text(encoding="utf-8")) models = data.get("ptz_models", {}) # 查找匹配型号 model_data = None if model_name and model_name in models: model_data = models[model_name] elif models: # 取第一个型号 model_data = next(iter(models.values())) if model_data: fov_table = model_data.get("fov_table", []) if fov_table: # 查找zoom对应的FOV，或插值 fov_h, fov_v = _interpolate_fov(fov_table, zoom) # 如果分辨率不同，按比例缩放 sensor = model_data.get("sensor", {}) native_w = sensor.get("resolution_w", img_w) native_h = sensor.get("resolution_h", img_h) if native_w != img_w or native_h != img_h: # FOV与分辨率无关（角度不变），但像素比例会变 # 这里返回原始FOV，像素转换会处理 pass return fov_h, fov_v except Exception: pass # 默认值 return 62.0, 35.0 def _interpolate_fov(fov_table: list[dict], zoom: int) -> tuple[float, float]: """从FOV表中插值FOV。 Args: fov_table: [{"zoom": 1, "fov_h": 62.0, "fov_v": 35.0}, ...] zoom: 当前变焦值 Returns: (fov_h, fov_v) """ if not fov_table: return 62.0, 35.0 # 按zoom排序 sorted_table = sorted(fov_table, key=lambda x: x.get("zoom", 1)) # 精确匹配 for entry in sorted_table: if entry.get("zoom") == zoom: return entry.get("fov_h", 62.0), entry.get("fov_v", 35.0) # 边界处理 if zoom <= sorted_table[0].get("zoom", 1): return sorted_table[0].get("fov_h", 62.0), sorted_table[0].get("fov_v", 35.0) if zoom >= sorted_table[-1].get("zoom", 1): return sorted_table[-1].get("fov_h", 62.0), sorted_table[-1].get("fov_v", 35.0) # 线性插值 for i in range(len(sorted_table) - 1): z0 = sorted_table[i].get("zoom", 1) z1 = sorted_table[i + 1].get("zoom", 2) if z0 <= zoom <= z1: t = (zoom - z0) / (z1 - z0) if z1 != z0 else 0 fov_h0 = sorted_table[i].get("fov_h", 62.0) fov_h1 = sorted_table[i + 1].get("fov_h", 62.0) fov_v0 = sorted_table[i].get("fov_v", 35.0) fov_v1 = sorted_table[i + 1].get("fov_v", 35.0) return fov_h0 + t * (fov_h1 - fov_h0), fov_v0 + t * (fov_v1 - fov_v0) return 62.0, 35.0 # ============================================================ # 数据持久化 # ============================================================ def load_click_point_data(mac_clean: str) -> dict: """加载点击指向数据。""" path = get_click_point_path(mac_clean) if path.exists(): try: return json.loads(path.read_text(encoding="utf-8")) except Exception: pass return {"points": [], "model_type": "none", "params": {}, "updated_at": 0} def save_click_point_data(mac_clean: str, model: ClickPointModel): """保存点击指向数据（带时间戳归档）。 每次保存都创建一个带时间戳的归档文件，同时更新主文件。 主文件始终保存最新数据，用于下次加载。 归档文件用于历史追溯。 """ path = get_click_point_path(mac_clean) data = { "points": model.points, "model_type": model.model_type, "params": model.params, "updated_at": model.updated_at, } # 保存主文件（最新数据） path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") # 创建归档文件（带时间戳） from datetime import datetime archive_dir = path.parent / "click_point_archive" archive_dir.mkdir(parents=True, exist_ok=True) timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") archive_path = archive_dir / f"click_point_{timestamp}.json" archive_path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") def model_from_data(data: dict) -> ClickPointModel: """从持久化数据恢复模型。""" m = ClickPointModel() m.points = data.get("points", []) m.model_type = data.get("model_type", "none") m.params = data.get("params", {}) m.updated_at = data.get("updated_at", 0) return m # ============================================================ # 会话管理 (内存中, 按 MAC) # ============================================================ _sessions: dict[str, ClickPointModel] = {} def get_session(mac_clean: str) -> ClickPointModel: """获取或创建点击指向会话。""" if mac_clean not in _sessions: data = load_click_point_data(mac_clean) _sessions[mac_clean] = model_from_data(data) return _sessions[mac_clean] def reset_session(mac_clean: str): """重置会话（v11.30: 仅内存，不写盘）。 延迟落盘语义：重来/取消只丢弃本轮内存数据，磁盘上的旧校准保持不变； 写盘仅由 finish_session 在用户点「完成」时执行。 """ _sessions[mac_clean] = ClickPointModel() def finish_session(mac_clean: str): """v11.30: 完成校准 — 内存会话落盘 + 清理待应用数据。""" m = _sessions.get(mac_clean) if m is None: return save_click_point_data(mac_clean, m) m._pending_click = None