diff --git a/src/api/routes.py b/src/api/routes.py index be73709..81cc611 100644 --- a/src/api/routes.py +++ b/src/api/routes.py @@ -12,6 +12,9 @@ from core.geometry_analyzer import GeometryAnalyzer from utils.file_handler import FileHandler from utils.html_generator import HTMLGenerator from services.storage_integration_rustfs import StorageIntegrationService +from services.redis_task_manager import redis_task_manager +from services.processing_service import processing_service +from services.task_query_service import TaskQueryService from database.database import get_db_session from utils.logger import get_logger from sqlalchemy.ext.asyncio import AsyncSession @@ -58,10 +61,12 @@ tasks = {} @router.get("/health") @router.post("/health") async def health(): + task_count = await redis_task_manager.get_task_count() return { "status": "healthy", "pythonocc": True, - "total_tasks": len(tasks) + "total_tasks": task_count, + "redis_connected": redis_task_manager.is_connected, } @@ -97,8 +102,8 @@ async def upload_stp( # 创建处理任务记录 await storage_service.create_processing_task(db_session, task_id, stp_file.id) - # 创建内存任务记录 - tasks[task_id] = create_task_info( + # 创建任务记录（Redis 为主，内存为兼容回退） + task_info = create_task_info( task_id=task_id, status=ProcessingStatus.PROCESSING, filename=file.filename, @@ -106,9 +111,14 @@ async def upload_stp( file_size=file_size, upload_time=str(datetime.now()) ) + await redis_task_manager.set_task(task_id, task_info) + tasks[task_id] = task_info - # 后台处理（包含数据库存储） - background_tasks.add_task(process_file_with_storage, task_id, file_path, stp_file.id, db_session, material) + # 后台处理（统一走 ProcessingService，使用独立数据库会话） + background_tasks.add_task( + processing_service.process_file_with_storage, + task_id, file_path, stp_file.id, material + ) return { "task_id": task_id, @@ -134,99 +144,12 @@ async def get_status(task_id: str, db_session: AsyncSession = Depends(get_db_ses 结构与内存任务保持尽量一致，便于前端集中展示总结性信息。 """ try: - # 1. 内存任务（进行中的任务） - if task_id in tasks: - task = tasks[task_id] - logger.info(f"返回内存任务状态：{task_id} - {task['status']}") - logger.info(f"内存任务 analysis_result: {task.get('analysis_result', 'None')}") - logger.info(f"内存任务 html_file: {task.get('html_file', 'None')}") - return task - - # 2. 持久化任务（已完成/失败，或服务重启后的任务） - from sqlalchemy import select - from models.database import ProcessingTask, STPFile, GeometryData, MeshData, MoldCavityData - storage_service = StorageIntegrationService() - - # 查询任务和文件元数据 - result = await db_session.execute( - select(ProcessingTask, STPFile) - .join(STPFile, ProcessingTask.stp_file_id == STPFile.id) - .where(ProcessingTask.task_id == task_id) - ) - row = result.first() - if not row: + task_view = await TaskQueryService.get_task_view(db_session, task_id) + if task_view is None: + # 兼容老的仅内存任务 + if task_id in tasks: + return tasks[task_id] raise HTTPException(404, "任务不存在") - - processing_task, stp_file = row - - # 从 RustFS 取几何 / 型腔 / 网格详细 JSON（小量数据，便于前端展示汇总） - try: - file_with_data = await storage_service.get_stp_file_with_data( - db_session, stp_file_id=stp_file.id - ) - except Exception as e: - logger.error(f"获取文件数据失败: {e}") - file_with_data = {} - - geometry_json: Optional[Dict[str, Any]] = None - if file_with_data.get("geometry_data"): - geo_raw = file_with_data["geometry_data"] - if isinstance(geo_raw, dict): - if "geometry_data" in geo_raw: - geometry_json = geo_raw["geometry_data"] - else: - geometry_json = geo_raw - - cavity_json: Optional[Dict[str, Any]] = file_with_data.get("mold_cavity_data") - features_json: List[Dict[str, Any]] = file_with_data.get("features", []) - recommendations_json: List[Dict[str, Any]] = file_with_data.get("recommendations", []) - - # 组装网格摘要 - mesh_summary = None - mesh_record = await db_session.execute( - select(MeshData).where(MeshData.stp_file_id == stp_file.id) - ) - mesh_record = mesh_record.scalar_one_or_none() - if mesh_record: - mesh_summary = { - "vertex_count": mesh_record.vertex_count, - "face_count": mesh_record.face_count, - "point_count": mesh_record.point_count, - "quality": mesh_record.quality, - } - - # 构造与内存任务兼容的任务视图 - task_view = { - "task_id": processing_task.task_id, - "status": processing_task.status, - "filename": stp_file.original_filename if stp_file else "", - "file_path": stp_file.file_path or "", - "file_size": stp_file.file_size if stp_file else 0, - "upload_time": processing_task.created_time.isoformat() - if processing_task.created_time - else "", - "completed_at": processing_task.completed_time.isoformat() - if processing_task.completed_time - else "", - "geometry_data": geometry_json, - "key_info": cavity_json, - "cavity_data": cavity_json, - "mesh_summary": mesh_summary, - "analysis_result": { - "geometry_data": geometry_json, - "detected_features": features_json, - "design_recommendations": recommendations_json, - "quality_metrics": { - "volume_utilization": file_with_data.get("analysis_metrics", {}).get("volume_utilization", 0), - "topology_complexity": file_with_data.get("analysis_metrics", {}).get("topology_complexity", 0), - "wall_uniformity": file_with_data.get("analysis_metrics", {}).get("wall_uniformity", 0) - }, - "analysis_summary": file_with_data.get("analysis_metrics", {}).get("analysis_summary", "分析完成") - } if geometry_json or features_json or recommendations_json else None, - "error": processing_task.error_message or stp_file.error_message or None, - } - - logger.info(f"返回持久化任务状态: {task_id} - {processing_task.status}") return task_view except HTTPException: raise @@ -239,9 +162,12 @@ async def get_status(task_id: str, db_session: AsyncSession = Depends(get_db_ses @router.post("/debug/tasks") async def debug_tasks(): """调试接口：查看所有任务""" + all_tasks = await redis_task_manager.get_all_tasks() return { - "total_tasks": len(tasks), - "tasks": tasks + "total_tasks": len(all_tasks), + "tasks": all_tasks, + "redis_connected": redis_task_manager.is_connected, + "memory_fallback_tasks": len(tasks), } diff --git a/src/core/aluminum_foam_mold.py b/src/core/aluminum_foam_mold.py index 18c2e22..3fa436a 100644 --- a/src/core/aluminum_foam_mold.py +++ b/src/core/aluminum_foam_mold.py @@ -28,6 +28,7 @@ from OCC.Core.BRepGProp import brepgprop from models.schemas import create_mold_cavity_data, create_mold_key_info from utils.logger import get_logger from core.base_mold_generator import BaseMoldGenerator +from core.side_action_designer import SideActionDesigner logger = get_logger(__name__) @@ -102,6 +103,7 @@ class AluminumFoamMoldGenerator(BaseMoldGenerator): self.cavity_count = 1 self.parting_precision = 0.1 self.cavity_match_rate = 95.0 + self.side_action_designer = SideActionDesigner() def set_foam_material(self, material: str): """设置铝泡沫材料""" @@ -147,8 +149,17 @@ class AluminumFoamMoldGenerator(BaseMoldGenerator): parting_result = self._detect_parting_surfaces(product_shape, analysis) primary_parting_surface = parting_result["primary_surface"] primary_parting_line = parting_result["primary_line"] + primary_parting_direction = parting_result["primary_direction"] - undercut_regions = self._detect_undercut_regions(product_shape, primary_parting_surface) + side_action_result = self.side_action_designer.analyze_and_design( + shape=product_shape, + parting_direction=primary_parting_direction, + mold_size=self._calculate_mold_size(analysis), + parting_surface=primary_parting_surface, + ) + undercut_regions = self._build_undercut_regions( + side_action_result.get("undercut_analysis", {}) + ) scaled_shape = self._apply_shrinkage_compensation(product_shape) @@ -170,6 +181,7 @@ class AluminumFoamMoldGenerator(BaseMoldGenerator): "mold_block": mold_block, "analysis": analysis, "undercut_regions": undercut_regions, + "side_actions": side_action_result, "parting_surfaces": parting_result, "material": self.foam_material, "shrinkage_applied": self.shrinkage_rate, @@ -224,6 +236,7 @@ class AluminumFoamMoldGenerator(BaseMoldGenerator): }, "quality_checks": { "undercut_regions": cavity_data.get("undercut_regions", []), + "side_actions": cavity_data.get("side_actions", {}), "parting_line_smoothness": self._assess_parting_line_smoothness( cavity_data.get("parting_line", []) ) @@ -265,6 +278,7 @@ class AluminumFoamMoldGenerator(BaseMoldGenerator): "quality_considerations": { "undercut_count": len(cavity_data.get("undercut_regions", [])), "undercut_regions": cavity_data.get("undercut_regions", []), + "side_action_summary": cavity_data.get("side_actions", {}).get("summary", {}), "sink_mark_risk": self._identify_sink_mark_risk(analysis), "warpage_risk": self._assess_warpage_risk(analysis), "venting_requirement": self._assess_venting_requirement(analysis) @@ -278,130 +292,27 @@ class AluminumFoamMoldGenerator(BaseMoldGenerator): def _analyze_product_geometry(self, shape: Any) -> Dict[str, Any]: """分析产品几何属性（扩展基类版本，增加法向量统计）""" result = super()._analyze_product_geometry(shape) - result["normal_statistics"] = self._analyze_face_normals(shape) + result["normal_statistics"] = self._analyze_parting_direction(shape) return result def _split_cavity_core(self, shape: Any, parting_surface: Any) -> Tuple[Any, Any]: """分离型腔和型芯（铝泡沫使用更大余量）""" return super()._split_cavity_core(shape, parting_surface, margin=25) - def _analyze_face_normals(self, shape: Any) -> Dict[str, Any]: - """ - 改进的法向量分析 - 使用高斯权重和多点采样 - """ - face_normals = [] - face_centers = [] - face_areas = [] - - explorer = TopExp_Explorer(shape, TopAbs_FACE) - - while explorer.More(): - face = TopoDS_Face(explorer.Current()) - try: - surface = BRepAdaptor_Surface(face) - - u_min, u_max = surface.FirstUParameter(), surface.LastUParameter() - v_min, v_max = surface.FirstVParameter(), surface.LastVParameter() - - sample_points = 4 - normal_sum = np.array([0.0, 0.0, 0.0]) - - for i in range(sample_points): - for j in range(sample_points): - u = u_min + (u_max - u_min) * i / (sample_points - 1) if sample_points > 1 else (u_min + u_max) / 2 - v = v_min + (v_max - v_min) * j / (sample_points - 1) if sample_points > 1 else (v_min + v_max) / 2 - - if surface.GetType() == 0: - normal = surface.Plane().Position().Direction() - normal_sum += np.array([normal.X(), normal.Y(), normal.Z()]) - break - if surface.GetType() == 0: - break - - bbox = Bnd_Box() - brepbndlib_Add(face, bbox) - center = bbox.Center() - - face_props = GProp_GProps() - brepgprop.SurfaceProperties(face, face_props) - area = face_props.Mass() - - length = np.linalg.norm(normal_sum) - if length > 0.001: - normal_sum /= length - - face_normals.append(normal_sum) - face_centers.append([center.X(), center.Y(), center.Z()]) - face_areas.append(area) - - except Exception as e: - logger.debug(f"面分析失败: {e}") - - explorer.Next() - - if not face_normals: - return { - "primary_direction": [0, 0, 1], - "confidence": 0.5, - "face_count": 0 - } - - total_area = sum(face_areas) - weighted_normal = np.array([0.0, 0.0, 0.0]) - - for i, normal in enumerate(face_normals): - weight = face_areas[i] / total_area if total_area > 0 else 1.0 / len(face_normals) - weighted_normal += normal * weight - - length = np.linalg.norm(weighted_normal) - if length > 0.001: - weighted_normal /= length - - dot_products = [] - for normal in face_normals: - dot = np.dot(normal, weighted_normal) - dot_products.append(abs(dot)) - - confidence = np.mean(dot_products) if dot_products else 0.5 - - return { - "primary_direction": weighted_normal.tolist(), - "confidence": float(confidence), - "face_count": len(face_normals), - "normal_distribution": face_normals - } - def _detect_parting_surfaces(self, shape: Any, analysis: Dict) -> Dict[str, Any]: """ 检测分型面（支持多分型面） """ - if self.ai_parting_detector is not None: - try: - ai_result = self.ai_parting_detector.detect(shape, analysis) - if ai_result: - return self._create_parting_surface_from_ai(ai_result, analysis, shape) - except Exception as e: - logger.warning(f"AI 分型面检测失败: {e}") + primary_parting = self._detect_primary_parting(shape, analysis) + parting_surface = primary_parting["surface"] + parting_line = self.optimize_parting_line(primary_parting["line"]) + primary_direction = primary_parting["direction"] - normal_stats = analysis.get("normal_statistics", {}) - primary_direction = normal_stats.get("primary_direction", [0, 0, 1]) + additional_surfaces = [] bbox = analysis["bounding_box"] center = bbox["center"] - dir_obj = gp_Dir(primary_direction[0], primary_direction[1], primary_direction[2]) - parting_plane = gp_Pln(gp_Pnt(center[0], center[1], center[2]), dir_obj) - - try: - parting_surface = BRepBuilderAPI_MakeFace(parting_plane).Face() - except Exception: - parting_plane = gp_Pln(gp_Pnt(0, 0, center[2]), gp_Dir(0, 0, 1)) - parting_surface = BRepBuilderAPI_MakeFace(parting_plane).Face() - - parting_line = self._calculate_parting_line(shape, parting_surface) - - additional_surfaces = [] - dims = bbox["dimensions"] max_dim = max(dims) min_dim = min(dims) @@ -422,59 +333,29 @@ class AluminumFoamMoldGenerator(BaseMoldGenerator): "primary_surface": parting_surface, "primary_line": parting_line, "primary_direction": primary_direction, - "confidence": normal_stats.get("confidence", 0.5), + "confidence": primary_parting["confidence"], + "method": primary_parting["method"], "additional_surfaces": additional_surfaces, "surface_count": 1 + len(additional_surfaces) } - def _detect_undercut_regions(self, shape: Any, parting_surface: Any) -> List[Dict]: - """ - 检测倒扣区域 - """ - undercut_regions = [] + def _build_undercut_regions(self, undercut_analysis: Dict[str, Any]) -> List[Dict[str, Any]]: + """将侧向机构分析结果转换为兼容旧结构的倒扣区域列表。""" + undercut_faces = undercut_analysis.get("undercut_faces", []) + regions = [] - try: - surface = BRepAdaptor_Surface(parting_surface) - parting_normal = surface.Plane().Position().Direction() + for face in undercut_faces: + regions.append({ + "type": "negative_draft", + "location": face.get("center", [0, 0, 0]), + "severity": face.get("severity", "medium"), + "area": face.get("area", 0), + "is_outer": face.get("is_outer", False), + "face_index": face.get("face_index"), + }) - explorer = TopExp_Explorer(shape, TopAbs_FACE) - - while explorer.More(): - face = TopoDS_Face(explorer.Current()) - - try: - face_surface = BRepAdaptor_Surface(face) - - if face_surface.GetType() == 0: - face_normal = face_surface.Plane().Position().Direction() - - dot = (face_normal.X() * parting_normal.X() + - face_normal.Y() * parting_normal.Y() + - face_normal.Z() * parting_normal.Z()) - - if dot < -0.7: - bbox = Bnd_Box() - brepbndlib_Add(face, bbox) - center = bbox.Center() - - if center.Z() < 0: - undercut_regions.append({ - "type": "negative_draft", - "location": [center.X(), center.Y(), center.Z()], - "severity": abs(dot) - }) - - except Exception as e: - logger.debug(f"倒扣检测失败: {e}") - - explorer.Next() - - logger.info(f"检测到 {len(undercut_regions)} 个倒扣区域") - - except Exception as e: - logger.warning(f"倒扣区域检测异常: {e}") - - return undercut_regions + logger.info(f"转换得到 {len(regions)} 个兼容倒扣区域") + return regions def _smooth_parting_line(self, parting_line: List[List[float]]) -> List[List[float]]: """ @@ -561,23 +442,13 @@ class AluminumFoamMoldGenerator(BaseMoldGenerator): def _extract_parting_surface_geometry(self, surface: Any) -> Dict[str, Any]: """提取分型面几何数据""" - try: - adaptor = BRepAdaptor_Surface(surface) - normal = adaptor.Plane().Position().Direction() - - return { - "type": "plane", - "normal": [normal.X(), normal.Y(), normal.Z()], - "origin": [0, 0, 0], - "bounds": {"u_range": [-200, 200], "v_range": [-200, 200]} - } - except Exception: - return { - "type": "plane", - "normal": [0, 0, 1], - "origin": [0, 0, 0], - "bounds": {"u_range": [-200, 200], "v_range": [-200, 200]} - } + metadata = self._extract_plane_metadata(surface) + return { + "type": "plane", + "normal": metadata["normal"], + "origin": metadata["origin"], + "bounds": metadata["bounds"], + } def _create_parting_surface_from_ai(self, ai_result: Dict, analysis: Dict, shape: Any = None) -> Dict: diff --git a/src/core/base_mold_generator.py b/src/core/base_mold_generator.py index 98feff9..f4aea9d 100644 --- a/src/core/base_mold_generator.py +++ b/src/core/base_mold_generator.py @@ -224,6 +224,180 @@ class BaseMoldGenerator: logger.error(f"产品几何分析失败: {e}") raise + def _analyze_parting_direction(self, shape: Any) -> Dict[str, Any]: + """ + 分析主分型方向。 + + 使用面积加权的面法向统计，作为普通模具与铝泡沫模具的统一几何回退。 + """ + face_normals = [] + face_areas = [] + explorer = TopExp_Explorer(shape, TopAbs_FACE) + + while explorer.More(): + face = TopoDS_Face(explorer.Current()) + explorer.Next() + + try: + normal = self._get_face_normal(face) + if normal is None: + continue + + face_props = GProp_GProps() + brepgprop.SurfaceProperties(face, face_props) + area = max(float(face_props.Mass()), 1e-6) + + face_normals.append(np.array([normal.X(), normal.Y(), normal.Z()], dtype=np.float64)) + face_areas.append(area) + except Exception as e: + logger.debug(f"分型方向面分析失败: {e}") + + if not face_normals: + return { + "primary_direction": [0.0, 0.0, 1.0], + "confidence": 0.5, + "face_count": 0, + } + + normals = np.array(face_normals, dtype=np.float64) + areas = np.array(face_areas, dtype=np.float64) + total_area = float(areas.sum()) + if total_area > 1e-6: + weights = areas / total_area + weighted_normal = np.sum(normals * weights[:, np.newaxis], axis=0) + else: + weighted_normal = np.mean(normals, axis=0) + + norm = np.linalg.norm(weighted_normal) + if norm > 1e-6: + weighted_normal /= norm + else: + weighted_normal = np.array([0.0, 0.0, 1.0], dtype=np.float64) + + confidence = float(np.mean(np.abs(np.dot(normals, weighted_normal)))) + + return { + "primary_direction": weighted_normal.tolist(), + "confidence": confidence, + "face_count": len(face_normals), + "normal_distribution": normals.tolist(), + } + + def _create_parting_surface_from_direction( + self, + shape: Any, + analysis: Dict[str, Any], + direction: Any, + extension: float = 10.0, + ) -> Any: + """按给定方向创建覆盖产品边界的分型面。""" + bbox = analysis.get("bounding_box", {}) + center = bbox.get("center", [0.0, 0.0, 0.0]) + dims = bbox.get("dimensions", [100.0, 100.0, 100.0]) + + dir_obj = self._normalize_direction(direction) + plane = gp_Pln(gp_Pnt(center[0], center[1], center[2]), dir_obj) + + span = max(max(dims), 1.0) + extension * 2 + try: + return BRepBuilderAPI_MakeFace(plane, -span, span, -span, span).Face() + except Exception: + return BRepBuilderAPI_MakeFace(plane).Face() + + def _normalize_direction(self, direction: Any) -> gp_Dir: + """归一化分型方向，异常时回退到 Z 轴。""" + try: + if isinstance(direction, gp_Dir): + return direction + + if isinstance(direction, np.ndarray): + values = direction.tolist() + else: + values = list(direction) + + if len(values) < 3: + raise ValueError("direction 维度不足") + + vec = np.array(values[:3], dtype=np.float64) + norm = np.linalg.norm(vec) + if norm <= 1e-6: + raise ValueError("direction 长度为 0") + + vec /= norm + return gp_Dir(float(vec[0]), float(vec[1]), float(vec[2])) + except Exception: + return gp_Dir(0, 0, 1) + + def _detect_primary_parting(self, shape: Any, analysis: Dict[str, Any]) -> Dict[str, Any]: + """ + 统一主分型面检测。 + + 返回统一结构，便于子类按需追加多分型面、倒扣与平滑逻辑。 + """ + if self.ai_parting_detector is not None: + try: + ai_result = self.ai_parting_detector.detect(shape, analysis) + if ai_result: + parting_surface = self._create_parting_surface_from_direction( + shape, + analysis, + ai_result.get("normal", [0, 0, 1]), + ) + parting_line = ai_result.get("parting_line") or self._calculate_parting_line(shape, parting_surface) + return { + "surface": parting_surface, + "line": parting_line, + "direction": ai_result.get("normal", [0, 0, 1]), + "confidence": ai_result.get("confidence", 0.8), + "method": ai_result.get("method", "ai"), + } + except Exception as e: + logger.warning(f"AI 分型面检测失败，回退到几何方法：{e}") + + normal_stats = self._analyze_parting_direction(shape) + parting_surface = self._create_parting_surface_from_direction( + shape, + analysis, + normal_stats.get("primary_direction", [0, 0, 1]), + ) + parting_line = self._calculate_parting_line(shape, parting_surface) + + return { + "surface": parting_surface, + "line": parting_line, + "direction": normal_stats.get("primary_direction", [0, 0, 1]), + "confidence": normal_stats.get("confidence", 0.5), + "method": "geometric", + } + + def _extract_plane_metadata(self, surface: Any, analysis: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: + """提取分型面法向、原点和参数范围。""" + default_origin = [0.0, 0.0, 0.0] + if analysis: + default_origin = analysis.get("bounding_box", {}).get("center", default_origin) + + try: + adaptor = BRepAdaptor_Surface(surface) + plane = adaptor.Plane() + origin = plane.Location() + normal = plane.Axis().Direction() + + return { + "normal": [float(normal.X()), float(normal.Y()), float(normal.Z())], + "origin": [float(origin.X()), float(origin.Y()), float(origin.Z())], + "bounds": { + "u_range": [float(adaptor.FirstUParameter()), float(adaptor.LastUParameter())], + "v_range": [float(adaptor.FirstVParameter()), float(adaptor.LastVParameter())], + }, + } + except Exception as e: + logger.warning(f"分型面几何提取失败，使用默认值: {e}") + return { + "normal": [0.0, 0.0, 1.0], + "origin": [float(default_origin[0]), float(default_origin[1]), float(default_origin[2])], + "bounds": {"u_range": [-200.0, 200.0], "v_range": [-200.0, 200.0]}, + } + def _split_cavity_core(self, shape: Any, parting_surface: Any, margin: int = 20) -> Tuple[Any, Any]: """ 分离型腔和型芯 diff --git a/src/core/mold_generator.py b/src/core/mold_generator.py index 1a8d572..a9c3a57 100644 --- a/src/core/mold_generator.py +++ b/src/core/mold_generator.py @@ -12,6 +12,7 @@ from OCC.Core.BRepBndLib import brepbndlib_Add from models.schemas import create_mold_cavity_data, create_mold_key_info from utils.logger import get_logger from core.base_mold_generator import BaseMoldGenerator +from core.side_action_designer import SideActionDesigner logger = get_logger(__name__) @@ -36,6 +37,7 @@ class MoldCavityGenerator(BaseMoldGenerator): self.parting_line_tolerance = 0.1 self.max_draft_angle = 5.0 + self.side_action_designer = SideActionDesigner() def set_material(self, material: str): """设置产品材料""" @@ -62,8 +64,19 @@ class MoldCavityGenerator(BaseMoldGenerator): try: analysis = self._analyze_product_geometry(product_shape) - parting_surface, parting_line = self._detect_parting_surface( - product_shape, analysis + parting_result = self._detect_primary_parting(product_shape, analysis) + parting_surface = parting_result["surface"] + parting_line = self.optimize_parting_line(parting_result["line"]) + parting_direction = parting_result["direction"] + + side_action_result = self.side_action_designer.analyze_and_design( + shape=product_shape, + parting_direction=parting_direction, + mold_size=self._calculate_mold_size(analysis), + parting_surface=parting_surface, + ) + undercut_regions = self._build_undercut_regions( + side_action_result.get("undercut_analysis", {}) ) scaled_shape = self._apply_shrinkage_compensation(product_shape) @@ -79,7 +92,9 @@ class MoldCavityGenerator(BaseMoldGenerator): "core": core, "parting_surface": parting_surface, "parting_line": parting_line, - "analysis": analysis + "analysis": analysis, + "undercut_regions": undercut_regions, + "side_actions": side_action_result, } except Exception as e: @@ -124,6 +139,10 @@ class MoldCavityGenerator(BaseMoldGenerator): "core": core_geometry }, "parting_surface": parting_geometry, + "quality_checks": { + "undercut_regions": cavity_data.get("undercut_regions", []), + "side_actions": cavity_data.get("side_actions", {}), + }, "manufacturing_info": { "estimated_mold_size": self._calculate_mold_size(analysis), "estimated_clamping_force": self._calculate_clamping_force(analysis), @@ -165,6 +184,8 @@ class MoldCavityGenerator(BaseMoldGenerator): "recommended_injection_pressure": "80-120 MPa" }, "quality_considerations": { + "undercut_count": len(cavity_data.get("undercut_regions", [])), + "side_action_summary": cavity_data.get("side_actions", {}).get("summary", {}), "potential_weld_lines": self._identify_weld_line_risk(analysis), "sink_mark_areas": self._identify_sink_mark_risk(analysis), "warpage_risk": self._assess_warpage_risk(analysis) @@ -184,26 +205,13 @@ class MoldCavityGenerator(BaseMoldGenerator): 2. 基于法向量分析的几何方法 3. 简化方法（基于边界框） """ - if self.ai_parting_detector is not None: - try: - logger.info("使用 AI 模型检测分型面") - ai_result = self.ai_parting_detector.detect(shape, analysis) - if ai_result: - return self._create_parting_surface_from_ai(ai_result, analysis, shape) - except Exception as e: - logger.warning(f"AI 分型面检测失败，回退到几何方法：{e}") - try: - logger.info("使用法向量分析检测分型面") - optimal_direction = self._analyze_face_normals(shape) - parting_plane = self._create_optimal_parting_plane( - shape, analysis, optimal_direction + parting_result = self._detect_primary_parting(shape, analysis) + logger.info( + f"使用 {parting_result['method']} 方法检测分型面，" + f"置信度={parting_result['confidence']:.3f}" ) - parting_surface = BRepBuilderAPI_MakeFace(parting_plane).Face() - - parting_line = self._calculate_parting_line(shape, parting_surface) - - return parting_surface, parting_line + return parting_result["surface"], self.optimize_parting_line(parting_result["line"]) except Exception as e: logger.warning(f"法向量分析失败，使用简化方法：{e}") @@ -211,6 +219,24 @@ class MoldCavityGenerator(BaseMoldGenerator): logger.info("使用简化方法检测分型面") return self._simple_parting_surface(shape, analysis) + def _build_undercut_regions(self, undercut_analysis: Dict[str, Any]) -> List[Dict[str, Any]]: + """将侧向机构分析结果转换为兼容旧结构的倒扣区域列表。""" + undercut_faces = undercut_analysis.get("undercut_faces", []) + regions = [] + + for face in undercut_faces: + regions.append({ + "type": "negative_draft", + "location": face.get("center", [0, 0, 0]), + "severity": face.get("severity", "medium"), + "area": face.get("area", 0), + "is_outer": face.get("is_outer", False), + "face_index": face.get("face_index"), + }) + + logger.info(f"转换得到 {len(regions)} 个兼容倒扣区域") + return regions + def _analyze_face_normals(self, shape: Any) -> gp_Dir: """ 分析产品表面的法向量分布，找出最优分型方向 @@ -335,27 +361,13 @@ class MoldCavityGenerator(BaseMoldGenerator): def _extract_parting_surface_geometry(self, surface: Any) -> Dict[str, Any]: """提取分型面几何数据""" - try: - adaptor = BRepAdaptor_Surface(surface) - u_min, u_max = adaptor.FirstUParameter(), adaptor.LastUParameter() - v_min, v_max = adaptor.FirstVParameter(), adaptor.LastVParameter() - - bounds = { - "u_range": [float(u_min), float(u_max)], - "v_range": [float(v_min), float(v_max)] - } - except Exception as e: - logger.warning(f"分型面边界提取失败，使用默认值: {e}") - bounds = { - "u_range": [-200, 200], - "v_range": [-200, 200] - } + metadata = self._extract_plane_metadata(surface) return { "type": "plane", - "normal": [0, 0, 1], - "origin": [0, 0, 0], - "bounds": bounds + "normal": metadata["normal"], + "origin": metadata["origin"], + "bounds": metadata["bounds"], } def _calculate_mold_size(self, analysis: Dict) -> Dict[str, float]: diff --git a/src/services/calculation_service.py b/src/services/calculation_service.py index d26af0f..cbe0ae4 100644 --- a/src/services/calculation_service.py +++ b/src/services/calculation_service.py @@ -221,6 +221,23 @@ class CalculationService: if key in mold_cavities: detailed_cavity_json["mold_cavities"][key] = mold_cavities[key] + # 合并分模附加信息，保持普通模具与铝泡沫模具输出结构一致 + if cavity_mesh_data: + if cavity_mesh_data.get("parting_surface"): + detailed_cavity_json["parting_surface"] = cavity_mesh_data["parting_surface"] + + quality_checks = cavity_mesh_data.get("quality_checks", {}) + if quality_checks: + detailed_cavity_json["quality_checks"] = quality_checks + + undercut_regions = quality_checks.get("undercut_regions") + if undercut_regions: + detailed_cavity_json["undercut_regions"] = undercut_regions + + side_actions = quality_checks.get("side_actions") + if side_actions: + detailed_cavity_json["side_actions"] = side_actions + # 添加型腔关键信息 detailed_cavity_json["mold_cavities"]["cavity_key_info"] = { "geometric_characteristics": { @@ -231,6 +248,8 @@ class CalculationService: "projected_area": f"{projected_area_cm2:.2f} cm²", }, "quality_considerations": { + "undercut_count": len(detailed_cavity_json.get("undercut_regions", [])), + "side_action_summary": detailed_cavity_json.get("side_actions", {}).get("summary", {}), "potential_weld_lines": "center" if cavity_count > 1 else "minimal", "sink_mark_areas": "thick_sections" if wall["wall_thickness_max"] > 4 else "minimal", "warpage_risk": "medium" if wall["wall_thickness_max"] > 5 else "low",