#!/usr/bin/env python3 """Interactive scan PTZ and overlap calibration assistant for smart-cam.""" from __future__ import annotations import argparse from datetime import datetime import json import math from pathlib import Path import subprocess import sys import time ROOT = Path(__file__).resolve().parent STATE_FILE = ROOT / "calibration.json" HELPER_SOURCE = ROOT / "src/measure_overlap.cpp" HELPER_CONTAINER_SOURCE = "/tmp/scan-calibrator-measure.cpp" HELPER_CONTAINER_BIN = "/tmp/scan-calibrator-measure" DEFAULT_TARGET_OVERLAP = 0.04 MAX_GENERATED_OVERLAP = 0.05 PHASE_MIN_RESPONSE = 0.20 def fail(message: str) -> "NoReturn": raise SystemExit(f"错误：{message}") def run(command: list[str], *, timeout: int = 90) -> subprocess.CompletedProcess[str]: result = subprocess.run(command, text=True, capture_output=True, timeout=timeout) if result.returncode: fail(f"命令失败：{' '.join(command)}\n{result.stderr.strip()}") return result def api(method: str, path: str, body: dict | None = None, timeout: int = 60) -> dict: command = ["docker", "exec", "smart-cam", "curl", "-sS", "--fail-with-body", "--max-time", str(timeout), "-X", method] if body is not None: command += ["-H", "Content-Type: application/json", "-d", json.dumps(body)] command.append(f"http://127.0.0.1:18080{path}") result = run(command, timeout=timeout + 10) try: return json.loads(result.stdout) except json.JSONDecodeError: fail(f"scan HTTP返回的不是JSON：{result.stdout}") 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")} 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")} def snapshot() -> str: return str(api("POST", "/api/v1/snapshot", timeout=30)["file"]) def load_state() -> dict: if not STATE_FILE.exists(): return {"points": {}, "steps": {}} try: state = json.loads(STATE_FILE.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: fail(f"无法读取{STATE_FILE}：{exc}") state.setdefault("points", {}) state.setdefault("steps", {}) return state def save_state(state: dict) -> None: state["updated_at"] = datetime.now().astimezone().isoformat(timespec="seconds") temporary = STATE_FILE.with_suffix(".json.tmp") temporary.write_text(json.dumps(state, ensure_ascii=False, indent=2), encoding="utf-8") temporary.replace(STATE_FILE) def ensure_helper() -> None: run(["docker", "cp", str(HELPER_SOURCE), f"smart-cam:{HELPER_CONTAINER_SOURCE}"]) run(["docker", "exec", "smart-cam", "sh", "-lc", f"g++ -O2 -std=c++17 {HELPER_CONTAINER_SOURCE} -o {HELPER_CONTAINER_BIN} " "$(pkg-config --cflags --libs opencv4)"], timeout=120) def measure_images(first: str, second: str) -> dict: ensure_helper() result = run(["docker", "exec", "smart-cam", HELPER_CONTAINER_BIN, first, second], timeout=60) try: return json.loads(result.stdout) except json.JSONDecodeError: fail(f"重叠分析器返回异常：{result.stdout} {result.stderr}") def shifted(origin: dict[str, int], axis: str, delta: int, direction: int) -> dict[str, int]: target = dict(origin) if axis == "pan": target["pan"] = (origin["pan"] + direction * delta) % 36000 else: target["tilt"] = origin["tilt"] + direction * delta return target def print_ptz(ptz: dict[str, int], prefix: str = "PTZ") -> None: print(f"{prefix}: pan={ptz['pan']} ({ptz['pan']/100:.2f}°), " f"tilt={ptz['tilt']} ({ptz['tilt']/100:.2f}°), " f"zoom={ptz['zoom']} ({ptz['zoom']/100:.2f}×)") def measure_delta(axis: str, delta: int, direction: int, wait_seconds: float, baseline: str | None = None, origin: dict[str, int] | None = None) -> dict: origin = origin or get_ptz() baseline = baseline or snapshot() target = shifted(origin, axis, delta, direction) try: actual = move(target) time.sleep(wait_seconds) candidate = snapshot() result = measure_images(baseline, candidate) if axis == "pan": actual_motion = abs( ((actual["pan"] - origin["pan"] + 18000) % 36000) - 18000 ) else: actual_motion = abs(actual["tilt"] - origin["tilt"]) result["actual_axis_motion"] = actual_motion result["requested_axis_motion"] = delta phase_primary_key = "phase_dx" if axis == "pan" else "phase_dy" phase_cross_key = "phase_dy" if axis == "pan" else "phase_dx" phase_primary = abs(float(result.get(phase_primary_key, 0.0))) phase_cross = abs(float(result.get(phase_cross_key, 0.0))) phase_response = float(result.get("phase_response", 0.0)) minimum_plausible_shift = delta / 18000.0 if ( not result.get("valid") and phase_response >= PHASE_MIN_RESPONSE and minimum_plausible_shift <= phase_primary <= 0.50 and phase_cross <= 0.05 ): result.update({ "valid": True, "reason": "phase_correlation_fallback", "match_method": "phase_correlation", "center_dx": float(result.get("phase_dx", 0.0)), "center_dy": float(result.get("phase_dy", 0.0)), }) # Absolute PTZ commands can be clamped by a camera's mechanical/firmware # limit. In that case identical images are not a feature-texture failure. if actual_motion < max(2, round(delta * 0.5)): result["valid"] = False result["reason"] = "ptz_motion_shortfall" if result.get("valid"): primary_key = "center_dx" if axis == "pan" else "center_dy" cross_key = "center_dy" if axis == "pan" else "center_dx" primary_shift = abs(float(result.get(primary_key, 0.0))) cross_shift = abs(float(result.get(cross_key, 0.0))) result["polygon_overlap"] = result.get("overlap") result["axis_shift"] = primary_shift result["cross_axis_shift"] = cross_shift # Step calibration is one-dimensional. Polygon intersection mixes # horizontal and vertical effects and is easily fooled by repeated # wall patterns. Use displacement along the commanded axis instead. result["overlap"] = max(0.0, min(1.0, 1.0 - primary_shift)) result["minimum_plausible_shift"] = minimum_plausible_shift if primary_shift < minimum_plausible_shift: result["valid"] = False result["reason"] = "axis_motion_too_small" elif cross_shift > 0.20 or primary_shift > 1.25: result["valid"] = False result["reason"] = "axis_motion_inconsistent" result.update({"axis": axis, "delta": delta, "direction": direction, "origin": origin, "target": target, "actual": actual, "baseline_file": baseline, "candidate_file": candidate}) return result finally: move(origin) def command_ptz(_: argparse.Namespace) -> None: print_ptz(get_ptz()) def command_mark(args: argparse.Namespace) -> None: state = load_state() ptz = get_ptz() state["points"][args.name] = ptz save_state(state) print_ptz(ptz, f"已记录 {args.name}") print(f"记录文件：{STATE_FILE}") def command_show(_: argparse.Namespace) -> None: state = load_state() print(json.dumps(state, ensure_ascii=False, indent=2)) def command_retarget(args: argparse.Namespace) -> None: if not 0.0 <= args.target_overlap <= MAX_GENERATED_OVERLAP: fail("当前项目要求target-overlap必须在0.00～0.05之间") state = load_state() for axis in ("pan", "tilt"): try: previous = state["steps"][axis] old_step = int(previous["value"]) old_overlap = float(previous["overlap"]) except (KeyError, TypeError, ValueError) as exc: fail(f"缺少可靠的{axis}历史标定：{exc}") if not str(previous.get("overlap_method", "")).startswith("axis_shift_v1"): fail(f"{axis}不是新版单轴位移标定结果，不能反算") if int(previous.get("inliers") or 0) < 50 or not 0.0 <= old_overlap < 0.90: fail(f"{axis}历史标定质量不足，不能安全反算") estimated_fov = old_step / (1.0 - old_overlap) new_step = max(1, round(estimated_fov * (1.0 - args.target_overlap))) state["steps"][axis] = { **previous, "value": new_step, "overlap": args.target_overlap, "target_overlap": args.target_overlap, "overlap_method": "axis_shift_v1_retargeted", "derived_from": { "value": old_step, "overlap": old_overlap, "estimated_fov": round(estimated_fov, 4), }, } print( f"{axis}: step {old_step}（实测{old_overlap*100:.1f}%）" f" → {new_step}（反算目标{args.target_overlap*100:.1f}%）" ) save_state(state) print("反算结果已写入calibration.json；它必须通过一次完整扫图复核。") def command_measure(args: argparse.Namespace) -> None: origin = get_ptz() print_ptz(origin, "标定起点") result = measure_delta(args.axis, args.delta, args.direction, args.wait) print(json.dumps(result, ensure_ascii=False, indent=2)) if result.get("valid"): print(f"候选step={args.delta}，实测画面重叠率={result['overlap']*100:.1f}%") if args.axis == "pan": view = "右" if float(result.get("center_dx", 0)) < 0 else "左" else: view = "下" if float(result.get("center_dy", 0)) < 0 else "上" print(f"按direction={args.direction}移动时，摄像头视野朝画面{view}侧推进。") else: print("当前画面纹理不足或位移过大，无法可靠匹配；请换到纹理丰富的中间位置或减小delta。") def command_calibrate(args: argparse.Namespace) -> None: if args.minimum <= 0 or args.maximum <= args.minimum: fail("minimum/maximum范围无效") if not 0.0 <= args.target_overlap <= MAX_GENERATED_OVERLAP: fail("当前项目要求target-overlap必须在0.00～0.05之间") origin = get_ptz() baseline = snapshot() print_ptz(origin, "自动标定起点") print(f"基准图：{baseline}") low, high = args.minimum, args.maximum tested = [] best = None for number in range(args.iterations): candidate = round((low + high) / 2) result = measure_delta(args.axis, candidate, args.direction, args.wait, baseline=baseline, origin=origin) tested.append(result) if result.get("valid"): overlap = float(result["overlap"]) error = abs(overlap - args.target_overlap) if best is None or error < best[0]: best = (error, candidate, overlap, result) print(f"[{number+1}/{args.iterations}] step={candidate} overlap={overlap*100:.1f}% " f"inliers={result.get('inliers', 0)}") if overlap > args.target_overlap: low = candidate + 1 else: high = candidate - 1 else: print(f"[{number+1}/{args.iterations}] step={candidate} 匹配失败，缩小步长") high = candidate - 1 if low > high: break if best is None: fail("所有候选步长都无法可靠匹配，请换到纹理丰富的位置并缩小maximum") # Binary search can fall back to a reliable result that is still far from # the requested overlap when larger candidates lose feature matches. # Estimate the one-axis FOV from the best measurement, calculate a corrected # step, and close the loop with up to three real captures. for refinement in range(3): _, current_step, current_overlap, _ = best denominator = max(0.05, 1.0 - current_overlap) corrected = round(current_step * (1.0 - args.target_overlap) / denominator) corrected = max(args.minimum, min(args.maximum, corrected)) if corrected == current_step: break result = measure_delta( args.axis, corrected, args.direction, args.wait, baseline=baseline, origin=origin ) if not result.get("valid"): print(f"[闭环{refinement+1}/3] step={corrected} 匹配失败，保留当前最优结果") break overlap = float(result["overlap"]) error = abs(overlap - args.target_overlap) print( f"[闭环{refinement+1}/3] step={corrected} " f"overlap={overlap*100:.1f}% inliers={result.get('inliers', 0)}" ) if error < best[0]: best = (error, corrected, overlap, result) else: break _, step, overlap, detail = best if abs(overlap - args.target_overlap) > 0.10: fail( f"最优可靠结果的重叠率仍为{overlap*100:.1f}%，" f"与目标{args.target_overlap*100:.1f}%偏差超过10个百分点；" "本次结果不保存，请更换非重复纹理位置或扩大minimum/maximum范围" ) state = load_state() state["steps"][args.axis] = { "value": step, "direction": args.direction, "overlap": overlap, "target_overlap": args.target_overlap, "overlap_method": "axis_shift_v1", "zoom": origin["zoom"], "measured_at": origin, "good_matches": detail.get("good_matches"), "inliers": detail.get("inliers"), "inlier_ratio": detail.get("inlier_ratio"), } if args.axis == "pan": state["steps"][args.axis]["visual_direction"] = ( "increasing" if args.direction == 1 else "decreasing" ) save_state(state) print(f"推荐 {args.axis}_step={step}（{step/100:.2f}°），实测重叠={overlap*100:.1f}%") print("推荐值已写入 calibration.json") def scan_pan_range( top_left: int, bottom_right: int, mount: str, visual_direction: str | None = None, ) -> tuple[int, int, int, str]: """Encode the user-selected visual left-to-right arc for the current scanner. visual_direction should be measured from signed image displacement on this exact camera model. mount remains only as a backward-compatible fallback for older manual calibration state. """ top_left %= 36000 bottom_right %= 36000 if visual_direction is None: visual_direction = "decreasing" if mount == "desktop" else "increasing" if visual_direction not in {"increasing", "decreasing"}: fail(f"未知pan视觉方向：{visual_direction}") if visual_direction == "decreasing": span = (top_left - bottom_right) % 36000 return bottom_right, top_left, span, "decreasing" span = (bottom_right - top_left) % 36000 return top_left, bottom_right, span, "increasing" def point_count(start: int, end: int, step: int, *, wrap: bool = False) -> int: distance = (end - start) % 36000 if wrap else end - start return distance // step + 1 + (1 if distance % step else 0) def expanded_grid(start: int, span: int, step: int, *, wrap: bool = False) -> tuple[int, int, int, int]: """Expand the end outward to a whole number of steps; never shrink coverage.""" if step <= 0: fail("step必须大于0") if span == 0: return start, step, 1, 0 intervals = max(1, math.ceil(span / step)) expanded_span = step * intervals extension = expanded_span - span end = start + expanded_span if wrap: end %= 36000 return end, step, intervals + 1, extension def command_generate(args: argparse.Namespace) -> None: state = load_state() try: top_left = state["points"]["top-left"] bottom_right = state["points"]["bottom-right"] pan_info = state["steps"]["pan"] tilt_info = state["steps"]["tilt"] except KeyError as exc: fail(f"缺少标定项{exc}；需要top-left、bottom-right、pan step和tilt step") for axis, info in (("pan", pan_info), ("tilt", tilt_info)): if not str(info.get("overlap_method", "")).startswith("axis_shift_v1"): fail( f"{axis}_step来自旧版二维重叠算法；" f"请使用新版工具重新执行calibrate {axis}" ) target_overlap = float(info.get("target_overlap", 1.0)) measured_overlap = float(info.get("overlap", 1.0)) if target_overlap > MAX_GENERATED_OVERLAP or measured_overlap > MAX_GENERATED_OVERLAP: fail( f"{axis}_step的目标/实测重叠为" f"{target_overlap*100:.1f}%/{measured_overlap*100:.1f}%，超过5%上限；" f"请使用--target-overlap {DEFAULT_TARGET_OVERLAP:.2f}重新标定" ) inliers = int(info.get("inliers") or 0) phase_response = float(info.get("phase_response") or 0.0) if inliers < 50 and phase_response < PHASE_MIN_RESPONSE: fail( f"{axis}_step标定质量不足（几何内点={inliers}，" f"相位响应={phase_response:.2f}）；" f"请在纹理丰富且尽量远离摄像头的位置重新执行calibrate {axis}" ) if abs(int(top_left["zoom"]) - int(bottom_right["zoom"])) > 10: fail("左上角和右下角zoom不一致，请固定zoom后重新记录两个角点") zoom = round((int(top_left["zoom"]) + int(bottom_right["zoom"])) / 2) pan_start, pan_end, pan_span, visual_pan_direction = scan_pan_range( int(top_left["pan"]), int(bottom_right["pan"]), args.mount, pan_info.get("visual_direction"), ) tilt_start = min(int(top_left["tilt"]), int(bottom_right["tilt"])) tilt_end = max(int(top_left["tilt"]), int(bottom_right["tilt"])) nominal_pan_step = int(pan_info["value"]) nominal_tilt_step = int(tilt_info["value"]) pan_end, pan_step, columns, pan_extension = expanded_grid( pan_start, pan_span, nominal_pan_step, wrap=True ) tilt_span = tilt_end - tilt_start tilt_end, tilt_step, rows, tilt_extension = expanded_grid( tilt_start, tilt_span, nominal_tilt_step ) config = { "pan_start": pan_start, "pan_end": pan_end, "pan_step": pan_step, # The current scan service expects an increasing numeric tilt range. # Physical/image orientation is handled by scan-panorama --mount. "tilt_start": tilt_start, "tilt_end": tilt_end, "tilt_step": tilt_step, "zoom": zoom, } result_dir = ROOT / "result" result_dir.mkdir(exist_ok=True) output = result_dir / ( f"scan_config_{args.mount}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json" ) output.write_text(json.dumps({"scan": config}, ensure_ascii=False, indent=2), encoding="utf-8") print(json.dumps(config, ensure_ascii=False, indent=2)) print(f"安装方式：{args.mount}") print(f"范围检查：视觉左→右pan={visual_pan_direction}，指定弧段={pan_span/100:.2f}°") print(f"整步扩边：pan_step={pan_step}，向终点外扩={pan_extension}（{pan_extension/100:.2f}°）") print(f"整步扩边：tilt_step={tilt_step}，向终点外扩={tilt_extension}（{tilt_extension/100:.2f}°）") print(f"预计网格：{columns}列 × {rows}行 = {columns * rows}张") print( "验证拼图：/home/zhangw/scan-calibrator/scan-panorama/" f"stitch.sh <扫描目录> --mount {args.mount}" ) print(f"配置片段：{output}") def parser() -> argparse.ArgumentParser: root = argparse.ArgumentParser(description="读取PTZ、记录扫描角点并按实拍重叠率标定scan步长。") sub = root.add_subparsers(dest="command", required=True) sub.add_parser("ptz", help="读取当前摄像头PTZ").set_defaults(func=command_ptz) mark = sub.add_parser("mark", help="记录当前PTZ为一个标定点") mark.add_argument("name", choices=["top-left", "bottom-right"]) mark.set_defaults(func=command_mark) sub.add_parser("show", help="显示已保存的角点和step").set_defaults(func=command_show) retarget = sub.add_parser("retarget", help="用可靠历史测量反算不超过百分之五的step") retarget.add_argument( "--target-overlap", type=float, default=DEFAULT_TARGET_OVERLAP, help="目标重叠率，默认0.04", ) retarget.set_defaults(func=command_retarget) for name, function in (("measure", command_measure), ("calibrate", command_calibrate)): item = sub.add_parser(name, help="测量单个候选step" if name == "measure" else "二分搜索推荐step") item.add_argument("axis", choices=["pan", "tilt"]) item.add_argument("--direction", type=int, choices=[-1, 1], required=True, help="从画面左到右/上到下移动时PTZ数值方向") item.add_argument("--wait", type=float, default=1.0, help="到位后抓图前等待秒数") if name == "measure": item.add_argument("--delta", type=int, required=True, help="候选step，单位0.01度") else: item.add_argument("--minimum", type=int, required=True) item.add_argument("--maximum", type=int, required=True) item.add_argument( "--target-overlap", type=float, default=DEFAULT_TARGET_OVERLAP, help="目标重叠率，默认0.04，为百分之五上限保留1%余量" ) item.add_argument("--iterations", type=int, default=6) item.set_defaults(func=function) generate = sub.add_parser("generate", help="根据角点和标定step生成scan配置片段") generate.add_argument( "--mount", choices=["desktop", "hanging"], required=True, help="desktop=桌面正放且后台旋转180度；hanging=野外倒挂", ) generate.set_defaults(func=command_generate) return root def main() -> None: args = parser().parse_args() args.func(args) if __name__ == "__main__": main()