262 lines
10 KiB
Python
262 lines
10 KiB
Python
"""coupler-panel 算法适配器 (v1.2.1)。
|
||
|
||
v1.2.1 变更：
|
||
- 算法包升级至 delivery_v1.2.1
|
||
- 新增项点 Excel 报表（write_inspection_report）：
|
||
项点名称/理论尺寸/公差要求/测量结果/误差结果/测量结论
|
||
- PYTHONPATH 指向 delivery_v1.2.1/coupler_panel_release
|
||
|
||
v1.2 变更：
|
||
- 新增 autotune 自动调参与 layer_select 分层选点
|
||
|
||
v1.1 变更：
|
||
- 入口从 compute_coupler_measurement → run_coupler_measurement_pipeline
|
||
- 新增 D_y_max_mm / D_z_max_mm 字段
|
||
- 支持可选 CAD STP 路径（cad_stp_path）和 ROI JSON（roi_json）
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import sys
|
||
import traceback
|
||
from pathlib import Path
|
||
|
||
sys.path.insert(0, str(Path(__file__).parent))
|
||
from _plugin_io import (
|
||
emit_progress,
|
||
first_pointcloud,
|
||
load_input,
|
||
write_error_result,
|
||
write_result,
|
||
)
|
||
|
||
|
||
def _status_for(value: float, tolerance: float | None) -> str:
|
||
if tolerance is None:
|
||
return "info"
|
||
if abs(value) <= tolerance:
|
||
return "pass"
|
||
if abs(value) <= tolerance * 1.5:
|
||
return "warning"
|
||
return "fail"
|
||
|
||
|
||
def main() -> int:
|
||
payload = load_input(sys.argv)
|
||
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, "加载 coupler_panel v1.2.1 算法包")
|
||
from coupler_panel.pipeline import run_coupler_measurement_pipeline # type: ignore
|
||
|
||
pc_path = first_pointcloud(payload)
|
||
emit_progress(0.15, f"读取点云：{os.path.basename(pc_path)}")
|
||
|
||
profile = str(config.get("profile") or "auto")
|
||
right_axis_sign = int(config.get("rightAxisSign") or 1)
|
||
|
||
# 可选 CAD STP 路径（前端传入 cadPath 时使用）
|
||
cad_stp_path = inputs.get("cadPath") or None
|
||
|
||
# ROI 裁剪：优先用 roiJson 文件，其次用前端填的 6 个坐标参数
|
||
roi_json = inputs.get("roiJson") or None
|
||
roi_spec = None
|
||
if not roi_json:
|
||
def _f(key: str) -> float | None:
|
||
v = config.get(key)
|
||
try:
|
||
return float(v) if v is not None and str(v).strip() != "" else None
|
||
except (TypeError, ValueError):
|
||
return None
|
||
|
||
x_min, x_max = _f("roiXMin"), _f("roiXMax")
|
||
y_min, y_max = _f("roiYMin"), _f("roiYMax")
|
||
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_select import ROISpec # type: ignore
|
||
|
||
roi_spec = ROISpec(
|
||
x_min=x_min if x_min is not None else float("-inf"),
|
||
x_max=x_max if x_max is not None else float("inf"),
|
||
y_min=y_min if y_min is not None else float("-inf"),
|
||
y_max=y_max if y_max is not None else float("inf"),
|
||
z_min=z_min if z_min is not None else float("-inf"),
|
||
z_max=z_max if z_max is not None else float("inf"),
|
||
)
|
||
emit_progress(0.20, f"ROI 裁剪：x[{x_min},{x_max}] y[{y_min},{y_max}] z[{z_min},{z_max}]")
|
||
|
||
emit_progress(0.25, "调用 run_coupler_measurement_pipeline")
|
||
|
||
def _on_progress(stage: str, fraction: float) -> None:
|
||
emit_progress(0.25 + min(1.0, max(0.0, fraction)) * 0.70, stage)
|
||
|
||
result = run_coupler_measurement_pipeline(
|
||
input_cloud=pc_path,
|
||
out_dir=str(out_dir),
|
||
profile=profile,
|
||
cad_stp_path=cad_stp_path,
|
||
roi_spec=roi_spec, # 优先用坐标参数构建的 ROISpec
|
||
roi_json=roi_json, # 备选：直接传 ROI JSON 文件
|
||
progress=_on_progress,
|
||
)
|
||
|
||
emit_progress(0.97, "整理结果")
|
||
|
||
# right_axis_sign 影响 theta_right_deg 的符号约定
|
||
theta_right = float(result.theta_right_deg) * right_axis_sign
|
||
theta_ref = float(result.theta_ref_deg)
|
||
|
||
# ── 项点 Excel 报表（v1.2.1 新增）────────────────────────────────────
|
||
# 用未截断的拟合平面法向量算原始角度（避免微小角被钳为 0），
|
||
# 公差按斜率形式给出：参考面 0.8/375mm，纵向中心 1/700mm。
|
||
# 产物 xlsx 由 write_result 自动发现并写入 files 字段。
|
||
try:
|
||
import math
|
||
|
||
from coupler_panel.reporting import InspectionItem, write_inspection_report # type: ignore
|
||
|
||
raw_right, raw_ref = theta_right, theta_ref
|
||
report_dict = result.report if isinstance(result.report, dict) else {}
|
||
normal = (report_dict.get("plane") or {}).get("normal")
|
||
if normal and len(normal) == 3:
|
||
nx, ny, nz = (float(v) for v in normal)
|
||
raw_right = math.degrees(math.atan2(ny, nx)) * right_axis_sign
|
||
raw_ref = math.degrees(math.atan2(nz, nx))
|
||
|
||
item_ref = InspectionItem(
|
||
name="车钩面板距参考面的右偏转角（头车一位端处车钩）",
|
||
theta_deg=raw_ref,
|
||
baseline_mm=375.0,
|
||
tol_value_mm=0.8,
|
||
)
|
||
item_long = InspectionItem(
|
||
name="车钩面板距纵向中心的右偏转角(头车一位端处车钩)",
|
||
theta_deg=raw_right,
|
||
baseline_mm=700.0,
|
||
tol_value_mm=1.0,
|
||
)
|
||
write_inspection_report(item_ref, out_dir / "report_ref_plane.xlsx")
|
||
write_inspection_report(item_long, out_dir / "report_longitudinal.xlsx")
|
||
except Exception:
|
||
# Excel 报表失败不影响主结果
|
||
traceback.print_exc()
|
||
dx_y = float(result.dx_y_max_mm)
|
||
dx_z = float(result.dx_z_max_mm)
|
||
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))
|
||
|
||
inspection_object = str((payload.get("context") or {}).get("inspectionObject") or "unknown")
|
||
longitudinal_task_no = "10" if inspection_object == "head" else "12" if inspection_object == "middle" else None
|
||
reference_task_no = "8" if inspection_object == "head" else "9" if inspection_object == "middle" else None
|
||
raw_selected_task_nos = config.get("demoTaskNos")
|
||
selected_task_nos = (
|
||
{str(int(value)) for value in raw_selected_task_nos if isinstance(value, (int, float)) and float(value).is_integer()}
|
||
if isinstance(raw_selected_task_nos, list)
|
||
else None
|
||
)
|
||
|
||
def _selected(task_no: str | None) -> bool:
|
||
return selected_task_nos is None or (task_no is not None and task_no in selected_task_nos)
|
||
|
||
measurements = []
|
||
if _selected(longitudinal_task_no):
|
||
longitudinal_measurement = {
|
||
"name": "车钩面板距纵向中心的右偏转角",
|
||
"value": round(offset_right, 6),
|
||
"unit": "mm",
|
||
"baseline": 700.0,
|
||
"tolerance": 1.0,
|
||
"status": _status_for(offset_right, 1.0),
|
||
}
|
||
if longitudinal_task_no:
|
||
longitudinal_measurement.update({
|
||
"taskNo": longitudinal_task_no,
|
||
"metricCode": "coupler.longitudinal_right_offset",
|
||
"name": "纵向中心右偏转",
|
||
})
|
||
measurements.append(longitudinal_measurement)
|
||
if _selected(reference_task_no):
|
||
reference_measurement = {
|
||
"name": "车钩面板距参考面的右偏转角",
|
||
"value": round(offset_ref, 6),
|
||
"unit": "mm",
|
||
"baseline": 375.0,
|
||
"tolerance": 0.8,
|
||
"status": _status_for(offset_ref, 0.8),
|
||
}
|
||
if reference_task_no:
|
||
reference_measurement.update({
|
||
"taskNo": reference_task_no,
|
||
"metricCode": "coupler.reference_right_offset",
|
||
"name": "参考面右偏转",
|
||
})
|
||
measurements.append(reference_measurement)
|
||
measurements.extend([
|
||
{
|
||
"name": "最大 x 偏差 dx_y",
|
||
"value": dx_y,
|
||
"unit": "mm",
|
||
"tolerance": 1.0,
|
||
"status": _status_for(dx_y, 1.0),
|
||
},
|
||
{
|
||
"name": "最大 x 偏差 dx_z",
|
||
"value": dx_z,
|
||
"unit": "mm",
|
||
"tolerance": 1.0,
|
||
"status": _status_for(dx_z, 1.0),
|
||
},
|
||
{
|
||
"name": "面板 y 向最大偏差 D_y",
|
||
"value": D_y,
|
||
"unit": "mm",
|
||
"status": "info",
|
||
},
|
||
{
|
||
"name": "面板 z 向最大偏差 D_z",
|
||
"value": D_z,
|
||
"unit": "mm",
|
||
"status": "info",
|
||
},
|
||
])
|
||
|
||
payload_out = {
|
||
"schemaVersion": getattr(result, "schema_version", "1.1"),
|
||
"status": "completed",
|
||
"measurements": measurements,
|
||
"overlay3d": {"kind": "none"},
|
||
"extra": {
|
||
"algorithmVersion": getattr(result, "algorithm_version", None),
|
||
"qualityFlags": list(result.quality_flags or []),
|
||
"normal": list(result.normal) if result.normal else None,
|
||
"centerC": list(result.center_C) if result.center_C else None,
|
||
"report": result.report if isinstance(result.report, dict) else None,
|
||
},
|
||
}
|
||
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"coupler-panel 计算失败：{exc}",
|
||
extra={"traceback": traceback.format_exc()},
|
||
)
|
||
return 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|