469 lines
18 KiB
Python
Executable File
469 lines
18 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""Fixed-dialogue field deployment wizard for scan calibration."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
from contextlib import redirect_stdout
|
||
from datetime import datetime
|
||
import io
|
||
import json
|
||
from pathlib import Path
|
||
import re
|
||
import subprocess
|
||
import sys
|
||
import tempfile
|
||
import time
|
||
|
||
import scan_calibrator as core
|
||
|
||
|
||
ROOT = Path(__file__).resolve().parent
|
||
CONFIG_PATH = "/smart-cam/config/config.json"
|
||
TARGET_OVERLAPS = {"pan": 0.01, "tilt": 0.04}
|
||
RED = "\033[31m"
|
||
GREEN = "\033[32m"
|
||
YELLOW = "\033[33m"
|
||
RESET = "\033[0m"
|
||
|
||
|
||
class Cancelled(Exception):
|
||
pass
|
||
|
||
|
||
def say(message: str = "", *, color: str = "") -> None:
|
||
print(f"{color}{message}{RESET if color else ''}", flush=True)
|
||
|
||
|
||
def choose(question: str, choices: set[str]) -> str:
|
||
while True:
|
||
answer = input(f"\n{question}\n> ").strip()
|
||
if answer == "3":
|
||
raise Cancelled
|
||
if answer in choices:
|
||
return answer
|
||
say(f"请输入 {' / '.join(sorted(choices | {'3'}))}。", color=YELLOW)
|
||
|
||
|
||
def confirm_position(question: str) -> None:
|
||
choose(f"{question}\n到位后回复 1；回复 3 随时退出。", {"1"})
|
||
|
||
|
||
def camera_ptz() -> dict[str, int]:
|
||
ptz = core.get_ptz()
|
||
core.print_ptz(ptz, "已读取")
|
||
return ptz
|
||
|
||
|
||
def remove_session_snapshots(paths: set[str]) -> None:
|
||
for path in sorted(paths):
|
||
if not path.startswith("/smart-cam/data/scan/calibration/"):
|
||
continue
|
||
subprocess.run(
|
||
["docker", "exec", "smart-cam", "rm", "-f", "--", path],
|
||
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False,
|
||
)
|
||
|
||
|
||
def candidate_deltas(axis: str) -> list[int]:
|
||
if axis == "pan":
|
||
return [150, 250, 350, 450, 600, 800, 1100, 1500, 2100]
|
||
return [80, 120, 170, 220, 280, 350, 450, 600, 850, 1150]
|
||
|
||
|
||
def choose_direction(axis: str, origin: dict[str, int], points: dict) -> int:
|
||
if axis == "pan":
|
||
# Pan is circular. The direction changes only the sign, not the FOV estimate.
|
||
return 1
|
||
low = min(int(points["top-left"]["tilt"]), int(points["bottom-right"]["tilt"]))
|
||
high = max(int(points["top-left"]["tilt"]), int(points["bottom-right"]["tilt"]))
|
||
return 1 if high - origin["tilt"] >= origin["tilt"] - low else -1
|
||
|
||
|
||
def estimate_axis(axis: str, origin: dict[str, int], points: dict,
|
||
snapshots: set[str]) -> dict:
|
||
direction = choose_direction(axis, origin, points)
|
||
baseline = core.snapshot()
|
||
snapshots.add(baseline)
|
||
reliable: list[tuple[int, float, float, dict]] = []
|
||
say(f"\n正在标定 {axis}_step，摄像头会自动移动并回到起点……")
|
||
|
||
for delta in candidate_deltas(axis):
|
||
result = core.measure_delta(
|
||
axis, delta, direction, 1.0, baseline=baseline, origin=origin
|
||
)
|
||
snapshots.add(str(result.get("candidate_file", "")))
|
||
if not result.get("valid"):
|
||
say(f" step={delta}: 当前图片匹配不足，继续尝试")
|
||
continue
|
||
overlap = float(result["overlap"])
|
||
inliers = int(result.get("inliers") or 0)
|
||
ratio = float(result.get("inlier_ratio") or 0)
|
||
say(f" step={delta}: 重叠约 {overlap*100:.1f}%，"
|
||
f"有效特征={inliers}，内点率={ratio*100:.1f}%")
|
||
if inliers >= 50 and ratio >= 0.45 and 0.08 <= overlap <= 0.80:
|
||
reliable.append((delta, overlap, delta / (1.0 - overlap), result))
|
||
|
||
if len(reliable) < 2:
|
||
core.move(origin)
|
||
core.fail(
|
||
f"{axis}方向只得到{len(reliable)}个可靠样本，至少需要2个不同移动量互相验证。"
|
||
"请把摄像头移到远处、纹理丰富、非重复图案的位置后重新运行向导"
|
||
)
|
||
|
||
# Fixed OSD text and repeated scenery can yield a convincing false
|
||
# homography. A real optical FOV must agree across different PTZ deltas.
|
||
clusters = [
|
||
[
|
||
sample for sample in reliable
|
||
if abs(sample[2] - seed[2]) / max(seed[2], 1.0) <= 0.12
|
||
]
|
||
for seed in reliable
|
||
]
|
||
consistent = max(
|
||
clusters, key=lambda cluster: (len(cluster), -max(x[2] for x in cluster))
|
||
)
|
||
if len(consistent) < 2:
|
||
core.move(origin)
|
||
values = ", ".join(f"{sample[2]:.0f}" for sample in reliable)
|
||
core.fail(
|
||
f"{axis}方向不同移动量反算的视场不一致（{values}），疑似重复图案误匹配；"
|
||
"本次结果不写入配置，请更换标定画面"
|
||
)
|
||
|
||
fovs = sorted(sample[2] for sample in consistent)
|
||
estimated_fov = (
|
||
fovs[len(fovs) // 2] if len(fovs) % 2
|
||
else (fovs[len(fovs) // 2 - 1] + fovs[len(fovs) // 2]) / 2
|
||
)
|
||
spread = (max(fovs) - min(fovs)) / estimated_fov
|
||
if len(consistent) == 2:
|
||
strong_pair = all(
|
||
int(sample[3].get("inliers") or 0) >= 100
|
||
and float(sample[3].get("inlier_ratio") or 0) >= 0.65
|
||
for sample in consistent
|
||
)
|
||
allowed_spread = 0.08
|
||
if not strong_pair:
|
||
core.move(origin)
|
||
core.fail(
|
||
f"{axis}方向目前只有2个一致样本，但特征质量不足以使用双样本规则；"
|
||
"需要第3个可靠样本，请更换标定画面"
|
||
)
|
||
else:
|
||
allowed_spread = 0.15
|
||
if spread > allowed_spread:
|
||
core.move(origin)
|
||
core.fail(
|
||
f"{axis}方向视场样本离散度为{spread*100:.1f}%，"
|
||
f"超过{allowed_spread*100:.0f}%；"
|
||
"本次结果不写入配置，请更换标定画面"
|
||
)
|
||
|
||
target_overlap = TARGET_OVERLAPS[axis]
|
||
step = max(1, round(estimated_fov * (1.0 - target_overlap)))
|
||
measured_step, measured_overlap, _, detail = max(
|
||
consistent, key=lambda sample: int(sample[3].get("inliers") or 0)
|
||
)
|
||
core.move(origin)
|
||
say(
|
||
f" {axis}视场一致性通过：{len(consistent)}个样本，"
|
||
f"估算视场={estimated_fov:.0f}，离散度={spread*100:.1f}%"
|
||
)
|
||
return {
|
||
"value": step,
|
||
"direction": direction,
|
||
"overlap": target_overlap,
|
||
"target_overlap": target_overlap,
|
||
"overlap_method": "axis_shift_v1_retargeted",
|
||
"zoom": origin["zoom"],
|
||
"measured_at": origin,
|
||
"good_matches": detail.get("good_matches"),
|
||
"inliers": detail.get("inliers"),
|
||
"inlier_ratio": detail.get("inlier_ratio"),
|
||
"derived_from": {
|
||
"value": measured_step,
|
||
"overlap": measured_overlap,
|
||
"estimated_fov": round(estimated_fov, 4),
|
||
"samples": [
|
||
{
|
||
"value": sample[0],
|
||
"overlap": round(sample[1], 6),
|
||
"estimated_fov": round(sample[2], 4),
|
||
"inliers": sample[3].get("inliers"),
|
||
"inlier_ratio": sample[3].get("inlier_ratio"),
|
||
}
|
||
for sample in consistent
|
||
],
|
||
"spread": round(spread, 6),
|
||
},
|
||
}
|
||
|
||
|
||
def calculate_steps(points: dict, snapshots: set[str], mount: str) -> None:
|
||
origin = camera_ptz()
|
||
zooms = [int(points[name]["zoom"]) for name in ("top-left", "bottom-right")]
|
||
if max(zooms + [origin["zoom"]]) - min(zooms + [origin["zoom"]]) > 10:
|
||
core.fail("三个位置的zoom不一致；请固定焦段后重新开始")
|
||
top_left = points["top-left"]
|
||
bottom_right = points["bottom-right"]
|
||
pan_start, _, pan_span, _ = core.scan_pan_range(
|
||
int(top_left["pan"]), int(bottom_right["pan"]), mount
|
||
)
|
||
pan_inside = (origin["pan"] - pan_start) % 36000 <= pan_span
|
||
tilt_low = min(int(top_left["tilt"]), int(bottom_right["tilt"]))
|
||
tilt_high = max(int(top_left["tilt"]), int(bottom_right["tilt"]))
|
||
tilt_inside = tilt_low <= origin["tilt"] <= tilt_high
|
||
if not pan_inside or not tilt_inside:
|
||
say(
|
||
"警告：当前标定位置不在刚才选择的扫描区域内。"
|
||
f"当前pan={origin['pan']}、tilt={origin['tilt']}；"
|
||
f"扫描区域tilt={tilt_low}～{tilt_high}。"
|
||
"同zoom下仍可尝试，但优先选择扫描区域内部的远处纹理。",
|
||
color=YELLOW,
|
||
)
|
||
state = core.load_state()
|
||
state["steps"]["pan"] = estimate_axis("pan", origin, points, snapshots)
|
||
state["steps"]["tilt"] = estimate_axis("tilt", origin, points, snapshots)
|
||
core.save_state(state)
|
||
|
||
|
||
def generate(mount: str) -> tuple[dict, Path, str]:
|
||
before = set((ROOT / "result").glob("scan_config_*.json"))
|
||
output = io.StringIO()
|
||
with redirect_stdout(output):
|
||
core.command_generate(argparse.Namespace(mount=mount))
|
||
after = set((ROOT / "result").glob("scan_config_*.json"))
|
||
created = sorted(after - before, key=lambda path: path.stat().st_mtime)
|
||
if not created:
|
||
core.fail("没有找到本次生成的配置文件")
|
||
result_file = created[-1]
|
||
scan = json.loads(result_file.read_text(encoding="utf-8"))["scan"]
|
||
return scan, result_file, output.getvalue()
|
||
|
||
|
||
def deploy_config(scan: dict) -> None:
|
||
raw = subprocess.run(
|
||
["docker", "exec", "smart-cam", "cat", CONFIG_PATH],
|
||
text=True, capture_output=True, check=True,
|
||
).stdout
|
||
config = json.loads(raw)
|
||
if not isinstance(config.get("scan"), dict):
|
||
raise RuntimeError(f"{CONFIG_PATH}缺少scan对象")
|
||
|
||
config["scan"].update(scan)
|
||
|
||
with tempfile.NamedTemporaryFile(
|
||
mode="w", encoding="utf-8", suffix=".json", delete=False
|
||
) as temporary:
|
||
json.dump(config, temporary, ensure_ascii=False, indent=2)
|
||
temporary.write("\n")
|
||
local_temp = Path(temporary.name)
|
||
try:
|
||
subprocess.run(
|
||
["docker", "cp", str(local_temp), f"smart-cam:{CONFIG_PATH}.new"],
|
||
check=True,
|
||
)
|
||
subprocess.run(
|
||
["docker", "exec", "smart-cam", "python3", "-m", "json.tool",
|
||
f"{CONFIG_PATH}.new"],
|
||
stdout=subprocess.DEVNULL, check=True,
|
||
)
|
||
subprocess.run(
|
||
["docker", "exec", "smart-cam", "mv", f"{CONFIG_PATH}.new", CONFIG_PATH],
|
||
check=True,
|
||
)
|
||
finally:
|
||
local_temp.unlink(missing_ok=True)
|
||
|
||
subprocess.run(["docker", "restart", "smart-cam"], check=True,
|
||
stdout=subprocess.DEVNULL)
|
||
deadline = time.monotonic() + 90
|
||
while time.monotonic() < deadline:
|
||
try:
|
||
core.api("GET", "/api/v1/scan/status", timeout=5)
|
||
return
|
||
except (SystemExit, subprocess.SubprocessError):
|
||
time.sleep(2)
|
||
raise RuntimeError("smart-cam重启后90秒内scan HTTP服务未就绪")
|
||
|
||
|
||
def run_test_scan() -> dict:
|
||
core.api("POST", "/api/v1/scan/start", timeout=20)
|
||
say("已启动测试扫描。", color=GREEN)
|
||
say("\n扫图日志：")
|
||
since = datetime.now().astimezone().isoformat(timespec="seconds")
|
||
saved_pattern = re.compile(r"\[scan\]\s+saved\b.*\bprogress=(\d+)/(\d+)\b")
|
||
last_progress: tuple[int, int] | None = None
|
||
process = subprocess.Popen(
|
||
["docker", "logs", "-f", "--since", since, "smart-cam"],
|
||
stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True,
|
||
)
|
||
try:
|
||
while True:
|
||
line = process.stdout.readline() if process.stdout else ""
|
||
match = saved_pattern.search(line)
|
||
if match:
|
||
progress = (int(match.group(1)), int(match.group(2)))
|
||
if progress != last_progress:
|
||
say(f"已保存第 {progress[0]}/{progress[1]} 张图")
|
||
last_progress = progress
|
||
status = core.api("GET", "/api/v1/scan/status", timeout=5).get("scan", {})
|
||
if status.get("state") in {"completed", "failed"}:
|
||
if status.get("state") == "completed":
|
||
say(
|
||
f"\n扫描完成，共保存 {status.get('completed', status.get('total', 0))} 张图。"
|
||
f"\n保存目录：{status.get('output_dir', '')}",
|
||
color=GREEN,
|
||
)
|
||
else:
|
||
say(f"\n扫描失败：{status.get('error') or '未知错误'}", color=RED)
|
||
return status
|
||
if not line:
|
||
time.sleep(1)
|
||
finally:
|
||
process.terminate()
|
||
try:
|
||
process.wait(timeout=3)
|
||
except subprocess.TimeoutExpired:
|
||
process.kill()
|
||
|
||
|
||
def run_panorama(scan_status: dict, mount: str) -> Path:
|
||
scan_dir = str(scan_status.get("output_dir") or "")
|
||
scan_id = str(scan_status.get("scan_id") or "")
|
||
# stitch_scan.py runs on the host. A scan_id resolves beneath
|
||
# /userdata/data/scan, while output_dir reported by the API is the
|
||
# container-side /smart-cam/data/scan path.
|
||
source = scan_id or scan_dir
|
||
if not source:
|
||
raise RuntimeError("扫描状态中没有output_dir或scan_id，无法确定拼图来源")
|
||
stitch = ROOT / "scan-panorama" / "stitch.sh"
|
||
result = subprocess.run(
|
||
[str(stitch), source, "--mount", mount],
|
||
text=True, capture_output=True,
|
||
)
|
||
if result.returncode:
|
||
raise RuntimeError(result.stderr.strip() or result.stdout.strip() or "拼图失败")
|
||
match = re.search(r"^结果：(.+)$", result.stdout, flags=re.MULTILINE)
|
||
if not match:
|
||
raise RuntimeError(f"拼图完成但没有返回结果路径：\n{result.stdout}")
|
||
output = Path(match.group(1).strip())
|
||
if not output.is_file() or output.stat().st_size == 0:
|
||
raise RuntimeError(f"拼图结果不存在或为空：{output}")
|
||
return output
|
||
|
||
|
||
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 会立即结束，并清理本次向导产生的标定痕迹。")
|
||
|
||
stage = choose(
|
||
"① 当前是【野外部署阶段】（回复 1）还是"
|
||
"【办公室测试阶段】（回复 2）？",
|
||
{"1", "2"},
|
||
)
|
||
mount = "hanging" if stage == "1" else "desktop"
|
||
|
||
confirm_position(
|
||
"② 请在Web界面中固定本次扫描使用的zoom，"
|
||
"再手动把摄像头移到扫描区域的【左上角】。"
|
||
)
|
||
top_left = camera_ptz()
|
||
state = core.load_state()
|
||
state["points"]["top-left"] = top_left
|
||
core.save_state(state)
|
||
|
||
confirm_position(
|
||
"③ 请保持zoom不变，手动把摄像头移到扫描区域的【右下角】。"
|
||
)
|
||
bottom_right = camera_ptz()
|
||
if abs(top_left["zoom"] - bottom_right["zoom"]) > 10:
|
||
core.fail("左右角的zoom不一致，请固定焦段后重新运行")
|
||
state = core.load_state()
|
||
state["points"]["bottom-right"] = bottom_right
|
||
core.save_state(state)
|
||
|
||
points = core.load_state()["points"]
|
||
while True:
|
||
confirm_position(
|
||
"④ 这一步用于测量当前zoom下的水平/垂直视场。\n"
|
||
"请保持zoom不变，优先把摄像头移到扫描区域内部、"
|
||
"距离较远且纹理分布在整幅画面的位置。\n"
|
||
"请避开纯天空、纯墙面、水面、玻璃、近距离物体和重复图案；"
|
||
"回复1后不要再操作Web云台，程序会自动移动、拍照并返回起点。"
|
||
)
|
||
say("\n正在计算，请不要操作摄像头或Web控制页面……")
|
||
try:
|
||
calculate_steps(points, snapshots, mount)
|
||
break
|
||
except SystemExit as exc:
|
||
say(f"\n本次视场标定没有通过：{exc}", color=RED)
|
||
choose(
|
||
"请换一个纹理更丰富的位置重新尝试第四步，两个角点无需重录。\n"
|
||
"重新尝试回复 1；结束对话回复 3。",
|
||
{"1"},
|
||
)
|
||
scan, generated, detail = generate(mount)
|
||
say("\n⑤ 本次得到的扫图参数：", color=GREEN)
|
||
say(json.dumps(scan, ensure_ascii=False, indent=2))
|
||
say(detail.strip())
|
||
|
||
install = choose(
|
||
"⑥ 是否替换 smart-cam 容器内的"
|
||
"/smart-cam/config/config.json？\n"
|
||
"替换并重启回复 1；不替换并结束回复 3。",
|
||
{"1"},
|
||
)
|
||
if install == "1":
|
||
deploy_config(scan)
|
||
say("替换完成，smart-cam已重启。", color=GREEN)
|
||
|
||
test = choose(
|
||
"⑦ 是否现在进行一次完整扫图测试？\n"
|
||
"开始测试回复 1；不测试并结束回复 3。",
|
||
{"1"},
|
||
)
|
||
if test == "1":
|
||
scan_status = run_test_scan()
|
||
if scan_status.get("state") != "completed":
|
||
raise RuntimeError("本次扫描没有成功完成，因此不进入拼图测试")
|
||
stitch = choose(
|
||
"⑧ 本次扫描已完成，是否立即进行拼图测试？\n"
|
||
"开始拼图回复 1；不拼图并结束回复 3。",
|
||
{"1"},
|
||
)
|
||
if stitch == "1":
|
||
panorama = run_panorama(scan_status, mount)
|
||
say(f"\n拼图完成。\n大图存储位置：{panorama}", color=GREEN)
|
||
success = True
|
||
say("\n配置向导已完成，对话结束。", color=GREEN)
|
||
except Cancelled:
|
||
say("\n已结束对话，正在清理本次向导痕迹……", color=YELLOW)
|
||
except (KeyboardInterrupt, EOFError):
|
||
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__":
|
||
main()
|