From ddea6d0578e9295edf9a51bc34e8d5b783efb704 Mon Sep 17 00:00:00 2001 From: Ankle Date: Tue, 15 Sep 2026 09:53:00 +0800 Subject: [PATCH] tiandytmd --- README.md | 34 +++++++- install.sh | 22 ++++++ interactive_calibrator.py | 143 +++++++++++++++++++++++++++++----- scan_calibrator.py | 103 ++++++++++++++++++++++-- tests/test_scan_calibrator.py | 119 ++++++++++++++++++++++++++++ 手动配置README.md | 2 +- 6 files changed, 389 insertions(+), 34 deletions(-) diff --git a/README.md b/README.md index 360172c..e80defd 100644 --- a/README.md +++ b/README.md @@ -1,7 +1,32 @@ # scan参数交互式配置向导 该工具用于在不同摄像头点位通过固定对话完成scan参数标定、配置替换、完整试扫和验证拼图， -适合批量部署RK3576和RK3588。详细的底层命令和原理见[手动配置README.md](./手动配置README.md)。 +适合批量部署RK3576和RK3588，并适用于smart-cam已支持的海康威视和天地伟业PTZ摄像头。 +详细的底层命令和原理见[手动配置README.md](./手动配置README.md)。 + +## 摄像头兼容性 + +标定工具不直接调用海康或天地伟业SDK，也不连接`80/8000/3000`等摄像头端口。 +它只调用smart-cam在容器内提供的统一HTTP接口： + +- `GET /api/v1/control/status`：检查实际厂商、驱动状态和能力； +- `GET /api/v1/ptz`：读取统一`pan/tilt/zoom`坐标； +- `POST /api/v1/ptz/absolute`：绝对移动； +- `POST /api/v1/snapshot`：抓取标定画面。 + +因此没有按型号写死的PTZ方向、视场角或zoom档位。程序会使用适配层实际回读的移动量， +通过实拍自动识别pan方向，并在zoom离散时向上选择最近的可达档位。pan或tilt某个方向遇到 +机械/固件限位时，会换反方向探测。 + +`install.sh`会在不移动摄像头的前提下检查`driver_ready`、`get_position`和 +`move_absolute`。只要当前型号已被当前smart-cam的对应厂商适配层正确支持， +本工具就无需增加型号分支。若驱动未就绪或型号不支持绝对PTZ，程序会在移动前明确拒绝标定， +不会盲目试转摄像头。 + +部分天地伟业型号的SDK不能像海康一样直接完成任意绝对坐标移动。smart-cam适配层会改用 +“短方向动作 -> 回读实际位置 -> 继续逼近”，所以肉眼看起来会一小段一小段移动。 +这是当前型号的控制方式，不是标定程序预先排了多个扫描点，也不能单凭这个现象判定电机故障。 +标定时应等待程序自己完成到位回读，不要同时在Web端控制云台。 ## 目录结构 @@ -217,9 +242,10 @@ smart-cam的HTTP接口读取PTZ并记录，等价于： 水平覆盖不一致；这种摄像头使用单一pan_step无法保证所有行同时零重叠且无缺口； 12. 根据多个水平样本的带符号位移，自动识别该型号摄像头从画面左到右时pan是递增还是递减； 13. 每次标定会话只编译一次OpenCV重叠分析器，后续step直接复用； -14. 如果smart-cam因RTSP采集看门狗而短暂重启，向导会等待scan HTTP恢复， +14. 标定开始前检查摄像头适配层已就绪且具备位置读取/绝对移动能力； +15. 如果smart-cam因RTSP采集看门狗而短暂重启，向导会等待scan HTTP恢复， 然后重试当前PTZ或拍照操作，不会直接丢失整个第四步； -15. 删除本次自动标定产生的临时截图。 +16. 删除本次自动标定产生的临时截图。 这一步不能省略。两个角点只能确定扫描范围，无法确定当前摄像头型号和zoom下的一幅画面 覆盖多少角度。 @@ -336,7 +362,7 @@ docker exec smart-cam curl -fsS -X POST \ - 每一个摄像头点位都要重新记录角点； - 摄像头型号或zoom变化时必须重新自动计算step； -- 不同型号的海康球机可能有相反的pan数值方向；向导会实拍识别，不能用desktop/hanging +- 不同厂商和型号的球机可能有相反的pan数值方向；向导会实拍识别，不能用desktop/hanging 或“默认走短弧”代替； - 标定期间不要同时从Web界面控制PTZ； - 向导按轴设置目标重叠：水平4.5%、垂直2%；最终以完整试扫的接缝结果为准； diff --git a/install.sh b/install.sh index e42121c..c25fd42 100755 --- a/install.sh +++ b/install.sh @@ -4,6 +4,7 @@ set -Eeuo pipefail ROOT_DIR="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)" CONTAINER_NAME="${SCAN_CALIBRATOR_CONTAINER:-smart-cam}" SCAN_STATUS_URL="http://127.0.0.1:18080/api/v1/scan/status" +CONTROL_STATUS_URL="http://127.0.0.1:18080/api/v1/control/status" HELPER_SOURCE="$ROOT_DIR/src/measure_overlap.cpp" HELPER_CONTAINER_SOURCE="/tmp/scan-calibrator-measure.cpp" HELPER_CONTAINER_BIN="/tmp/scan-calibrator-measure" @@ -127,6 +128,27 @@ if not isinstance(payload.get("scan"), dict): ' <<<"$status" || fail "scan状态接口没有返回合法的scan JSON" ok "scan HTTP服务正常" + info "检查摄像头厂商适配层和标定所需PTZ能力" + local control_status vendor + control_status="$(docker exec "$CONTAINER_NAME" curl -fsS --max-time 15 \ + "$CONTROL_STATUS_URL")" || + fail "摄像头控制状态接口不可用：$CONTROL_STATUS_URL" + vendor="$(python3 -c ' +import json +import sys +payload = json.load(sys.stdin) +vendor = str(payload.get("vendor") or "unknown").strip().lower() +if payload.get("driver_ready") is not True: + detail = payload.get("driver_error") or "未返回具体原因" + raise SystemExit(f"摄像头驱动未就绪（vendor={vendor}）：{detail}") +capabilities = {str(item).strip().lower() for item in payload.get("capabilities", [])} +missing = sorted({"get_position", "move_absolute"} - capabilities) +if missing: + raise SystemExit("缺少PTZ能力：" + ", ".join(missing)) +print(vendor) +' <<<"$control_status")" || fail "smart-cam摄像头适配层不满足标定要求" + ok "摄像头适配层正常：vendor=$vendor，支持位置读取和绝对PTZ移动" + info "检查ImageMagick拼图命令" identify -version >/dev/null montage -version >/dev/null diff --git a/interactive_calibrator.py b/interactive_calibrator.py index dfb79fd..4e49462 100755 --- a/interactive_calibrator.py +++ b/interactive_calibrator.py @@ -284,12 +284,14 @@ def candidate_deltas(axis: str) -> list[int]: return [25, 40, 60, 90, 130, 190, 280, 420, 620, 850, 1150] -def fov_guided_deltas(delta: int, overlap: float, axis: str) -> list[int]: +def fov_guided_deltas(actual_delta: int, overlap: float, axis: str) -> list[int]: """Choose probes that retain enough shared image for robust matching.""" shift = 1.0 - overlap - if not 0.04 <= shift <= 0.92: + # Tiny shifts amplify PTZ quantization and image-matching error. Wait for a + # 15%..75% real image shift before replacing the generic probe sequence. + if not 0.15 <= shift <= 0.75: return [] - estimated_fov = delta / shift + estimated_fov = actual_delta / shift minimum = 20 if axis == "pan" else 12 # Fractions are image displacement/FOV, hence target overlaps are roughly # 82%, 72%, 60%, 48% and 35%. The middle probes provide the strongest @@ -300,6 +302,15 @@ def fov_guided_deltas(delta: int, overlap: float, axis: str) -> list[int]: }) +def safe_probe_cap(actual_delta: int, overlap: float) -> int | None: + """Keep later probes inside the FOV already observed by the camera.""" + shift = 1.0 - overlap + if actual_delta <= 0 or not 0.03 <= shift <= 0.85: + return None + estimated_fov = actual_delta / shift + return max(1, round(estimated_fov * 0.85)) + + def choose_direction(axis: str, origin: dict[str, int], points: dict) -> int: if axis == "pan": # Pan is circular. The direction changes only the sign, not the FOV estimate. @@ -437,12 +448,12 @@ def configured_image_aspect_ratio() -> float: def failure_detail(result: dict) -> str: reason = str(result.get("reason") or "unknown") labels = { - "ptz_motion_shortfall": "云台受限，实际移动量不足", - "not_enough_features": "两张图片可检测特征不足", - "not_enough_matches": "两张图片的公共特征不足", - "homography_failed": "公共特征无法形成可靠几何变换", - "axis_motion_too_small": "匹配到的主轴画面位移过小", - "axis_motion_inconsistent": "匹配位移方向异常或串轴过大", + "ptz_motion_shortfall": "摄像头实际移动小于请求值，本样本自动跳过", + "not_enough_features": "画面纹理太少，无法测量，本样本自动跳过", + "not_enough_matches": "移动前后相同画面太少，本样本自动跳过", + "homography_failed": "移动前后相同画面太少或纹理重复，本样本自动跳过", + "axis_motion_too_small": "画面位移太小，本样本自动跳过", + "axis_motion_inconsistent": "画面同时出现过大横纵位移，本样本自动跳过", } fields = [] if result.get("actual_axis_motion") is not None: @@ -476,6 +487,7 @@ def estimate_axis(axis: str, origin: dict[str, int], points: dict, pending = candidate_deltas(axis) tested: set[int] = set() adaptive_seeded = False + direction_switched = False while pending: delta = pending.pop(0) if delta in tested: @@ -485,14 +497,20 @@ def estimate_axis(axis: str, origin: dict[str, int], points: dict, axis, delta, direction, 1.0, baseline=baseline, origin=origin ) snapshots.add(str(result.get("candidate_file", ""))) + actual_motion = int(result.get("actual_axis_motion") or 0) + switch_threshold = 150 if axis == "pan" else 90 if ( - axis == "tilt" - and not reliable + not reliable + and not direction_switched and result.get("reason") == "ptz_motion_shortfall" + and delta >= switch_threshold + and actual_motion < max(2, round(delta * 0.15)) ): direction *= -1 + direction_switched = True say( - f" step={delta}: 原方向受云台限位，自动改用相反方向继续标定" + f" step={delta}: {axis}在原方向基本没有移动，" + "可能接近限位；程序只切换一次反方向再试" ) result = core.measure_delta( axis, delta, direction, 1.0, baseline=baseline, origin=origin @@ -501,7 +519,20 @@ def estimate_axis(axis: str, origin: dict[str, int], points: dict, if not result.get("valid"): reason = str(result.get("reason") or "unknown") failure_reasons[reason] = failure_reasons.get(reason, 0) + 1 - say(f" step={delta}: {failure_detail(result)}，继续尝试") + stop_expanding = ( + bool(reliable) + and reason in {"not_enough_matches", "homography_failed"} + ) + say( + f" step={delta}: {failure_detail(result)}" + f"，{'停止扩大步长' if stop_expanding else '继续尝试'}" + ) + if stop_expanding: + say( + " 移动前后已没有足够相同画面；" + "更大step只会移得更远，现在停止扩大步长。" + ) + break continue overlap = float(result["overlap"]) inliers = int(result.get("inliers") or 0) @@ -511,8 +542,12 @@ def estimate_axis(axis: str, origin: dict[str, int], points: dict, else: quality = f"有效特征={inliers}，内点率={ratio*100:.1f}%" say(f" step={delta}: 重叠约 {overlap*100:.1f}%，{quality}") + actual_delta = int(result.get("actual_axis_motion") or delta) + cap = safe_probe_cap(actual_delta, overlap) + if cap is not None: + pending = [value for value in pending if value <= max(delta, cap)] if not adaptive_seeded: - guided = fov_guided_deltas(delta, overlap, axis) + guided = fov_guided_deltas(actual_delta, overlap, axis) if guided: # Keep only a small safety tail beyond the guided range. Very # large fixed probes cannot match once the current optical FOV @@ -533,11 +568,9 @@ def estimate_axis(axis: str, origin: dict[str, int], points: dict, and float(result.get("phase_response") or 0) >= core.PHASE_MIN_RESPONSE ) if (feature_reliable or phase_reliable) and 0.08 <= overlap <= 0.90: - # Hikvision absolute PTZ movement is quantized and can also have - # backlash. At high optical zoom, using the requested delta - # instead of the SDK-reported actual movement creates a large, - # systematic FOV error. Shortfall samples have already been - # rejected above, so the actual movement is authoritative here. + # Absolute PTZ movement may be quantized or affected by backlash on + # either vendor. Use the adapter-reported motion, not the request. + # Shortfall samples have already been rejected above. actual_delta = int(result.get("actual_axis_motion") or delta) reliable.append(( actual_delta, overlap, @@ -564,7 +597,10 @@ def estimate_axis(axis: str, origin: dict[str, int], points: dict, ) core.fail( f"{axis}方向只得到{len(reliable)}个可靠样本，至少需要2个不同移动量互相验证。" - f"失败统计：{summary or '无'}。请根据上方每个step的具体原因处理后重试" + f"失败统计：{summary or '无'}。\n" + "现场处理：1. 把镜头移到扫描区域中部；" + "2. 画面中保留岸线、树木、建筑等固定纹理；" + "3. 不要操作Web云台；4. 回复1重试。" ) # Fixed OSD text and repeated scenery can yield a convincing false @@ -575,7 +611,9 @@ def estimate_axis(axis: str, origin: dict[str, int], points: dict, values = ", ".join(f"{sample[2]:.0f}" for sample in reliable) core.fail( f"{axis}方向不同移动量反算的视场不一致（{values}），疑似重复图案误匹配；" - "本次结果不写入配置，请更换标定画面" + "本次结果不写入配置。\n" + "现场处理：把镜头移到扫描区域中部，对准距离较远、" + "整幅都有不重复固定纹理的场景，然后回复1重试。" ) estimated_fov, spread = fov_statistics(consistent) @@ -1003,10 +1041,72 @@ def reusable_step_axes(state: dict, points: dict) -> set[str]: return reusable +def acquire_calibration_control() -> bool: + """Pause an active patrol and wait for its in-flight PTZ command to end.""" + status, camera = core.get_control_status() + host = str(camera["camera"].get("sdk_host") or "").strip() + detail = f"，摄像头={host}" if host else "" + say( + f"已识别摄像头适配层：vendor={camera['vendor']}{detail}；" + "PTZ位置读取和绝对移动能力正常。" + ) + control = status.get("control", {}) + if not isinstance(control, dict): + core.fail("scan HTTP未返回合法的control状态") + if bool(control.get("scan_active")): + core.fail("当前正在执行完整扫图，请等待扫图结束后再运行标定向导") + + should_resume = bool( + control.get("patrol_active") + and not control.get("patrol_paused_by_operator") + ) + if should_resume: + core.api( + "POST", "/api/v1/patrol/stop", + {"source": "scan_calibrator"}, timeout=15, + ) + say("已临时暂停巡航；标定结束或正常中断时会自动恢复。", color=YELLOW) + + try: + # This is a synchronization barrier: GET waits behind any PTZ command + # already in flight when patrol was paused, then claims manual control. + core.get_ptz() + except BaseException: + if should_resume: + try: + core.api( + "POST", "/api/v1/patrol/start", + {"source": "scan_calibrator"}, timeout=15, + ) + except BaseException as resume_error: + core.warn(f"标定准备失败后未能自动恢复巡航：{resume_error}") + raise + return should_resume + + +def release_calibration_control(should_resume: bool) -> None: + if not should_resume: + return + try: + core.api( + "POST", "/api/v1/patrol/start", + {"source": "scan_calibrator"}, timeout=15, + ) + say("已恢复进入向导前的巡航状态。", color=YELLOW) + except BaseException as exc: + core.warn( + f"未能自动恢复巡航：{exc}；请执行 " + "docker exec smart-cam curl -sS -X POST " + "http://127.0.0.1:18080/api/v1/patrol/start" + ) + + def main() -> None: snapshots: set[str] = set() + resume_patrol = False try: say("您好！我们现在开始配置扫图参数！", color=GREEN) + resume_patrol = acquire_calibration_control() say( "标定进度会逐步保存；中途回复 3 或按 Ctrl+C 退出后，" "下次运行仍可选择本次记录继续。" @@ -1150,6 +1250,7 @@ def main() -> None: say(f"\n向导失败：{exc}", color=RED) finally: remove_session_snapshots(snapshots) + release_calibration_control(resume_patrol) if __name__ == "__main__": diff --git a/scan_calibrator.py b/scan_calibrator.py index 379c6cd..5a1512c 100755 --- a/scan_calibrator.py +++ b/scan_calibrator.py @@ -8,6 +8,7 @@ from datetime import datetime import json import math from pathlib import Path +import re import subprocess import sys import time @@ -26,6 +27,7 @@ API_RECOVERY_TIMEOUT = 90.0 API_RECOVERY_POLL_INTERVAL = 2.0 TRANSIENT_CURL_EXIT_CODES = {7, 28, 52, 55, 56} _HELPER_READY = False +REQUIRED_PTZ_CAPABILITIES = {"get_position", "move_absolute"} def fail(message: str) -> "NoReturn": @@ -69,6 +71,45 @@ def transient_api_failure(result: subprocess.CompletedProcess[str]) -> bool: )) +def api_error_detail(result: subprocess.CompletedProcess[str]) -> str: + """Turn scan-service JSON errors into field-operator instructions.""" + message = "" + code = "" + try: + payload = json.loads(result.stdout) + error = payload.get("error", {}) + if isinstance(error, dict): + code = str(error.get("code") or "") + message = str(error.get("message") or "") + except (json.JSONDecodeError, AttributeError, TypeError): + message = result.stdout.strip() + + reached = re.search( + r"move_not_reached\s+pan_error=(-?\d+)\s+tilt_error=(-?\d+)\s+" + r"zoom_error=(-?\d+)\s+target=(-?\d+),(-?\d+),(-?\d+)\s+" + r"actual=(-?\d+),(-?\d+),(-?\d+)", + message, + ) + if reached: + pe, te, ze, tp, tt, tz, ap, at, az = ( + int(value) for value in reached.groups() + ) + return ( + "摄像头在限定时间内没有完全到位。\n" + f" 目标：pan={tp} ({tp / 100:.2f}°), " + f"tilt={tt} ({tt / 100:.2f}°), zoom={tz} ({tz / 100:.2f}×)\n" + f" 实际：pan={ap} ({ap / 100:.2f}°), " + f"tilt={at} ({at / 100:.2f}°), zoom={az} ({az / 100:.2f}×)\n" + f" 未到位差值：pan={pe / 100:.2f}°, " + f"tilt={te / 100:.2f}°, zoom={ze / 100:.2f}×\n" + "现场处理：请勿再操作Web云台，把第四步标定位置放在" + "扫描区域中部后重试；程序会自动限制后续测试步长。" + ) + if message: + return f"摄像头控制失败（{code or 'unknown'}）：{message}" + return result.stderr.strip() or f"请求退出码={result.returncode}" + + def scan_http_ready() -> bool: try: state = subprocess.run( @@ -114,9 +155,7 @@ def api(method: str, path: str, body: dict | None = None, timeout: int = 60) -> except json.JSONDecodeError: fail(f"scan HTTP返回的不是JSON：{result.stdout}") - last_detail = result.stderr.strip() or result.stdout.strip() or ( - f"退出码={result.returncode}" - ) + last_detail = api_error_detail(result) if not transient_api_failure(result) or attempt == API_MAX_ATTEMPTS: break warn( @@ -131,17 +170,65 @@ def api(method: str, path: str, body: dict | None = None, timeout: int = 60) -> # the in-container overlap helper before the next image comparison. _HELPER_READY = False - fail(f"命令失败：{' '.join(command)}\n{last_detail}") + fail(f"scan HTTP请求失败：{method} {path}\n{last_detail}") + + +def validate_control_status(payload: dict) -> dict: + """Validate the vendor-neutral PTZ contract exposed by smart-cam.""" + if not isinstance(payload, dict): + fail("scan HTTP控制状态不是JSON对象") + vendor = str(payload.get("vendor") or "unknown").strip().lower() + if payload.get("driver_ready") is not True: + detail = str(payload.get("driver_error") or "未返回具体原因") + fail(f"摄像头驱动未就绪（vendor={vendor}）：{detail}") + capabilities = { + str(item).strip().lower() + for item in payload.get("capabilities", []) + if str(item).strip() + } + missing = sorted(REQUIRED_PTZ_CAPABILITIES - capabilities) + if missing: + fail( + f"当前smart-cam的摄像头适配层缺少标定必需能力：" + f"{', '.join(missing)}（vendor={vendor}）" + ) + camera = payload.get("camera") + return { + "vendor": vendor, + "capabilities": capabilities, + "camera": camera if isinstance(camera, dict) else {}, + } + + +def get_control_status() -> tuple[dict, dict]: + payload = api("GET", "/api/v1/control/status", timeout=15) + return payload, validate_control_status(payload) + + +def parse_ptz_payload(payload: dict) -> dict[str, int]: + try: + ptz = payload["ptz"] + result = {key: int(ptz[key]) for key in ("pan", "tilt", "zoom")} + except (KeyError, TypeError, ValueError): + fail(f"scan HTTP未返回合法PTZ坐标：{payload}") + if not 0 <= result["pan"] <= 36000: + fail(f"scan HTTP返回的pan超出统一坐标范围：{result['pan']}") + if not -18000 <= result["tilt"] <= 18000 or result["zoom"] <= 0: + fail(f"scan HTTP返回的tilt/zoom无效：{result}") + return result def get_ptz() -> dict[str, int]: - payload = api("GET", "/api/v1/ptz", timeout=15) - return {key: int(payload["ptz"][key]) for key in ("pan", "tilt", "zoom")} + payload = api("GET", "/api/v1/ptz?source=scan_calibrator", timeout=50) + return parse_ptz_payload(payload) def move(ptz: dict[str, int]) -> dict[str, int]: - payload = api("POST", "/api/v1/ptz/absolute", ptz, timeout=50) - return {key: int(payload["ptz"][key]) for key in ("pan", "tilt", "zoom")} + payload = api( + "POST", "/api/v1/ptz/absolute", + {**ptz, "source": "scan_calibrator"}, timeout=50, + ) + return parse_ptz_payload(payload) def snapshot() -> str: diff --git a/tests/test_scan_calibrator.py b/tests/test_scan_calibrator.py index ee0198b..f7fd67b 100644 --- a/tests/test_scan_calibrator.py +++ b/tests/test_scan_calibrator.py @@ -51,6 +51,29 @@ class ApiRecoveryTests(unittest.TestCase): self.assertEqual(payload["scan"]["state"], "idle") wait_mock.assert_called_once_with() + def test_move_not_reached_error_is_explained_for_operator(self): + result = completed( + 22, + stdout=json.dumps({ + "error": { + "code": "sdk_error", + "message": ( + "ERR move_not_reached pan_error=-25 tilt_error=56 " + "zoom_error=0 target=25892,2044,4620 " + "actual=25917,1988,4620" + ), + } + }), + stderr="curl: (22) The requested URL returned error: 500", + ) + + detail = core.api_error_detail(result) + + self.assertIn("没有完全到位", detail) + self.assertIn("tilt=2044 (20.44°)", detail) + self.assertIn("实际", detail) + self.assertIn("现场处理", detail) + @mock.patch("scan_calibrator.wait_for_scan_http") @mock.patch("scan_calibrator.subprocess.run") def test_non_transient_http_error_is_not_retried(self, run_mock, wait_mock): @@ -111,8 +134,104 @@ class ApiRecoveryTests(unittest.TestCase): self.assertIn("恢复起点失败", stderr.getvalue()) self.assertEqual(move_mock.call_count, 2) + @mock.patch("scan_calibrator.api") + def test_calibration_ptz_requests_identify_their_control_source(self, api_mock): + api_mock.return_value = {"ptz": {"pan": 1, "tilt": 2, "zoom": 3}} + + self.assertEqual(core.get_ptz(), {"pan": 1, "tilt": 2, "zoom": 3}) + api_mock.assert_called_with( + "GET", "/api/v1/ptz?source=scan_calibrator", timeout=50 + ) + + core.move({"pan": 4, "tilt": 5, "zoom": 6}) + api_mock.assert_called_with( + "POST", "/api/v1/ptz/absolute", + {"pan": 4, "tilt": 5, "zoom": 6, + "source": "scan_calibrator"}, + timeout=50, + ) + + def test_hikvision_control_contract_is_supported(self): + result = core.validate_control_status({ + "vendor": "hikvision", + "driver_ready": True, + "capabilities": ["get_position", "move_absolute", "jog"], + "camera": {"sdk_host": "192.168.2.10", "sdk_port": 8000}, + }) + + self.assertEqual(result["vendor"], "hikvision") + self.assertIn("move_absolute", result["capabilities"]) + + def test_tiandy_control_contract_is_supported(self): + result = core.validate_control_status({ + "vendor": "tiandy", + "driver_ready": True, + "capabilities": ["get_position", "move_absolute", "presets"], + "camera": {"sdk_host": "192.168.100.64", "sdk_port": 3000}, + }) + + self.assertEqual(result["vendor"], "tiandy") + self.assertIn("get_position", result["capabilities"]) + + def test_control_contract_rejects_unready_driver(self): + with self.assertRaisesRegex(SystemExit, "login failed"): + core.validate_control_status({ + "vendor": "tiandy", + "driver_ready": False, + "driver_error": "login failed", + "capabilities": [], + }) + + def test_control_contract_rejects_missing_absolute_move(self): + with self.assertRaisesRegex(SystemExit, "move_absolute"): + core.validate_control_status({ + "vendor": "hikvision", + "driver_ready": True, + "capabilities": ["get_position", "jog"], + }) + + def test_ptz_response_must_use_normalized_contract(self): + with self.assertRaisesRegex(SystemExit, "pan超出"): + core.parse_ptz_payload({ + "ptz": {"pan": 40000, "tilt": 100, "zoom": 200} + }) + class InteractiveWizardTests(unittest.TestCase): + def test_guided_steps_use_actual_motion_not_requested_motion(self): + self.assertEqual( + wizard.fov_guided_deltas(62, 0.802, "tilt"), + [56, 88, 125, 163, 204], + ) + + def test_guided_steps_wait_for_measurable_image_shift(self): + self.assertEqual(wizard.fov_guided_deltas(33, 0.952, "tilt"), []) + + def test_probe_cap_uses_camera_reported_motion(self): + self.assertEqual(wizard.safe_probe_cap(33, 0.952), 584) + self.assertEqual(wizard.safe_probe_cap(62, 0.802), 266) + + @mock.patch("interactive_calibrator.core.get_ptz") + @mock.patch("interactive_calibrator.core.api") + def test_calibration_pauses_active_patrol_before_ptz_barrier( + self, api_mock, get_ptz_mock + ): + status = {"control": {"scan_active": False, "patrol_active": True, + "patrol_paused_by_operator": False}} + camera = {"vendor": "tiandy", "camera": {}, "capabilities": set()} + api_mock.return_value = {"success": True} + + with mock.patch( + "interactive_calibrator.core.get_control_status", + return_value=(status, camera), + ): + self.assertTrue(wizard.acquire_calibration_control()) + api_mock.assert_called_once_with( + "POST", "/api/v1/patrol/stop", + {"source": "scan_calibrator"}, timeout=15, + ) + get_ptz_mock.assert_called_once_with() + def test_manual_ptz_accepts_annotated_display_format(self): self.assertEqual( wizard.parse_ptz_input( diff --git a/手动配置README.md b/手动配置README.md index 3c6c1cb..caf59b7 100644 --- a/手动配置README.md +++ b/手动配置README.md @@ -2,7 +2,7 @@ 这个程序用于不同摄像头点位的scan现场标定： -- 读取当前海康摄像头PTZ； +- 通过smart-cam统一接口读取当前海康威视或天地伟业摄像头PTZ； - 记录扫描区域左上角和右下角； - 在固定zoom下移动一个候选角度并抓取两张图； - 用ORB特征和RANSAC单应矩阵定位公共特征，再按当前移动轴的画面位移计算重叠率；