cjw/geMoldInsight
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

x

Browse Source
This commit is contained in:
SZCJW
2026-04-23 00:49:22 +08:00
parent 882756412e
commit cccb0920ae
5 changed files with 314 additions and 312 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/api/routes.py
+25 -99
View File
@@ -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),
}
src/core/aluminum_foam_mold.py
+45 -174
View File
@@ -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:
src/core/base_mold_generator.py
+174
View File
@@ -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]:
"""
分离型腔和型芯
src/core/mold_generator.py
+51 -39
View File
@@ -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]:
src/services/calculation_service.py
+19
View File
@@ -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",