635 lines
26 KiB
Python
Executable File
635 lines
26 KiB
Python
Executable File
#!/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
|
||
API_MAX_ATTEMPTS = 3
|
||
API_RECOVERY_TIMEOUT = 90.0
|
||
API_RECOVERY_POLL_INTERVAL = 2.0
|
||
TRANSIENT_CURL_EXIT_CODES = {7, 28, 52, 55, 56}
|
||
_HELPER_READY = False
|
||
|
||
|
||
def fail(message: str) -> "NoReturn":
|
||
raise SystemExit(f"错误：{message}")
|
||
|
||
|
||
def warn(message: str) -> None:
|
||
print(f"警告：{message}", file=sys.stderr, flush=True)
|
||
|
||
|
||
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_command(method: str, path: str, body: dict | None, timeout: int) -> list[str]:
|
||
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}")
|
||
return command
|
||
|
||
|
||
def transient_api_failure(result: subprocess.CompletedProcess[str]) -> bool:
|
||
if result.returncode in TRANSIENT_CURL_EXIT_CODES:
|
||
return True
|
||
detail = f"{result.stdout}\n{result.stderr}".lower()
|
||
return any(marker in detail for marker in (
|
||
"container is not running",
|
||
"container is restarting",
|
||
"no such container",
|
||
"couldn't connect to server",
|
||
"connection reset by peer",
|
||
"empty reply from server",
|
||
"returned error: 502",
|
||
"returned error: 503",
|
||
"returned error: 504",
|
||
))
|
||
|
||
|
||
def scan_http_ready() -> bool:
|
||
try:
|
||
state = subprocess.run(
|
||
["docker", "inspect", "-f", "{{.State.Running}}", "smart-cam"],
|
||
text=True, capture_output=True, timeout=5,
|
||
)
|
||
if state.returncode or state.stdout.strip() != "true":
|
||
return False
|
||
health = subprocess.run(
|
||
["docker", "exec", "smart-cam", "curl", "-sS", "--fail",
|
||
"--max-time", "5", "-o", "/dev/null",
|
||
"http://127.0.0.1:18080/health"],
|
||
text=True, capture_output=True, timeout=10,
|
||
)
|
||
return health.returncode == 0
|
||
except (OSError, subprocess.TimeoutExpired):
|
||
return False
|
||
|
||
|
||
def wait_for_scan_http(timeout: float = API_RECOVERY_TIMEOUT) -> bool:
|
||
deadline = time.monotonic() + timeout
|
||
while time.monotonic() < deadline:
|
||
if scan_http_ready():
|
||
return True
|
||
time.sleep(API_RECOVERY_POLL_INTERVAL)
|
||
return False
|
||
|
||
|
||
def api(method: str, path: str, body: dict | None = None, timeout: int = 60) -> dict:
|
||
global _HELPER_READY
|
||
command = api_command(method, path, body, timeout)
|
||
last_detail = ""
|
||
for attempt in range(1, API_MAX_ATTEMPTS + 1):
|
||
try:
|
||
result = subprocess.run(
|
||
command, text=True, capture_output=True, timeout=timeout + 10,
|
||
)
|
||
except subprocess.TimeoutExpired:
|
||
result = subprocess.CompletedProcess(command, 28, "", "请求执行超时")
|
||
if result.returncode == 0:
|
||
try:
|
||
return json.loads(result.stdout)
|
||
except json.JSONDecodeError:
|
||
fail(f"scan HTTP返回的不是JSON：{result.stdout}")
|
||
|
||
last_detail = result.stderr.strip() or result.stdout.strip() or (
|
||
f"退出码={result.returncode}"
|
||
)
|
||
if not transient_api_failure(result) or attempt == API_MAX_ATTEMPTS:
|
||
break
|
||
warn(
|
||
f"scan HTTP或smart-cam容器短暂不可用（{last_detail}），"
|
||
f"正在等待恢复后重试当前操作 "
|
||
f"[{attempt}/{API_MAX_ATTEMPTS - 1}]"
|
||
)
|
||
if not wait_for_scan_http():
|
||
last_detail += f"；等待{API_RECOVERY_TIMEOUT:.0f}秒后scan HTTP仍未恢复"
|
||
break
|
||
# The container might have been replaced instead of restarted. Recheck
|
||
# the in-container overlap helper before the next image comparison.
|
||
_HELPER_READY = False
|
||
|
||
fail(f"命令失败：{' '.join(command)}\n{last_detail}")
|
||
|
||
|
||
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:
|
||
global _HELPER_READY
|
||
if _HELPER_READY:
|
||
existing = subprocess.run(
|
||
["docker", "exec", "smart-cam", "test", "-x", HELPER_CONTAINER_BIN],
|
||
text=True, capture_output=True, timeout=10,
|
||
)
|
||
if existing.returncode == 0:
|
||
return
|
||
_HELPER_READY = False
|
||
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)
|
||
_HELPER_READY = True
|
||
|
||
|
||
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})
|
||
except BaseException:
|
||
try:
|
||
move(origin)
|
||
except BaseException as restore_error:
|
||
warn(f"当前样本失败后暂时无法返回标定起点：{restore_error}")
|
||
raise
|
||
else:
|
||
move(origin)
|
||
return result
|
||
|
||
|
||
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}"
|
||
)
|
||
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"]),
|
||
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()
|