from __future__ import annotations import importlib.util import sys import traceback import uuid from dataclasses import dataclass from pathlib import Path from typing import Any from ..models import read_json, sha256_json, write_json from ..parameter_solver import load_requirement from .datum_inference import infer_step_datum_axis from .generator import JointModuleError, run_joint_module from .models import JointModuleManifest, JointPlatformProfile PACKAGE_ROOT = Path(__file__).resolve().parents[2] DEFAULT_PLATFORM_PROFILE = ( PACKAGE_ROOT / "input" / "assets" / "platforms" / "actuator_v2_fixed_platform.json" ) AUTO_CONTROLLED_OUTPUTS = [ "auto_design_input.json", "auto_design_report.json", "generated_reducer_requirement.json", "generated_joint_requirement.json", "auto_generation_error.json", ] class AutoJointDesignError(RuntimeError): pass @dataclass(frozen=True) class AutoJointEnvelope: motor_radial_diameter_mm: float motor_axis_length_mm: float housing_inner_diameter_mm: float housing_outer_diameter_mm: float housing_axis_length_mm: float reducer_max_outer_diameter_mm: float target_ratio: float module_candidates_mm: list[float] planet_count_candidates: list[int] def _patch_cadquery_visual_imports() -> None: try: import OCP.IVtkOCC as ivtkocc # type: ignore class _Dummy: def __init__(self, *args: object, **kwargs: object) -> None: pass def __getattr__(self, name: str) -> object: raise RuntimeError("CadQuery VTK visualization bridge is unavailable") for name in ["IVtkOCC_Shape", "IVtkOCC_ShapeMesher"]: if not hasattr(ivtkocc, name): setattr(ivtkocc, name, _Dummy) try: import OCP.IVtkVTK as ivtkvtk # type: ignore if not hasattr(ivtkvtk, "IVtkVTK_ShapeData"): setattr(ivtkvtk, "IVtkVTK_ShapeData", _Dummy) except Exception: # noqa: BLE001 return except Exception: # noqa: BLE001 return def _load_housing_module(path: Path) -> Any: _patch_cadquery_visual_imports() spec = importlib.util.spec_from_file_location("auto_joint_housing_source", path) if spec is None or spec.loader is None: raise AutoJointDesignError(f"cannot_load_housing_python: {path}") module = importlib.util.module_from_spec(spec) sys.modules["auto_joint_housing_source"] = module spec.loader.exec_module(module) return module def _resolve_path(path_text: str, *, requirement_path: Path | None = None) -> Path: path = Path(path_text).expanduser() if path.is_absolute(): return path if requirement_path is not None: relative_to_requirement = requirement_path.parent / path if relative_to_requirement.exists(): return relative_to_requirement.resolve() relative_to_package = PACKAGE_ROOT / path if relative_to_package.exists(): return relative_to_package.resolve() return path.resolve() def load_platform_assets(platform_profile: Path) -> tuple[JointPlatformProfile, dict[str, Path]]: if not platform_profile.exists(): raise AutoJointDesignError(f"missing_joint_platform_profile: {platform_profile}") profile = JointPlatformProfile.model_validate(read_json(platform_profile)) paths = { "motor_step": _resolve_path(profile.motor_step, requirement_path=platform_profile), "housing_step": _resolve_path(profile.housing_step, requirement_path=platform_profile), } for field in ( "motor_reference_assembly_step", "motor_stator_housing_step", "motor_rotor_housing_step", "motor_output_shaft_step", "motor_windings_step", ): value = getattr(profile, field) if value is not None: paths[field] = _resolve_path(value, requirement_path=platform_profile) if profile.chip_step is not None: paths["chip_step"] = _resolve_path(profile.chip_step, requirement_path=platform_profile) if profile.reference_assembly_step is not None: paths["reference_assembly_step"] = _resolve_path( profile.reference_assembly_step, requirement_path=platform_profile, ) missing = {name: str(path) for name, path in paths.items() if not path.exists()} if missing: raise AutoJointDesignError(f"fixed_platform_asset_missing: {missing}") return profile, paths def _axis_extent(metrics: Any) -> float: axis = metrics.inferred_axis if axis is None: raise AutoJointDesignError(f"{metrics.component_id}: missing inferred axis") selected = str(axis.details.get("selected_axis_dimension", "z")) xmin, ymin, zmin, xmax, ymax, zmax = metrics.bbox extents = { "x": xmax - xmin, "y": ymax - ymin, "z": zmax - zmin, } return float(extents[selected]) def _radial_diameter(metrics: Any) -> float: axis = metrics.inferred_axis if axis is None: raise AutoJointDesignError(f"{metrics.component_id}: missing inferred axis") selected = str(axis.details.get("selected_axis_dimension", "z")) xmin, ymin, zmin, xmax, ymax, zmax = metrics.bbox extents = { "x": xmax - xmin, "y": ymax - ymin, "z": zmax - zmin, } radial_axes = [axis_name for axis_name in ["x", "y", "z"] if axis_name != selected] return max(float(extents[radial_axes[0]]), float(extents[radial_axes[1]])) def _ratio_from_envelope(envelope: AutoJointEnvelope) -> float: if envelope.reducer_max_outer_diameter_mm < 46.0: return 3.0 if envelope.reducer_max_outer_diameter_mm < 58.0: return 5.0 if envelope.reducer_max_outer_diameter_mm < 76.0: return 7.5 return 10.0 def _module_candidates_for_outer_diameter(max_outer_diameter_mm: float) -> list[float]: if max_outer_diameter_mm < 45.0: return [0.3, 0.35, 0.4, 0.5] if max_outer_diameter_mm < 65.0: return [0.35, 0.4, 0.5, 0.6, 0.75] return [0.35, 0.4, 0.5, 0.75, 1.0] def compute_auto_envelope( *, motor_step: Path, housing_source: Path, radial_clearance_mm: float, reducer_axial_allowance_mm: float, reducer_radial_clearance_mm: float, ) -> tuple[AutoJointEnvelope, dict[str, Any], Any]: motor_metrics = infer_step_datum_axis(component_id="motor", step_path=motor_step, role="motor") motor_radial_diameter = _radial_diameter(motor_metrics) motor_axis_length = _axis_extent(motor_metrics) housing_module = _load_housing_module(housing_source) if not all(hasattr(housing_module, name) for name in ["scale_from_motor", "scaled_dimensions"]): raise AutoJointDesignError( "housing source must provide scale_from_motor(...) and scaled_dimensions(...)" ) scale = housing_module.scale_from_motor( motor_radial_diameter_mm=motor_radial_diameter, motor_axis_length_mm=motor_axis_length, radial_clearance_mm=radial_clearance_mm, reducer_axial_allowance_mm=reducer_axial_allowance_mm, ) dimensions = housing_module.scaled_dimensions(scale) housing_inner_diameter = float(dimensions["inner_wall_radius"]) * 2.0 housing_outer_diameter = float(dimensions["top_outer_radius"]) * 2.0 housing_axis_length = float(dimensions["height"]) reducer_max_outer = max(24.0, housing_inner_diameter - 2.0 * reducer_radial_clearance_mm) envelope = AutoJointEnvelope( motor_radial_diameter_mm=motor_radial_diameter, motor_axis_length_mm=motor_axis_length, housing_inner_diameter_mm=housing_inner_diameter, housing_outer_diameter_mm=housing_outer_diameter, housing_axis_length_mm=housing_axis_length, reducer_max_outer_diameter_mm=reducer_max_outer, target_ratio=0.0, module_candidates_mm=_module_candidates_for_outer_diameter(reducer_max_outer), planet_count_candidates=[3, 4], ) envelope = AutoJointEnvelope( **{ **envelope.__dict__, "target_ratio": _ratio_from_envelope(envelope), } ) return envelope, { "motor_metrics": motor_metrics.model_dump(mode="json"), "housing_scale": { "radial_scale": float(scale.radial_scale), "axial_scale": float(scale.axial_scale), "source": str(scale.source), }, "housing_dimensions": dimensions, }, housing_module def compute_fixed_platform_envelope( *, platform_profile: Path, radial_clearance_mm: float, reducer_radial_clearance_mm: float, ) -> tuple[AutoJointEnvelope, dict[str, Any], JointPlatformProfile]: profile, paths = load_platform_assets(platform_profile) motor_metrics = infer_step_datum_axis( component_id="motor", step_path=paths["motor_step"], role="motor", ) housing_metrics = infer_step_datum_axis( component_id="housing", step_path=paths["housing_step"], role="housing", ) motor_radial_diameter = _radial_diameter(motor_metrics) motor_axis_length = _axis_extent(motor_metrics) housing_radial_diameter = _radial_diameter(housing_metrics) housing_axis_length = _axis_extent(housing_metrics) housing_inner_diameter = max(24.0, housing_radial_diameter - 2.0 * radial_clearance_mm) reducer_max_outer = max(24.0, housing_inner_diameter - 2.0 * reducer_radial_clearance_mm) envelope = AutoJointEnvelope( motor_radial_diameter_mm=motor_radial_diameter, motor_axis_length_mm=motor_axis_length, housing_inner_diameter_mm=housing_inner_diameter, housing_outer_diameter_mm=housing_radial_diameter, housing_axis_length_mm=housing_axis_length, reducer_max_outer_diameter_mm=reducer_max_outer, target_ratio=0.0, module_candidates_mm=_module_candidates_for_outer_diameter(reducer_max_outer), planet_count_candidates=[3, 4], ) envelope = AutoJointEnvelope( **{ **envelope.__dict__, "target_ratio": _ratio_from_envelope(envelope), } ) return envelope, { "platform_profile": str(platform_profile.resolve()), "platform_id": profile.platform_id, "asset_paths": {name: str(path.resolve()) for name, path in paths.items()}, "motor_metrics": motor_metrics.model_dump(mode="json"), "housing_metrics": housing_metrics.model_dump(mode="json"), "housing_dimensions": { "estimated_inner_diameter_mm": housing_inner_diameter, "outer_diameter_mm": housing_radial_diameter, "axis_length_mm": housing_axis_length, }, }, profile def export_generated_housing( *, housing_module: Any, scale_payload: dict[str, Any], out_path: Path, ) -> None: if not hasattr(housing_module, "HousingScale") or not hasattr(housing_module, "export_scaled"): raise AutoJointDesignError("housing source must provide HousingScale and export_scaled") scale = housing_module.HousingScale( radial_scale=float(scale_payload["radial_scale"]), axial_scale=float(scale_payload["axial_scale"]), source=str(scale_payload.get("source", "computed_from_motor_bbox")), ) out_path.parent.mkdir(parents=True, exist_ok=True) housing_module.export_scaled(out_path, scale) if not out_path.exists() or out_path.stat().st_size <= 0: raise AutoJointDesignError(f"generated_housing_step_missing_or_empty: {out_path}") def _cascade_requirement_template() -> dict[str, Any]: template = ( PACKAGE_ROOT / "input" / "requirements" / "reducers" / "simple_2k_h_cascade" / "simple_2kh_cascade_ratio_9_spur_compact.json" ) return read_json(template) def build_reducer_requirement( envelope: AutoJointEnvelope, *, tooth_form: str = "spur", topology_family: str = "simple_2k_h", ) -> dict[str, Any]: if topology_family == "simple_2k_h_cascade": payload = _cascade_requirement_template() payload["tooth_form"] = tooth_form payload.setdefault("constraints", {})["max_outer_diameter_mm"] = ( envelope.reducer_max_outer_diameter_mm ) return payload if topology_family != "simple_2k_h": raise AutoJointDesignError(f"unsupported_auto_joint_reducer_family: {topology_family}") return { "target_ratio": envelope.target_ratio, "ratio_tolerance": 0.000001, "topology_family": "simple_2k_h", "detail_level": "industrial_core", "input": "sun", "fixed": "ring", "output": "carrier", "tooth_form": tooth_form, "planet_count_candidates": envelope.planet_count_candidates, "module_candidates_mm": envelope.module_candidates_mm, "pressure_angle_deg": 20.0, "face_width_mm": min(8.0, max(5.0, envelope.housing_axis_length_mm * 0.10)), "backlash_mm": 0.03, "addendum_coeff": 1.0, "clearance_coeff": 0.25, "ring_rim_thickness_mm": max(1.8, envelope.reducer_max_outer_diameter_mm * 0.035), "constraints": { "max_outer_diameter_mm": envelope.reducer_max_outer_diameter_mm, "min_planet_neighbor_clearance_mm": 0.3, "min_sun_teeth": 18, "max_sun_teeth": 96, "min_planet_teeth": 18, "max_planet_teeth": 96, }, } def build_joint_requirement( *, platform_profile: Path, reducer_requirement: Path, topology_family: str = "simple_2k_h", ) -> dict[str, Any]: if topology_family == "simple_2k_h_cascade": motor_relation = "motor_output_to_s1_sun_input" housing_relation = "stage_rings_fixed_to_housing" joint_output = "s2_carrier_output" else: motor_relation = "motor_output_to_sun_keyed_input" housing_relation = "ring_fixed_to_housing" joint_output = "carrier_output_shaft" return { "module_type": "planetary_joint_module", "platform_profile": str(platform_profile.resolve()), "reducer_requirement": str(reducer_requirement.resolve()), "placement_policy": "infer_then_write_editable_json", "motor_to_reducer_relation": motor_relation, "housing_relation": housing_relation, "joint_output": joint_output, "motor_placement_policy": "external_flush_input_end", "input_shaft_insertion_depth_mm": 4.0, "output_extension_mm": 14.0, } def prepare_auto_dir(out_dir: Path) -> None: out_dir.mkdir(parents=True, exist_ok=True) for name in AUTO_CONTROLLED_OUTPUTS: path = out_dir / name if path.exists(): path.unlink() def run_auto_joint_design( *, out_dir: Path, platform_profile: Path = DEFAULT_PLATFORM_PROFILE, motor_step: Path | None = None, housing_source: Path | None = None, tooth_form: str = "spur", radial_clearance_mm: float = 14.0, reducer_axial_allowance_mm: float = 28.0, reducer_radial_clearance_mm: float = 8.0, reducer_requirement_overrides: dict[str, Any] | None = None, ) -> JointModuleManifest: prepare_auto_dir(out_dir) platform_profile = platform_profile.resolve() envelope, evidence, _profile = compute_fixed_platform_envelope( platform_profile=platform_profile, radial_clearance_mm=radial_clearance_mm, reducer_radial_clearance_mm=reducer_radial_clearance_mm, ) reducer_requirement_path = out_dir / "generated_reducer_requirement.json" joint_requirement_path = out_dir / "generated_joint_requirement.json" overrides = reducer_requirement_overrides or {} topology_family = str(overrides.get("topology_family") or "simple_2k_h") reducer_requirement_payload = build_reducer_requirement( envelope, tooth_form=tooth_form, topology_family=topology_family, ) for name in { "target_ratio", "tooth_form", "helix_angle_deg", "pressure_angle_deg", "face_width_mm", "backlash_mm", "ring_rim_thickness_mm", }: if name in overrides: reducer_requirement_payload[name] = overrides[name] if "module_mm" in overrides: reducer_requirement_payload["module_candidates_mm"] = [float(overrides["module_mm"])] if "planet_count" in overrides: reducer_requirement_payload["planet_count_candidates"] = [int(overrides["planet_count"])] if "max_outer_diameter_mm" in overrides: reducer_requirement_payload["constraints"]["max_outer_diameter_mm"] = float( overrides["max_outer_diameter_mm"] ) if reducer_requirement_payload.get("tooth_form") == "helical": reducer_requirement_payload.setdefault("helix_angle_deg", 15.0) else: reducer_requirement_payload.pop("helix_angle_deg", None) write_json(reducer_requirement_path, reducer_requirement_payload) # Validate before launching the heavy generation path. load_requirement(reducer_requirement_path) joint_requirement_payload = build_joint_requirement( platform_profile=platform_profile, reducer_requirement=reducer_requirement_path, topology_family=str(reducer_requirement_payload.get("topology_family") or topology_family), ) write_json(joint_requirement_path, joint_requirement_payload) write_json( out_dir / "auto_design_input.json", { "platform_profile": str(platform_profile.resolve()), "ignored_legacy_motor_step": str(motor_step.resolve()) if motor_step else None, "ignored_legacy_housing_source": str(housing_source.resolve()) if housing_source else None, "tooth_form": tooth_form, "radial_clearance_mm": radial_clearance_mm, "reducer_axial_allowance_mm": reducer_axial_allowance_mm, "reducer_radial_clearance_mm": reducer_radial_clearance_mm, }, ) write_json( out_dir / "auto_design_report.json", { "run_id": f"auto_{uuid.uuid4().hex[:12]}", "requirement_hash": sha256_json(joint_requirement_payload), "envelope": envelope.__dict__, "evidence": evidence, "generated_files": { "reducer_requirement": str(reducer_requirement_path.resolve()), "joint_requirement": str(joint_requirement_path.resolve()), }, }, ) try: return run_joint_module(requirement_path=joint_requirement_path, out_dir=out_dir / "joint") except Exception as exc: # noqa: BLE001 write_json( out_dir / "auto_generation_error.json", { "stage": "joint", "error_type": type(exc).__name__, "message": str(exc), "traceback": traceback.format_exc(), }, ) raise