Files
scan-calibrator/scan_calibrator.py
T
2026-08-04 17:46:08 +08:00

520 lines
22 KiB
Python
Executable File
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/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}"
)
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()