lipan
29f1d45d66
算法更新: delivery v1.2.1、平整度v1.5、测距v2.2、整车配准v1.3 适配器: vehicle-registration 新增、各算法 manifest 和 adapter 更新 后端: taskCadFiles/source 字段迁移、derivedPointClouds 注册、取消任务联动 前端: ElapsedTime 时长显示、两步算法流程、canRun 逻辑修复 文档: 新增5份算法性能分析报告
321 lines
13 KiB
Python
321 lines
13 KiB
Python
"""End-to-end single-beam measurement: stream-crop a cloud by ROI, measure,
|
|
write artifacts. This is the DEFAULT pipeline for the "ROI already isolates
|
|
one beam" workflow.
|
|
|
|
Each call handles ONE beam (one ROI -> one result). For left+right, call it
|
|
twice with the two ROI files; outputs land in separate sub-directories.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import numpy as np
|
|
|
|
from .io import ensure_output_dir, read_point_cloud_cropped, write_csv, write_json
|
|
from .roi import load_roi_aabb, normalize_aabb
|
|
from .single_beam import SingleBeamResult, measure_single_beam
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
def run_single_beam_pipeline(
|
|
input_cloud: str | Path,
|
|
roi_path: str | Path,
|
|
out_dir: str | Path,
|
|
*,
|
|
side: str | None = None,
|
|
tolerance_mm: float | None = None,
|
|
config: dict | None = None,
|
|
cad_path: str | Path | None = None,
|
|
enable_registration: bool = False,
|
|
skip_excel: bool = False,
|
|
skip_plot: bool = False,
|
|
_registration: dict | None = None,
|
|
) -> dict:
|
|
"""Measure one beam and write report.json/csv/xlsx (+ png if available).
|
|
|
|
Args:
|
|
input_cloud: PLY/XYZ path. Large binary PLYs are stream-cropped.
|
|
roi_path: JSON AABB isolating the beam (native frame x=len, y=height,
|
|
z=width).
|
|
out_dir: output directory for this beam's artifacts.
|
|
side: ``"left"`` / ``"right"``. If None, inferred from the ROI's
|
|
lateral (z) centre — looking along +x, z>0 is left, z<0 is right.
|
|
tolerance_mm: pass/fail tolerance. If None, defaults by side
|
|
(left 7 mm, right 12 mm).
|
|
cad_path: theoretical CAD reference. Recorded for provenance; parsed
|
|
and used only when ``enable_registration`` is True.
|
|
enable_registration: if True, the cloud is rigidly aligned to the CAD
|
|
before measurement. Default False — cloud assumed pre-registered.
|
|
_registration: internal; precomputed registration metadata.
|
|
"""
|
|
out_dir_path = ensure_output_dir(out_dir)
|
|
roi_aabb = load_roi_aabb(roi_path)
|
|
aabb = normalize_aabb(roi_aabb)
|
|
# Side is determined by the ROI lateral position (x-forward convention),
|
|
# not by the file name, so it cannot be set wrong by a mislabelled file.
|
|
if side is None:
|
|
z_center = 0.5 * (aabb["z_min"] + aabb["z_max"])
|
|
side = "left" if z_center >= 0 else "right"
|
|
|
|
reg_meta = _registration
|
|
transform = None
|
|
if reg_meta is None:
|
|
reg_meta, transform = _maybe_register(input_cloud, cad_path, enable_registration, config)
|
|
elif reg_meta.get("performed"):
|
|
transform = (np.asarray(reg_meta["R"], dtype=np.float64), np.asarray(reg_meta["t"], dtype=np.float64))
|
|
|
|
points = read_point_cloud_cropped(input_cloud, aabb)
|
|
if transform is not None:
|
|
from .registration import apply_transform
|
|
# ROI is defined in the registered frame; crop already-registered
|
|
# points would be wrong, so when registering we crop a slightly
|
|
# padded box, transform, then re-crop. Handled in _maybe_register's
|
|
# contract: here transform is applied then re-cropped.
|
|
points = apply_transform(points, transform[0], transform[1])
|
|
from .io import _crop_points
|
|
points = _crop_points(points, (
|
|
aabb["x_min"], aabb["x_max"], aabb["y_min"], aabb["y_max"], aabb["z_min"], aabb["z_max"]))
|
|
|
|
result = measure_single_beam(points, aabb, side=side, tolerance_mm=tolerance_mm, config=config)
|
|
|
|
result.report["input_cloud"] = str(Path(input_cloud).resolve())
|
|
result.report["roi_source"] = str(Path(roi_path).resolve())
|
|
result.report["cad_reference"] = str(Path(cad_path).resolve()) if cad_path else None
|
|
result.report["registration"] = reg_meta
|
|
|
|
write_json(out_dir_path / "report.json", result.report)
|
|
_write_curve_csv(result, out_dir_path / "deflection_curve.csv")
|
|
|
|
if not skip_plot:
|
|
_plot_safely(result, out_dir_path / "deflection_curve.png")
|
|
if not skip_excel:
|
|
_excel_safely(result, out_dir_path / "report.xlsx", input_cloud, roi_path, cad_path, reg_meta)
|
|
|
|
return result.report
|
|
|
|
|
|
def run_both_beams(
|
|
input_cloud: str | Path,
|
|
left_roi: str | Path,
|
|
right_roi: str | Path,
|
|
out_dir: str | Path,
|
|
*,
|
|
config: dict | None = None,
|
|
cad_path: str | Path | None = None,
|
|
enable_registration: bool = False,
|
|
left_tolerance_mm: float | None = None,
|
|
right_tolerance_mm: float | None = None,
|
|
skip_excel: bool = False,
|
|
skip_plot: bool = False,
|
|
) -> dict:
|
|
"""Measure both beams; each side a fully independent run (one Excel each).
|
|
|
|
Side (left/right) and tolerance are determined from each ROI's lateral
|
|
position, NOT from the ``left_roi`` / ``right_roi`` argument names — so a
|
|
swapped file cannot apply the wrong tolerance. Looking along +x:
|
|
z>0 is left (default tol 7 mm), z<0 is right (default tol 12 mm).
|
|
``left_tolerance_mm`` / ``right_tolerance_mm`` override the side that the
|
|
corresponding ROI resolves to.
|
|
|
|
When ``enable_registration`` is True the whole-train cloud is registered to
|
|
the CAD once and the transform reused for both sides.
|
|
"""
|
|
out_dir_path = ensure_output_dir(out_dir)
|
|
reg_meta, _ = _maybe_register(input_cloud, cad_path, enable_registration, config)
|
|
results = {}
|
|
for roi, tol in ((left_roi, left_tolerance_mm), (right_roi, right_tolerance_mm)):
|
|
side = _resolve_side(roi)
|
|
rep = run_single_beam_pipeline(
|
|
input_cloud, roi, out_dir_path / side, side=side, tolerance_mm=tol,
|
|
config=config, cad_path=cad_path, enable_registration=enable_registration,
|
|
skip_excel=skip_excel, skip_plot=skip_plot, _registration=reg_meta,
|
|
)
|
|
results[side] = rep
|
|
write_json(out_dir_path / "summary.json", results)
|
|
return results
|
|
|
|
|
|
def _resolve_side(roi_path) -> str:
|
|
"""left/right from the ROI lateral (z) centre — x-forward: z>0 left, z<0 right."""
|
|
aabb = normalize_aabb(load_roi_aabb(roi_path))
|
|
return "left" if 0.5 * (aabb["z_min"] + aabb["z_max"]) >= 0 else "right"
|
|
|
|
|
|
def _maybe_register(input_cloud, cad_path, enable_registration, config):
|
|
"""Return (registration_meta, transform_or_None).
|
|
|
|
When ``enable_registration`` is False (default), no transform is applied
|
|
and the meta records that the cloud is assumed pre-registered. When True,
|
|
the whole-train cloud (downsampled) is rigidly aligned to the CAD; the CAD
|
|
may be a point cloud / mesh (.ply/.xyz/.stl/.obj). STEP (.stp) is recorded
|
|
but must be converted to a sampled cloud first (see tools/cad_sampling/).
|
|
"""
|
|
if not enable_registration:
|
|
return {"performed": False, "reason": "disabled (cloud assumed pre-registered)",
|
|
"cad_reference": str(Path(cad_path).resolve()) if cad_path else None}, None
|
|
if cad_path is None:
|
|
return {"performed": False, "reason": "no cad_path supplied"}, None
|
|
|
|
suffix = Path(cad_path).suffix.lower()
|
|
if suffix in {".stp", ".step", ".igs", ".iges"}:
|
|
return {"performed": False,
|
|
"reason": f"STEP not parsed; convert to sampled .ply/.stl first ({suffix})",
|
|
"cad_reference": str(Path(cad_path).resolve())}, None
|
|
|
|
from .cad_io import read_cad_reference
|
|
from .config import load_config, resolve_numeric_parameters, cloud_geometry
|
|
from .preprocess import estimate_spacing, voxel_downsample
|
|
from .registration import register_to_cad
|
|
|
|
cad = read_cad_reference(cad_path)
|
|
# Downsample the (possibly huge) train cloud for registration.
|
|
scan_full = read_point_cloud_cropped(input_cloud, {
|
|
"x_min": -1e12, "x_max": 1e12, "y_min": -1e12, "y_max": 1e12, "z_min": -1e12, "z_max": 1e12})
|
|
base = load_config(None, "auto")
|
|
if config:
|
|
from .config import deep_merge
|
|
base = deep_merge(base, config)
|
|
spacing = estimate_spacing(scan_full, base["preprocess"]["spacing_sample_size"])
|
|
rc = resolve_numeric_parameters(base, cloud_geometry(scan_full), spacing)
|
|
scan_ds = voxel_downsample(scan_full, max(rc["preprocess"]["voxel_mm"], 10.0))
|
|
reg = register_to_cad(scan_ds, cad, rc)
|
|
meta = {
|
|
"performed": True, "rms_mm": reg.rms_mm, "iterations": reg.iterations,
|
|
"inlier_ratio": reg.inlier_ratio, "converged": reg.converged,
|
|
"R": reg.R.tolist(), "t": reg.t.tolist(),
|
|
"cad_reference": str(Path(cad_path).resolve()),
|
|
}
|
|
return meta, (reg.R, reg.t)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Writers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _write_curve_csv(result: SingleBeamResult, path: Path) -> None:
|
|
rows = []
|
|
for i in range(len(result.x_mm)):
|
|
rows.append({
|
|
"x_mm": float(result.x_mm[i]),
|
|
"underside_height_mm": float(result.underside_height_mm[i]),
|
|
"reference_height_mm": float(result.reference_height_mm[i]),
|
|
"deflection_mm": float(result.deflection_mm[i]),
|
|
"kept": bool(result.kept_mask[i]) if len(result.kept_mask) else False,
|
|
})
|
|
write_csv(
|
|
path,
|
|
fieldnames=["x_mm", "underside_height_mm", "reference_height_mm", "deflection_mm", "kept"],
|
|
rows=rows,
|
|
)
|
|
|
|
|
|
def _plot_safely(result: SingleBeamResult, path: Path) -> None:
|
|
if not result.available or len(result.x_mm) == 0:
|
|
return
|
|
try:
|
|
import matplotlib
|
|
matplotlib.use("Agg")
|
|
import matplotlib.pyplot as plt
|
|
except ImportError:
|
|
log.warning("matplotlib not installed; skipping single-beam plot")
|
|
return
|
|
x = result.x_mm
|
|
fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(11, 6), sharex=True)
|
|
ax1.scatter(x, result.underside_height_mm, s=8, c="tab:blue", label="underside")
|
|
ax1.plot(x, result.reference_height_mm, "r-", lw=1.5, label="reference line")
|
|
ax1.set_ylabel("height (mm)")
|
|
ax1.legend(loc="best")
|
|
ax1.set_title(f"{result.side} beam — underside vs straight reference")
|
|
ax2.axhline(0.0, color="k", lw=0.8)
|
|
ax2.plot(x, result.deflection_mm, "tab:green", lw=1.2)
|
|
if result.selected_x_mm is not None:
|
|
ax2.plot([result.selected_x_mm], [result.selected_signed_mm], "ro",
|
|
label=f"max |dev| = {result.selected_absolute_mm:.2f} mm")
|
|
ax2.legend(loc="best")
|
|
ax2.set_xlabel("x — longitudinal (mm)")
|
|
ax2.set_ylabel("deflection (mm)")
|
|
fig.tight_layout()
|
|
fig.savefig(path, dpi=150)
|
|
plt.close(fig)
|
|
|
|
|
|
def _fmt(v: float | None, ndigits: int = 4) -> str:
|
|
"""Format a mm value for the report; '-' when not available."""
|
|
if v is None:
|
|
return "-"
|
|
return f"{float(v):.{ndigits}f}"
|
|
|
|
|
|
def _excel_safely(result: SingleBeamResult, path: Path, input_cloud, roi_path,
|
|
cad_path=None, reg_meta=None) -> None:
|
|
try:
|
|
from openpyxl import Workbook
|
|
except ImportError:
|
|
log.warning("openpyxl not installed; skipping single-beam xlsx")
|
|
return
|
|
reg_meta = reg_meta or {}
|
|
wb = Workbook()
|
|
|
|
# --- Sheet 1: inspection report (the 6 deliverable columns) ---
|
|
ws = wb.active
|
|
ws.title = "检测报表"
|
|
headers = ["项点名称", "理论尺寸", "公差要求", "测量结果", "误差结果", "测量结论"]
|
|
ws.append(headers)
|
|
tol = result.tolerance_mm
|
|
ws.append([
|
|
result.item_name, # 项点名称
|
|
_fmt(result.nominal_mm, 1), # 理论尺寸 (0)
|
|
"-" if tol is None else f"±{float(tol):.1f}mm", # 公差要求
|
|
_fmt(result.measured_mm), # 测量结果
|
|
_fmt(result.error_mm), # 误差结果
|
|
result.verdict or "-", # 测量结论 (合格/不合格)
|
|
])
|
|
widths = [30, 12, 12, 14, 14, 12]
|
|
for i, w in enumerate(widths):
|
|
ws.column_dimensions[chr(ord("A") + i)].width = w
|
|
|
|
# --- Sheet 2: detail / provenance (for traceability, optional read) ---
|
|
ws2 = wb.create_sheet("明细")
|
|
d = result.report.get("deflection", {})
|
|
detail = [
|
|
("side(左/右,以x正向为准)", result.side),
|
|
("input_cloud", str(Path(input_cloud).resolve())),
|
|
("roi_source", str(Path(roi_path).resolve())),
|
|
("cad_reference", str(Path(cad_path).resolve()) if cad_path else ""),
|
|
("registration_performed", reg_meta.get("performed", False)),
|
|
("registration_note", reg_meta.get("reason", "")),
|
|
("roi_point_count", result.roi_point_count),
|
|
("n_slices", result.n_slices),
|
|
("n_kept", result.n_kept),
|
|
("reference_slope_mm_per_m", result.reference_slope_mm_per_m),
|
|
("selected_x_mm", d.get("selected_x_mm")),
|
|
("max_downward_mm", d.get("max_downward_mm")),
|
|
("max_upward_mm", d.get("max_upward_mm")),
|
|
("rms_mm", d.get("rms_mm")),
|
|
("quality_flags", "; ".join(result.quality_flags)),
|
|
]
|
|
ws2.append(["key", "value"])
|
|
for k, v in detail:
|
|
ws2.append([k, "" if v is None else v])
|
|
ws2.column_dimensions["A"].width = 30
|
|
ws2.column_dimensions["B"].width = 46
|
|
|
|
# --- Sheet 3: deflection curve ---
|
|
ws3 = wb.create_sheet("挠度曲线")
|
|
ws3.append(["x_mm", "underside_height_mm", "reference_height_mm", "deflection_mm", "kept"])
|
|
for i in range(len(result.x_mm)):
|
|
ws3.append([
|
|
float(result.x_mm[i]), float(result.underside_height_mm[i]),
|
|
float(result.reference_height_mm[i]), float(result.deflection_mm[i]),
|
|
bool(result.kept_mask[i]) if len(result.kept_mask) else False,
|
|
])
|
|
wb.save(path)
|
|
|
|
|
|
__all__ = ["run_single_beam_pipeline", "run_both_beams"]
|