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

Merge branch 'main' of http://szcjw:10001/cjw/geMoldInsight

Browse Source 
# Conflicts:
#	src/utils/html_generator.py
This commit is contained in:
SZCJW
2026-05-02 00:41:05 +08:00
parent d1a3f22785 0f4c17e67f
commit 58d045354a
4 changed files with 239 additions and 119 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/api/v1/advanced_router.py
+135 -119
View File
@@ -4,48 +4,73 @@ import os
from fastapi import APIRouter, Depends, HTTPException, Request
from services.auth_service import get_current_active_user
from services.redis_task_manager import redis_task_manager
from models.database import User
from utils.logger import get_logger
from api.routes import (
tasks,
cavity_layout_optimizer,
mold_system_designer,
side_action_designer,
mold_cam_designer,
collision_detector,
toolpath_optimizer,
edm_designer,
machining_simulator,
cad_exporter,
)
logger = get_logger(__name__)
router = APIRouter()
_api_routes_cache = {}
def _get_cached(key):
global _api_routes_cache
if key not in _api_routes_cache:
try:
from api import routes
except Exception as e:
logger.warning(f"api.routes 模块加载失败: {e}")
_api_routes_cache["__error__"] = str(e)
return None
_api_routes_cache.clear()
_api_routes_cache.update({
"tasks": routes.tasks,
"cavity_layout_optimizer": routes.cavity_layout_optimizer,
"mold_system_designer": routes.mold_system_designer,
"side_action_designer": routes.side_action_designer,
"mold_cam_designer": routes.mold_cam_designer,
"collision_detector": routes.collision_detector,
"toolpath_optimizer": routes.toolpath_optimizer,
"edm_designer": routes.edm_designer,
"machining_simulator": routes.machining_simulator,
"cad_exporter": routes.cad_exporter,
})
return _api_routes_cache.get(key)
async def _get_task_data(task_id: str) -> dict:
task = await redis_task_manager.get_task(task_id)
if task:
return task
tasks = _get_cached("tasks")
if tasks and task_id in tasks:
return tasks[task_id]
return None
@router.post("/optimize-layout")
async def optimize_cavity_layout(
request: Request,
current_user: User = Depends(get_current_active_user),
):
"""多型腔布局优化"""
body = await request.json()
product_bbox = body.get("product_bbox", {"dimensions": [100, 100, 50]})
cavity_count = body.get("cavity_count", 1)
mold_base_size = body.get("mold_base_size")
layout_type = body.get("layout_type", "auto")
if cavity_count < 1 or cavity_count > 64:
raise HTTPException(400, "型腔数量必须在 1-64 之间")
result = cavity_layout_optimizer.optimize_layout(
optimizer = _get_cached("cavity_layout_optimizer")
if not optimizer:
raise HTTPException(503, "服务不可用：核心模块未加载")
result = optimizer.optimize_layout(
product_bbox=product_bbox,
cavity_count=cavity_count,
mold_base_size=mold_base_size,
layout_type=layout_type,
)
return {"status": "success", "data": result}
@@ -54,25 +79,19 @@ async def design_cooling_system(
request: Request,
current_user: User = Depends(get_current_active_user),
):
"""冷却系统设计"""
body = await request.json()
mold_size = body.get("mold_size", {"length": 300, "width": 300, "height": 200})
product_bbox = body.get("product_bbox", {"dimensions": [100, 100, 50]})
material = body.get("material", "ABS")
cavity_count = body.get("cavity_count", 1)
cycle_time_target = body.get("cycle_time_target")
from core.mold_system_designer import CoolingSystemDesigner
designer = CoolingSystemDesigner()
result = designer.design_cooling_system(
mold_size=mold_size,
product_bbox=product_bbox,
material=material,
cavity_count=cavity_count,
mold_size=mold_size, product_bbox=product_bbox,
material=material, cavity_count=cavity_count,
cycle_time_target=cycle_time_target,
)
return {"status": "success", "data": result}
@@ -81,25 +100,19 @@ async def design_gating_system(
request: Request,
current_user: User = Depends(get_current_active_user),
):
"""浇注系统设计"""
body = await request.json()
product_bbox = body.get("product_bbox", {"dimensions": [100, 100, 50]})
material = body.get("material", "ABS")
cavity_count = body.get("cavity_count", 1)
gate_type = body.get("gate_type", "auto")
layout_positions = body.get("layout_positions")
from core.mold_system_designer import GatingSystemDesigner
designer = GatingSystemDesigner()
result = designer.design_gating_system(
product_bbox=product_bbox,
material=material,
cavity_count=cavity_count,
gate_type=gate_type,
product_bbox=product_bbox, material=material,
cavity_count=cavity_count, gate_type=gate_type,
layout_positions=layout_positions,
)
return {"status": "success", "data": result}
@@ -108,7 +121,6 @@ async def design_complete_mold_system(
request: Request,
current_user: User = Depends(get_current_active_user),
):
"""综合模具系统设计（冷却+浇注）"""
body = await request.json()
mold_size = body.get("mold_size", {"length": 300, "width": 300, "height": 200})
product_bbox = body.get("product_bbox", {"dimensions": [100, 100, 50]})
@@ -117,17 +129,15 @@ async def design_complete_mold_system(
gate_type = body.get("gate_type", "auto")
cycle_time_target = body.get("cycle_time_target")
layout_positions = body.get("layout_positions")
result = mold_system_designer.design_complete_system(
mold_size=mold_size,
product_bbox=product_bbox,
material=material,
cavity_count=cavity_count,
gate_type=gate_type,
cycle_time_target=cycle_time_target,
ds = _get_cached("mold_system_designer")
if not ds:
raise HTTPException(503, "服务不可用：核心模块未加载")
result = ds.design_complete_system(
mold_size=mold_size, product_bbox=product_bbox,
material=material, cavity_count=cavity_count,
gate_type=gate_type, cycle_time_target=cycle_time_target,
layout_positions=layout_positions,
)
return {"status": "success", "data": result}
@@ -136,20 +146,17 @@ async def ai_parting_surface_detect(
request: Request,
current_user: User = Depends(get_current_active_user),
):
"""AI 分型面检测"""
body = await request.json()
task_id = body.get("task_id")
if not task_id or task_id not in tasks:
if not task_id:
raise HTTPException(404, "缺少 task_id")
task_data = await _get_task_data(task_id)
if not task_data:
raise HTTPException(404, "任务不存在")
task_data = tasks[task_id]
geometry_data = task_data.get("geometry_data")
if not geometry_data:
raise HTTPException(400, "该任务尚未完成几何分析")
from core.ai_parting_detector import AIPartingSurfaceDetectorV2
detector = AIPartingSurfaceDetectorV2(use_gnn=True)
result = detector._detect_with_geometry(None, geometry_data)
return {"status": "success", "data": result}
@@ -160,24 +167,20 @@ async def detect_undercuts(
request: Request,
current_user: User = Depends(get_current_active_user),
):
"""倒扣区域检测与滑块/斜顶机构设计"""
body = await request.json()
task_id = body.get("task_id")
parting_direction = body.get("parting_direction", [0, 0, 1])
mold_size = body.get("mold_size", {"length": 300, "width": 300, "height": 200})
if not task_id or task_id not in tasks:
if not task_id:
raise HTTPException(404, "缺少 task_id")
task_data = await _get_task_data(task_id)
if not task_data:
raise HTTPException(404, "任务不存在")
task_data = tasks[task_id]
geometry_data = task_data.get("geometry_data")
if not geometry_data:
raise HTTPException(400, "该任务尚未完成几何分析")
result = side_action_designer.analyze_and_design(
shape=None,
parting_direction=parting_direction,
mold_size=mold_size,
sd = _get_cached("side_action_designer")
if not sd:
raise HTTPException(503, "服务不可用：核心模块未加载")
result = sd.analyze_and_design(
shape=None, parting_direction=parting_direction, mold_size=mold_size,
)
return {"status": "success", "data": result}
@@ -187,22 +190,20 @@ async def design_mold_cam(
request: Request,
current_user: User = Depends(get_current_active_user),
):
"""模具CAM刀路设计"""
body = await request.json()
cavity_bbox = body.get("cavity_bbox", {"dimensions": [100, 100, 50], "min": [-50, -50, -25], "max": [50, 50, 25]})
stock_bbox = body.get("stock_bbox", {"dimensions": [150, 150, 100], "min": [-75, -75, -50], "max": [75, 75, 50]})
mold_steel = body.get("mold_steel", "P20")
surface_quality = body.get("surface_quality", "standard")
controller = body.get("controller", "fanuc")
result = mold_cam_designer.design_mold_cam(
cavity_bbox=cavity_bbox,
stock_bbox=stock_bbox,
mold_steel=mold_steel,
surface_quality=surface_quality,
cam = _get_cached("mold_cam_designer")
if not cam:
raise HTTPException(503, "服务不可用：核心模块未加载")
result = cam.design_mold_cam(
cavity_bbox=cavity_bbox, stock_bbox=stock_bbox,
mold_steel=mold_steel, surface_quality=surface_quality,
controller=controller,
)
return {"status": "success", "data": result}
@@ -211,16 +212,15 @@ async def check_toolpath_collision(
request: Request,
current_user: User = Depends(get_current_active_user),
):
"""刀路碰撞检测"""
body = await request.json()
toolpath_points = body.get("toolpath_points", [[0, 0, 50], [10, 10, -5], [20, 20, -10]])
tool = body.get("tool", {"diameter": 10, "flute_length": 30, "shank_diameter": 10})
stock_bbox = body.get("stock_bbox", {"min": [-50, -50, -25], "max": [50, 50, 25]})
clamp_positions = body.get("clamp_positions")
result = collision_detector.check_toolpath_safety(
toolpath_points, tool, stock_bbox, clamp_positions
)
cd = _get_cached("collision_detector")
if not cd:
raise HTTPException(503, "服务不可用：核心模块未加载")
result = cd.check_toolpath_safety(toolpath_points, tool, stock_bbox, clamp_positions)
return {"status": "success", "data": result}
@@ -229,15 +229,14 @@ async def optimize_toolpath(
request: Request,
current_user: User = Depends(get_current_active_user),
):
"""刀路优化"""
body = await request.json()
toolpath_points = body.get("toolpath_points", [[0, 0, 50], [10, 10, -5], [20, 20, -10]])
cutting_params = body.get("cutting_params", {"feed_rate_mm_min": 500})
stock_bbox = body.get("stock_bbox")
result = toolpath_optimizer.optimize_toolpath(
toolpath_points, cutting_params, stock_bbox
)
to = _get_cached("toolpath_optimizer")
if not to:
raise HTTPException(503, "服务不可用：核心模块未加载")
result = to.optimize_toolpath(toolpath_points, cutting_params, stock_bbox)
return {"status": "success", "data": result}
@@ -246,17 +245,16 @@ async def design_edm_electrodes(
request: Request,
current_user: User = Depends(get_current_active_user),
):
"""EDM电极设计"""
body = await request.json()
undercut_regions = body.get("undercut_regions", [{"center": [0, 0, 0], "area": 100, "type": "undercut"}])
cavity_bbox = body.get("cavity_bbox", {"dimensions": [100, 100, 50]})
material = body.get("material", "copper")
spark_gap = body.get("spark_gap", 0.05)
overburn = body.get("overburn", 0.1)
result = edm_designer.design_electrodes(
undercut_regions, cavity_bbox, material, spark_gap, overburn
)
ed = _get_cached("edm_designer")
if not ed:
raise HTTPException(503, "服务不可用：核心模块未加载")
result = ed.design_electrodes(undercut_regions, cavity_bbox, material, spark_gap, overburn)
return {"status": "success", "data": result}
@@ -265,15 +263,14 @@ async def simulate_machining(
request: Request,
current_user: User = Depends(get_current_active_user),
):
"""加工仿真"""
body = await request.json()
operations = body.get("operations", [{"strategy": "z_level_roughing", "levels": [{"z": -5}]}])
stock_bbox = body.get("stock_bbox", {"dimensions": [100, 100, 50], "min": [-50, -50, -25], "max": [50, 50, 25]})
resolution = body.get("resolution", 2.0)
result = machining_simulator.simulate_machining(
operations, stock_bbox, resolution
)
ms = _get_cached("machining_simulator")
if not ms:
raise HTTPException(503, "服务不可用：核心模块未加载")
result = ms.simulate_machining(operations, stock_bbox, resolution)
return {"status": "success", "data": result}
@@ -282,28 +279,40 @@ async def export_mold_results(
request: Request,
current_user: User = Depends(get_current_active_user),
):
"""导出模具设计结果（STEP/IGES/STL/BRep）"""
body = await request.json()
task_id = body.get("task_id")
formats = body.get("formats", ["step", "stl"])
components = body.get("components", ["cavity", "core"])
if not task_id or task_id not in tasks:
if not task_id:
raise HTTPException(404, "缺少 task_id")
task_data = await _get_task_data(task_id)
if not task_data:
raise HTTPException(404, "任务不存在")
task_data = tasks[task_id]
cavity_shapes = task_data.get("cavity_shapes")
filename = task_data.get("filename", f"mold_{task_id}")
if not cavity_shapes:
file_path = task_data.get("file_path")
if file_path and os.path.exists(str(file_path)):
cavity_shapes = await _reparse_stp_for_export(str(file_path), task_data.get("material", "ABS"))
if not cavity_shapes:
raise HTTPException(400, "该任务尚未完成模具生成或形状数据不可用")
base_filename = Path(task_data.get("filename", f"mold_{task_id}")).stem
result = cad_exporter.export_mold_results(
exporter = _get_cached("cad_exporter")
if not exporter:
raise HTTPException(503, "服务不可用：核心模块未加载，请检查 PythonOCC 环境")
base_filename = Path(filename).stem
result = exporter.export_mold_results(
cavity_data=cavity_shapes,
base_filename=base_filename,
formats=formats,
components=components,
)
return {"status": "success", "data": result}
@@ -312,34 +321,23 @@ async def download_export_file(
filepath: str,
current_user: User = Depends(get_current_active_user),
):
"""下载导出的CAD文件"""
from fastapi.responses import FileResponse
full_path = os.path.join(cad_exporter.output_dir, filepath)
exporter = _get_cached("cad_exporter")
if not exporter:
raise HTTPException(503, "服务不可用")
full_path = os.path.join(exporter.output_dir, filepath)
if not os.path.exists(full_path):
raise HTTPException(404, "文件不存在")
if not os.path.abspath(full_path).startswith(os.path.abspath(cad_exporter.output_dir)):
if not os.path.abspath(full_path).startswith(os.path.abspath(exporter.output_dir)):
raise HTTPException(403, "禁止访问")
media_types = {
".step": "application/step",
".stp": "application/step",
".iges": "application/iges",
".igs": "application/iges",
".stl": "model/stl",
".brep": "application/octet-stream",
".step": "application/step", ".stp": "application/step",
".iges": "application/iges", ".igs": "application/iges",
".stl": "model/stl", ".brep": "application/octet-stream",
}
ext = Path(full_path).suffix.lower()
media_type = media_types.get(ext, "application/octet-stream")
return FileResponse(
full_path,
media_type=media_type,
filename=os.path.basename(full_path),
)
return FileResponse(full_path, media_type=media_type, filename=os.path.basename(full_path))
@router.get("/export-recommendations")
@@ -347,6 +345,24 @@ async def get_export_recommendations(
target: str = "ug",
current_user: User = Depends(get_current_active_user),
):
"""获取导出格式建议（UG/FreeCAD/SolidWorks）"""
result = cad_exporter.get_export_recommendations(target)
exporter = _get_cached("cad_exporter")
if not exporter:
raise HTTPException(503, "服务不可用")
result = exporter.get_export_recommendations(target)
return {"status": "success", "data": result}
async def _reparse_stp_for_export(file_path: str, material: str = "ABS") -> dict:
try:
from core.stp_parser import STPParser
from core.mold_generator import MoldCavityGenerator
stp_parser = STPParser()
shape = stp_parser.load_step_file(Path(file_path))
mold_gen = MoldCavityGenerator(shrinkage_rate=0.005)
mold_gen.set_material(material)
cavity_result = mold_gen.generate_mold_cavities(shape)
logger.info(f"重新解析 STP 用于导出: {file_path}")
return cavity_result
except Exception as e:
logger.warning(f"重新解析 STP 导出失败: {e}")
return None
src/core/aluminum_foam_mold.py
+30
View File
@@ -298,6 +298,36 @@ class AluminumFoamMoldGenerator(BaseMoldGenerator):
result["normal_statistics"] = self._analyze_parting_direction(shape)
return result
def _analyze_parting_direction(self, shape: Any) -> Dict[str, float]:
"""分析产品法向量分布，按面积加权统计各轴方向强度"""
stats = {"X": 0.0, "Y": 0.0, "Z": 0.0}
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
props = GProp_GProps()
brepgprop.SurfaceProperties(face, props)
area = max(float(props.Mass()), 1.0)
stats["X"] += abs(float(normal.X())) * area
stats["Y"] += abs(float(normal.Y())) * area
stats["Z"] += abs(float(normal.Z())) * area
except Exception:
continue
total = stats["X"] + stats["Y"] + stats["Z"]
if total <= 0:
return {"X": 33.3, "Y": 33.3, "Z": 33.4}
return {
axis: round(value / total * 100, 2)
for axis, value in stats.items()
}
def _split_cavity_core(self, shape: Any, parting_surface: Any) -> Tuple[Any, Any]:
"""分离型腔和型芯（铝泡沫使用更大余量）"""
return super()._split_cavity_core(shape, parting_surface, margin=25)
src/core/base_mold_generator.py
+28
View File
@@ -606,6 +606,34 @@ class BaseMoldGenerator:
"face_count": 0,
}
def _extract_plane_metadata(self, surface: Any) -> Dict[str, Any]:
"""从分型面提取平面元数据（法向量、原点、边界）"""
metadata = {
"normal": [0.0, 0.0, 1.0],
"origin": [0.0, 0.0, 0.0],
"bounds": {"min": [0.0, 0.0, 0.0], "max": [0.0, 0.0, 0.0]},
}
try:
surface_adaptor = BRepAdaptor_Surface(surface)
if surface_adaptor.GetType() == 0:
plane = surface_adaptor.Plane()
axis = plane.Axis()
normal = axis.Direction()
origin = plane.Location()
metadata["normal"] = [float(normal.X()), float(normal.Y()), float(normal.Z())]
metadata["origin"] = [float(origin.X()), float(origin.Y()), float(origin.Z())]
bbox = Bnd_Box()
brepbndlib.Add(surface, bbox)
xmin, ymin, zmin, xmax, ymax, zmax = bbox.Get()
metadata["bounds"] = {
"min": [float(xmin), float(ymin), float(zmin)],
"max": [float(xmax), float(ymax), float(zmax)],
}
except Exception as e:
logger.warning(f"提取平面元数据失败: {e}")
return metadata
def _calculate_product_weight(self, analysis: Dict) -> str:
volume_cm3 = analysis.get("volume", 0) / 1000
weight_g = volume_cm3 * self.material_density
src/core/mold_generator.py
+46
View File
@@ -219,6 +219,52 @@ class MoldCavityGenerator(BaseMoldGenerator):
logger.info("使用简化方法检测分型面")
return self._simple_parting_surface(shape, analysis)
def _detect_primary_parting(self, shape: Any, 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)
dims = analysis.get("bounding_box", {}).get("dimensions", [100, 100, 100])
span = max(dims) * 1.5 + 30
parting_surface = BRepBuilderAPI_MakeFace(
parting_plane, -span, span, -span, span
).Face()
parting_surface = self.extend_parting_surface(parting_surface, shape, extension=30.0)
parting_line = self._calculate_parting_line(shape, parting_surface)
return {
"surface": parting_surface,
"line": parting_line,
"direction": [float(normal_dir.X()), float(normal_dir.Y()), float(normal_dir.Z())],
"method": "face_normal_analysis",
"confidence": 0.85,
}
except Exception as e:
logger.warning(f"法向量分析失败，使用简化方法：{e}")
surface, line = self._simple_parting_surface(shape, analysis)
return {
"surface": surface,
"line": line,
"direction": [0, 0, 1],
"method": "simple",
"confidence": 0.6,
}
def _build_undercut_regions(self, undercut_analysis: Dict[str, Any]) -> List[Dict[str, Any]]:
"""将侧向机构分析结果转换为兼容旧结构的倒扣区域列表。"""
undercut_faces = undercut_analysis.get("undercut_faces", [])