From f7a782dcefc0268d73cfbf89bbc2c182681773ef Mon Sep 17 00:00:00 2001 From: Ankle Date: Thu, 10 Sep 2026 17:06:56 +0800 Subject: [PATCH] gengxin --- .gitignore | 1 + README.md | 55 +++- interactive_calibrator.py | 455 +++++++++++++++++++++++++++++----- scan_calibrator.py | 14 +- tests/test_scan_calibrator.py | 132 ++++++++++ 5 files changed, 584 insertions(+), 73 deletions(-) diff --git a/.gitignore b/.gitignore index a11115a..5fdb98f 100644 --- a/.gitignore +++ b/.gitignore @@ -3,6 +3,7 @@ __pycache__/ # Per-device calibration state and generated artifacts /calibration.json +/history/ /result/* !/result/.gitkeep /scan-panorama/result/* diff --git a/README.md b/README.md index 3d2ef0f..360172c 100644 --- a/README.md +++ b/README.md @@ -14,6 +14,7 @@ scan-calibrator/ ├── interactive_calibrator.py ├── calibrate.sh # 手动配置入口 ├── scan_calibrator.py +├── history/ # 每次交互向导的完整或中途标定记录（运行时生成） ├── src/measure_overlap.cpp ├── result/ # 运行时生成的scan配置片段 └── scan-panorama/ @@ -41,14 +42,19 @@ cd /home/zhangw/scan-calibrator 运行`install.sh`会自动补齐。ImageMagick安装在RK3588宿主机，而不是smart-cam容器内， 因为验证拼图程序运行在宿主机。 -只需要输入： +常用输入： - `1`：确认、执行或选择野外部署； - `2`：只在第一个问题中表示办公室测试； -- `3`：结束对话，并清理本次未完成向导生成的标定状态和临时标定截图； +- 历史记录选择处输入`1`、`2`、`3`等序号：复用对应记录，输入`0`新建，输入`q`退出； +- 左上角和右下角既可回复`1`读取摄像头，也可直接粘贴 + `pan=21720 (217.20°), tilt=1268 (12.68°), zoom=4580 (45.80×)`； +- 除历史记录选择处外，`3`：结束对话并保留当前标定进度； - `Ctrl+C`：等同于退出。 -退出不会删除历史扫描目录，也不会回滚已经明确确认替换并重启的`config.json`。 +每次启动都会建立一条独立历史记录。退出不会删除已记录的角点、已测得的单轴step、 +已生成的配置、历史扫描目录，也不会回滚已经明确确认替换并重启的`config.json`。 +临时标定截图仍会清理，避免长期占用设备磁盘。 ## 完整对话示例 @@ -57,21 +63,27 @@ cd /home/zhangw/scan-calibrator ```text 您好！我们现在开始配置扫图参数！ -在接下来的对话中，回复 3 会立即结束，并清理本次向导产生的标定痕迹。 +标定进度会逐步保存；中途回复 3 或按 Ctrl+C 退出后，下次运行仍可选择本次记录继续。 ① 当前是【野外部署阶段】（回复 1）还是【办公室测试阶段】（回复 2）？ > 2 +在第②步之前发现以下历史标定记录（最新的排在前面）： +1. 2026-09-10T10:20:30+08:00，安装方式=办公室正放，左上角=有，右下角=有，pan_step=有，tilt_step=无 +2. 2026-09-09T16:10:00+08:00，安装方式=野外倒挂，左上角=有，右下角=无，pan_step=无，tilt_step=无 +输入对应序号复用；回复 0 不复用并新建；回复 q 退出。 +> 0 + ② 请在Web界面中设置左上角zoom，再手动把摄像头移到扫描区域的【左上角】。 -到位后回复 1；回复 3 随时退出。 +摄像头到位后回复 1 读取当前位置；也可以直接输入pan、tilt、zoom；回复 3 随时退出。 > 1 已读取: pan=19120 (191.20°), tilt=990 (9.90°), zoom=810 (8.10×) ③ 请手动把摄像头移到扫描区域的【右下角】。 -右下角可以使用不同zoom，程序会取两个角点zoom的平均值作为最终扫描zoom。 -到位后回复 1；回复 3 随时退出。 -> 1 -已读取: pan=13420 (134.20°), tilt=-340 (-3.40°), zoom=810 (8.10×) +右下角可以使用不同zoom，程序会从两个角点zoom的平均值向上选择最近可达的最终扫描zoom。 +摄像头到位后回复 1 读取当前位置；也可以直接输入pan、tilt、zoom；回复 3 随时退出。 +> pan=13420 (134.20°), tilt=-340 (-3.40°), zoom=810 (8.10×) +已采用手动输入: pan=13420 (134.20°), tilt=-340 (-3.40°), zoom=810 (8.10×) ④ 这一步用于测量当前zoom下的水平/垂直视场。 请把摄像头移到扫描区域内部、距离较远且纹理分布在整幅画面的位置。 @@ -176,8 +188,11 @@ smart-cam的HTTP接口读取PTZ并记录，等价于： ./calibrate.sh mark bottom-right ``` -向导会显示两个角点的zoom，并取算术平均值作为最终扫描zoom。第四步开始测量前， -程序会自动把镜头切换到该平均zoom，因此视场标定与最终配置使用同一焦段。 +向导会显示两个角点的zoom，并以算术平均值作为选择最终扫描zoom的下界。第四步开始测量前， +程序会自动把镜头切换到该平均zoom；即使摄像头当前已经显示为这个zoom，也会明确下发一次 +绝对zoom命令并回读。若算术平均值不是摄像头可达档位，程序会在平均值上方查找最近的可达档位， +例如平均值`4565`只能落到`4550`或`4580`时采用`4580`。视场标定和最终生成配置都会使用 +回读确认的同一个可达zoom，不会再把不可达平均值写入配置。 ### ④ 自动计算pan_step和tilt_step @@ -211,8 +226,22 @@ smart-cam的HTTP接口读取PTZ并记录，等价于： 不要把画面对准纯天空、纯墙、纯水面、玻璃、近距离物体或重复图案。水面场景应让岸线、 浮标等固定物体横向分布在画面中，不能只依赖不断变化的水纹。 -测量过程中程序会锁定两个角点的平均zoom；自动对焦不影响zoom。标定失败时可以只更换第四步的位置重试， -不需要重新记录左上角和右下角。 +测量过程中程序会锁定上述实际可达的扫图zoom；自动对焦不影响zoom。标定失败时可以只更换 +第四步的位置重试，不需要重新记录左上角和右下角。pan或tilt任一方向已经成功时会立刻保存， +另一方向失败或退出后，下次可以只继续尚未完成的方向。 + +### 历史记录复用 + +进入第②步前，向导会读取`history/session_*.json`并按时间倒序编号，逐条显示是否已经包含： + +- 左上角位置； +- 右下角位置； +- `pan_step`； +- `tilt_step`。 + +输入序号后，已有角点会直接跳过；只有一个角点时只补录另一个；只有一个轴的step时只补测另一个。 +如果四项都齐全，向导会询问直接复用还是重新测量step。复用会建立一条新的本次记录，原历史文件 +不会被覆盖。升级前已有的`calibration.json`也会自动导入历史记录一次。 ### ⑤ 显示最终参数 diff --git a/interactive_calibrator.py b/interactive_calibrator.py index 4f28d3a..dfb79fd 100755 --- a/interactive_calibrator.py +++ b/interactive_calibrator.py @@ -5,6 +5,7 @@ from __future__ import annotations import argparse from contextlib import redirect_stdout +from copy import deepcopy from datetime import datetime import io import json @@ -21,6 +22,7 @@ import scan_calibrator as core ROOT = Path(__file__).resolve().parent +HISTORY_DIR = ROOT / "history" CONFIG_PATH = "/smart-cam/config/config.json" # The grid contact sheet is the final calibration check. Horizontal seams on # the tested PTZ camera retain more real-world overlap than the center-shift @@ -31,6 +33,7 @@ RED = "\033[31m" GREEN = "\033[32m" YELLOW = "\033[33m" RESET = "\033[0m" +ACTIVE_HISTORY_FILE: Path | None = None class Cancelled(Exception): @@ -51,6 +54,58 @@ def choose(question: str, choices: set[str]) -> str: say(f"请输入 {' / '.join(sorted(choices | {'3'}))}。", color=YELLOW) +def valid_ptz(ptz: object) -> bool: + if not isinstance(ptz, dict): + return False + try: + pan = int(ptz["pan"]) + tilt = int(ptz["tilt"]) + zoom = int(ptz["zoom"]) + except (KeyError, TypeError, ValueError): + return False + return 0 <= pan <= 36000 and -18000 <= tilt <= 18000 and zoom > 0 + + +def parse_ptz_input(value: str) -> dict[str, int]: + """Parse PTZ copied from this tool, with or without display annotations.""" + result: dict[str, int] = {} + for name in ("pan", "tilt", "zoom"): + match = re.search( + rf"(?:^|[,，;；\s]){name}\s*=\s*(-?\d+)", + value, + flags=re.IGNORECASE, + ) + if match: + result[name] = int(match.group(1)) + if not valid_ptz(result): + raise ValueError( + "格式应为 pan=21720, tilt=1268, zoom=4580；" + "也可以直接粘贴带角度和倍数的完整输出" + ) + return result + + +def capture_position(question: str) -> dict[str, int]: + while True: + answer = input( + f"\n{question}\n" + "摄像头到位后回复 1 读取当前位置；也可以直接输入 " + "pan=21720 (217.20°), tilt=1268 (12.68°), zoom=4580 (45.80×)；" + "回复 3 随时退出。\n> " + ).strip() + if answer == "3": + raise Cancelled + if answer == "1": + return camera_ptz() + try: + ptz = parse_ptz_input(answer) + except ValueError as exc: + say(f"输入无效：{exc}", color=YELLOW) + continue + core.print_ptz(ptz, "已采用手动输入") + return ptz + + def confirm_position(question: str) -> None: choose(f"{question}\n到位后回复 1；回复 3 随时退出。", {"1"}) @@ -61,6 +116,148 @@ def camera_ptz() -> dict[str, int]: return ptz +def substantive_state(state: dict) -> dict: + """Return reusable calibration data, excluding per-session metadata.""" + return { + key: deepcopy(state[key]) + for key in ("points", "steps", "scan_zoom", "scan_zoom_average", "mount") + if key in state + } + + +def state_fingerprint(state: dict) -> str: + return json.dumps(substantive_state(state), ensure_ascii=False, sort_keys=True) + + +def write_json_atomic(path: Path, value: dict) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + temporary = path.with_name(f".{path.name}.tmp") + temporary.write_text( + json.dumps(value, ensure_ascii=False, indent=2), encoding="utf-8" + ) + temporary.replace(path) + + +def load_history_records() -> list[tuple[Path, dict]]: + """Load sessions and import a pre-upgrade calibration.json when needed.""" + HISTORY_DIR.mkdir(parents=True, exist_ok=True) + records: list[tuple[Path, dict]] = [] + fingerprints: set[str] = set() + for path in sorted(HISTORY_DIR.glob("session_*.json")): + try: + state = json.loads(path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as exc: + say(f"忽略无法读取的历史记录 {path.name}：{exc}", color=YELLOW) + continue + if not isinstance(state, dict): + continue + records.append((path, state)) + fingerprints.add(state_fingerprint(state)) + + if core.STATE_FILE.exists(): + try: + current = json.loads(core.STATE_FILE.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + current = None + if ( + isinstance(current, dict) + and any(substantive_state(current).get(key) for key in ("points", "steps")) + and state_fingerprint(current) not in fingerprints + ): + stamp = datetime.now().strftime("%Y%m%d_%H%M%S_%f") + recovered = deepcopy(current) + recovered.setdefault("_session", {}) + recovered["_session"].update({ + "id": f"recovered_{stamp}", + "created_at": recovered.get("updated_at") + or datetime.now().astimezone().isoformat(timespec="seconds"), + "source": "calibration.json升级前记录", + }) + path = HISTORY_DIR / f"session_{stamp}_recovered.json" + write_json_atomic(path, recovered) + records.append((path, recovered)) + + return sorted( + records, + key=lambda item: str( + item[1].get("_session", {}).get("created_at") + or item[1].get("updated_at") + or item[0].name + ), + reverse=True, + ) + + +def state_item_exists(state: dict, section: str, name: str) -> bool: + value = state.get(section, {}).get(name) + if section == "points": + return valid_ptz(value) + return isinstance(value, dict) and bool(value) + + +def select_history_record( + records: list[tuple[Path, dict]], +) -> tuple[Path, dict] | None: + if not records: + say("\n未发现可复用的历史标定记录，将开始一条新记录。") + return None + + say("\n在第②步之前发现以下历史标定记录（最新的排在前面）：") + for index, (path, state) in enumerate(records, start=1): + session = state.get("_session", {}) + when = session.get("created_at") or state.get("updated_at") or path.stem + mount = {"hanging": "野外倒挂", "desktop": "办公室正放"}.get( + state.get("mount"), "未记录" + ) + say( + f"{index}. {when}，安装方式={mount}，" + f"左上角={'有' if state_item_exists(state, 'points', 'top-left') else '无'}，" + f"右下角={'有' if state_item_exists(state, 'points', 'bottom-right') else '无'}，" + f"pan_step={'有' if state_item_exists(state, 'steps', 'pan') else '无'}，" + f"tilt_step={'有' if state_item_exists(state, 'steps', 'tilt') else '无'}" + ) + while True: + answer = input( + "\n输入对应序号复用；回复 0 不复用并新建；回复 q 退出。" + "此处数字 3 表示第3条记录（如果存在）。\n> " + ).strip().lower() + if answer in {"q", "quit", "退出"}: + raise Cancelled + if answer == "0": + return None + if answer.isdigit() and 1 <= int(answer) <= len(records): + return records[int(answer) - 1] + say(f"请输入 0、1～{len(records)} 或 q。", color=YELLOW) + + +def start_history_session( + mount: str, selected: tuple[Path, dict] | None +) -> dict: + global ACTIVE_HISTORY_FILE + stamp = datetime.now().strftime("%Y%m%d_%H%M%S_%f") + source_path, source_state = selected if selected else (None, {}) + state = substantive_state(source_state) + state.setdefault("points", {}) + state.setdefault("steps", {}) + state["mount"] = mount + state["_session"] = { + "id": stamp, + "created_at": datetime.now().astimezone().isoformat(timespec="seconds"), + **({"reused_from": source_path.name} if source_path else {}), + } + ACTIVE_HISTORY_FILE = HISTORY_DIR / f"session_{stamp}.json" + save_working_state(state) + return state + + +def save_working_state(state: dict) -> None: + session = state.setdefault("_session", {}) + session["updated_at"] = datetime.now().astimezone().isoformat(timespec="seconds") + core.save_state(state) + if ACTIVE_HISTORY_FILE is not None: + write_json_atomic(ACTIVE_HISTORY_FILE, state) + + def remove_session_snapshots(paths: set[str]) -> None: for path in sorted(paths): if not path.startswith(( @@ -492,22 +689,77 @@ def estimate_axis(axis: str, origin: dict[str, int], points: dict, } -def calculate_steps(points: dict, snapshots: set[str], mount: str) -> None: +def read_zoom_after_move(origin: dict[str, int], requested_zoom: int) -> dict[str, int]: + core.move({**origin, "zoom": requested_zoom}) + time.sleep(1.0) + return core.get_ptz() + + +def move_to_scan_zoom(points: dict) -> tuple[dict[str, int], float]: + """Always move to the smallest reachable zoom at or above the mean.""" origin = camera_ptz() zooms = [int(points[name]["zoom"]) for name in ("top-left", "bottom-right")] - target_zoom = round(sum(zooms) / len(zooms)) - if origin["zoom"] != target_zoom: - say( - f"两个角点zoom的平均值为{target_zoom}（{target_zoom/100:.2f}×）；" - "正在自动切换到该zoom后再测量视场……" - ) - origin = core.move({**origin, "zoom": target_zoom}) - time.sleep(1.0) - if abs(int(origin["zoom"]) - target_zoom) > 10: + average_zoom = sum(zooms) / len(zooms) + minimum_zoom = math.ceil(average_zoom) + say( + f"两个角点zoom的平均值为{average_zoom:g}（{average_zoom/100:.3f}×）；" + "正在自动切换到该zoom。若平均值不可达，将向上取最近的可达档位……" + ) + + # Sending the command is intentional even when the current zoom appears + # equal: Step 4 must start from a confirmed camera-reported scan zoom. + actual = read_zoom_after_move(origin, minimum_zoom) + if int(actual["zoom"]) < minimum_zoom: + low_request = minimum_zoom + high_request = max(zooms) + if high_request <= low_request: core.fail( - f"摄像头未能到达平均zoom：目标={target_zoom}，实际={origin['zoom']}" + f"平均zoom={average_zoom:g}不可达，且没有更大的角点zoom可用于查找" ) - core.print_ptz(origin, "平均zoom到位") + + high_actual = read_zoom_after_move(origin, high_request) + if int(high_actual["zoom"]) < minimum_zoom: + core.fail( + f"无法找到大于平均值的可达zoom：平均={average_zoom:g}，" + f"请求上界={high_request}，实际={high_actual['zoom']}" + ) + + # Absolute zoom response is monotonic. Find the first request that + # crosses the mean, while retaining the smallest camera-reported zoom. + best_zoom = int(high_actual["zoom"]) + while low_request + 1 < high_request: + probe = (low_request + high_request) // 2 + probe_actual = read_zoom_after_move(origin, probe) + observed = int(probe_actual["zoom"]) + if observed >= minimum_zoom: + high_request = probe + best_zoom = min(best_zoom, observed) + else: + low_request = probe + actual = read_zoom_after_move(origin, best_zoom) + + selected_zoom = int(actual["zoom"]) + if selected_zoom < minimum_zoom: + core.fail( + f"摄像头回读zoom={selected_zoom}，小于要求的平均值{average_zoom:g}" + ) + core.print_ptz(actual, "扫图zoom到位") + if selected_zoom != average_zoom: + say( + f"平均zoom={average_zoom:g}不是可达档位；" + f"已向上采用最近可达zoom={selected_zoom}（{selected_zoom/100:.2f}×）。" + ) + return actual, average_zoom + + +def calculate_steps( + points: dict, + snapshots: set[str], + mount: str, + axes: set[str] | None = None, +) -> None: + requested_axes = set(axes or {"pan", "tilt"}) + origin, average_zoom = move_to_scan_zoom(points) top_left = points["top-left"] bottom_right = points["bottom-right"] # This check is only advisory and runs before signed pan direction has @@ -533,8 +785,30 @@ def calculate_steps(points: dict, snapshots: set[str], mount: str) -> None: color=YELLOW, ) state = core.load_state() - pan_info = estimate_axis("pan", origin, points, snapshots) - tilt_info = estimate_axis("tilt", origin, points, snapshots) + state["scan_zoom"] = int(origin["zoom"]) + state["scan_zoom_average"] = average_zoom + for axis in ("pan", "tilt"): + existing_zoom = state.get("steps", {}).get(axis, {}).get("zoom") + if axis not in requested_axes and ( + existing_zoom is None or int(existing_zoom) != int(origin["zoom"]) + ): + say( + f"历史{axis}_step是在zoom={existing_zoom or '未知'}下测得，" + f"与本次扫图zoom={origin['zoom']}不一致，将重新标定。", + color=YELLOW, + ) + requested_axes.add(axis) + save_working_state(state) + + if "pan" in requested_axes: + state["steps"]["pan"] = estimate_axis("pan", origin, points, snapshots) + save_working_state(state) + if "tilt" in requested_axes: + state["steps"]["tilt"] = estimate_axis("tilt", origin, points, snapshots) + save_working_state(state) + + pan_info = state["steps"]["pan"] + tilt_info = state["steps"]["tilt"] pan_info, cross_axis_message = constrain_pan_with_tilt( pan_info, tilt_info, configured_image_aspect_ratio() ) @@ -543,7 +817,7 @@ def calculate_steps(points: dict, snapshots: set[str], mount: str) -> None: )) state["steps"]["pan"] = pan_info state["steps"]["tilt"] = tilt_info - core.save_state(state) + save_working_state(state) def generate(mount: str) -> tuple[dict, Path, str]: @@ -690,14 +964,54 @@ def run_panorama(scan_status: dict, mount: str, pan_direction: str) -> Path: return output +def reusable_step_axes(state: dict, points: dict) -> set[str]: + average_zoom = sum( + int(points[name]["zoom"]) for name in ("top-left", "bottom-right") + ) / 2 + minimum_zoom = math.ceil(average_zoom) + scan_zoom = state.get("scan_zoom") + if scan_zoom is None: + measured_zooms: set[int] = set() + for info in state.get("steps", {}).values(): + if not isinstance(info, dict) or info.get("zoom") is None: + continue + try: + measured_zooms.add(int(info["zoom"])) + except (TypeError, ValueError): + continue + if len(measured_zooms) == 1 and next(iter(measured_zooms)) >= minimum_zoom: + scan_zoom = next(iter(measured_zooms)) + state["scan_zoom"] = scan_zoom + state["scan_zoom_average"] = average_zoom + try: + scan_zoom = int(scan_zoom) + except (TypeError, ValueError): + return set() + if scan_zoom < minimum_zoom: + return set() + reusable: set[str] = set() + for axis in ("pan", "tilt"): + info = state.get("steps", {}).get(axis) + if not isinstance(info, dict): + continue + try: + measured_zoom = int(info.get("zoom", -1)) + except (TypeError, ValueError): + continue + if measured_zoom == scan_zoom: + reusable.add(axis) + return reusable + + def main() -> None: - original_state = core.STATE_FILE.read_bytes() if core.STATE_FILE.exists() else None snapshots: set[str] = set() - generated: Path | None = None - success = False try: say("您好！我们现在开始配置扫图参数！", color=GREEN) - say("在接下来的对话中，回复 3 会立即结束，并清理本次向导产生的标定痕迹。") + say( + "标定进度会逐步保存；中途回复 3 或按 Ctrl+C 退出后，" + "下次运行仍可选择本次记录继续。" + ) + history_records = load_history_records() stage = choose( "① 当前是【野外部署阶段】（回复 1）还是" @@ -705,53 +1019,88 @@ def main() -> None: {"1", "2"}, ) mount = "hanging" if stage == "1" else "desktop" + selected = select_history_record(history_records) + state = start_history_session(mount, selected) + if selected: + say(f"已复用历史记录 {selected[0].name}；本次进度另存为一条新记录。") - confirm_position( - "② 请在Web界面中设置左上角需要的zoom，" - "再手动把摄像头移到扫描区域的【左上角】。" - ) - top_left = camera_ptz() - state = core.load_state() - state["points"]["top-left"] = top_left - core.save_state(state) + if valid_ptz(state["points"].get("top-left")): + top_left = state["points"]["top-left"] + core.print_ptz(top_left, "② 已复用左上角") + else: + top_left = capture_position( + "② 请在Web界面中设置左上角需要的zoom，" + "再手动把摄像头移到扫描区域的【左上角】。" + ) + state["points"]["top-left"] = top_left + state["steps"] = {} + state.pop("scan_zoom", None) + state.pop("scan_zoom_average", None) + save_working_state(state) - confirm_position( - "③ 请手动把摄像头移到扫描区域的【右下角】。\n" - "右下角可以使用不同zoom，程序会取两个角点zoom的平均值作为最终扫描zoom。" - ) - bottom_right = camera_ptz() - average_zoom = round((top_left["zoom"] + bottom_right["zoom"]) / 2) + if valid_ptz(state["points"].get("bottom-right")): + bottom_right = state["points"]["bottom-right"] + core.print_ptz(bottom_right, "③ 已复用右下角") + else: + bottom_right = capture_position( + "③ 请手动把摄像头移到扫描区域的【右下角】。\n" + "右下角可以使用不同zoom；第四步会取两个角点zoom的平均值，" + "若平均值不可达则向上取最近可达档位。" + ) + state["points"]["bottom-right"] = bottom_right + state["steps"] = {} + state.pop("scan_zoom", None) + state.pop("scan_zoom_average", None) + save_working_state(state) + + average_zoom = (top_left["zoom"] + bottom_right["zoom"]) / 2 say( f"两个角点zoom：左上={top_left['zoom']/100:.2f}×，" f"右下={bottom_right['zoom']/100:.2f}×；" - f"最终取平均={average_zoom/100:.2f}×。" + f"算术平均={average_zoom/100:.3f}×。" ) - state = core.load_state() - state["points"]["bottom-right"] = bottom_right - core.save_state(state) + points = state["points"] + available_axes = reusable_step_axes(state, points) + if available_axes == {"pan", "tilt"}: + save_working_state(state) + use_steps = choose( + "④ 该历史记录已有同一扫图zoom下的pan_step和tilt_step。\n" + "直接复用回复 1；在当前画面重新测量回复 2；回复 3 退出。", + {"1", "2"}, + ) + if use_steps == "2": + state["steps"] = {} + state.pop("scan_zoom", None) + state.pop("scan_zoom_average", None) + save_working_state(state) + available_axes = set() - points = core.load_state()["points"] - while True: + while available_axes != {"pan", "tilt"}: + missing_axes = {"pan", "tilt"} - available_axes confirm_position( - "④ 这一步用于测量当前zoom下的水平/垂直视场。\n" + "④ 这一步用于测量当前扫图zoom下的水平/垂直视场。\n" "请优先把摄像头移到扫描区域内部、" "距离较远且纹理分布在整幅画面的位置。\n" "请避开纯天空、纯墙面、纯水面、玻璃、近距离物体和重复图案；" "水面场景需要让岸线、浮标或其他固定物体横向分布在画面中；" - "回复1后不要再操作Web云台，程序会先切换到两个角点的平均zoom，" - "再自动移动、拍照并返回起点。" + "回复1后不要再操作Web云台。无论当前zoom是多少，程序都会先切换到" + "两个角点的平均zoom；若平均值不可达就向上取最近可达档位，" + f"然后测量{'+'.join(sorted(missing_axes))}并返回起点。" ) say("\n正在计算，请不要操作摄像头或Web控制页面……") try: - calculate_steps(points, snapshots, mount) - break + calculate_steps(points, snapshots, mount, missing_axes) + state = core.load_state() + available_axes = reusable_step_axes(state, points) except SystemExit as exc: say(f"\n本次视场标定没有通过：{exc}", color=RED) choose( "请根据上方逐项失败原因调整位置后重试第四步，两个角点无需重录。\n" - "重新尝试回复 1；结束对话回复 3。", + "已成功完成的单轴结果已经保存。重新尝试回复 1；结束对话回复 3。", {"1"}, ) + state = core.load_state() + available_axes = reusable_step_axes(state, points) scan, generated, detail = generate(mount) say("\n⑤ 本次得到的扫图参数：", color=GREEN) say(json.dumps(scan, ensure_ascii=False, indent=2)) @@ -792,27 +1141,15 @@ def main() -> None: raise RuntimeError("缺少本次实测pan方向，无法可靠排列拼图左右顺序") panorama = run_panorama(scan_status, mount, pan_direction) say(f"\n拼图完成。\n大图存储位置：{panorama}", color=GREEN) - success = True say("\n配置向导已完成，对话结束。", color=GREEN) except Cancelled: - say("\n已结束对话，正在清理本次向导痕迹……", color=YELLOW) + say("\n已结束对话；当前标定进度和已生成结果均已保留，后续可以复用。", color=YELLOW) except (KeyboardInterrupt, EOFError): - say("\n对话被中断，正在清理本次向导痕迹……", color=YELLOW) + say("\n对话被中断；当前标定进度和已生成结果均已保留，后续可以复用。", color=YELLOW) except Exception as exc: say(f"\n向导失败：{exc}", color=RED) finally: remove_session_snapshots(snapshots) - if not success: - if original_state is None: - core.STATE_FILE.unlink(missing_ok=True) - else: - current_state = ( - core.STATE_FILE.read_bytes() if core.STATE_FILE.exists() else None - ) - if current_state != original_state: - core.STATE_FILE.write_bytes(original_state) - if generated is not None: - generated.unlink(missing_ok=True) if __name__ == "__main__": diff --git a/scan_calibrator.py b/scan_calibrator.py index 9bf366a..379c6cd 100755 --- a/scan_calibrator.py +++ b/scan_calibrator.py @@ -525,7 +525,19 @@ def command_generate(args: argparse.Namespace) -> None: f"相位响应={phase_response:.2f}）；" f"请在纹理丰富且尽量远离摄像头的位置重新执行calibrate {axis}" ) - zoom = round((int(top_left["zoom"]) + int(bottom_right["zoom"])) / 2) + arithmetic_zoom = (int(top_left["zoom"]) + int(bottom_right["zoom"])) / 2 + # The interactive wizard records the exact camera-reported zoom used for + # FOV measurement. Prefer it because many PTZ cameras expose quantized zoom + # positions and cannot reach the arithmetic mean exactly. + try: + zoom = int(state["scan_zoom"]) + except (KeyError, TypeError, ValueError): + zoom = round(arithmetic_zoom) + if "scan_zoom" in state and zoom < math.ceil(arithmetic_zoom): + fail( + f"保存的scan_zoom={zoom}小于两个角点平均值{arithmetic_zoom:g}；" + "请重新执行交互向导第四步" + ) pan_start, pan_end, pan_span, visual_pan_direction = scan_pan_range( int(top_left["pan"]), int(bottom_right["pan"]), diff --git a/tests/test_scan_calibrator.py b/tests/test_scan_calibrator.py index 052a5f6..ee0198b 100644 --- a/tests/test_scan_calibrator.py +++ b/tests/test_scan_calibrator.py @@ -1,11 +1,16 @@ from __future__ import annotations +import argparse from contextlib import redirect_stderr import io +import json +from pathlib import Path import subprocess +import tempfile import unittest from unittest import mock +import interactive_calibrator as wizard import scan_calibrator as core @@ -107,5 +112,132 @@ class ApiRecoveryTests(unittest.TestCase): self.assertEqual(move_mock.call_count, 2) +class InteractiveWizardTests(unittest.TestCase): + def test_manual_ptz_accepts_annotated_display_format(self): + self.assertEqual( + wizard.parse_ptz_input( + "pan=21720 (217.20°), tilt=1268 (12.68°), " + "zoom=4580 (45.80×)" + ), + {"pan": 21720, "tilt": 1268, "zoom": 4580}, + ) + + def test_manual_ptz_rejects_incomplete_input(self): + with self.assertRaisesRegex(ValueError, "pan=21720"): + wizard.parse_ptz_input("pan=21720, tilt=1268") + + @mock.patch("interactive_calibrator.core.print_ptz") + @mock.patch("interactive_calibrator.camera_ptz") + @mock.patch("interactive_calibrator.read_zoom_after_move") + def test_unreachable_average_uses_nearest_higher_zoom( + self, move_mock, camera_mock, _print_mock + ): + origin = {"pan": 23052, "tilt": 1298, "zoom": 4510} + camera_mock.return_value = origin + + def quantized(_origin, requested): + return { + **origin, + "zoom": 4550 if requested <= 4565 else 4580, + } + + move_mock.side_effect = quantized + actual, average = wizard.move_to_scan_zoom({ + "top-left": {**origin, "zoom": 4550}, + "bottom-right": {**origin, "zoom": 4580}, + }) + + self.assertEqual(average, 4565) + self.assertEqual(actual["zoom"], 4580) + self.assertEqual(move_mock.call_args_list[0].args[1], 4565) + self.assertEqual(move_mock.call_args_list[-1].args[1], 4580) + + @mock.patch("interactive_calibrator.core.print_ptz") + @mock.patch("interactive_calibrator.camera_ptz") + @mock.patch("interactive_calibrator.read_zoom_after_move") + def test_step_four_always_sends_zoom_even_when_already_equal( + self, move_mock, camera_mock, _print_mock + ): + origin = {"pan": 100, "tilt": 200, "zoom": 4580} + camera_mock.return_value = origin + move_mock.return_value = origin + + actual, _average = wizard.move_to_scan_zoom({ + "top-left": origin, + "bottom-right": origin, + }) + + self.assertEqual(actual["zoom"], 4580) + move_mock.assert_called_once_with(origin, 4580) + + def test_partial_session_is_saved_to_history(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + history = root / "history" + state_file = root / "calibration.json" + with ( + mock.patch.object(wizard, "HISTORY_DIR", history), + mock.patch.object(wizard, "ACTIVE_HISTORY_FILE", None), + mock.patch.object(core, "STATE_FILE", state_file), + ): + state = wizard.start_history_session("hanging", None) + state["points"]["top-left"] = { + "pan": 21720, + "tilt": 1268, + "zoom": 4580, + } + wizard.save_working_state(state) + + records = wizard.load_history_records() + + self.assertEqual(len(records), 1) + self.assertTrue( + wizard.state_item_exists(records[0][1], "points", "top-left") + ) + self.assertFalse( + wizard.state_item_exists(records[0][1], "points", "bottom-right") + ) + + +class GenerateZoomTests(unittest.TestCase): + def test_generate_uses_camera_reachable_scan_zoom(self): + state = { + "points": { + "top-left": {"pan": 1000, "tilt": 200, "zoom": 4550}, + "bottom-right": {"pan": 900, "tilt": 100, "zoom": 4580}, + }, + "steps": { + "pan": { + "value": 50, + "overlap": 0.04, + "target_overlap": 0.04, + "overlap_method": "axis_shift_v1_retargeted", + "inliers": 50, + "visual_direction": "decreasing", + }, + "tilt": { + "value": 50, + "overlap": 0.04, + "target_overlap": 0.04, + "overlap_method": "axis_shift_v1_retargeted", + "inliers": 50, + }, + }, + "scan_zoom": 4580, + } + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + with ( + mock.patch.object(core, "ROOT", root), + mock.patch.object(core, "load_state", return_value=state), + ): + with mock.patch("sys.stdout", new=io.StringIO()): + core.command_generate(argparse.Namespace(mount="desktop")) + result = next((root / "result").glob("scan_config_*.json")) + generated = json.loads(result.read_text(encoding="utf-8")) + + self.assertEqual(generated["scan"]["zoom"], 4580) + + if __name__ == "__main__": unittest.main()