refactor(moldinsight): 引入可插拔注册表(MoldGenerator + FeatureDetector)

- 新增 MoldGeneratorRegistry:multi_scheme_planner 消除 if-else,按 mold_type 选生成器
- 新增 FeatureDetectorRegistry:geometry_analyzer._detect_features 消除 6 个检测器硬编码
- 新增模具类型/特征检测器只需 register 一行
- 移除 OCC ThreadPoolExecutor(注册表本身串行执行)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
2026-07-20 11:03:16 +08:00
parent f3d2951dc3
commit 66ba6e00f5
4 changed files with 92 additions and 28 deletions
@@ -0,0 +1,34 @@
"""特征检测器注册表
特征检测器通过 FeatureDetectorRegistry 注册,GeometryAnalyzer 遍历注册表执行,
不再硬编码检测器列表。新增检测器只需 register 一个 (name, fn, requires_shape)。
"""
from typing import Callable, List, Dict, Any, Optional, Tuple
class FeatureDetectorRegistry:
"""特征检测器注册表
每个检测器是一个 (name, fn, requires_shape) 条目:
- fn: callable(geometry_data, shape) -> List[Dict],返回检测到的特征列表
- requires_shape: True 表示仅当 shape 非 None 时才执行(如曲率/圆角检测需 OCC Shape)
串行执行(OCC 非线程安全),单个检测器失败不影响其他。
"""
def __init__(self):
self._entries: List[Tuple[str, Callable, bool]] = []
def register(self, name: str, fn: Callable, requires_shape: bool = False) -> None:
self._entries.append((name, fn, requires_shape))
def detect_all(self, geometry_data: Dict[str, Any], shape: Optional[Any]) -> List[Dict[str, Any]]:
features: List[Dict[str, Any]] = []
for _name, fn, requires_shape in self._entries:
if requires_shape and shape is None:
continue
try:
features.extend(fn(geometry_data, shape))
except Exception:
pass
return features
+14 -20
View File
@@ -1,6 +1,5 @@
from typing import Dict, List, Any, Optional
import math
from concurrent.futures import ThreadPoolExecutor, as_completed
import numpy as np
from OCC.Core.TopoDS import TopoDS_Shape
from shared.models.schemas import (
@@ -10,6 +9,7 @@ from shared.models.schemas import (
)
from shared.utils.logger import get_logger
from moldinsight.services.material_service import MaterialService
from moldinsight.core.feature_detector_registry import FeatureDetectorRegistry
logger = get_logger(__name__)
@@ -32,6 +32,15 @@ class GeometryAnalyzer:
"H13_Steel": {"thermal_conductivity": 25, "hardness": "HRC48", "cost": "high"}
}
# 特征检测器注册表 - 新增检测器只需在此 register
self._feature_detectors = FeatureDetectorRegistry()
self._feature_detectors.register("wall", lambda gd, s: self._detect_wall_features(gd, s))
self._feature_detectors.register("rib", lambda gd, s: self._detect_rib_features(gd, s))
self._feature_detectors.register("boss", lambda gd, s: self._detect_boss_features(gd, s))
self._feature_detectors.register("draft", lambda gd, s: self._analyze_draft_angles(gd, s))
self._feature_detectors.register("curvature", lambda gd, s: self._detect_curvature_features(s), requires_shape=True)
self._feature_detectors.register("fillet", lambda gd, s: self._detect_fillet_features(s), requires_shape=True)
def analyze_mold_design(self, geometry_data: Dict[str, Any],
product_material: str = "ABS",
mold_material: str = "Aluminum",
@@ -69,26 +78,11 @@ class GeometryAnalyzer:
def _detect_features(self, geometry_data: Dict[str, Any],
shape: Optional[TopoDS_Shape] = None) -> List[Dict[str, Any]]:
"""检测模具特征 — 独立检测并行执行"""
features: List[Dict[str, Any]] = []
with ThreadPoolExecutor(max_workers=1, thread_name_prefix="feat") as pool:
futures = {
pool.submit(self._detect_wall_features, geometry_data, shape): "wall",
pool.submit(self._detect_rib_features, geometry_data, shape): "rib",
pool.submit(self._detect_boss_features, geometry_data, shape): "boss",
pool.submit(self._analyze_draft_angles, geometry_data, shape): "draft",
}
if shape is not None:
futures[pool.submit(self._detect_curvature_features, shape)] = "curvature"
futures[pool.submit(self._detect_fillet_features, shape)] = "fillet"
for future in as_completed(futures):
try:
features.extend(future.result())
except Exception:
pass
"""检测模具特征 - 串行执行(OCC 非线程安全)
检测器通过 FeatureDetectorRegistry 注册,新增检测器只需在 __init__ 中 register。
"""
features = self._feature_detectors.detect_all(geometry_data, shape)
logger.info(f"检测到 {len(features)} 个特征")
return features
@@ -0,0 +1,42 @@
"""模具生成器注册表
按 mold_type 注册生成器实例,MultiSchemeMoldPlanner 通过 mold_type 查询,
不再硬编码 if-else 选择生成器。新增模具类型只需 `register` 一个新生成器,无需改 planner。
"""
from typing import Dict
from moldinsight.core.base_mold_generator import BaseMoldGenerator
from moldinsight.core.mold_generator import MoldCavityGenerator
from moldinsight.core.aluminum_foam_mold import AluminumFoamMoldGenerator
class MoldGeneratorRegistry:
"""模具生成器注册表"""
def __init__(self):
self._generators: Dict[str, BaseMoldGenerator] = {}
def register(self, mold_type: str, generator: BaseMoldGenerator) -> None:
"""注册一个模具生成器"""
self._generators[mold_type] = generator
def get_by_type(self, mold_type: str) -> BaseMoldGenerator:
"""按 mold_type 获取生成器,未注册则抛错"""
gen = self._generators.get(mold_type)
if gen is None:
raise ValueError(
f"未注册的模具类型: {mold_type},已注册: {list(self._generators.keys())}"
)
return gen
def list_types(self):
return list(self._generators.keys())
# 全局单例,注册默认生成器
mold_generator_registry = MoldGeneratorRegistry()
mold_generator_registry.register("injection", MoldCavityGenerator(shrinkage_rate=0.005))
mold_generator_registry.register(
"aluminum_foam",
AluminumFoamMoldGenerator(shrinkage_rate=0.015, draft_angle=3.0),
)
+2 -8
View File
@@ -8,8 +8,7 @@ from OCC.Core.TopAbs import TopAbs_FACE
from OCC.Core.TopExp import TopExp_Explorer
from OCC.Core.TopoDS import TopoDS_Face, TopoDS_Shape, topods
from moldinsight.core.mold_generator import MoldCavityGenerator
from moldinsight.core.aluminum_foam_mold import AluminumFoamMoldGenerator
from moldinsight.core.mold_generator_registry import mold_generator_registry
from moldinsight.core.parting_candidate_generator import PartingCandidateGenerator
from moldinsight.core.parting_scheme_scorer import PartingSchemeScorer
from shared.utils.logger import get_logger
@@ -23,11 +22,6 @@ class MultiSchemeMoldPlanner:
def __init__(self):
self.candidate_generator = PartingCandidateGenerator()
self.scheme_scorer = PartingSchemeScorer()
self.mold_generator = MoldCavityGenerator(shrinkage_rate=0.005)
self.aluminum_foam_generator = AluminumFoamMoldGenerator(
shrinkage_rate=0.015,
draft_angle=3.0,
)
def generate_plan(
self,
@@ -37,7 +31,7 @@ class MultiSchemeMoldPlanner:
max_schemes: int = 3,
process_params: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
generator = self.aluminum_foam_generator if is_foam_material else self.mold_generator
generator = mold_generator_registry.get_by_type("aluminum_foam" if is_foam_material else "injection")
generator.set_material(material["name"])
self._apply_process_params(generator, material, process_params)