@@ -603,7 +606,8 @@ const state = reactive({
surface_quality: 'standard',
controller: 'fanuc',
include_gcode: false
- }
+ },
+ creatingProduct: false
})
const camSteelOptions = [
@@ -644,6 +648,19 @@ const loadTask = async () => {
}
}
+const createProductFromAnalysis = async () => {
+ const taskId = route.params.taskId as string
+ try {
+ state.creatingProduct = true
+ const product = await apiRequest
(`/api/products/from-task/${taskId}`, { method: 'POST' })
+ addNotification(`已创建成品：${product.name}（SKU: ${product.sku}）`, 'success')
+ } catch (e) {
+ handleApiError(e, '创建成品')
+ } finally {
+ state.creatingProduct = false
+ }
+}
+
onMounted(() => {
if (!appStore.user) {
router.push('/login')
diff --git a/src/inventory/api/product_routes.py b/src/inventory/api/product_routes.py
index 96032af..550d5cb 100644
--- a/src/inventory/api/product_routes.py
+++ b/src/inventory/api/product_routes.py
@@ -14,10 +14,11 @@ from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, or_, func, delete
from typing import Optional, List, Dict
from decimal import Decimal
+from pathlib import Path
from shared.database.database import get_db_session
from shared.services.auth_service import get_current_active_user, get_current_admin_user
-from shared.models.database import User, Product, ProductMaterial
+from shared.models.database import User, Product, ProductMaterial, STPFile, ProcessingTask
from ..schemas import (
ProductCreate,
ProductResponse,
@@ -118,6 +119,70 @@ async def create_product(
return _build_product_response(product, 0)
+@router.post("/from-task/{task_id}", response_model=ProductResponse, status_code=201)
+async def create_product_from_task(
+ task_id: str,
+ db_session: AsyncSession = Depends(get_db_session),
+ current_user: User = Depends(get_current_active_user),
+):
+ """从模具分析任务创建进销存成品，回写 stp_files.product_id（P2-1）"""
+ task_result = await db_session.execute(select(ProcessingTask).where(ProcessingTask.task_id == task_id))
+ task = task_result.scalar_one_or_none()
+ if not task:
+ raise HTTPException(status_code=404, detail="分析任务不存在")
+ stp_result = await db_session.execute(select(STPFile).where(STPFile.id == task.stp_file_id))
+ stp_file = stp_result.scalar_one_or_none()
+ if not stp_file:
+ raise HTTPException(status_code=404, detail="STP 分析记录不存在")
+
+ # 已关联成品则直接返回（幂等）
+ if stp_file.product_id:
+ existed = await db_session.execute(select(Product).where(Product.id == stp_file.product_id))
+ product = existed.scalar_one_or_none()
+ if product:
+ return _build_product_response(product, 0)
+
+ # 生成唯一 SKU：MI{stp_file_id}，冲突则追加序号
+ base_sku = f"MI{stp_file_id}"
+ sku = base_sku
+ n = 1
+ while True:
+ conflict = await db_session.execute(select(Product).where(Product.sku == sku))
+ if not conflict.scalar_one_or_none():
+ break
+ n += 1
+ sku = f"{base_sku}-{n}"
+
+ name = Path(stp_file.original_filename or f"mold_{stp_file_id}").stem or f"模具分析-{stp_file_id}"
+ desc_parts = []
+ if stp_file.volume:
+ desc_parts.append(f"体积 {stp_file.volume:.1f} mm³")
+ if stp_file.product_weight:
+ desc_parts.append(f"重量 {stp_file.product_weight:.2f} g")
+ if stp_file.surface_area:
+ desc_parts.append(f"表面积 {stp_file.surface_area:.1f} mm²")
+ description = "由模具分析创建" + ("：" + "；".join(desc_parts) if desc_parts else "")
+
+ product = Product(
+ sku=sku,
+ name=name,
+ description=description,
+ category="模具成品",
+ unit="件",
+ item_type="finished",
+ cost_price=0,
+ sale_price=0,
+ min_stock=0,
+ max_stock=0,
+ )
+ db_session.add(product)
+ await db_session.flush()
+ stp_file.product_id = product.id
+ await db_session.commit()
+ await db_session.refresh(product)
+ return _build_product_response(product, 0)
+
+
@router.put("/{product_id}", response_model=ProductResponse)
async def update_product(
product_id: int,
diff --git a/src/moldinsight/api/advanced_router.py b/src/moldinsight/api/advanced_router.py
index 395e852..df4606e 100644
--- a/src/moldinsight/api/advanced_router.py
+++ b/src/moldinsight/api/advanced_router.py
@@ -279,27 +279,6 @@ async def design_complete_mold_system(
return {"status": "success", "data": result}
-@router.post("/ai-parting-detect")
-async def ai_parting_surface_detect(
- request: Request,
- current_user: User = Depends(get_current_active_user),
-):
- body = await request.json()
- task_id = body.get("task_id")
- if not task_id:
- raise HTTPException(404, "缺少 task_id")
- task_data = await _get_task_data(task_id)
- if not task_data:
- raise HTTPException(404, "任务不存在")
- geometry_data = task_data.get("geometry_data")
- if not geometry_data:
- raise HTTPException(400, "该任务尚未完成几何分析")
- from moldinsight.core.ai_parting_detector import AIPartingSurfaceDetectorV2
- detector = AIPartingSurfaceDetectorV2(use_gnn=True)
- result = detector._detect_with_geometry(None, geometry_data)
- return {"status": "success", "data": result}
-
-
@router.post("/detect-undercuts")
async def detect_undercuts(
request: Request,
@@ -323,6 +302,34 @@ async def detect_undercuts(
return {"status": "success", "data": result}
+@router.post("/cost-estimate")
+async def estimate_cost(
+ request: Request,
+ current_user: User = Depends(get_current_active_user),
+):
+ """LLM 模具成本估算（P2-2：真 AI 落地，需启用 LLM）"""
+ body = await request.json()
+ task_id = body.get("task_id")
+ if not task_id:
+ raise HTTPException(404, "缺少 task_id")
+ task_data = await _get_task_data(task_id)
+ if not task_data:
+ raise HTTPException(404, "任务不存在")
+ analysis_result = task_data.get("analysis_result")
+ if not analysis_result:
+ raise HTTPException(400, "该任务尚未完成分析")
+ detailed_context = {
+ "candidate_schemes": task_data.get("candidate_schemes", []),
+ "geometry_data": task_data.get("geometry_data", {}),
+ "metadata": {"selected_material": task_data.get("material")},
+ }
+ from moldinsight.services.llm_service import llm_service
+ result = await llm_service.estimate_cost(analysis_result, detailed_context)
+ if result is None:
+ raise HTTPException(503, "成本估算不可用（LLM 未启用或生成失败）")
+ return {"status": "success", "data": result}
+
+
@router.post("/design-cam")
async def design_mold_cam(
request: Request,
diff --git a/src/moldinsight/core/ai_mold_assistant.py b/src/moldinsight/core/ai_mold_assistant.py
deleted file mode 100644
index 620ff58..0000000
--- a/src/moldinsight/core/ai_mold_assistant.py
+++ /dev/null
@@ -1,209 +0,0 @@
-"""
-AI 分模辅助模型接口示例
-
-此文件展示了如何创建 AI 模型来辅助分模过程。
-实际使用时需要替换为真实的 AI 模型。
-"""
-from typing import Dict, Any, Optional
-import numpy as np
-from OCC.Core.TopoDS import TopoDS_Shape, TopoDS_Face
-
-
-class AIPartingSurfaceDetector:
- """
- AI 分型面检测器（示例接口）
-
- 功能：
- - 分析产品 3D 几何
- - 预测最优分型面位置和方向
- - 识别倒扣区域
- """
-
- def __init__(self, model_path: Optional[str] = None):
- """
- 初始化 AI 分型面检测器
-
- Args:
- model_path: 训练好的模型路径
- """
- self.model_path = model_path
- self.model = None
-
- # 如果提供了模型路径，加载模型
- if model_path:
- self._load_model(model_path)
-
- def _load_model(self, model_path: str):
- """加载训练好的 AI 模型"""
- # TODO: 实现模型加载逻辑
- # 示例：
- # import torch
- # self.model = torch.load(model_path)
- print(f"AI 模型加载：{model_path}")
-
- def detect(self, product_shape: TopoDS_Shape, analysis: Dict) -> Optional[Dict]:
- """
- 检测最优分型面
-
- Args:
- product_shape: OpenCASCADE 形状对象
- analysis: 几何分析结果（包含 bounding_box, volume 等）
-
- Returns:
- {
- "origin": [x, y, z], # 分型面原点
- "normal": [nx, ny, nz], # 分型面法向量
- "confidence": 0.95, # 置信度
- "parting_line": [...] # 可选的分型线
- }
- """
- # TODO: 使用 AI 模型进行预测
- # 这里是示例返回
-
- # 1. 将产品形状转换为 AI 模型输入
- # - 体素化 (voxelization)
- # - 点云 (point cloud)
- # - 多视图 (multi-view images)
- input_data = self._preprocess_shape(product_shape, analysis)
-
- # 2. 使用模型预测
- # prediction = self.model.predict(input_data)
-
- # 3. 返回预测结果
- return {
- "origin": [0, 0, analysis["bounding_box"]["center"][2]],
- "normal": [0, 0, 1], # Z 方向
- "confidence": 0.85,
- "undercut_regions": [] # 倒扣区域
- }
-
- def _preprocess_shape(self, shape: TopoDS_Shape, analysis: Dict) -> TopoDS_Shape:
- """
- 预处理产品形状为 AI 模型输入
-
- 可能的预处理方式：
- 1. 体素化：将 3D 模型转换为 3D 网格
- 2. 点云：采样表面点
- 3. 多视图：渲染多个角度的 2D 图像
- """
- # TODO: 实现预处理逻辑
- return None
-
-
-class AIDraftAnalyzer:
- """
- AI 拔模分析器（示例接口）
-
- 功能：
- - 分析哪些面需要拔模
- - 预测最优拔模角度
- - 检测脱模干涉
- """
-
- def __init__(self, model_path: Optional[str] = None):
- self.model_path = model_path
- self.model = None
-
- if model_path:
- self._load_model(model_path)
-
- def _load_model(self, model_path: str):
- """加载训练好的 AI 模型"""
- print(f"AI 拔模分析模型加载：{model_path}")
-
- def analyze(self, product_shape: TopoDS_Shape, parting_surface: TopoDS_Face,
- base_draft_angle: float) -> Optional[Dict]:
- """
- 分析拔模需求
-
- Args:
- product_shape: 产品形状
- parting_surface: 分型面
- base_draft_angle: 基础拔模角（度）
-
- Returns:
- {
- "drafted_shape": ..., # 应用拔模后的形状
- "draft_angles": {...}, # 各面的拔模角
- "interference_areas": [...], # 干涉区域
- "recommendations": [...] # 优化建议
- }
- """
- # TODO: 使用 AI 模型分析拔模
-
- # 示例返回
- return {
- "drafted_shape": product_shape, # 简化：返回原始形状
- "draft_angles": {"default": base_draft_angle},
- "interference_areas": [],
- "recommendations": ["建议增加圆角", "壁厚均匀化"]
- }
-
-
-class AICavityLayoutOptimizer:
- """
- AI 型腔布局优化器（示例接口）
-
- 功能：
- - 优化多型腔排列
- - 设计流道系统
- - 平衡材料流动
- """
-
- def __init__(self, model_path: Optional[str] = None):
- self.model_path = model_path
- self.model = None
-
- if model_path:
- self._load_model(model_path)
-
- def optimize(self, product_shape: TopoDS_Shape, cavity_count: int,
- mold_base_size: Dict) -> Optional[Dict]:
- """
- 优化型腔布局
-
- Args:
- product_shape: 产品形状
- cavity_count: 型腔数量
- mold_base_size: 模架尺寸
-
- Returns:
- {
- "cavity_positions": [...], # 各型腔位置
- "runner_system": {...}, # 流道系统设计
- "balance_score": 0.92, # 流动平衡评分
- "material_efficiency": 0.85 # 材料利用率
- }
- """
- # TODO: 使用 AI 优化型腔布局
-
- return {
- "cavity_positions": [[0, 0, 0]], # 示例
- "runner_system": {"type": "cold_runner"},
- "balance_score": 0.85,
- "material_efficiency": 0.80
- }
-
-
-# ==================== 使用示例 ====================
-
-if __name__ == "__main__":
- # 示例：如何使用 AI 模型接口
-
- # 1. 创建 AI 模型实例
- parting_detector = AIPartingSurfaceDetector(model_path="models/parting_surface.pth")
- draft_analyzer = AIDraftAnalyzer(model_path="models/draft_analysis.pth")
-
- # 2. 设置到 MoldCavityGenerator
- from moldinsight.core.mold_generator import MoldCavityGenerator
-
- generator = MoldCavityGenerator()
- generator.set_ai_model(
- parting_detector=parting_detector,
- draft_analyzer=draft_analyzer
- )
-
- # 3. 使用（AI 模型会自动介入）
- # result = generator.generate_mold_cavities(product_shape)
-
- print("AI 模型接口已配置，分模时将自动使用 AI 辅助")
diff --git a/src/moldinsight/core/ai_parting_detector.py b/src/moldinsight/core/ai_parting_detector.py
deleted file mode 100644
index 4a5cd91..0000000
--- a/src/moldinsight/core/ai_parting_detector.py
+++ /dev/null
@@ -1,547 +0,0 @@
-"""
-AI 分型面检测模块 - 基于 GNN 的分型面预测框架
-
-架构设计：
-1. ShapeGraphBuilder - 将 OCC 形状转换为图表示（面为节点，共享边为图边）
-2. PartingSurfaceGNN - 图神经网络模型定义
-3. AIPartingSurfaceDetectorV2 - 增强版分型面检测器（集成 GNN）
-
-图构建策略：
-- 节点：每个 TopoDS_Face 作为一个节点
-- 节点特征：法向量(3) + 面积(1) + 曲率(2) + 面类型(1) = 7维
-- 边：共享 TopoDS_Edge 的面之间建立边
-- 边特征：共享边长度(1) + 二面角(1) = 2维
-
-GNN 模型：
-- 3层 GraphConv + 全局池化 + MLP 分类头
-- 输出：每个面的分型面归属概率 + 分型方向
-
-依赖：
-- PyTorch + PyTorch Geometric（可选，缺失时回退到几何方法）
-"""
-
-from typing import Dict, List, Any, Optional, Tuple
-import numpy as np
-from OCC.Core.TopoDS import TopoDS_Shape
-from shared.utils.logger import get_logger
-
-logger = get_logger(__name__)
-
-_TORCH_AVAILABLE = False
-_TORCH_GEOMETRIC_AVAILABLE = False
-
-try:
- import torch
- import torch.nn as nn
- import torch.nn.functional as F
- _TORCH_AVAILABLE = True
- try:
- from torch_geometric.nn import GCNConv, global_mean_pool
- from torch_geometric.data import Data
- _TORCH_GEOMETRIC_AVAILABLE = True
- except ImportError:
- logger.info("PyTorch Geometric 未安装，GNN 模型不可用")
-except ImportError:
- logger.info("PyTorch 未安装，AI 分型面检测将使用几何回退方法")
-
-
-class ShapeGraphBuilder:
- """将 OCC 形状转换为图表示"""
-
- def build_graph(self, shape: TopoDS_Shape) -> Optional[Dict]:
- """
- 从 OCC 形状构建图数据
-
- Returns:
- {
- "node_features": np.ndarray (N, 7),
- "edge_index": np.ndarray (2, E),
- "edge_features": np.ndarray (E, 2),
- "face_map": List[TopoDS_Face],
- "num_nodes": int,
- "num_edges": int
- }
- """
- try:
- from OCC.Core.TopExp import TopExp_Explorer
- from OCC.Core.TopAbs import TopAbs_FACE, TopAbs_EDGE
- from OCC.Core.BRepAdaptor import BRepAdaptor_Surface
- 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.TopTools import TopTools_IndexedDataMapOfShapeListOfShape
- from OCC.Core.TopExp import topexp_MapShapesAndAncestors
- from OCC.Core.TopoDS import TopoDS_Face, TopoDS_Edge, topods
-
- faces = []
- face_features = []
-
- explorer = TopExp_Explorer(shape, TopAbs_FACE)
- while explorer.More():
- face = topods.Face(explorer.Current())
- features = self._extract_face_features(face)
- if features is not None:
- faces.append(face)
- face_features.append(features)
- explorer.Next()
-
- if not faces:
- logger.warning("未找到面，无法构建图")
- return None
-
- node_features = np.array(face_features, dtype=np.float32)
-
- edge_map = TopTools_IndexedDataMapOfShapeListOfShape()
- topexp_MapShapesAndAncestors(shape, TopAbs_EDGE, TopAbs_FACE, edge_map)
-
- edge_list = []
- edge_features_list = []
-
- for i in range(1, edge_map.Extent() + 1):
- edge = topods.Edge(edge_map.FindKey(i))
- face_list = edge_map.FindFromIndex(i)
-
- connected_faces = []
- it = face_list.begin()
- while it != face_list.end():
- f = topods.Face(it.Value())
- try:
- idx = faces.index(f)
- connected_faces.append(idx)
- except ValueError:
- pass
- it.next_ptr()
-
- if len(connected_faces) >= 2:
- edge_feat = self._extract_edge_features(edge, connected_faces, faces)
- for j in range(len(connected_faces)):
- for k in range(j + 1, len(connected_faces)):
- edge_list.append([connected_faces[j], connected_faces[k]])
- edge_features_list.append(edge_feat)
-
- if not edge_list:
- logger.warning("未找到边连接，返回无图边的图")
- edge_index = np.zeros((2, 0), dtype=np.int64)
- edge_features_arr = np.zeros((0, 2), dtype=np.float32)
- else:
- edge_index = np.array(edge_list, dtype=np.int64).T
- rev_edges = np.array([[e[1], e[0]] for e in edge_list], dtype=np.int64).T
- edge_index = np.concatenate([edge_index, rev_edges], axis=1)
- edge_features_arr = np.array(edge_features_list, dtype=np.float32)
- edge_features_arr = np.concatenate([edge_features_arr, edge_features_arr], axis=0)
-
- return {
- "node_features": node_features,
- "edge_index": edge_index,
- "edge_features": edge_features_arr,
- "face_map": faces,
- "num_nodes": len(faces),
- "num_edges": edge_index.shape[1]
- }
-
- except Exception as e:
- logger.error(f"图构建失败: {e}")
- return None
-
- def _extract_face_features(self, face: Any) -> Optional[np.ndarray]:
- """
- 提取面特征：[nx, ny, nz, area, u_curvature, v_curvature, face_type]
- """
- try:
- from OCC.Core.BRepAdaptor import BRepAdaptor_Surface
- from OCC.Core.GProp import GProp_GProps
- from OCC.Core.BRepGProp import brepgprop
-
- surface = BRepAdaptor_Surface(face)
-
- u = (surface.FirstUParameter() + surface.LastUParameter()) / 2
- v = (surface.FirstVParameter() + surface.LastVParameter()) / 2
-
- if surface.GetType() == 0:
- normal = surface.Plane().Position().Direction()
- face_type = 0.0
- u_curv = 0.0
- v_curv = 0.0
- elif surface.GetType() == 1:
- normal = surface.Cylinder().Position().Direction()
- face_type = 1.0
- radius = surface.Cylinder().Radius()
- u_curv = 1.0 / radius if radius > 0.001 else 0.0
- v_curv = 0.0
- elif surface.GetType() == 2:
- normal = surface.Cone().Position().Direction()
- face_type = 2.0
- u_curv = 0.0
- v_curv = 0.0
- elif surface.GetType() == 3:
- normal = surface.Sphere().Position().Direction()
- face_type = 3.0
- radius = surface.Sphere().Radius()
- u_curv = 1.0 / radius if radius > 0.001 else 0.0
- v_curv = 1.0 / radius if radius > 0.001 else 0.0
- elif surface.GetType() == 4:
- normal = surface.Torus().Position().Direction()
- face_type = 4.0
- u_curv = 0.0
- v_curv = 0.0
- else:
- from OCC.Core.BRepLProp import BRepLProp_SLProps
- props = BRepLProp_SLProps(surface, 2, 0.001)
- props.SetParameters(u, v)
- if props.IsNormalDefined():
- normal = props.Normal()
- else:
- normal = gp_Dir(0, 0, 1)
- face_type = 5.0
- u_curv = 0.0
- v_curv = 0.0
-
- face_props = GProp_GProps()
- brepgprop.SurfaceProperties(face, face_props)
- area = face_props.Mass()
-
- return np.array([
- normal.X(), normal.Y(), normal.Z(),
- area,
- u_curv, v_curv,
- face_type
- ], dtype=np.float32)
-
- except Exception as e:
- logger.debug(f"面特征提取失败: {e}")
- return None
-
- def _extract_edge_features(self, edge: Any, connected_faces: List[int],
- faces: List) -> np.ndarray:
- """
- 提取边特征：[edge_length, dihedral_angle]
- """
- try:
- from OCC.Core.BRepAdaptor import BRepAdaptor_Curve
- from OCC.Core.GProp import GProp_GProps
- from OCC.Core.BRepGProp import brepgprop
-
- curve = BRepAdaptor_Curve(edge)
- first = curve.FirstParameter()
- last = curve.LastParameter()
-
- edge_len = abs(last - first)
-
- dihedral = 0.0
- if len(connected_faces) >= 2:
- n1 = self._get_face_normal_fast(faces[connected_faces[0]])
- n2 = self._get_face_normal_fast(faces[connected_faces[1]])
- if n1 is not None and n2 is not None:
- dot = np.clip(np.dot(n1, n2), -1.0, 1.0)
- dihedral = np.arccos(dot)
-
- return np.array([edge_len, dihedral], dtype=np.float32)
-
- except Exception:
- return np.array([0.0, 0.0], dtype=np.float32)
-
- def _get_face_normal_fast(self, face: Any) -> Optional[np.ndarray]:
- """快速获取面法向量（numpy数组）"""
- try:
- from OCC.Core.BRepAdaptor import BRepAdaptor_Surface
- surface = BRepAdaptor_Surface(face)
- if surface.GetType() == 0:
- n = surface.Plane().Position().Direction()
- return np.array([n.X(), n.Y(), n.Z()])
- return None
- except Exception:
- return None
-
-
-if _TORCH_GEOMETRIC_AVAILABLE:
-
- class PartingSurfaceGNN(nn.Module):
- """
- 分型面检测 GNN 模型
-
- 架构：
- - 3层 GCNConv (hidden_dim=64)
- - 全局平均池化
- - 3层 MLP 分类头
- - 输出：每个面的分型面归属概率 (0-1)
- """
-
- def __init__(self, input_dim: int = 7, hidden_dim: int = 64,
- num_layers: int = 3, dropout: float = 0.3):
- super().__init__()
-
- self.input_dim = input_dim
- self.hidden_dim = hidden_dim
- self.num_layers = num_layers
-
- self.input_proj = nn.Linear(input_dim, hidden_dim)
-
- self.convs = nn.ModuleList()
- self.bns = nn.ModuleList()
- for _ in range(num_layers):
- self.convs.append(GCNConv(hidden_dim, hidden_dim))
- self.bns.append(nn.BatchNorm1d(hidden_dim))
-
- self.dropout = dropout
-
- self.mlp = nn.Sequential(
- nn.Linear(hidden_dim, hidden_dim),
- nn.ReLU(),
- nn.Dropout(dropout),
- nn.Linear(hidden_dim, hidden_dim // 2),
- nn.ReLU(),
- nn.Dropout(dropout),
- nn.Linear(hidden_dim // 2, 1),
- )
-
- def forward(self, data: Data) -> torch.Tensor:
- x, edge_index = data.x, data.edge_index
-
- x = self.input_proj(x)
- x = F.relu(x)
-
- for conv, bn in zip(self.convs, self.bns):
- x = conv(x, edge_index)
- x = bn(x)
- x = F.relu(x)
- x = F.dropout(x, p=self.dropout, training=self.training)
-
- out = self.mlp(x)
- return torch.sigmoid(out).squeeze(-1)
-
- class PartingDirectionHead(nn.Module):
- """
- 分型方向预测头
-
- 基于全局池化的面特征，预测分型方向向量
- """
-
- def __init__(self, hidden_dim: int = 64):
- super().__init__()
- self.direction_mlp = nn.Sequential(
- nn.Linear(hidden_dim, hidden_dim),
- nn.ReLU(),
- nn.Linear(hidden_dim, 3),
- )
-
- def forward(self, node_embeddings: torch.Tensor,
- batch: torch.Tensor) -> torch.Tensor:
- pooled = global_mean_pool(node_embeddings, batch)
- direction = self.direction_mlp(pooled)
- direction = F.normalize(direction, p=2, dim=-1)
- return direction
-
-
-class AIPartingSurfaceDetectorV2:
- """
- 增强版 AI 分型面检测器
-
- 支持：
- 1. GNN 模型推理（需要 PyTorch + PyG）
- 2. 几何方法回退（无需任何 AI 依赖）
- 3. 模型训练数据收集
- """
-
- def __init__(self, model_path: Optional[str] = None,
- use_gnn: bool = True,
- device: str = "cpu"):
- self.model = None
- self.direction_head = None
- self.graph_builder = ShapeGraphBuilder()
- self.device = device
- self.use_gnn = use_gnn and _TORCH_GEOMETRIC_AVAILABLE
-
- if model_path and self.use_gnn:
- self._load_model(model_path)
-
- def _load_model(self, model_path: str):
- """加载训练好的 GNN 模型"""
- if not _TORCH_GEOMETRIC_AVAILABLE:
- logger.warning("PyTorch Geometric 不可用，无法加载 GNN 模型")
- return
-
- try:
- checkpoint = torch.load(model_path, map_location=self.device)
- self.model = PartingSurfaceGNN(
- input_dim=checkpoint.get("input_dim", 7),
- hidden_dim=checkpoint.get("hidden_dim", 64),
- )
- self.model.load_state_dict(checkpoint["model_state_dict"])
- self.model.to(self.device)
- self.model.eval()
-
- if "direction_head_state_dict" in checkpoint:
- self.direction_head = PartingDirectionHead(
- hidden_dim=checkpoint.get("hidden_dim", 64)
- )
- self.direction_head.load_state_dict(checkpoint["direction_head_state_dict"])
- self.direction_head.to(self.device)
- self.direction_head.eval()
-
- logger.info(f"GNN 模型加载成功: {model_path}")
- except Exception as e:
- logger.error(f"GNN 模型加载失败: {e}")
- self.model = None
-
- def detect(self, product_shape: TopoDS_Shape, analysis: Dict) -> Optional[Dict]:
- """
- 检测最优分型面
-
- Args:
- product_shape: OpenCASCADE 形状对象
- analysis: 几何分析结果
-
- Returns:
- {
- "origin": [x, y, z],
- "normal": [nx, ny, nz],
- "confidence": float,
- "parting_line": [...],
- "method": "gnn" | "geometric"
- }
- """
- if self.use_gnn and self.model is not None:
- result = self._detect_with_gnn(product_shape, analysis)
- if result is not None:
- return result
-
- return self._detect_with_geometry(product_shape, analysis)
-
- def _detect_with_gnn(self, shape: TopoDS_Shape, analysis: Dict) -> Optional[Dict]:
- """使用 GNN 模型检测分型面"""
- if not _TORCH_GEOMETRIC_AVAILABLE:
- return None
-
- try:
- graph_data = self.graph_builder.build_graph(shape)
- if graph_data is None:
- return None
-
- node_features = torch.tensor(
- graph_data["node_features"], dtype=torch.float32
- ).to(self.device)
- edge_index = torch.tensor(
- graph_data["edge_index"], dtype=torch.long
- ).to(self.device)
-
- data = Data(x=node_features, edge_index=edge_index)
-
- with torch.no_grad():
- face_probs = self.model(data)
-
- if self.direction_head is not None:
- batch = torch.zeros(
- data.num_nodes, dtype=torch.long, device=self.device
- )
- direction = self.direction_head(data.x, batch)
- normal = direction.cpu().numpy().tolist()
- else:
- normal = [0, 0, 1]
-
- parting_face_mask = face_probs.cpu().numpy() > 0.5
- confidence = float(face_probs.mean().cpu().numpy())
-
- bbox = analysis.get("bounding_box", {})
- center = bbox.get("center", [0, 0, 0])
-
- return {
- "origin": center,
- "normal": normal,
- "confidence": confidence,
- "method": "gnn",
- "face_probabilities": face_probs.cpu().numpy().tolist(),
- "parting_face_count": int(parting_face_mask.sum()),
- }
-
- except Exception as e:
- logger.warning(f"GNN 检测失败，回退到几何方法: {e}")
- return None
-
- def _detect_with_geometry(self, shape: TopoDS_Shape, analysis: Dict) -> Dict:
- """几何方法回退：基于法向量统计的分型面检测"""
- try:
- graph_data = self.graph_builder.build_graph(shape)
- if graph_data is not None:
- node_features = graph_data["node_features"]
- normals = node_features[:, :3]
- areas = node_features[:, 3]
-
- total_area = areas.sum()
- if total_area > 0:
- weights = areas / total_area
- weighted_normal = np.sum(normals * weights[:, np.newaxis], axis=0)
- else:
- weighted_normal = np.mean(normals, axis=0)
-
- length = np.linalg.norm(weighted_normal)
- if length > 0.001:
- weighted_normal /= length
- else:
- weighted_normal = np.array([0, 0, 1])
-
- dot_products = np.abs(np.dot(normals, weighted_normal))
- confidence = float(np.mean(dot_products))
-
- bbox = analysis.get("bounding_box", {})
- center = bbox.get("center", [0, 0, 0])
-
- return {
- "origin": center,
- "normal": weighted_normal.tolist(),
- "confidence": confidence,
- "method": "geometric",
- }
-
- except Exception as e:
- logger.warning(f"几何方法检测失败: {e}")
-
- bbox = analysis.get("bounding_box", {})
- center = bbox.get("center", [0, 0, 0])
- return {
- "origin": center,
- "normal": [0, 0, 1],
- "confidence": 0.5,
- "method": "fallback",
- }
-
- def collect_training_sample(self, shape: TopoDS_Shape, analysis: Dict,
- ground_truth_normal: List[float],
- ground_truth_origin: List[float]) -> Optional[Dict]:
- """
- 收集训练样本
-
- Args:
- shape: OCC 形状
- analysis: 几何分析
- ground_truth_normal: 人工标注的分型方向
- ground_truth_origin: 人工标注的分型面原点
-
- Returns:
- 可序列化的训练样本
- """
- graph_data = self.graph_builder.build_graph(shape)
- if graph_data is None:
- return None
-
- return {
- "node_features": graph_data["node_features"].tolist(),
- "edge_index": graph_data["edge_index"].tolist(),
- "edge_features": graph_data["edge_features"].tolist(),
- "label_normal": ground_truth_normal,
- "label_origin": ground_truth_origin,
- "bounding_box": analysis.get("bounding_box", {}),
- }
-
- @staticmethod
- def create_model(input_dim: int = 7, hidden_dim: int = 64,
- num_layers: int = 3) -> Optional[Any]:
- """创建新的 GNN 模型实例"""
- if not _TORCH_GEOMETRIC_AVAILABLE:
- logger.warning("PyTorch Geometric 不可用，无法创建模型")
- return None
- return PartingSurfaceGNN(
- input_dim=input_dim,
- hidden_dim=hidden_dim,
- num_layers=num_layers,
- )
diff --git a/src/moldinsight/core/aluminum_foam_mold.py b/src/moldinsight/core/aluminum_foam_mold.py
index 63a8595..b49f883 100644
--- a/src/moldinsight/core/aluminum_foam_mold.py
+++ b/src/moldinsight/core/aluminum_foam_mold.py
@@ -462,37 +462,6 @@ class AluminumFoamMoldGenerator(BaseMoldGenerator):
"bounds": metadata["bounds"],
}
- def _create_parting_surface_from_ai(self, ai_result: Dict, analysis: Dict,
- shape: Optional[TopoDS_Shape] = None) -> Dict:
- """从 AI 结果创建分型面"""
- origin = ai_result.get("origin", [0, 0, 0])
- normal = ai_result.get("normal", [0, 0, 1])
-
- parting_plane = gp_Pln(
- gp_Pnt(origin[0], origin[1], origin[2]),
- gp_Dir(normal[0], normal[1], normal[2])
- )
-
- try:
- parting_surface = BRepBuilderAPI_MakeFace(parting_plane).Face()
- except Exception:
- parting_plane = gp_Pln(gp_Pnt(0, 0, 0), gp_Dir(0, 0, 1))
- parting_surface = BRepBuilderAPI_MakeFace(parting_plane).Face()
-
- if shape is not None:
- parting_line = self._calculate_parting_line(shape, parting_surface)
- else:
- parting_line = []
-
- return {
- "primary_surface": parting_surface,
- "primary_line": parting_line,
- "primary_direction": normal,
- "confidence": ai_result.get("confidence", 0.8),
- "additional_surfaces": [],
- "surface_count": 1
- }
-
# ==================== 辅助方法 ====================
def _calculate_mold_size(self, analysis: Dict) -> Dict[str, float]:
diff --git a/src/moldinsight/core/base_mold_generator.py b/src/moldinsight/core/base_mold_generator.py
index 437f7a4..7cf53bd 100644
--- a/src/moldinsight/core/base_mold_generator.py
+++ b/src/moldinsight/core/base_mold_generator.py
@@ -33,14 +33,6 @@ class BaseMoldGenerator:
self.draft_angle = draft_angle
self.material_density = material_density
- self.ai_parting_detector: Optional[Any] = None
- self.ai_draft_analyzer: Optional[Any] = None
-
- def set_ai_model(self, parting_detector: Any = None, draft_analyzer: Any = None):
- self.ai_parting_detector = parting_detector
- self.ai_draft_analyzer = draft_analyzer
- logger.info("AI 模型接口已设置")
-
def _apply_shrinkage_compensation(self, shape: TopoDS_Shape) -> TopoDS_Shape:
scale_factor = 1.0 + self.shrinkage_rate
trsf = gp_Trsf()
diff --git a/src/moldinsight/core/mold_generator.py b/src/moldinsight/core/mold_generator.py
index a884280..6397bec 100644
--- a/src/moldinsight/core/mold_generator.py
+++ b/src/moldinsight/core/mold_generator.py
@@ -214,21 +214,6 @@ class MoldCavityGenerator(BaseMoldGenerator):
def _detect_primary_parting(self, shape: TopoDS_Shape, analysis: Dict) -> Dict[str, Any]:
"""检测主分型面（AI优先 → 几何法向量 → 简化回退）"""
- if self.ai_parting_detector is not None:
- try:
- ai_result = self.ai_parting_detector.detect(shape, analysis)
- if ai_result is not None:
- surface, line = self._create_parting_surface_from_ai(ai_result, analysis, shape)
- return {
- "surface": surface,
- "line": line,
- "direction": ai_result.get("normal", [0, 0, 1]),
- "method": ai_result.get("method", "ai"),
- "confidence": ai_result.get("confidence", 0.8),
- }
- except Exception as e:
- logger.warning(f"AI 分型面检测失败: {e}")
-
try:
normal_dir = self._analyze_face_normals(shape)
parting_plane = self._create_optimal_parting_plane(shape, analysis, normal_dir)
@@ -364,40 +349,6 @@ class MoldCavityGenerator(BaseMoldGenerator):
return parting_surface, parting_line
- def _create_parting_surface_from_ai(self, ai_result: Dict,
- analysis: Dict, shape: Optional[TopoDS_Shape] = None) -> Tuple[TopoDS_Face, List]:
- """
- 从 AI 模型结果创建分型面（预留接口）
-
- Args:
- ai_result: AI 模型输出，应包含：
- - origin: [x, y, z] 平面原点
- - normal: [nx, ny, nz] 法向量
- analysis: 几何分析结果
- shape: 产品形状（用于计算分型线）
-
- Returns:
- (parting_surface, parting_line)
- """
- origin = ai_result.get("origin", [0, 0, 0])
- normal = ai_result.get("normal", [0, 0, 1])
-
- parting_plane = gp_Pln(
- gp_Pnt(origin[0], origin[1], origin[2]),
- gp_Dir(normal[0], normal[1], normal[2])
- )
- parting_surface = BRepBuilderAPI_MakeFace(parting_plane).Face()
-
- if "parting_line" in ai_result:
- parting_line = ai_result["parting_line"]
- elif shape is not None:
- parting_line = self._calculate_parting_line(shape, parting_surface)
- else:
- parting_line = []
-
- logger.info(f"从 AI 结果创建分型面：原点={origin}, 法向量={normal}")
- return parting_surface, parting_line
-
def _extract_parting_surface_geometry(self, surface: TopoDS_Face) -> Dict[str, Any]:
"""提取分型面几何数据"""
metadata = self._extract_plane_metadata(surface)
diff --git a/src/moldinsight/services/llm_service.py b/src/moldinsight/services/llm_service.py
index 159c2df..2632758 100644
--- a/src/moldinsight/services/llm_service.py
+++ b/src/moldinsight/services/llm_service.py
@@ -179,6 +179,33 @@ _PARTING_USER = """请评估以下候选分模方向并推荐最优方案：
请综合评估制造可行性、成本和风险，给出推荐。"""
+_COST_ESTIMATE_SYSTEM = """你是一位资深模具报价工程师，擅长根据产品几何与模具设计方案估算模具造价与单件成本。
+
+要求：
+1. 使用中文，金额用人民币（¥）
+2. 基于给定数据合理估算，数据不足时给出区间并标注假设
+3. 综合考虑：模具材料、加工复杂度（滑块/斜顶/镶件）、型腔数、产品材料用量、成型周期
+
+严格输出 JSON，不要输出其他内容。JSON 格式：
+{
+ "mold_cost": {
+ "material": "¥XX（P20 钢，约 XX kg）",
+ "machining": "¥XX（含 CNC/EDM/线切割）",
+ "complexity_factor": "1.2（含 X 个滑块/斜顶）",
+ "subtotal": "¥XX"
+ },
+ "part_cost": {
+ "material": "¥XX（ABS，约 XX g）",
+ "cycle_time": "30 s",
+ "cost_per_part": "¥XX"
+ },
+ "total_mold_cost": "¥XX",
+ "cost_per_part": "¥XX",
+ "confidence": 0.7,
+ "assumptions": ["假设模具寿命 50 万模次", "假设..."]
+}"""
+
+
class LLMService:
"""LLM 增强分析服务（单例）"""
@@ -308,6 +335,61 @@ class LLMService:
logger.warning("LLM 分型推荐失败（不影响主流程）: %s", e)
return None
+ async def estimate_cost(
+ self,
+ analysis_result: Dict[str, Any],
+ detailed_cavity_json: Optional[Dict[str, Any]] = None,
+ ) -> Optional[Dict[str, Any]]:
+ """估算模具造价与单件成本 (结构化 JSON)"""
+ if not self._enabled:
+ return None
+ try:
+ prompt = self._build_cost_estimate_prompt(analysis_result, detailed_cavity_json)
+ response = await self._chat(
+ _COST_ESTIMATE_SYSTEM,
+ prompt,
+ min(self._max_tokens, 1200),
+ expect_json=True,
+ )
+ if not response:
+ return None
+ result = self._parse_json_response(response)
+ if result:
+ logger.info("LLM 成本估算生成成功: total=%s", result.get("total_mold_cost"))
+ return result
+ except Exception as e:
+ logger.warning("LLM 成本估算失败（不影响主流程）: %s", e)
+ return None
+
+ def _build_cost_estimate_prompt(self, analysis_result: Dict[str, Any], detailed_cavity_json: Optional[Dict[str, Any]]) -> str:
+ geometry_data = analysis_result.get("geometry_data", {}) or (detailed_cavity_json or {}).get("geometry_data", {})
+ volume = geometry_data.get("volume", 0) or 0
+ bbox = geometry_data.get("bounding_box", {}) or {}
+ dims = bbox.get("dimensions", [0, 0, 0])
+ meta = (detailed_cavity_json or {}).get("metadata", {}) or {}
+ material = meta.get("selected_material") or analysis_result.get("material") or "ABS"
+ schemes = (detailed_cavity_json or {}).get("candidate_schemes", []) or []
+ best = schemes[0] if schemes else {}
+ cd = best.get("cavity_data", {}) if isinstance(best, dict) else {}
+ mfg = cd.get("manufacturing_info", {}) if isinstance(cd, dict) else {}
+ features = analysis_result.get("detected_features", []) or []
+ complexity_hints = [
+ f.get("description", f.get("feature_type", ""))
+ for f in features
+ if f.get("feature_type") in ("undercut", "side_action", "insert")
+ ]
+ return (
+ f"产品材料：{material}\n"
+ f"体积：{float(volume):.1f} mm³\n"
+ f"边界框尺寸（长×宽×高）：{float(dims[0]):.1f} × {float(dims[1]):.1f} × {float(dims[2]):.1f} mm\n"
+ f"预估锁模力：{mfg.get('estimated_clamping_force', '未知')}\n"
+ f"预估模具尺寸：{json.dumps(mfg.get('estimated_mold_size', {}), ensure_ascii=False)}\n"
+ f"预估成型周期：{mfg.get('estimated_cycle_time', '未知')}\n"
+ f"型腔数：{best.get('cavity_count', 1) if isinstance(best, dict) else 1}\n"
+ f"模具结构：{best.get('mold_structure_type', '未知') if isinstance(best, dict) else '未知'}\n"
+ f"复杂度线索：{', '.join(complexity_hints) if complexity_hints else '无明显倒扣/滑块'}\n"
+ )
+
def _build_design_report_prompt(self, analysis_result, detailed_cavity_json) -> str:
features = json.dumps(analysis_result.get("detected_features", []), ensure_ascii=False, indent=2)
if len(features) > 4000:
diff --git a/src/shared/models/database.py b/src/shared/models/database.py
index c7120d5..12b6d69 100644
--- a/src/shared/models/database.py
+++ b/src/shared/models/database.py
@@ -125,6 +125,8 @@ class STPFile(Base):
id = Column(Integer, primary_key=True, index=True)
user_id = Column(Integer, ForeignKey("users.id"), nullable=True, index=True)
+ # 关联进销存成品（P2-1：分析结果可一键创建为成品并回写）
+ product_id = Column(Integer, ForeignKey("products.id"), nullable=True, index=True)
# 对象存储信息
object_key = Column(String(500), nullable=False, index=True) # MinIO对象键
@@ -160,6 +162,7 @@ class STPFile(Base):
# 关联关系
user = relationship("User", back_populates="stp_files")
+ product = relationship("Product") # P2-1: 关联的进销存成品
geometry_data = relationship("GeometryData", back_populates="stp_file", uselist=False)
mesh_data = relationship("MeshData", back_populates="stp_file", uselist=False)
mold_cavity_data = relationship("MoldCavityData", back_populates="stp_file", uselist=False)