196 lines
7.3 KiB
Python
Executable File
196 lines
7.3 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""Build a deterministic scan verification mosaic from scan/meta.json."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import os
|
||
from pathlib import Path
|
||
import re
|
||
import shutil
|
||
import subprocess
|
||
import sys
|
||
|
||
|
||
PROGRAM_DIR = Path(__file__).resolve().parent
|
||
DEFAULT_SCAN_ROOT = Path("/userdata/data/scan")
|
||
RESULT_DIR = PROGRAM_DIR / "result"
|
||
SAFE_IMAGE_DIMENSION = 16000
|
||
|
||
|
||
def fail(message: str) -> "NoReturn":
|
||
raise SystemExit(f"错误：{message}")
|
||
|
||
|
||
def resolve_scan_dir(value: str) -> Path:
|
||
candidate = Path(value).expanduser()
|
||
if not candidate.is_absolute():
|
||
candidate = DEFAULT_SCAN_ROOT / candidate
|
||
candidate = candidate.resolve()
|
||
if not candidate.is_dir():
|
||
fail(f"扫描目录不存在：{candidate}")
|
||
return candidate
|
||
|
||
|
||
def load_grid(scan_dir: Path, reverse_rows: bool, reverse_cols: bool) -> tuple[str, int, int, list[Path]]:
|
||
meta_path = scan_dir / "meta.json"
|
||
if not meta_path.is_file():
|
||
fail(f"缺少meta.json：{meta_path}")
|
||
|
||
try:
|
||
meta = json.loads(meta_path.read_text(encoding="utf-8"))
|
||
except (OSError, json.JSONDecodeError) as exc:
|
||
fail(f"无法读取meta.json：{exc}")
|
||
|
||
records = meta.get("images")
|
||
if not isinstance(records, list) or not records:
|
||
fail("meta.json中没有images记录")
|
||
|
||
cells: dict[tuple[int, int], Path] = {}
|
||
for index, record in enumerate(records):
|
||
try:
|
||
row, col = int(record["row"]), int(record["col"])
|
||
relative = Path(record["file"])
|
||
except (KeyError, TypeError, ValueError) as exc:
|
||
fail(f"images[{index}]缺少有效的row/col/file：{exc}")
|
||
if relative.is_absolute() or ".." in relative.parts:
|
||
fail(f"images[{index}]包含不安全的文件路径：{relative}")
|
||
key = (row, col)
|
||
if key in cells:
|
||
fail(f"meta.json存在重复网格位置：row={row}, col={col}")
|
||
image = scan_dir / relative
|
||
if not image.is_file() or image.stat().st_size == 0:
|
||
fail(f"图片缺失或为空：{image}")
|
||
cells[key] = image
|
||
|
||
rows = sorted({row for row, _ in cells})
|
||
cols = sorted({col for _, col in cells})
|
||
if rows != list(range(len(rows))) or cols != list(range(len(cols))):
|
||
fail(f"row/col必须从0连续编号，实际rows={rows} cols={cols}")
|
||
missing = [(row, col) for row in rows for col in cols if (row, col) not in cells]
|
||
if missing:
|
||
fail(f"扫描网格不完整，缺少{len(missing)}个位置，例如：{missing[:8]}")
|
||
|
||
scan_id = str(meta.get("scan_id") or scan_dir.name)
|
||
safe_id = re.sub(r"[^0-9A-Za-z._-]+", "_", scan_id).strip("._-")
|
||
if not safe_id:
|
||
safe_id = re.sub(r"[^0-9A-Za-z._-]+", "_", scan_dir.name)
|
||
layout_rows = list(reversed(rows)) if reverse_rows else rows
|
||
layout_cols = list(reversed(cols)) if reverse_cols else cols
|
||
ordered = [cells[(row, col)] for row in layout_rows for col in layout_cols]
|
||
return safe_id, len(rows), len(cols), ordered
|
||
|
||
|
||
def identify_dimensions(images: list[Path]) -> tuple[int, int]:
|
||
identify = shutil.which("identify")
|
||
if not identify:
|
||
fail("未找到ImageMagick identify，请先安装imagemagick")
|
||
result = subprocess.run(
|
||
[identify, "-format", "%w %h\\n", *map(str, images)],
|
||
text=True,
|
||
capture_output=True,
|
||
)
|
||
if result.returncode != 0:
|
||
fail(f"图片格式校验失败：{result.stderr.strip()}")
|
||
dimensions = set()
|
||
for line in result.stdout.splitlines():
|
||
try:
|
||
width, height = map(int, line.split())
|
||
except ValueError:
|
||
fail(f"identify返回了无法识别的尺寸：{line!r}")
|
||
dimensions.add((width, height))
|
||
if len(dimensions) != 1:
|
||
fail(f"小图尺寸不一致：{sorted(dimensions)}")
|
||
return dimensions.pop()
|
||
|
||
|
||
def build_mosaic(scan_dir: Path, tile_width: int, quality: int, mount: str) -> Path:
|
||
# Office testing: the dome stands on a desk and the camera firmware rotates
|
||
# the encoded picture by 180 degrees. Field deployment: the dome is hung.
|
||
reverse_rows = mount == "desktop"
|
||
reverse_cols = mount == "desktop"
|
||
scan_id, rows, cols, images = load_grid(scan_dir, reverse_rows, reverse_cols)
|
||
source_width, source_height = identify_dimensions(images)
|
||
if tile_width <= 0:
|
||
tile_width = source_width
|
||
requested_tile_width = tile_width
|
||
max_width_for_columns = SAFE_IMAGE_DIMENSION // cols
|
||
max_width_for_rows = (SAFE_IMAGE_DIMENSION * source_width) // (rows * source_height)
|
||
tile_width = min(tile_width, max_width_for_columns, max_width_for_rows)
|
||
if tile_width <= 0:
|
||
fail(f"网格{cols}列×{rows}行过大，无法在{SAFE_IMAGE_DIMENSION}px尺寸限制内生成")
|
||
tile_height = max(1, round(source_height * tile_width / source_width))
|
||
|
||
montage = shutil.which("montage")
|
||
if not montage:
|
||
fail("未找到ImageMagick montage，请先安装imagemagick")
|
||
RESULT_DIR.mkdir(parents=True, exist_ok=True)
|
||
output = RESULT_DIR / f"{scan_id}_{mount}_scan_grid_panorama.jpg"
|
||
temporary = RESULT_DIR / f".{output.name}.{os.getpid()}.tmp.jpg"
|
||
command = [
|
||
montage,
|
||
*map(str, images),
|
||
"-thumbnail", f"{tile_width}x{tile_height}!",
|
||
"-tile", f"{cols}x{rows}",
|
||
"-geometry", "+0+0",
|
||
"-background", "black",
|
||
"-quality", str(quality),
|
||
str(temporary),
|
||
]
|
||
try:
|
||
result = subprocess.run(command, text=True, capture_output=True)
|
||
if result.returncode != 0:
|
||
fail(f"拼图失败：{result.stderr.strip()}")
|
||
if not temporary.is_file() or temporary.stat().st_size == 0:
|
||
fail("拼图命令未生成有效结果")
|
||
temporary.replace(output)
|
||
finally:
|
||
temporary.unlink(missing_ok=True)
|
||
|
||
print(f"扫描目录：{scan_dir}")
|
||
print(f"网格：{cols}列 × {rows}行，共{len(images)}张")
|
||
print(f"安装方式：{mount}（行={'反向' if reverse_rows else '正向'}，列={'反向' if reverse_cols else '正向'}）")
|
||
print(f"小图：{source_width}×{source_height}")
|
||
if tile_width != requested_tile_width:
|
||
print(f"尺寸保护：单格宽度由{requested_tile_width}自动缩小到{tile_width}，避免超过ImageMagick限制")
|
||
print(f"大图：{cols * tile_width}×{rows * tile_height}")
|
||
print(f"结果：{output}")
|
||
return output
|
||
|
||
|
||
def main() -> None:
|
||
parser = argparse.ArgumentParser(
|
||
description="读取scan目录的meta.json，按row/col生成端侧快速验证拼图。"
|
||
)
|
||
parser.add_argument(
|
||
"scan_dir",
|
||
help="扫描目录名（如2026-07-22_18-25-09）或绝对路径",
|
||
)
|
||
parser.add_argument(
|
||
"--mount",
|
||
choices=["desktop", "hanging"],
|
||
required=True,
|
||
help="desktop=办公室桌面正放且后台旋转180度；hanging=野外倒挂",
|
||
)
|
||
parser.add_argument(
|
||
"--tile-width",
|
||
type=int,
|
||
default=640,
|
||
help="每张小图在大图中的宽度，默认640；传0使用原图宽度",
|
||
)
|
||
parser.add_argument(
|
||
"--quality",
|
||
type=int,
|
||
default=92,
|
||
choices=range(1, 101),
|
||
metavar="1-100",
|
||
help="输出JPEG质量，默认92",
|
||
)
|
||
args = parser.parse_args()
|
||
build_mosaic(resolve_scan_dir(args.scan_dir), args.tile_width, args.quality, args.mount)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|