first commit

scan_calibrator.py

@@ -0,0 +1,471 @@
+#!/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
+
+
+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 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))
+
+            minimum_plausible_shift = delta / 18000.0
+            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"),
+    }
+    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) -> tuple[int, int, int, str]:
+    """Encode the user-selected visual left-to-right arc for the current scanner.
+
+    With the tested desktop setup, firmware rotates the image by 180 degrees, so
+    visual left-to-right is decreasing PTZ pan. The current scanner only executes
+    increasing pan, therefore reversing both endpoints traverses the exact same
+    selected arc in the opposite order. A normally hanging camera uses the native
+    increasing direction.
+    """
+    top_left %= 36000
+    bottom_right %= 36000
+    if mount == "desktop":
+        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)
+        if inliers < 50:
+            fail(
+                f"{axis}_step标定只有{inliers}个几何内点，低于可靠阈值50；"
+                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
+    )
+    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()