355 lines
13 KiB
Python
Executable File
355 lines
13 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
|
||
import tempfile
|
||
|
||
|
||
PROGRAM_DIR = Path(__file__).resolve().parent
|
||
DEFAULT_SCAN_ROOT = Path("/userdata/data/scan")
|
||
RESULT_DIR = PROGRAM_DIR / "result"
|
||
SAFE_IMAGE_DIMENSION = 16000
|
||
RESOURCE_ERROR_MARKERS = (
|
||
"cache resources exhausted",
|
||
"memory allocation failed",
|
||
"unable to extend cache",
|
||
"no space left on device",
|
||
)
|
||
|
||
|
||
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 run_imagemagick(command: list[str], temporary_dir: Path) -> None:
|
||
env = os.environ.copy()
|
||
env["MAGICK_TEMPORARY_PATH"] = str(temporary_dir)
|
||
result = subprocess.run(command, text=True, capture_output=True, env=env)
|
||
if result.returncode == 0:
|
||
return
|
||
detail = (result.stderr or result.stdout).strip()
|
||
raise RuntimeError(detail or f"ImageMagick命令返回{result.returncode}")
|
||
|
||
|
||
def is_resource_error(message: str) -> bool:
|
||
lowered = message.lower()
|
||
return any(marker in lowered for marker in RESOURCE_ERROR_MARKERS)
|
||
|
||
|
||
def build_staged_mosaic(
|
||
images: list[Path],
|
||
rows: int,
|
||
cols: int,
|
||
tile_width: int,
|
||
tile_height: int,
|
||
quality: int,
|
||
output: Path,
|
||
) -> None:
|
||
"""Resize one source at a time, join rows, then join the finished rows.
|
||
|
||
A single ImageMagick montage invocation keeps every 4K source image open.
|
||
Large scan grids therefore exceed the deliberately conservative cache
|
||
policy on edge devices. Staging bounds the working set to one source
|
||
image while resizing and one already-small row while joining.
|
||
"""
|
||
convert = shutil.which("convert")
|
||
montage = shutil.which("montage")
|
||
if not convert or not montage:
|
||
fail("未找到ImageMagick convert/montage，请先安装imagemagick")
|
||
|
||
with tempfile.TemporaryDirectory(
|
||
prefix=f".{output.stem}.stitch.", dir=RESULT_DIR
|
||
) as temporary_name:
|
||
temporary_dir = Path(temporary_name)
|
||
row_outputs: list[Path] = []
|
||
for row in range(rows):
|
||
tile_outputs: list[Path] = []
|
||
for col in range(cols):
|
||
source = images[row * cols + col]
|
||
tile = temporary_dir / f"tile_{row:04d}_{col:04d}.jpg"
|
||
run_imagemagick(
|
||
[
|
||
convert,
|
||
str(source),
|
||
"-thumbnail", f"{tile_width}x{tile_height}!",
|
||
"-strip",
|
||
"-quality", str(quality),
|
||
str(tile),
|
||
],
|
||
temporary_dir,
|
||
)
|
||
tile_outputs.append(tile)
|
||
|
||
row_output = temporary_dir / f"row_{row:04d}.jpg"
|
||
run_imagemagick(
|
||
[
|
||
montage,
|
||
*map(str, tile_outputs),
|
||
"-tile", f"{cols}x1",
|
||
"-geometry", "+0+0",
|
||
"-background", "black",
|
||
"-quality", str(quality),
|
||
str(row_output),
|
||
],
|
||
temporary_dir,
|
||
)
|
||
if not row_output.is_file() or row_output.stat().st_size == 0:
|
||
raise RuntimeError(f"第{row + 1}行没有生成有效结果")
|
||
row_outputs.append(row_output)
|
||
for tile in tile_outputs:
|
||
tile.unlink(missing_ok=True)
|
||
|
||
temporary_output = temporary_dir / "panorama.jpg"
|
||
run_imagemagick(
|
||
[
|
||
montage,
|
||
*map(str, row_outputs),
|
||
"-tile", f"1x{rows}",
|
||
"-geometry", "+0+0",
|
||
"-background", "black",
|
||
"-quality", str(quality),
|
||
str(temporary_output),
|
||
],
|
||
temporary_dir,
|
||
)
|
||
if not temporary_output.is_file() or temporary_output.stat().st_size == 0:
|
||
raise RuntimeError("分阶段拼图没有生成有效结果")
|
||
shutil.copyfile(temporary_output, output)
|
||
|
||
|
||
def build_mosaic(
|
||
scan_dir: Path,
|
||
tile_width: int,
|
||
quality: int,
|
||
mount: str,
|
||
pan_direction: str,
|
||
) -> Path:
|
||
# Office testing: the dome stands on a desk and the camera firmware rotates
|
||
# the encoded picture by 180 degrees, which reverses the vertical grid.
|
||
# Horizontal ordering is camera/firmware-specific and is measured during
|
||
# calibration; it must not be inferred from the physical mount.
|
||
reverse_rows = mount == "desktop"
|
||
if pan_direction == "auto":
|
||
# Backward-compatible behavior for scans made before calibration began
|
||
# recording the actual signed pan direction.
|
||
pan_direction = "decreasing" if mount == "desktop" else "increasing"
|
||
reverse_cols = pan_direction == "decreasing"
|
||
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))
|
||
|
||
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"
|
||
attempt_widths: list[int] = []
|
||
for scale in (1.0, 0.75, 0.5):
|
||
width = max(64, int(tile_width * scale))
|
||
if width not in attempt_widths:
|
||
attempt_widths.append(width)
|
||
|
||
final_tile_width = tile_width
|
||
last_error = ""
|
||
try:
|
||
for attempt, candidate_width in enumerate(attempt_widths, start=1):
|
||
candidate_height = max(
|
||
1, round(source_height * candidate_width / source_width)
|
||
)
|
||
temporary.unlink(missing_ok=True)
|
||
try:
|
||
build_staged_mosaic(
|
||
images,
|
||
rows,
|
||
cols,
|
||
candidate_width,
|
||
candidate_height,
|
||
quality,
|
||
temporary,
|
||
)
|
||
final_tile_width = candidate_width
|
||
tile_height = candidate_height
|
||
break
|
||
except RuntimeError as exc:
|
||
last_error = str(exc)
|
||
if not is_resource_error(last_error) or attempt == len(attempt_widths):
|
||
fail(f"拼图失败：{last_error}")
|
||
next_width = attempt_widths[attempt]
|
||
print(
|
||
"提示：盒子可用内存或ImageMagick缓存不足，"
|
||
f"程序正在把单格宽度从{candidate_width}缩小到{next_width}后自动重试……",
|
||
flush=True,
|
||
)
|
||
else:
|
||
fail(f"拼图失败：{last_error}")
|
||
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 '正向'}）"
|
||
f"，实测pan方向：{pan_direction}（列={'反向' if reverse_cols else '正向'}）"
|
||
)
|
||
print(f"小图：{source_width}×{source_height}")
|
||
if final_tile_width != requested_tile_width:
|
||
print(
|
||
f"尺寸保护：单格宽度由{requested_tile_width}自动缩小到"
|
||
f"{final_tile_width}，避免尺寸或缓存超限"
|
||
)
|
||
print(f"拼图方式：分阶段低内存处理（逐张缩图 → 逐行合并 → 合并全部行）")
|
||
print(f"大图：{cols * final_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(
|
||
"--pan-direction",
|
||
choices=["auto", "increasing", "decreasing"],
|
||
default="auto",
|
||
help=(
|
||
"画面左到右时pan数值方向；交互式向导会自动传入。"
|
||
"旧扫描未记录方向时默认auto沿用历史mount规则"
|
||
),
|
||
)
|
||
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,
|
||
args.pan_direction,
|
||
)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|