lipan/massive-cloud
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: update algorithm integrations and point cloud overlays

Browse Source
This commit is contained in:
lipan
2026-06-30 16:46:36 +08:00
parent ddca7e5436
commit 7e2e56a50d
64 changed files with 4356 additions and 2781 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
server/plugins/adapters/coupler_panel_run.py
View File

@@ -78,7 +78,7 @@ def main() -> int:
z_min, z_max = _f("roiZMin"), _f("roiZMax")
# 至少有一对轴的范围被填写时才构建 ROI
if any(v is not None for v in (x_min, x_max, y_min, y_max, z_min, z_max)):
from coupler_panel.roi import ROISpec # type: ignore
from coupler_panel.roi_select import ROISpec # type: ignore
roi_spec = ROISpec(
x_min=x_min if x_min is not None else float("-inf"),
@@ -150,20 +150,27 @@ def main() -> int:
D_y = float(result.D_y_max_mm)
D_z = float(result.D_z_max_mm)
# 斜率形式：offset_mm = baseline * tan(theta)，与 Excel 报表一致
import math as _math
offset_right = 700.0 * _math.tan(_math.radians(raw_right)) if 'raw_right' in dir() else 700.0 * _math.tan(_math.radians(theta_right))
offset_ref = 375.0 * _math.tan(_math.radians(raw_ref)) if 'raw_ref' in dir() else 375.0 * _math.tan(_math.radians(theta_ref))
measurements = [
{
"name": "偏转角 θ_right (右向)",
"value": theta_right,
"unit": "deg",
"tolerance": 0.5,
"status": _status_for(theta_right, 0.5),
"name": "车钩面板距纵向中心的右偏转角",
"value": round(offset_right, 6),
"unit": "mm",
"baseline": 700.0,
"tolerance": 1.0,
"status": _status_for(offset_right, 1.0),
},
{
"name": "偏转角 θ_ref (参考向)",
"value": theta_ref,
"unit": "deg",
"tolerance": 0.5,
"status": _status_for(theta_ref, 0.5),
"name": "车钩面板距参考面的右偏转角",
"value": round(offset_ref, 6),
"unit": "mm",
"baseline": 375.0,
"tolerance": 0.8,
"status": _status_for(offset_ref, 0.8),
},
{
"name": "最大 x 偏差 dx_y",

142
server/plugins/adapters/flatness_suite_run.py
View File

@@ -7,15 +7,16 @@ v1.5 变更：
- pipeline 新增 result_template_path / combined_context_pointcloud_path 参数
约定：
- module=all：plane + curvature（contour 需要 CAD STL，缺失时跳过）
- module=contour：轮廓/平整度，需要 CAD STEP（推荐，算法方基准 3min25s）
- module=plane：只跑平面拟合
- module=curvature：只跑曲率估计
- module=contour：需要 CAD STL，缺失时报错
- module=curvature：只跑曲率估计（大点云极慢，慎用）
- module=all：plane + curvature + contour（已从前端选项移除，大点云不可用）
"""
from __future__ import annotations
import copy
import hashlib
import os
import sys
import traceback
@@ -39,6 +40,39 @@ def _safe_float(v: object) -> float | None:
return None
def _mesh_cache_path(cad_path: str, mesh_deflection: float) -> Path:
"""根据 STEP 路径 + 文件指纹 + 三角化精度生成缓存路径。
优先返回配准阶段已生成的 mesh 缓存（_mesh.ply），
其次返回 flatness 自己的缓存路径，均放在 CAD 同目录的 .cad_sample_cache/ 下。
"""
stat = os.stat(cad_path)
cache_key = "|".join([
os.path.abspath(cad_path),
str(stat.st_size),
str(stat.st_mtime_ns),
f"{mesh_deflection:.12g}",
])
digest = hashlib.sha1(cache_key.encode()).hexdigest()[:16]
cad_stem = Path(cad_path).stem
shared_dir = Path(cad_path).parent / ".cad_sample_cache"
# 配准阶段已缓存的 mesh（与配准采样缓存同目录，命名规则：<stem>_<digest>_mesh.ply）
registration_mesh = shared_dir / f"{cad_stem}_{digest}_mesh.ply"
if registration_mesh.exists():
return registration_mesh
# flatness 自己写缓存的目标路径
try:
shared_dir.mkdir(parents=True, exist_ok=True)
test = shared_dir / ".write_test"
test.touch()
test.unlink()
return shared_dir / f"{cad_stem}_{digest}_mesh.ply"
except OSError:
return Path(cad_path).parent / f"{cad_stem}_{digest}_mesh.ply"
def _find_alg_dir(alg_root: str) -> Path:
"""定位平整度算法目录，支持多个历史版本路径，优先使用最新版。"""
candidates = [
@@ -99,11 +133,12 @@ def main() -> int:
try:
alg_root = os.environ.get("ALG_ROOT") or os.environ.get("ALG_ROOT_HOST") or "/alg"
emit_progress(0.05, "定位 point_porj 算法目录")
emit_progress(0.05, "定位算法目录")
alg_dir = _find_alg_dir(alg_root)
# ── 判断走单点云模式还是 datasets 批量模式 ──────────────────────────────
pc_path_raw = first_pointcloud(payload)
emit_progress(0.10, f"读取点云：{os.path.basename(pc_path_raw)}")
use_batch_mode = bool(config.get("useBatchDatasets", True))
# ── ROI 裁剪（整车点云场景下必须填写，否则算法极慢）──────────────────────
def _f(key: str) -> float | None:
@@ -147,10 +182,61 @@ def main() -> int:
# v1.5 用 o3d.io.read_point_cloud 读取，直接支持二进制 PLY，无需转 ASCII
pc_path = pc_path_raw
module = str(config.get("module") or "all")
module = str(config.get("module") or "contour")
emit_progress(0.15, f"初始化 MeasurementPipeline（module={module}）")
pipeline = _build_pipeline(alg_dir, pc_path, out_dir, config)
pipeline = _build_pipeline(alg_dir, pc_path if not use_batch_mode else str(alg_dir), out_dir, config)
# ── datasets 批量模式（useBatchDatasets=true 时） ──────────────────────
if use_batch_mode and module == "contour":
emit_progress(0.20, "批量模式：从 datasets/Noise_PointCloud/ 读取所有面板点云")
data_root = alg_dir / "datasets"
# 所选整车点云作为可视化背景；用户也可在 contextPointCloudPath 里手动指定
context_pc = (inputs.get("contextPointCloudPath")
or pc_path_raw
or None)
save_dir = str(out_dir / "contour_batch")
excel_path = str(out_dir / "contour.xlsx")
res = pipeline.run_contour_batch(
all_tasks=True,
data_root=str(data_root),
pointcloud_kind="noise",
point_chunk_size=8192,
triangle_chunk_size=512,
save_path=save_dir,
excel_output_path=excel_path,
combined_context_pointcloud_path=context_pc,
no_visualize=True,
)
emit_progress(0.95, "整理批量结果")
batch_stats = res.get("batch_stats") or {}
measurements: list[dict] = []
for task_id, task_res in (res.get("results") or {}).items():
s = task_res.get("distance_stats") or {}
if s:
measurements.append({
"name": f"任务{task_id} 轮廓偏差均值",
"value": round(float(s.get("mean", 0)), 4),
"unit": "mm", "status": "info",
})
# 各面板热力图 PLY 列表（不含背景点，全部是有意义的颜色）
panel_plys = sorted([
f"contour_batch/{p.name}"
for p in (out_dir / "contour_batch").glob("*_contour.ply")
if p.name != "combined_contour.ply"
])
if panel_plys:
overlay3d = {"kind": "ply", "file": panel_plys[0], "files": panel_plys}
else:
overlay3d = {"kind": "none"}
write_result(out_dir, {
"status": "completed",
"message": f"批量完成 {len(res.get('results', {}))} 个面板测量",
"measurements": measurements,
"overlay3d": overlay3d,
"extra": {"module": "contour_batch", "excelPath": "contour.xlsx"},
})
return 0
measurements: list[dict] = []
extra: dict[str, Any] = {"module": module}
@@ -226,7 +312,7 @@ def main() -> int:
# ── contour ────────────────────────────────────────────────────────────
if module in ("all", "contour"):
emit_progress(0.75, "轮廓-平整度（需 CAD STL）")
emit_progress(0.75, "轮廓-平整度（需 CAD STEP）")
cad_path = inputs.get("cadPath") or None
mesh_path = inputs.get("meshPath") or None
if not cad_path and not mesh_path:
@@ -235,14 +321,54 @@ def main() -> int:
try:
save_path = str(out_dir / "contour.ply")
excel_path = str(out_dir / "contour.xlsx")
# STEP mesh 缓存：避免每次重新三角化 442 MB+ 整车 STEP
cached_mesh_ply: str | None = None
if cad_path and not mesh_path:
_deflection = float(
_safe_float(config.get("meshDeflection")) or 0.1
)
_cache_ply = _mesh_cache_path(cad_path, _deflection)
if _cache_ply.exists():
emit_progress(0.76, "使用已缓存 CAD mesh，跳过 STEP 三角化")
mesh_path = str(_cache_ply)
cad_path = None
else:
# 先单独做三角化，完成后立即缓存，再跑点距计算
# 这样即使点距阶段 OOM，下次也不用重新三角化
emit_progress(0.76, "首次三角化 CAD STEP，结果将缓存供后续复用…")
try:
import open3d as o3d # type: ignore
_orig_cad = cad_path
sys.path.insert(0, str(alg_dir))
from cad_converter import CADConverter # type: ignore
_converter = CADConverter()
_verts, _faces = _converter.step_to_numpy_mesh(
_orig_cad, mesh_deflection=_deflection
)
_mesh_o3d = o3d.geometry.TriangleMesh()
_mesh_o3d.vertices = o3d.utility.Vector3dVector(_verts)
_mesh_o3d.triangles = o3d.utility.Vector3iVector(_faces.astype(int))
o3d.io.write_triangle_mesh(str(_cache_ply), _mesh_o3d)
emit_progress(0.80, f"CAD mesh 已缓存：{_cache_ply.name}，开始点距计算…")
mesh_path = str(_cache_ply)
cad_path = None
except Exception as _tri_exc:
emit_progress(0.76, f"三角化缓存写入失败（{_tri_exc}），继续直接跑…")
context_pc_path = (inputs.get("contextPointCloudPath") or None)
res = pipeline.run_contour(
pointcloud_path=pc_path,
cad_path=cad_path,
mesh_path=mesh_path,
point_chunk_size=8192,
triangle_chunk_size=512,
save_path=save_path,
excel_output_path=excel_path,
combined_context_pointcloud_path=context_pc_path,
no_visualize=True,
)
stats = res.get("distance_stats")
for key, label, unit in [
("mean", "轮廓偏差均值", "mm"),

179
server/plugins/adapters/train_measurement_run.py
View File

@@ -5,14 +5,14 @@
v2.2 变更：
- 算法路径从 alg/点云测距v1.1 → alg/train_measurement_system_v2.2
- 新增孔中心拟合（hole_center_fitter）、测量日志、性能日志
- 内置配置文件改为 config/measurements_0618.yaml
- TRAIN_MEAS_ROOT 指向 train_measurement_system_v2.2/（用于定位 config/）
- 内置配置文件改为 config/measurements.yaml（v3.2）
- TRAIN_MEAS_ROOT 指向 train_measurement_system_v3.2_0629/（用于定位 config/）
- 算法内置 _read_binary_ply，直接支持二进制 PLY，无需 ASCII 转换
约定：
- 第一个点云路径作为 pointcloud_file 注入到测量配置 YAML 中；
- 若用户在 UI 选择「自定义 YAML」并提供 customYaml，则使用之；
否则使用算法仓库自带的 config/measurements_0618.yaml 作为模板。
否则使用算法仓库自带的 config/measurements.yaml（v3）作为模板。
- 所有任务结果合成成 measurements[] 列表，附带可视化 PLY 路径到 extra。
"""
@@ -33,6 +33,57 @@ from _plugin_io import (
)
def _to_float(value):
if value is None:
return None
if isinstance(value, (int, float)):
return float(value)
try:
return float(str(value).strip())
except (TypeError, ValueError):
return None
def _load_excel_measurement_rows(out_dir: Path) -> dict[str, dict]:
"""读取算法导出的 Excel 结论，作为前端 measurements 的权威来源。"""
try:
from openpyxl import load_workbook
excel_files = sorted(
out_dir.glob("measurement_results_*.xlsx"),
key=lambda p: p.stat().st_mtime,
)
if not excel_files:
return {}
wb = load_workbook(excel_files[-1], data_only=True, read_only=True)
ws = wb.active
rows: dict[str, dict] = {}
for row in ws.iter_rows(min_row=2, values_only=True):
name = str(row[0] or "").strip()
if not name:
continue
conclusion = str(row[6] or "").strip()
status = None
if conclusion == "通过":
status = "pass"
elif conclusion in ("不通过", "失败"):
status = "fail"
elif conclusion == "未配置":
status = "info"
rows[name] = {
"value": _to_float(row[3]),
"toleranceLabel": str(row[2]) if row[2] is not None else None,
"status": status,
}
return rows
except Exception as exc:
print(f"读取 Excel 结论失败，继续使用适配器判定：{exc}", file=sys.stderr)
return {}
def main() -> int:
payload = load_input(sys.argv)
inputs = payload.get("inputs") or {}
@@ -41,14 +92,13 @@ def main() -> int:
out_dir.mkdir(parents=True, exist_ok=True)
try:
emit_progress(0.05, "加载 train_measurement v2.2 算法包")
emit_progress(0.05, "加载 train_measurement v3.2 算法包")
# TRAIN_MEAS_ROOT 由 manifest envExtra 注入，指向 train_measurement_system_v2.2/
# TRAIN_MEAS_ROOT 由 manifest envExtra 注入，指向 train_measurement_system_v3.2_0629/
train_root = Path(os.environ.get("TRAIN_MEAS_ROOT") or "")
if not train_root.exists():
# 回退：从 ALG_ROOT 推导
alg_root = Path(os.environ.get("ALG_ROOT") or "/alg")
train_root = alg_root / "train_measurement_system_v2.2" / "train_measurement_system_v2.2"
train_root = alg_root / "train_measurement_system_v3.2_0629"
if not train_root.exists():
raise RuntimeError(
@@ -78,8 +128,8 @@ def main() -> int:
# 已经是 YAML 格式，直接写盘
yaml_path.write_text(custom_yaml_str, encoding="utf-8")
else:
# 使用算法仓库内置的 config/measurements_0618.yaml
yaml_path = train_root / "config" / "measurements_0618.yaml"
# 使用算法仓库内置的 config/measurements.yaml（v3）
yaml_path = train_root / "config" / "measurements.yaml"
if not yaml_path.exists():
raise RuntimeError(f"内置配置文件不存在：{yaml_path}")
@@ -87,16 +137,25 @@ def main() -> int:
# ── 初始化 TaskManager ────────────────────────────────────────────────
tm = TaskManager()
# 先注入点云路径，让 load_config 跳过 YAML 里的 pointcloud_file（开发测试路径）
tm.pointcloud_path_override = pc_path
# 把输出目录重定向到 outDir，避免写到只读的 /alg 目录
tm.output_dir_override = str(out_dir)
emit_progress(0.25, f"读取点云（{os.path.basename(pc_path)}，可能需要 1~2 分钟）")
if not tm.load_config(str(yaml_path)):
raise RuntimeError(f"TaskManager 加载配置失败：{yaml_path}")
# 注入点云路径（覆盖 YAML 中的 pointcloud_file）
tm.config["pointcloud_file"] = pc_path
# 过滤 enabled=False 的任务（v3 TaskManager 不处理 enabled 字段）
import yaml as _yaml_check
_raw = _yaml_check.safe_load(yaml_path.read_text(encoding="utf-8"))
_enabled_names = {
t["name"] for t in (_raw.get("tasks") or _raw.get("measurements") or [])
if t.get("enabled", True)
}
tm.tasks = [t for t in tm.tasks if t.name in _enabled_names]
# 注入可视化输出目录
viz_dir = out_dir / "visualizations"
viz_dir.mkdir(parents=True, exist_ok=True)
@@ -104,10 +163,37 @@ def main() -> int:
if hasattr(task, "visualization_dir"):
task.visualization_dir = str(viz_dir)
emit_progress(0.30, "加载点云（已在 load_config 中完成）")
emit_progress(0.40, f"点云已加载，开始执行 {len(tm.tasks)} 个测量任务")
emit_progress(0.40, f"执行 {len(tm.tasks)} 个测量任务")
# ── 从 YAML 构建公差映射（name → {theoretical, upper, lower}） ─────────
import yaml as _yaml
try:
raw_yaml = _yaml.safe_load(yaml_path.read_text(encoding="utf-8"))
# v3 用 'tasks' key，v2 用 'measurements' key，兼容两者
task_cfgs = raw_yaml.get("tasks") or raw_yaml.get("measurements") or []
tol_map: dict[str, dict] = {}
for tc in task_cfgs:
tname = tc.get("name")
if not tname:
continue
tol = tc.get("tolerance")
theoretical = tc.get("theoretical_dimension")
if isinstance(tol, dict):
tol_map[tname] = {
"theoretical": theoretical,
"upper": float(tol.get("upper", 0)),
"lower": float(tol.get("lower", 0)),
}
elif tol is not None:
tol_map[tname] = {
"theoretical": theoretical,
"upper": float(tol),
"lower": float(tol),
}
except Exception:
tol_map = {}
results = tm.run_all_tasks()
report_rows = _load_excel_measurement_rows(out_dir)
emit_progress(0.92, "整理结果")
measurements: list[dict] = []
@@ -133,39 +219,41 @@ def main() -> int:
value = clean_data.get("mean_distance")
if value is None:
value = clean_data.get("distance")
fval = float(value) if isinstance(value, (int, float)) else None
# v3.2 data 里直接有 is_qualified 判定结果（line_to_plane/line_to_line 也支持）
data_pass = clean_data.get("is_qualified")
if data_pass is None:
data_pass = clean_data.get("pass") # 兼容旧字段名
tol_cfg = tol_map.get(name)
tolerance = None
meas_status = "info"
if data_pass is not None:
# 算法已判定，直接用
meas_status = "pass" if data_pass else "fail"
if tol_cfg:
tolerance = tol_cfg.get("upper")
elif tol_cfg and fval is not None:
theoretical = tol_cfg.get("theoretical")
upper = tol_cfg["upper"]
lower = tol_cfg["lower"]
tolerance = upper
if theoretical is not None:
error = fval - float(theoretical)
meas_status = "pass" if (-lower <= error <= upper) else "fail"
measurements.append(
{
"name": name,
"value": float(value) if isinstance(value, (int, float)) else float("nan"),
"value": fval if fval is not None else float("nan"),
"unit": "mm",
"status": "info",
"tolerance": tolerance,
"status": meas_status,
"passLabel": "通过",
"failLabel": "不通过",
}
)
std = clean_data.get("std_distance")
if isinstance(std, (int, float)):
measurements.append(
{
"name": f"{name} · 标准差",
"value": float(std),
"unit": "mm",
"status": "info",
}
)
# v1.1 新增字段：min/max distance
for field_name, label in [
("min_distance", "最小距离"),
("max_distance", "最大距离"),
]:
v = clean_data.get(field_name)
if isinstance(v, (int, float)):
measurements.append(
{
"name": f"{name} · {label}",
"value": float(v),
"unit": "mm",
"status": "info",
}
)
else:
measurements.append(
{
@@ -176,6 +264,17 @@ def main() -> int:
}
)
for measurement in measurements:
report_row = report_rows.get(measurement["name"])
if not report_row:
continue
if report_row.get("value") is not None:
measurement["value"] = report_row["value"]
if report_row.get("toleranceLabel"):
measurement["toleranceLabel"] = report_row["toleranceLabel"]
if report_row.get("status"):
measurement["status"] = report_row["status"]
success_count = sum(1 for r in results if r.get("status") == "success")
payload_out = {
"schemaVersion": 1,

231
server/plugins/adapters/underframe_deflection_run.py
View File

@@ -1,18 +1,14 @@
"""underframe-deflection 算法适配器 (v1.2.1)。
调用 alg/delivery_v1.2.1/underframe_deflection 的 run_both_beams：
- 输入：整车配准点云 + 左梁 ROI JSON + 右梁 ROI JSON + CAD STP（溯源用）
- 输出：left/ right/ 各一份 report.xlsx + report.json + deflection_curve.csv/png
- measurements 展示左右梁最大挠度及检验结论
v1.2.1 变更：
- 算法包升级至 delivery_v1.2.1（单梁结果新增项点检验字段：公差/测量/误差/结论）
- report.xlsx 等 Excel 产物通过 result.json 的 files 字段暴露给前端下载
- PYTHONPATH 指向 delivery_v1.2.1
v1.2 变更：
- 新增 single_beam 单梁管线
v1.1 变更：
- 入口改为 run_measurement_pipeline（返回 dict，同时落盘 report.json / xlsx / PLY）
- 新增 cad_path（CAD 基准 PLY/STL）和 roi_path（ROI JSON）可选输入
- 新增 skip_registration / enable_bilateral 配置项
- 结果从 report dict 中读取，而非 DeflectionResult 对象字段
- 改用 run_both_beams 双梁并行入口，与算法方命令行一致
- leftRoiPath / rightRoiPath 参数默认指向 test_data ROI
- skipRegistration 默认 True（点云已配准）
"""
from __future__ import annotations
@@ -35,142 +31,139 @@ from _plugin_io import (
def _status_for(value: float | None, tolerance: float | None) -> str:
if value is None or tolerance is None:
return "info"
if abs(value) <= tolerance:
abs_val = abs(value)
if abs_val <= tolerance:
return "pass"
if abs(value) <= tolerance * 1.5:
if abs_val <= tolerance * 1.5:
return "warning"
return "fail"
def _safe_float(v: object) -> float | None:
try:
return float(v) # type: ignore[arg-type]
except (TypeError, ValueError):
return None
def main() -> int:
payload = load_input(sys.argv)
inputs = payload.get("inputs") or {}
config = inputs.get("config") or {}
inputs = payload.get("inputs") or {}
config = inputs.get("config") or {}
out_dir = Path(payload["outDir"])
out_dir.mkdir(parents=True, exist_ok=True)
try:
emit_progress(0.05, "加载 underframe_deflection v1.2.1 算法包")
from underframe_deflection.pipeline import run_measurement_pipeline # type: ignore
emit_progress(0.05, "加载挠度算法包 v1.2.1")
pc_path = first_pointcloud(payload)
emit_progress(0.15, f"读取点云：{os.path.basename(pc_path)}")
# ── 输入文件 ────────────────────────────────────────────────────────
pointcloud_path = first_pointcloud(payload)
if not pointcloud_path:
write_error_result(out_dir, "未提供点云文件")
return 1
profile = str(config.get("profile") or "auto")
skip_registration = bool(config.get("skipRegistration") or False)
enable_bilateral = bool(config.get("enableBilateral") if "enableBilateral" in config else True)
# 可选 CAD 基准（PLY/STL）和 ROI JSON
cad_path = inputs.get("cadPath") or None
roi_path = inputs.get("roiPath") or None
emit_progress(0.20, "调用 run_measurement_pipeline")
left_roi = config.get("leftRoiPath") or \
"/alg/delivery_v1.2.1/underframe_deflection/test_data/roi_left_beam.json"
right_roi = config.get("rightRoiPath") or \
"/alg/delivery_v1.2.1/underframe_deflection/test_data/roi_right_beam.json"
# run_measurement_pipeline 内部会打印进度到 stdout，
# 我们在调用前后各报一次进度即可。
report: dict = run_measurement_pipeline(
input_cloud=pc_path,
out_dir=str(out_dir),
profile=profile,
cad_path=cad_path,
roi_path=roi_path,
skip_registration=skip_registration,
enable_bilateral=enable_bilateral,
skip_excel=False,
for p, label in [(left_roi, "左梁 ROI"), (right_roi, "右梁 ROI")]:
if not os.path.isfile(p):
write_error_result(out_dir, f"{label} 文件不存在：{p}")
return 1
tolerance_mm = float(config.get("toleranceMm") or 3.0)
left_tolerance_mm = float(config.get("leftToleranceMm") or tolerance_mm or 7.0)
right_tolerance_mm = float(config.get("rightToleranceMm") or tolerance_mm or 12.0)
skip_registration = bool(config.get("skipRegistration", True))
enable_bilateral = bool(config.get("enableBilateral", True))
emit_progress(0.10, f"开始左梁挠度测量（容差 {left_tolerance_mm} mm）")
from underframe_deflection import run_single_beam_pipeline # type: ignore
left_out = out_dir / "left"
left_out.mkdir(exist_ok=True)
left_report = run_single_beam_pipeline(
input_cloud = pointcloud_path,
roi_path = left_roi,
out_dir = str(left_out),
side = "left",
tolerance_mm = left_tolerance_mm,
cad_path = cad_path,
enable_registration= not skip_registration,
config = {"enable_bilateral": enable_bilateral},
)
emit_progress(0.92, "整理结果")
emit_progress(0.55, f"左梁完成，开始右梁挠度测量（容差 {right_tolerance_mm} mm）")
right_out = out_dir / "right"
right_out.mkdir(exist_ok=True)
right_report = run_single_beam_pipeline(
input_cloud = pointcloud_path,
roi_path = right_roi,
out_dir = str(right_out),
side = "right",
tolerance_mm = right_tolerance_mm,
cad_path = cad_path,
enable_registration= not skip_registration,
config = {"enable_bilateral": enable_bilateral},
)
# ── 从 report dict 中提取核心测量值 ──────────────────────────────────
deflection = report.get("deflection") or {}
left_sel = (deflection.get("left") or {}).get("selected_measurement") or {}
right_sel = (deflection.get("right") or {}).get("selected_measurement") or {}
result = {"left": left_report, "right": right_report}
left_mm = _safe_float(left_sel.get("signed_deflection_mm"))
right_mm = _safe_float(right_sel.get("signed_deflection_mm"))
emit_progress(0.95, "整理结果")
registration = report.get("registration") or {}
reg_rms = _safe_float(registration.get("rms_mm"))
def _beam_measurements(report: dict, label: str) -> list:
insp = report.get("inspection") or {}
item_name = insp.get("item_name") or "纵向梁到直线的垂直偏差（挠度）"
measured = insp.get("measured_mm")
tol = insp.get("tolerance_mm")
verdict = insp.get("verdict") or ""
return [{
"name" : f"{label} · {item_name}",
"value" : round(float(measured), 4) if measured is not None else 0.0,
"unit" : "mm",
"tolerance": float(tol) if tol is not None else None,
"status" : "pass" if verdict == "合格"
else "fail" if verdict == "不合格"
else "info",
}]
quality_flags: list[str] = report.get("quality_flags") or []
measurements = (
_beam_measurements(left_report, "左梁") +
_beam_measurements(right_report, "右梁")
)
# 容差：高铁底架挠度通常要求 ≤ 3 mm（可在 manifest inputs 中暴露）
tol = _safe_float(config.get("toleranceMm")) or 3.0
# 收集产物文件（xlsx/csv/png）- write_result 会自动扫 xlsx，
# 这里额外加 csv/png 让前端也能下载
extra_files: list[str] = []
for subdir in ("left", "right"):
sub = out_dir / subdir
if not sub.exists():
continue
for fname in sorted(sub.iterdir()):
if fname.suffix in (".csv", ".png"):
extra_files.append(str(fname.relative_to(out_dir)))
measurements = [
# ── 汇总消息 ────────────────────────────────────────────────────────
left_v = (left_report.get("inspection") or {}).get("verdict") or "—"
right_v = (right_report.get("inspection") or {}).get("verdict") or "—"
left_d = (left_report.get("inspection") or {}).get("measured_mm")
right_d = (right_report.get("inspection") or {}).get("measured_mm")
msg = (
f"挠度测量完成。"
f"左梁 {left_d:.3f} mm（{left_v}），"
f"右梁 {right_d:.3f} mm（{right_v}）。"
if left_d is not None and right_d is not None
else "挠度测量完成。"
)
write_result(
out_dir,
{
"name": "左梁挠度",
"value": left_mm if left_mm is not None else float("nan"),
"unit": "mm",
"tolerance": tol,
"status": _status_for(left_mm, tol),
"message": msg,
"measurements": measurements,
"files": extra_files,
"extra": {"leftReport": left_report, "rightReport": right_report},
},
{
"name": "右梁挠度",
"value": right_mm if right_mm is not None else float("nan"),
"unit": "mm",
"tolerance": tol,
"status": _status_for(right_mm, tol),
},
]
if reg_rms is not None:
measurements.append(
{
"name": "配准 RMS",
"value": reg_rms,
"unit": "mm",
"status": "pass" if reg_rms <= 1.0 else "warning" if reg_rms <= 2.0 else "fail",
}
)
# ── 3D 叠加层：优先 out_dir/ply/ 子目录，回退到 out_dir 根目录 ──────────
overlay3d: dict = {"kind": "none"}
ply_files: list = []
ply_dir = out_dir / "ply"
if ply_dir.exists():
ply_files = sorted(ply_dir.glob("*.ply"))
if not ply_files:
# 算法直接输出到 out_dir 根目录；优先取 full overlay
for name in ("roi_measurement_overlay_full.ply", "roi_cropped_cloud.ply"):
candidate = out_dir / name
if candidate.exists():
ply_files = [candidate]
break
if not ply_files:
ply_files = sorted(out_dir.glob("*.ply"))
if ply_files:
rel = ply_files[0].relative_to(out_dir)
overlay3d = {
"kind": "ply",
"file": str(rel),
}
payload_out = {
"schemaVersion": report.get("schema_version", "1.1"),
"status": "completed",
"measurements": measurements,
"overlay3d": overlay3d,
"extra": {
"algorithmVersion": report.get("algorithm_version"),
"qualityFlags": quality_flags,
"registration": registration,
"deflectionFull": deflection,
},
}
write_result(out_dir, payload_out)
emit_progress(1.0, "完成")
)
return 0
except Exception as exc:
traceback.print_exc()
write_error_result(
out_dir,
f"underframe-deflection 计算失败：{exc}",