# core/stp_parser.py from pathlib import Path from typing import Dict, Any, Optional, List import numpy as np import json from shared.utils.logger import get_logger from OCC.Core.GProp import GProp_GProps from OCC.Core.BRepGProp import brepgprop from OCC.Core.TopoDS import TopoDS_Shape logger = get_logger(__name__) class STPParser: """STP文件解析器""" def __init__(self): # 强制要求PythonOCC必须可用 self._verify_occ_availability() def _verify_occ_availability(self): """验证PythonOCC是否可用，不可用则抛出异常""" try: from OCC.Core.STEPControl import STEPControl_Reader from OCC.Core.IFSelect import IFSelect_RetDone logger.info("PythonOCC验证通过") except ImportError as e: logger.error("PythonOCC不可用，服务无法运行") raise RuntimeError("PythonOCC未安装，请安装PythonOCC后再运行服务") from e def load_step_file(self, file_path: Path) -> TopoDS_Shape: """加载STP文件""" try: from OCC.Core.STEPControl import STEPControl_Reader from OCC.Core.IFSelect import IFSelect_RetDone logger.info(f"加载STP文件: {file_path}") reader = STEPControl_Reader() status = reader.ReadFile(str(file_path)) if status == IFSelect_RetDone: reader.TransferRoots() shape = reader.OneShape() logger.info("STP文件加载成功") return shape else: raise ValueError(f"STP文件读取失败，状态码: {status}") except Exception as e: logger.error(f"STP解析失败: {e}") raise def analyze_geometry(self, shape: TopoDS_Shape) -> Dict[str, Any]: """分析几何属性""" try: from OCC.Core.GProp import GProp_GProps from OCC.Core.BRepGProp import brepgprop from OCC.Core.Bnd import Bnd_Box from OCC.Core.BRepBndLib import brepbndlib from OCC.Core.TopExp import TopExp_Explorer from OCC.Core.TopAbs import TopAbs_FACE, TopAbs_EDGE, TopAbs_VERTEX logger.info("开始几何分析...") # 计算边界框 bbox = self._compute_bounding_box(shape) # 计算体积和表面积 volume = self._compute_volume(shape) area = self._compute_surface_area(shape) # 分析拓扑 topology = self._analyze_topology(shape) # 计算质心 center_of_mass = self._compute_center_of_mass(shape) # 计算惯性属性 inertia_properties = self._compute_inertia_properties(shape) result = { "bounding_box": bbox, "volume": float(volume), "surface_area": float(area), "topology": topology, "center_of_mass": center_of_mass, "inertia_properties": inertia_properties, "analysis_method": "pythonocc" } logger.info("几何分析完成") return result except Exception as e: logger.error(f"几何分析失败: {e}") raise def _compute_bounding_box(self, shape: TopoDS_Shape) -> Dict[str, Any]: """计算边界框""" try: from OCC.Core.Bnd import Bnd_Box from OCC.Core.BRepBndLib import brepbndlib bbox = Bnd_Box() brepbndlib.Add(shape, bbox) xmin, ymin, zmin, xmax, ymax, zmax = bbox.Get() return { "min": [float(xmin), float(ymin), float(zmin)], "max": [float(xmax), float(ymax), float(zmax)], "dimensions": [ float(xmax - xmin), float(ymax - ymin), float(zmax - zmin) ], "center": [ float((xmin + xmax) / 2), float((ymin + ymax) / 2), float((zmin + zmax) / 2) ] } except Exception as e: logger.error(f"边界框计算失败: {e}") return self._default_bounding_box() def _compute_volume(self, shape: TopoDS_Shape) -> float: """计算体积""" try: from OCC.Core.GProp import GProp_GProps from OCC.Core.BRepGProp import brepgprop props = GProp_GProps() brepgprop.VolumeProperties(shape, props) volume = props.Mass() if volume <= 0: raise ValueError("计算得到的体积为0或负数，形状可能无效") return volume except Exception as e: logger.error(f"体积计算失败: {e}") raise RuntimeError(f"体积计算失败: {e}") from e def _compute_surface_area(self, shape: TopoDS_Shape) -> float: """计算表面积""" try: from OCC.Core.GProp import GProp_GProps from OCC.Core.BRepGProp import brepgprop props = GProp_GProps() brepgprop.SurfaceProperties(shape, props) area = props.Mass() # 如果计算结果为0，使用备选估算方法 if area <= 0: logger.warning("表面积计算结果为0，使用边界框估算") raise ValueError("Surface area is zero") return area except ValueError: # 基于边界框估算表面积 try: bbox = self._compute_bounding_box(shape) dims = bbox.get("dimensions", [0, 0, 0]) if any(d <= 0 for d in dims): raise RuntimeError("边界框尺寸无效，无法估算表面积") # 简化的估算公式：2*(lw + lh + wh) estimated_area = 2 * (dims[0]*dims[1] + dims[0]*dims[2] + dims[1]*dims[2]) logger.warning(f"使用边界框估算表面积: {estimated_area:.2f} mm²") return estimated_area except Exception as e: logger.error(f"表面积估算失败: {e}") raise RuntimeError(f"表面积计算失败: {e}") from e except Exception as e: logger.error(f"表面积计算失败: {e}") raise RuntimeError(f"表面积计算失败: {e}") from e def _compute_center_of_mass(self, shape: TopoDS_Shape) -> List[float]: """计算质心""" try: from OCC.Core.GProp import GProp_GProps from OCC.Core.BRepGProp import brepgprop props = GProp_GProps() brepgprop.VolumeProperties(shape, props) center = props.CentreOfMass() return [float(center.X()), float(center.Y()), float(center.Z())] except Exception as e: logger.error(f"质心计算失败: {e}") # 回退到边界框中心 try: bbox = self._compute_bounding_box(shape) return bbox.get("center", [0.0, 0.0, 0.0]) except Exception: raise RuntimeError(f"质心计算失败且边界框回退也失败: {e}") from e def _compute_inertia_properties(self, shape: TopoDS_Shape) -> Dict[str, Any]: """计算惯性属性""" try: from OCC.Core.GProp import GProp_GProps from OCC.Core.BRepGProp import brepgprop props = GProp_GProps() brepgprop.VolumeProperties(shape, props) inertia = props.MatrixOfInertia() return { "mass": float(props.Mass()), "moment_of_inertia": [ [float(inertia.Value(1, 1)), float(inertia.Value(1, 2)), float(inertia.Value(1, 3))], [float(inertia.Value(2, 1)), float(inertia.Value(2, 2)), float(inertia.Value(2, 3))], [float(inertia.Value(3, 1)), float(inertia.Value(3, 2)), float(inertia.Value(3, 3))] ] } except Exception as e: logger.error(f"惯性属性计算失败: {e}") return {} def _analyze_topology(self, shape: TopoDS_Shape) -> Dict[str, int]: """分析拓扑""" try: from OCC.Core.TopExp import TopExp_Explorer from OCC.Core.TopAbs import TopAbs_FACE, TopAbs_EDGE, TopAbs_VERTEX def count_elements(element_type): explorer = TopExp_Explorer(shape, element_type) count = 0 while explorer.More(): count += 1 explorer.Next() return count return { "faces": count_elements(TopAbs_FACE), "edges": count_elements(TopAbs_EDGE), "vertices": count_elements(TopAbs_VERTEX) } except Exception as e: logger.error(f"拓扑分析失败: {e}") raise def _default_bounding_box(self) -> Dict[str, Any]: """默认边界框（边界框计算失败时的回退值，标注为估算）""" return { "min": [0.0, 0.0, 0.0], "max": [0.0, 0.0, 0.0], "dimensions": [0.0, 0.0, 0.0], "center": [0.0, 0.0, 0.0], "estimated": True } def export_to_json(self, geometry_data: Dict[str, Any], output_path: Path) -> str: """将几何数据导出为JSON文件""" try: # 确保输出目录存在 output_path.parent.mkdir(parents=True, exist_ok=True) # 添加元数据 json_data = { "metadata": { "export_time": str(np.datetime64('now')), "analysis_method": geometry_data.get("analysis_method", "unknown"), "version": "1.0.0" }, "geometry_data": geometry_data } # 保存JSON文件 with open(output_path, 'w', encoding='utf-8') as f: json.dump(json_data, f, indent=2, ensure_ascii=False) logger.info(f"几何数据已导出到: {output_path}") return str(output_path) except Exception as e: logger.error(f"JSON导出失败: {e}") raise def get_json_data(self, geometry_data: Dict[str, Any]) -> Dict[str, Any]: """获取JSON格式的几何数据""" return { "metadata": { "export_time": str(np.datetime64('now')), "analysis_method": geometry_data.get("analysis_method", "unknown"), "version": "1.0.0" }, "geometry_data": geometry_data }