This commit is contained in:
2026-08-14 10:26:28 +08:00
parent f051146486
commit 7d12459a32
44 changed files with 900211 additions and 15295 deletions
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
@@ -51,7 +51,7 @@ def scale_from_motor(
reducer_axial_allowance_mm: float = 42.0,
min_radial_scale: float = 0.45,
max_radial_scale: float = 1.50,
min_axial_scale: float = 0.55,
min_axial_scale: float = 0.42,
max_axial_scale: float = 1.80,
) -> HousingScale:
target_inner_radius = motor_radial_diameter_mm / 2.0 + radial_clearance_mm
@@ -0,0 +1,255 @@
from pathlib import Path
import math
import cadquery as cq
AXIS = cq.Vector(0.0, 1.0, 0.0)
def _point_on_xz(radius: float, angle_degrees: float, y: float) -> cq.Vector:
angle = math.radians(angle_degrees)
return cq.Vector(radius * math.cos(angle), y, radius * math.sin(angle))
def _cylinder(radius: float, y0: float, length: float, x: float = 0.0, z: float = 0.0):
return cq.Solid.makeCylinder(radius, length, cq.Vector(x, y0, z), AXIS)
def _ring(outer_radius: float, inner_radius: float, y0: float, length: float):
return _cylinder(outer_radius, y0, length).cut(_cylinder(inner_radius, y0, length))
def _conical_ring(
outer_start: float,
outer_end: float,
inner_start: float,
inner_end: float,
y0: float,
length: float,
):
outer = cq.Solid.makeCone(outer_start, outer_end, length, cq.Vector(0.0, y0, 0.0), AXIS)
inner = cq.Solid.makeCone(inner_start, inner_end, length, cq.Vector(0.0, y0, 0.0), AXIS)
return outer.cut(inner)
def _sector_cut(angle_center: float, angle_width: float, y0: float, length: float, radius: float = 100.0):
start = math.radians(angle_center - angle_width / 2.0)
end = math.radians(angle_center + angle_width / 2.0)
points = [
(0.0, 0.0),
(radius * math.cos(start), radius * math.sin(start)),
(radius * math.cos(end), radius * math.sin(end)),
]
plane = cq.Plane(origin=(0.0, y0, 0.0), xDir=(1.0, 0.0, 0.0), normal=(0.0, 1.0, 0.0))
return cq.Workplane(plane).polyline(points).close().extrude(length).val()
def _annular_sector_cut(
angle_center: float,
angle_width: float,
inner_radius: float,
outer_radius: float,
y0: float,
length: float,
profile_fillet_radius: float = 0.0,
axial_fillet_radius: float = 0.0,
):
start = math.radians(angle_center - angle_width / 2.0)
middle = math.radians(angle_center)
end = math.radians(angle_center + angle_width / 2.0)
inner_start = (inner_radius * math.cos(start), inner_radius * math.sin(start))
inner_middle = (inner_radius * math.cos(middle), inner_radius * math.sin(middle))
inner_end = (inner_radius * math.cos(end), inner_radius * math.sin(end))
outer_start = (outer_radius * math.cos(start), outer_radius * math.sin(start))
outer_middle = (outer_radius * math.cos(middle), outer_radius * math.sin(middle))
outer_end = (outer_radius * math.cos(end), outer_radius * math.sin(end))
plane = cq.Plane(origin=(0.0, y0, 0.0), xDir=(1.0, 0.0, 0.0), normal=(0.0, 1.0, 0.0))
cutter = (
cq.Workplane(plane)
.moveTo(*inner_start)
.lineTo(*outer_start)
.threePointArc(outer_middle, outer_end)
.lineTo(*inner_end)
.threePointArc(inner_middle, inner_start)
.close()
.extrude(length)
)
if profile_fillet_radius > 0.0:
# Round the corners of the annular profile (fillet axes parallel to Y).
cutter = cutter.edges("|Y").fillet(profile_fillet_radius)
if axial_fillet_radius > 0.0:
# Round only the straight radial edges at the two axial ends. This
# produces the R10/R5/R2 window run-outs present in the source STEP.
end_edges = [
edge
for edge in cutter.edges("%LINE").vals()
if abs(edge.Center().y - y0) < 1.0e-6
or abs(edge.Center().y - (y0 + length)) < 1.0e-6
]
cutter = cutter.newObject(end_edges).fillet(axial_fillet_radius)
return cutter.val()
def _radial_bar(angle_degrees: float, y0: float, length: float, inner_radius: float, outer_radius: float, width: float):
radial_length = outer_radius - inner_radius
bar = (
cq.Workplane("XY")
.box(radial_length, length, width)
.translate(((inner_radius + outer_radius) / 2.0, y0 + length / 2.0, 0.0))
)
return bar.rotate((0.0, 0.0, 0.0), (0.0, 1.0, 0.0), -angle_degrees).val()
def _rounded_radial_slot(angle_degrees: float, y0: float, length: float):
"""A 40 mm wide radial window with R2 corners in its axial section."""
plane = cq.Plane(
origin=(40.0, y0 + length / 2.0, 0.0),
xDir=(0.0, 1.0, 0.0),
normal=(1.0, 0.0, 0.0),
)
slot = (
cq.Workplane(plane)
.rect(length - 4.0, 36.0)
.offset2D(2.0)
.extrude(42.0)
.rotate((0.0, 0.0, 0.0), (0.0, 1.0, 0.0), -angle_degrees)
)
return slot.val()
def _support_blend(angle_degrees: float, mirrored: bool):
"""One of the six R2 blends between the R64 and R29.2 support arcs."""
sharp = (-62.80715975198, 12.298808230444)
inner_tangent = (-62.1855700706093, 15.131254904777)
lobe_tangent = (-63.6707499640005, 16.8512614004542)
inner_center = (0.0, 0.0)
blend_center = (-61.6824120316291, 17.0669284278203)
lobe_center = (-34.6410161514, 20.0)
def transform(point):
x, z = point
if mirrored:
axis = math.radians(150.0)
ux, uz = math.cos(axis), math.sin(axis)
projection = x * ux + z * uz
x, z = 2.0 * projection * ux - x, 2.0 * projection * uz - z
angle = math.radians(angle_degrees)
return (
x * math.cos(angle) - z * math.sin(angle),
x * math.sin(angle) + z * math.cos(angle),
)
def arc_middle(center, start, end, radius):
ux = (start[0] - center[0]) / radius + (end[0] - center[0]) / radius
uz = (start[1] - center[1]) / radius + (end[1] - center[1]) / radius
scale = radius / math.hypot(ux, uz)
return center[0] + ux * scale, center[1] + uz * scale
points = [transform(point) for point in (sharp, inner_tangent, lobe_tangent)]
centers = [transform(point) for point in (inner_center, blend_center, lobe_center)]
plane = cq.Plane(origin=(0.0, 90.2, 0.0), xDir=(1.0, 0.0, 0.0), normal=(0.0, -1.0, 0.0))
return (
cq.Workplane(plane)
.moveTo(*points[0])
.threePointArc(arc_middle(centers[0], points[0], points[1], 64.0), points[1])
.threePointArc(arc_middle(centers[1], points[1], points[2], 2.0), points[2])
.threePointArc(arc_middle(centers[2], points[2], points[0], 29.2), points[0])
.close()
.extrude(10.0)
.val()
)
def build():
# Front cylindrical shell and its smaller continuation.
part = cq.Workplane(obj=_ring(79.4, 74.4, -10.0, 22.0))
part = part.union(_ring(78.4, 74.4, 12.0, 43.2))
# Nine radial R2 holes through the front lip, spaced every 40 degrees.
for angle in (90.0, 50.0, 10.0, -30.0, -70.0, -110.0, -150.0, -190.0, -230.0):
start = _point_on_xz(72.0, angle, -5.0)
direction = _point_on_xz(1.0, angle, 0.0)
direction = cq.Vector(direction.x, 0.0, direction.z)
part = part.cut(cq.Solid.makeCylinder(2.0, 10.0, start, direction))
# Six long ventilation windows, 40-degree openings with 20-degree ribs.
for center in (-110.0, -50.0, 10.0, 70.0, 130.0, 190.0):
part = part.cut(
_annular_sector_cut(center, 40.0, 70.0, 82.0, 12.0, 40.0, axial_fillet_radius=10.0)
)
# Taper from the large front shell to the narrower middle shell.
part = part.union(_conical_ring(78.4, 73.2, 74.4, 69.2, 55.2, 20.0))
part = part.union(_ring(73.2, 60.0, 75.2, 5.0))
# Three small axial holes in the first shoulder ring.
for angle in (30.0, 150.0, 270.0):
p = _point_on_xz(64.0, angle, 75.2)
part = part.cut(cq.Solid.makeCylinder(2.0, 5.0, p, AXIS))
# Three-lobed support plate: a disk with three large circular openings.
support = cq.Workplane(obj=_ring(73.2, 64.0, 80.2, 10.0))
for x, z in ((34.6410161514, 20.0), (-34.6410161514, 20.0), (0.0, -40.0)):
support = support.cut(_cylinder(29.2, 80.2, 10.0, x, z))
part = part.union(support)
# Thin middle shell and its second, angularly shifted six-window pattern.
part = part.union(_ring(73.2, 69.2, 90.2, 30.0))
for center in (-150.0, -90.0, -30.0, 30.0, 90.0, 150.0):
part = part.cut(
_annular_sector_cut(center, 40.0, 65.0, 76.0, 90.2, 20.0, axial_fillet_radius=5.0)
)
# Rear shell is one continuous 2:7 taper. Its inner cavity changes from
# a steep R69.2 -> R63.6 cone to the matching shallow taper at Y=125.8.
taper_outer = cq.Solid.makeCone(
73.2, 64.91428571428571, 29.0, cq.Vector(0.0, 120.2, 0.0), AXIS
)
taper_inner_a = cq.Solid.makeCone(69.2, 63.6, 5.6, cq.Vector(0.0, 120.2, 0.0), AXIS)
taper_inner_b = cq.Solid.makeCone(63.6, 59.2, 15.4, cq.Vector(0.0, 125.8, 0.0), AXIS)
part = part.union(taper_outer.cut(taper_inner_a).cut(taper_inner_b))
hub = cq.Workplane(obj=_ring(38.0, 26.0, 129.2, 4.0)).union(
_ring(38.0, 30.0, 133.2, 16.0)
)
part = part.union(hub)
# Stop 0.0001 mm before the coincident rear-plate face so OpenCascade
# keeps the six ribs in one valid solid.
for center in (30.0, 90.0, 150.0, 210.0, 270.0, 330.0):
part = part.cut(_rounded_radial_slot(center, 131.2, 9.9999))
# Central bore through the solid rear plate, then its six fan windows.
part = part.cut(_cylinder(30.0, 141.2, 8.0))
for center in (45.0, 105.0, 165.0, 225.0, 285.0, 345.0):
cutter = _annular_sector_cut(
center, 30.0, 38.0, 59.2, 141.2, 8.0, profile_fillet_radius=2.0
)
part = part.cut(cutter)
# These holes pass through both the end plate and the previously added hub.
for angle in (-90.0, -30.0, 30.0, 90.0, 150.0, 210.0):
p = _point_on_xz(34.0, angle, 141.2)
part = part.cut(cq.Solid.makeCylinder(1.81198, 8.0, p, AXIS))
# Add the small support blends last so they cannot interfere with fusing
# the face-connected rear shell and end plate.
for mirrored in (False, True):
for angle in (0.0, 120.0, 240.0):
part = part.union(_support_blend(angle, mirrored))
# Remove only the six 0.0001 mm run-out residuals after the complete part
# is connected. This reaches the exact Y=141.2 boundary without causing
# the earlier coincident-face split into six solids.
for center in (30.0, 90.0, 150.0, 210.0, 270.0, 330.0):
full_slot = _rounded_radial_slot(center, 131.2, 10.0)
stable_slot = _rounded_radial_slot(center, 131.2, 9.9999)
part = part.cut(full_slot.cut(stable_slot))
return part
if __name__ == "__main__":
output = Path(__file__).resolve().with_name("外壳框架.step")
cq.exporters.export(build(), str(output))
print(output)
File diff suppressed because one or more lines are too long
@@ -0,0 +1,6 @@
{
"platform_id": "actuator_v2_fixed_platform",
"motor_step": "../motors/3500_motor_part2_standard.step",
"housing_step": "../housings/actuator_v2_knee_abd_main_case_2.step",
"reference_assembly_step": "../references/actuator_v2_knee_abd_reference.step"
}
File diff suppressed because one or more lines are too long
@@ -1,14 +1,14 @@
{
"module_type": "planetary_joint_module",
"motor_step": "../../assets/motors/3500_motor_part2.step",
"housing_step": "../../assets/housings/generated_housing_baseline.step",
"reducer_requirement": "../reducers/simple_2k_h/simple_2kh_industrial_ratio_7p5_spur.json",
"platform_profile": "../../assets/platforms/actuator_v2_fixed_platform.json",
"reducer_requirement": "../reducers/simple_2k_h/simple_2kh_industrial_ratio_7p5_spur_compact.json",
"placement_policy": "infer_then_write_editable_json",
"motor_to_reducer_relation": "motor_output_to_sun_keyed_input",
"housing_relation": "ring_fixed_to_housing",
"joint_output": "carrier_output_shaft",
"output_extension_mm": 8.0,
"coupling_clearance_mm": 1.0,
"motor_placement_policy": "external_flush_input_end",
"output_extension_mm": 2.0,
"coupling_clearance_mm": 0.5,
"input_shaft_insertion_depth_mm": 4.0,
"motor_output_side": "axis_max",
"axis_angle_tolerance_deg": 0.5,
@@ -1,15 +1,15 @@
{
"module_type": "planetary_joint_module",
"motor_step": "../../assets/motors/3500_motor_part2.step",
"housing_step": "../../assets/housings/generated_housing_baseline.step",
"reducer_requirement": "../reducers/simple_2k_h_cascade/simple_2kh_cascade_ratio_9_spur.json",
"platform_profile": "../../assets/platforms/actuator_v2_fixed_platform.json",
"reducer_requirement": "../reducers/simple_2k_h_cascade/simple_2kh_cascade_ratio_9_spur_compact.json",
"placement_policy": "infer_then_write_editable_json",
"motor_to_reducer_relation": "motor_output_to_s1_sun_input",
"housing_relation": "stage_rings_fixed_to_housing",
"joint_output": "s2_carrier_output",
"motor_placement_policy": "external_flush_input_end",
"output_extension_mm": 2.0,
"coupling_clearance_mm": 0.5,
"input_shaft_insertion_depth_mm": 1.0,
"input_shaft_insertion_depth_mm": 4.0,
"motor_output_side": "axis_max",
"axis_angle_tolerance_deg": 0.5,
"axis_distance_tolerance_mm": 0.25,
@@ -0,0 +1,26 @@
{
"target_ratio": 7.5,
"ratio_tolerance": 0.01,
"topology_family": "simple_2k_h",
"detail_level": "industrial_core",
"input": "sun",
"fixed": "ring",
"output": "carrier",
"tooth_form": "spur",
"planet_count_candidates": [3],
"module_candidates_mm": [0.3],
"pressure_angle_deg": 20.0,
"face_width_mm": 6.0,
"backlash_mm": 0.03,
"addendum_coeff": 1.0,
"clearance_coeff": 0.25,
"ring_rim_thickness_mm": 1.5,
"constraints": {
"max_outer_diameter_mm": 60.0,
"min_planet_neighbor_clearance_mm": 0.3,
"min_sun_teeth": 20,
"max_sun_teeth": 20,
"min_planet_teeth": 55,
"max_planet_teeth": 55
}
}
@@ -0,0 +1,44 @@
{
"target_ratio": 9.0,
"ratio_tolerance": 0.000001,
"topology_family": "simple_2k_h_cascade",
"tooth_form": "spur",
"planet_count_candidates": [3],
"module_candidates_mm": [0.5],
"pressure_angle_deg": 20.0,
"face_width_mm": 6.0,
"backlash_mm": 0.03,
"addendum_coeff": 1.0,
"clearance_coeff": 0.25,
"ring_rim_thickness_mm": 2.5,
"constraints": {
"max_outer_diameter_mm": 100.0,
"min_planet_neighbor_clearance_mm": 0.3,
"min_sun_teeth": 36,
"max_sun_teeth": 36,
"min_planet_teeth": 18,
"max_planet_teeth": 18
},
"stages": [
{
"stage_id": "s1",
"input": "sun",
"fixed": "ring",
"output": "carrier",
"planet_count_candidates": [3],
"module_candidates_mm": [0.5]
},
{
"stage_id": "s2",
"input": "sun",
"fixed": "ring",
"output": "carrier",
"planet_count_candidates": [3],
"module_candidates_mm": [0.5]
}
],
"cascade": {
"coupling_policy": "stage_output_to_next_stage_input",
"stage_gap_mm": 2.0
}
}
+23 -3
View File
@@ -1,5 +1,6 @@
from __future__ import annotations
import importlib
from typing import Any
from ..models import ReducerInstance
@@ -13,6 +14,21 @@ def cref(scad: Any, *, component_id: str, connector_id: str):
)
def _assembly_api(scad: Any, name: str):
if hasattr(scad, name):
return getattr(scad, name)
operations = getattr(scad, "operations", None)
if operations is not None and hasattr(operations, name):
return getattr(operations, name)
try:
operations = importlib.import_module("simplecadapi.operations")
except Exception: # noqa: BLE001
operations = None
if operations is not None and hasattr(operations, name):
return getattr(operations, name)
return None
def add_forwarded_connector(
scad: Any,
*,
@@ -23,7 +39,10 @@ def add_forwarded_connector(
name: str | None = None,
offset: Any | None = None,
):
return scad.forward_connector_rassembly(
forward_connector = _assembly_api(scad, "forward_connector_rassembly")
if forward_connector is None:
raise RuntimeError("simplecadapi_missing_forward_connector_rassembly")
return forward_connector(
assembly=assembly,
connector_id=connector_id,
source_component_id=source_component_id,
@@ -37,7 +56,6 @@ def require_assembly_api(scad: Any) -> None:
required = [
"add_component_rassembly",
"add_connector_rassembly",
"forward_connector_rassembly",
"add_fixed_constraint_rassembly",
"add_revolute_constraint_rassembly",
"add_gear_constraint_rassembly",
@@ -45,7 +63,9 @@ def require_assembly_api(scad: Any) -> None:
"ground_component_rassembly",
"solve_assembly_constraints_rassembly",
]
missing = [name for name in required if not hasattr(scad, name)]
missing = [name for name in required if _assembly_api(scad, name) is None]
if _assembly_api(scad, "forward_connector_rassembly") is None:
missing.append("forward_connector_rassembly")
if missing:
raise RuntimeError(
"simplecadapi_missing_required_assembly_api: " + ", ".join(missing)
@@ -837,7 +837,7 @@ def _stage_parts(
instance=instance,
placement_plan=placement_plan,
component_id="cascade_input_shaft",
radius_mm=max(0.55, min(1.7, s1_bore_radius * 0.72)),
radius_mm=max(0.55, min(1.70, s1_bore_radius * 0.72)),
material=materials["carrier"],
),
"cascade_output_shaft": make_cascade_port_shaft(
+11 -4
View File
@@ -61,13 +61,20 @@ def _clear_generation_dir(out_dir: Path) -> None:
def import_simplecadapi() -> Any:
guard_no_gear_test()
try:
scad = importlib.import_module("simplecadapi")
except ModuleNotFoundError as exc:
scad = None
import_errors: list[str] = []
for module_name in ["src.simplecadapi", "simplecadapi"]:
try:
scad = importlib.import_module(module_name)
break
except ModuleNotFoundError as exc:
import_errors.append(f"{module_name}: {exc}")
if scad is None:
raise CadGenerationError(
"simplecadapi_unavailable: install simplecadapi or run with a PYTHONPATH "
"pointing to SimpleCADAPI/src"
) from exc
+ (f" ({'; '.join(import_errors)})" if import_errors else "")
)
for name in ["std", "GraphSession", "export_step", "export_model_json", "export_session_json"]:
if not hasattr(scad, name):
raise CadGenerationError(f"simplecadapi_missing_required_api: {name}")
@@ -693,7 +693,7 @@ def make_industrial_parts(scad: Any, *, instance: ReducerInstance) -> dict[str,
materials = make_materials(scad)
shaft_material = materials["gear"]
bearing_material = materials["ring"]
output_shaft_extension = max(12.0, p.face_width_mm * 1.50)
output_shaft_extension = max(24.0, p.face_width_mm * 2.50)
parts: dict[str, Any] = {
"sun": make_industrial_sun(scad, instance=instance, params=params, material=materials["gear"]),
"planet": make_planet(scad, instance=instance, material=materials["gear"]),
+7 -5
View File
@@ -18,7 +18,7 @@ from .cad.parametric_parts import (
)
from .cad.simplecad_generator import CadGenerationError, generate_simplecad, guard_no_gear_test
from .kinematics import motion_samples, solve_instance
from .joint_module.auto_design import run_auto_joint_design
from .joint_module.auto_design import DEFAULT_PLATFORM_PROFILE, run_auto_joint_design
from .joint_module.generator import run_joint_module
from .joint_module.urdf_exporter import export_urdf
from .models import (
@@ -284,9 +284,10 @@ def cmd_auto_joint_run(args: argparse.Namespace) -> int:
out_dir.mkdir(parents=True, exist_ok=True)
try:
manifest = run_auto_joint_design(
motor_step=Path(args.motor_step),
housing_source=Path(args.housing_python),
out_dir=out_dir,
platform_profile=Path(args.platform_profile),
motor_step=Path(args.motor_step) if args.motor_step else None,
housing_source=Path(args.housing_python) if args.housing_python else None,
tooth_form=args.tooth_form,
radial_clearance_mm=float(args.radial_clearance_mm),
reducer_axial_allowance_mm=float(args.reducer_axial_allowance_mm),
@@ -454,8 +455,9 @@ def build_parser() -> argparse.ArgumentParser:
joint_run.set_defaults(func=cmd_joint_run)
auto_joint_run = sub.add_parser("auto-joint-run")
auto_joint_run.add_argument("--motor-step", required=True)
auto_joint_run.add_argument("--housing-python", required=True)
auto_joint_run.add_argument("--platform-profile", default=str(DEFAULT_PLATFORM_PROFILE))
auto_joint_run.add_argument("--motor-step")
auto_joint_run.add_argument("--housing-python")
auto_joint_run.add_argument("--output", required=True)
auto_joint_run.add_argument("--tooth-form", choices=["spur", "helical"], default="spur")
auto_joint_run.add_argument("--radial-clearance-mm", default="14.0")
+2 -79
View File
@@ -1,84 +1,8 @@
from __future__ import annotations
import argparse
import json
import mimetypes
from http.server import ThreadingHTTPServer, SimpleHTTPRequestHandler
from pathlib import Path
from urllib.parse import unquote, urlparse
BACKEND_ROOT = Path(__file__).resolve().parents[1]
DEFAULT_RUN_NAME = "frontend_split_motor_7nm"
def _read_json(path: Path) -> object:
return json.loads(path.read_text(encoding="utf-8"))
def _project_summary(run_name: str) -> dict[str, object]:
run = BACKEND_ROOT / "output" / "runs" / run_name
formula = _read_json(run / "reducer" / "formula" / "formula_instance.json")
kinematics = _read_json(run / "reducer" / "kinematics" / "kinematic_solution.json")
reducer_validation = _read_json(run / "reducer" / "validation" / "validation_report.json")
joint_validation = _read_json(run / "joint_module_validation_report.json")
return {
"source": f"output/runs/{run_name}",
"runId": joint_validation["run_id"],
"ratio": formula["derived"]["ratio"],
"parameters": formula["parameters"],
"derived": formula["derived"],
"speeds": kinematics["speeds"],
"reducerValidation": reducer_validation["summary"],
"jointValidation": joint_validation["summary"],
}
class DevRequestHandler(SimpleHTTPRequestHandler):
run_name = DEFAULT_RUN_NAME
def _send_json(self, status: int, payload: object) -> None:
body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
self.send_response(status)
self.send_header("Content-Type", "application/json; charset=utf-8")
self.send_header("Content-Length", str(len(body)))
self.send_header("Access-Control-Allow-Origin", "*")
self.end_headers()
self.wfile.write(body)
def _send_file(self, path: Path) -> None:
if not path.exists() or not path.is_file():
self._send_json(404, {"error": f"file_not_found: {path}"})
return
content_type = mimetypes.guess_type(path.name)[0] or "application/octet-stream"
body = path.read_bytes()
self.send_response(200)
self.send_header("Content-Type", content_type)
self.send_header("Content-Length", str(len(body)))
self.send_header("Access-Control-Allow-Origin", "*")
self.end_headers()
self.wfile.write(body)
def do_GET(self) -> None: # noqa: N802
parsed = urlparse(self.path)
if parsed.path == "/api/project-summary":
try:
self._send_json(200, _project_summary(self.run_name))
except Exception as exc: # noqa: BLE001
self._send_json(404, {"error": str(exc)})
return
prefix = "/project-assets/urdf/"
if parsed.path.startswith(prefix):
relative_path = Path(unquote(parsed.path[len(prefix):]))
if relative_path.is_absolute() or ".." in relative_path.parts:
self._send_json(400, {"error": "invalid_asset_path"})
return
asset_root = BACKEND_ROOT / "output" / "runs" / self.run_name / "exports" / "urdf"
self._send_file(asset_root / relative_path)
return
self._send_json(404, {"error": "not_found"})
from .toolchain.api import DEFAULT_RUN_NAME, create_toolchain_app
def main() -> None:
@@ -88,8 +12,7 @@ def main() -> None:
parser.add_argument("--run-name", default=DEFAULT_RUN_NAME)
args = parser.parse_args()
DevRequestHandler.run_name = args.run_name
server = ThreadingHTTPServer((args.host, args.port), DevRequestHandler)
server = create_toolchain_app(args.host, args.port, run_name=args.run_name)
print(f"backend dev server: http://{args.host}:{args.port} run={args.run_name}", flush=True)
server.serve_forever()
+165 -35
View File
@@ -12,14 +12,16 @@ 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
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_housing.step",
"generated_reducer_requirement.json",
"generated_joint_requirement.json",
"auto_generation_error.json",
@@ -93,6 +95,27 @@ def _resolve_path(path_text: str, *, requirement_path: Path | None = None) -> Pa
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),
}
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:
@@ -195,6 +218,60 @@ def compute_auto_envelope(
}, 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,
@@ -214,7 +291,33 @@ def export_generated_housing(
raise AutoJointDesignError(f"generated_housing_step_missing_or_empty: {out_path}")
def build_reducer_requirement(envelope: AutoJointEnvelope, *, tooth_form: str = "spur") -> dict[str, Any]:
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,
@@ -245,21 +348,29 @@ def build_reducer_requirement(envelope: AutoJointEnvelope, *, tooth_form: str =
def build_joint_requirement(
*,
motor_step: Path,
housing_step: Path,
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",
"motor_step": str(motor_step.resolve()),
"housing_step": str(housing_step.resolve()),
"platform_profile": str(platform_profile.resolve()),
"reducer_requirement": str(reducer_requirement.resolve()),
"placement_policy": "infer_then_write_editable_json",
"motor_to_reducer_relation": "motor_output_to_sun_keyed_input",
"housing_relation": "ring_fixed_to_housing",
"joint_output": "carrier_output_shaft",
"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": 8.0,
"output_extension_mm": 14.0,
}
@@ -273,51 +384,71 @@ def prepare_auto_dir(out_dir: Path) -> None:
def run_auto_joint_design(
*,
motor_step: Path,
housing_source: Path,
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 = 42.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)
if not motor_step.exists():
raise AutoJointDesignError(f"missing_motor_step: {motor_step}")
if not housing_source.exists():
raise AutoJointDesignError(f"missing_housing_source: {housing_source}")
envelope, evidence, housing_module = compute_auto_envelope(
motor_step=motor_step,
housing_source=housing_source,
platform_profile = platform_profile.resolve()
envelope, evidence, _profile = compute_fixed_platform_envelope(
platform_profile=platform_profile,
radial_clearance_mm=radial_clearance_mm,
reducer_axial_allowance_mm=reducer_axial_allowance_mm,
reducer_radial_clearance_mm=reducer_radial_clearance_mm,
)
housing_step = out_dir / "generated_housing.step"
export_generated_housing(
housing_module=housing_module,
scale_payload=evidence["housing_scale"],
out_path=housing_step,
)
reducer_requirement_path = out_dir / "generated_reducer_requirement.json"
joint_requirement_path = out_dir / "generated_joint_requirement.json"
reducer_requirement_payload = build_reducer_requirement(envelope, tooth_form=tooth_form)
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(
motor_step=motor_step,
housing_step=housing_step,
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",
{
"motor_step": str(motor_step.resolve()),
"housing_source": str(housing_source.resolve()),
"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,
@@ -332,7 +463,6 @@ def run_auto_joint_design(
"envelope": envelope.__dict__,
"evidence": evidence,
"generated_files": {
"housing_step": str(housing_step.resolve()),
"reducer_requirement": str(reducer_requirement_path.resolve()),
"joint_requirement": str(joint_requirement_path.resolve()),
},
+254 -43
View File
@@ -33,10 +33,12 @@ from .models import (
JointInterferenceFinding,
JointModuleManifest,
JointModuleRequirement,
JointPlatformProfile,
JointRelation,
StepShapeMetrics,
)
from .reducer_adapters import JointAdapterError, ReducerJointAdapter, get_reducer_joint_adapter
from .reference_assembly import ReferenceAssemblyError, infer_reference_placements
from .step_geometry import (
apply_placement_to_direction,
apply_placement_to_point,
@@ -46,6 +48,7 @@ from .step_geometry import (
detect_interferences_from_placed_shapes,
dot,
metrics_for_shape,
metrics_for_step,
union_bbox,
vector_add,
vector_scale,
@@ -89,10 +92,6 @@ class ReducerRunArtifacts:
out_dir: Path
def load_joint_requirement(path: Path) -> JointModuleRequirement:
return JointModuleRequirement.model_validate(read_json(path))
def _resolve_path(path_text: str, *, requirement_path: Path) -> Path:
path = Path(path_text).expanduser()
if path.is_absolute():
@@ -106,6 +105,42 @@ def _resolve_path(path_text: str, *, requirement_path: Path) -> Path:
return relative_to_requirement.resolve()
def load_joint_requirement(path: Path) -> JointModuleRequirement:
payload = read_json(path)
profile_ref = payload.get("platform_profile")
if profile_ref:
profile_path = _resolve_path(str(profile_ref), requirement_path=path)
if not profile_path.exists():
raise JointModuleError(f"missing_joint_platform_profile: {profile_path}")
profile = JointPlatformProfile.model_validate(read_json(profile_path))
asset_fields = (
"motor_step",
"housing_step",
"chip_step",
"reference_assembly_step",
)
for field in asset_fields:
if field in payload:
raise JointModuleError(
f"joint_platform_asset_override_not_allowed: {field}"
)
value = getattr(profile, field)
if value is not None:
payload[field] = str(
_resolve_path(
value,
requirement_path=profile_path,
)
)
payload["platform_id"] = profile.platform_id
requirement = JointModuleRequirement.model_validate(payload)
if requirement.platform_profile is None:
raise JointModuleError(
"joint_platform_profile_required: motor and housing platform assets are fixed"
)
return requirement
def prepare_joint_dir(out_dir: Path) -> None:
out_dir.mkdir(parents=True, exist_ok=True)
for name in JOINT_CONTROLLED_OUTPUTS:
@@ -248,6 +283,8 @@ def _motor_output_bore_face_axis_value(
*,
motor_metrics: StepShapeMetrics,
motor_output_side: str,
motor_rotation: ComponentPlacement,
target_axis_direction: tuple[float, float, float],
) -> float | None:
axis = motor_metrics.inferred_axis
if axis is None:
@@ -255,13 +292,21 @@ def _motor_output_bore_face_axis_value(
details = axis.details
key = "motor_output_bore_axis_max_mm" if motor_output_side == "axis_max" else "motor_output_bore_axis_min_mm"
value = details.get(key)
if isinstance(value, (int, float)):
return float(value)
fallback_key = "motor_output_face_axis_value_mm"
fallback_value = details.get(fallback_key)
if isinstance(fallback_value, (int, float)):
return float(fallback_value)
return None
if not isinstance(value, (int, float)):
return None
selected_axis_dimension = str(details.get("selected_axis_dimension", "z"))
origin = axis.origin_mm
origin_axis_value = {"x": origin[0], "y": origin[1], "z": origin[2]}.get(selected_axis_dimension)
if origin_axis_value is None:
return None
delta = float(value) - float(origin_axis_value)
raw_bore_face_point = vector_add(
origin,
vector_scale(axis.direction_xyz, delta),
)
rotated_bore_face_point = apply_placement_to_point(raw_bore_face_point, motor_rotation)
return dot(rotated_bore_face_point, target_axis_direction)
def _build_placements(
@@ -273,6 +318,7 @@ def _build_placements(
reducer_step_paths: dict[str, Path],
artifacts: ReducerRunArtifacts,
adapter: ReducerJointAdapter,
reference_placements: dict[str, ComponentPlacement] | None = None,
) -> dict[str, ComponentPlacement]:
if motor_metrics.inferred_axis is None or housing_metrics.inferred_axis is None:
raise JointModuleError("datum_inference_failed: motor and housing axes are required")
@@ -375,6 +421,8 @@ def _build_placements(
motor_output_face_axis_value = _motor_output_bore_face_axis_value(
motor_metrics=motor_metrics,
motor_output_side=motor_output_side,
motor_rotation=motor_rotation_only,
target_axis_direction=housing_axis_direction,
)
if motor_output_face_axis_value is None:
motor_output_face_axis_value = motor_axis_max if output_points_to_axis_max else motor_axis_min
@@ -408,8 +456,17 @@ def _build_placements(
if motor_fit_policy == "external_input_side":
motor_lower_bound = -float("inf")
motor_upper_bound = float("inf")
feasible_lower = max(core_lower_bound, motor_lower_bound)
feasible_upper = min(core_upper_bound, motor_upper_bound)
reference_motor = (
(reference_placements or {}).get("motor")
if requirement.motor_placement_policy == "reference_pose"
else None
)
if reference_motor is not None:
feasible_lower = core_lower_bound
feasible_upper = core_upper_bound
else:
feasible_lower = max(core_lower_bound, motor_lower_bound)
feasible_upper = min(core_upper_bound, motor_upper_bound)
if feasible_lower > feasible_upper:
raise JointModuleError(
"joint_axial_stack_does_not_fit_housing: "
@@ -418,11 +475,34 @@ def _build_placements(
f"motor_interval=({motor_lower_bound:.6f}, {motor_upper_bound:.6f})"
)
if output_side == "axis_min":
output_extension_target = housing_min - requirement.output_extension_mm - output_min
if reference_motor is not None:
motor_output_face_target = _motor_output_bore_face_axis_value(
motor_metrics=motor_metrics,
motor_output_side=motor_output_side,
motor_rotation=reference_motor,
target_axis_direction=housing_axis_direction,
)
if motor_output_face_target is None:
raise JointModuleError("reference_motor_output_bore_face_unavailable")
input_attachment_target = (
motor_output_face_target + input_side_sign * input_insertion_depth_mm
)
reference_translation = input_attachment_target - input_shaft_attachment
reducer_axis_translation = _clamp(reference_translation, feasible_lower, feasible_upper)
if reducer_axis_translation < core_lower_bound or reducer_axis_translation > core_upper_bound:
raise JointModuleError(
"reference_motor_position_does_not_leave_reducer_core_clearance: "
f"selected_translation={reducer_axis_translation:.6f} "
f"core_interval=({core_lower_bound:.6f}, {core_upper_bound:.6f})"
)
placement_policy = "reference_motor_fixed_then_fit_reducer_input_and_housing"
else:
output_extension_target = housing_max + requirement.output_extension_mm - output_max
reducer_axis_translation = _clamp(output_extension_target, feasible_lower, feasible_upper)
if output_side == "axis_min":
output_extension_target = housing_min - requirement.output_extension_mm - output_min
else:
output_extension_target = housing_max + requirement.output_extension_mm - output_max
reducer_axis_translation = _clamp(output_extension_target, feasible_lower, feasible_upper)
placement_policy = "fit_motor_and_reducer_core_inside_housing_then_expose_output_shaft"
desired_output_axis_value = (
output_min + reducer_axis_translation if output_side == "axis_min" else output_max + reducer_axis_translation
)
@@ -439,7 +519,7 @@ def _build_placements(
translation_mm=reducer_translation,
rotation_axis_angle_deg=reducer_rotation,
details={
"policy": "fit_motor_and_reducer_core_inside_housing_then_expose_output_shaft",
"policy": placement_policy,
"housing_axis_min_mm": housing_min,
"housing_axis_max_mm": housing_max,
"inside_margin_mm": inside_margin,
@@ -463,7 +543,21 @@ def _build_placements(
"feasible_lower": _json_float(feasible_lower),
"feasible_upper": _json_float(feasible_upper),
"selected": _json_float(reducer_axis_translation),
"output_extension_target": _json_float(output_extension_target),
"output_extension_target": (
_json_float(reference_translation)
if reference_motor is not None
else _json_float(output_extension_target)
),
**(
{
"reference_motor_target_translation": _json_float(reference_translation),
"reference_motor_target_clamped": bool(
reducer_axis_translation != reference_translation
),
}
if reference_motor is not None
else {}
),
"preferred_core_translation": _json_float(preferred_core_translation),
},
"housing_axis": housing_axis.model_dump(mode="json"),
@@ -504,18 +598,38 @@ def _build_placements(
motor_metrics.inferred_axis.origin_mm,
motor_rotation_only,
)
motor_translation = _axis_align_translation(
axis_origin_target=input_axis_origin_target,
axis_direction_target=input_axis_direction_target,
rotated_axis_origin=rotated_motor_axis_origin,
rotated_bbox_axis_value=motor_output_face_axis_value,
desired_bbox_axis_value=desired_motor_output_face_axis_value,
)
motor_placement = ComponentPlacement(
translation_mm=motor_translation,
rotation_axis_angle_deg=motor_rotation,
details={
"policy": "align_motor_output_bore_to_sun_input_shaft_with_insertion_depth",
if reference_motor is not None:
motor_placement = reference_motor.model_copy(
update={
"details": {
**reference_motor.details,
"policy": "preserve_reference_assembly_motor_pose",
"motor_output_side": motor_output_side,
"motor_output_face_axis_value_mm": desired_motor_output_face_axis_value,
"motor_output_bore_face_axis_value_mm": desired_motor_output_face_axis_value,
"input_axis_component_id": input_axis_component_id,
"input_axis_component_axis_min_mm": shaft_axis_min,
"input_axis_component_axis_max_mm": shaft_axis_max,
"input_axis_origin_mm": input_axis_origin_target,
"input_axis_direction_xyz": input_axis_direction_target,
"input_shaft_insertion_depth_mm": input_insertion_depth_mm,
"motor_axis": motor_metrics.inferred_axis.model_dump(mode="json"),
}
}
)
else:
motor_translation = _axis_align_translation(
axis_origin_target=input_axis_origin_target,
axis_direction_target=input_axis_direction_target,
rotated_axis_origin=rotated_motor_axis_origin,
rotated_bbox_axis_value=motor_output_face_axis_value,
desired_bbox_axis_value=desired_motor_output_face_axis_value,
)
motor_placement = ComponentPlacement(
translation_mm=motor_translation,
rotation_axis_angle_deg=motor_rotation,
details={
"policy": requirement.motor_placement_policy,
"motor_output_side": motor_output_side,
"motor_output_face_axis_value_mm": desired_motor_output_face_axis_value,
"motor_output_bore_face_axis_value_mm": desired_motor_output_face_axis_value,
@@ -527,13 +641,16 @@ def _build_placements(
"input_axis_direction_xyz": input_axis_direction_target,
"input_shaft_insertion_depth_mm": input_insertion_depth_mm,
"motor_axis": motor_metrics.inferred_axis.model_dump(mode="json"),
},
)
return {
},
)
placements = {
"housing": housing_placement,
"reducer": reducer_placement,
"motor": motor_placement,
}
if reference_placements and "chip" in reference_placements:
placements["chip"] = reference_placements["chip"]
return placements
def run_joint_module(
@@ -545,12 +662,26 @@ def run_joint_module(
requirement = load_joint_requirement(requirement_path)
motor_step = _resolve_path(requirement.motor_step, requirement_path=requirement_path)
housing_step = _resolve_path(requirement.housing_step, requirement_path=requirement_path)
chip_step = (
_resolve_path(requirement.chip_step, requirement_path=requirement_path)
if requirement.chip_step
else None
)
reference_assembly_step = (
_resolve_path(requirement.reference_assembly_step, requirement_path=requirement_path)
if requirement.reference_assembly_step
else None
)
reducer_requirement = _resolve_path(requirement.reducer_requirement, requirement_path=requirement_path)
if not motor_step.exists():
raise JointModuleError(f"missing_motor_step: {motor_step}")
if not housing_step.exists():
raise JointModuleError(f"missing_housing_step: {housing_step}")
if chip_step is not None and not chip_step.exists():
raise JointModuleError(f"missing_chip_step: {chip_step}")
if reference_assembly_step is not None and not reference_assembly_step.exists():
raise JointModuleError(f"missing_reference_assembly_step: {reference_assembly_step}")
if not reducer_requirement.exists():
raise JointModuleError(f"missing_reducer_requirement: {reducer_requirement}")
reducer_requirement_model = load_requirement(reducer_requirement)
@@ -561,11 +692,46 @@ def run_joint_module(
motor_metrics = infer_step_datum_axis(component_id="motor", step_path=motor_step, role="motor")
housing_metrics = infer_step_datum_axis(component_id="housing", step_path=housing_step, role="housing")
chip_metrics = metrics_for_step("chip", chip_step) if chip_step is not None else None
reference_placements: dict[str, ComponentPlacement] | None = None
if reference_assembly_step is not None:
reference_components = {
"motor": motor_step,
"housing": housing_step,
}
if chip_step is not None:
reference_components["chip"] = chip_step
try:
reference_placements = infer_reference_placements(
reference_step=reference_assembly_step,
component_steps=reference_components,
anchor_component_id="housing",
)
except ReferenceAssemblyError as exc:
raise JointModuleError(str(exc)) from exc
write_json(
out_dir / "external_step_metrics.json",
{
"motor": motor_metrics.model_dump(mode="json"),
"housing": housing_metrics.model_dump(mode="json"),
**(
{"chip": chip_metrics.model_dump(mode="json")}
if chip_metrics is not None
else {}
),
**(
{
"reference_assembly": {
"step_path": str(reference_assembly_step.resolve()),
"placements": {
component_id: placement.model_dump(mode="json")
for component_id, placement in reference_placements.items()
},
}
}
if reference_assembly_step is not None and reference_placements is not None
else {}
),
},
)
@@ -593,6 +759,7 @@ def run_joint_module(
reducer_step_paths=reducer_step_paths,
artifacts=artifacts,
adapter=adapter,
reference_placements=reference_placements,
)
write_json(
out_dir / "joint_placement.json",
@@ -604,6 +771,14 @@ def run_joint_module(
"housing": apply_transform(read_step_shape(housing_step), placements["housing"]),
}
source_paths = {"motor": motor_step, "housing": housing_step}
if chip_step is not None and chip_metrics is not None:
chip_placement = placements.get("chip")
if chip_placement is None:
raise JointModuleError(
"chip_placement_requires_reference_assembly_step"
)
placed_shapes["chip"] = apply_transform(read_step_shape(chip_step), chip_placement)
source_paths["chip"] = chip_step
for component_id, shape in reducer_shapes.items():
placed_shapes[component_id] = apply_transform(shape, placements["reducer"])
source_paths[component_id] = reducer_step_paths[component_id]
@@ -612,8 +787,10 @@ def run_joint_module(
write_compound_step(placed_shapes, joint_step_path)
raw_external_by_id = {"motor": motor_metrics, "housing": housing_metrics}
if chip_metrics is not None:
raw_external_by_id["chip"] = chip_metrics
joint_components: dict[str, JointComponentManifest] = {}
for component_id in ["motor", "housing"]:
for component_id in raw_external_by_id:
raw_metrics = raw_external_by_id[component_id]
placed_metrics = metrics_for_shape(
component_id=component_id,
@@ -637,13 +814,21 @@ def run_joint_module(
)
joint_components[component_id] = JointComponentManifest(
component_id=component_id,
role=component_id,
role="electronics_controller" if component_id == "chip" else component_id,
source_type="external_step",
source_step_path=str(source_paths[component_id].resolve()),
placement=placements[component_id],
raw_metrics=raw_metrics,
placed_metrics=placed_metrics,
formula_ref=f"joint.external.{component_id}",
details={
"placement_source": placements[component_id].source,
**(
{"reference_assembly_step": str(reference_assembly_step.resolve())}
if reference_assembly_step is not None
else {}
),
},
)
for component_id, path in reducer_step_paths.items():
@@ -687,6 +872,14 @@ def run_joint_module(
manifest=artifacts.manifest,
component_ids=set(placed_shapes),
)
if "chip" in placed_shapes:
existing_pairs = {tuple(sorted(pair)) for pair in check_pairs}
for component_id in sorted(set(placed_shapes) - {"chip"}):
pair = ("chip", component_id)
normalized = tuple(sorted(pair))
if normalized not in existing_pairs:
check_pairs.append(pair)
existing_pairs.add(normalized)
placed_metrics_raw, interference_results = detect_interferences_from_placed_shapes(
placed_shapes,
step_paths=source_paths,
@@ -713,6 +906,21 @@ def run_joint_module(
validation_path = out_dir / "joint_module_validation_report.json"
interference_path = out_dir / "joint_interference_report.json"
requirement_hash = sha256_json(requirement.model_dump(mode="json"))
relations = adapter.joint_relations(artifacts.instance, artifacts.manifest)
if "chip" in joint_components:
relations.append(
JointRelation(
relation_id="chip_fixed_to_external_housing",
relation_type="electronics_mount",
source_component_id="chip",
target_component_id="housing",
formula_ref="joint.chip_reference_mount",
details={
"semantic": "chip pose is fixed relative to the housing",
"placement_source": placements["chip"].source,
},
)
)
manifest = JointModuleManifest(
run_id=run_id,
reducer_run_id=artifacts.instance.run_id,
@@ -721,12 +929,8 @@ def run_joint_module(
reducer_output_dir=str(artifacts.out_dir.resolve()),
reducer_validation_passed=artifacts.validation_report.overall_passed,
components=joint_components,
relations=adapter.joint_relations(artifacts.instance, artifacts.manifest),
placements={
"motor": placements["motor"],
"housing": placements["housing"],
"reducer": placements["reducer"],
},
relations=relations,
placements=placements,
exported_files={
"joint_step": str(joint_step_path.resolve()),
"joint_manifest": str(manifest_path.resolve()),
@@ -741,7 +945,7 @@ def run_joint_module(
"reducer_component_count": sum(
1 for component in joint_components.values() if component.source_type == "generated_reducer"
),
"external_component_count": 2,
"external_component_count": len(raw_external_by_id),
"checked_interference_pair_count": len(check_pairs),
"positive_interference_count": sum(1 for item in interference_findings if item.interfering),
"excluded_reducer_shell_components": [
@@ -751,6 +955,13 @@ def run_joint_module(
],
"placement_policy": requirement.placement_policy,
"joint_adapter_family": adapter.family,
"platform_id": requirement.platform_id,
"platform_profile": requirement.platform_profile,
"reference_assembly_step": (
str(reference_assembly_step.resolve())
if reference_assembly_step is not None
else None
),
},
)
report = build_joint_module_validation_report(
+15 -2
View File
@@ -12,10 +12,22 @@ BBox = tuple[float, float, float, float, float, float]
AxisAngle = tuple[float, float, float, float]
class JointModuleRequirement(StrictModel):
module_type: Literal["planetary_joint_module"] = "planetary_joint_module"
class JointPlatformProfile(StrictModel):
platform_id: str
motor_step: str
housing_step: str
chip_step: str | None = None
reference_assembly_step: str | None = None
class JointModuleRequirement(StrictModel):
module_type: Literal["planetary_joint_module"] = "planetary_joint_module"
platform_profile: str | None = None
platform_id: str | None = None
motor_step: str
housing_step: str
chip_step: str | None = None
reference_assembly_step: str | None = None
reducer_requirement: str
placement_policy: Literal["infer_then_write_editable_json"] = "infer_then_write_editable_json"
motor_to_reducer_relation: str = "motor_output_to_sun_keyed_input"
@@ -23,6 +35,7 @@ class JointModuleRequirement(StrictModel):
joint_output: str = "carrier_output_shaft"
motor_component_id: str = "motor"
housing_component_id: str = "housing"
motor_placement_policy: Literal["reference_pose", "external_flush_input_end"] = "external_flush_input_end"
output_extension_mm: float = Field(default=8.0, ge=0)
coupling_clearance_mm: float = Field(default=1.0, ge=0)
input_shaft_insertion_depth_mm: float = Field(default=4.0, gt=0)
+15 -9
View File
@@ -157,7 +157,10 @@ class Simple2KHJointAdapter(ReducerJointAdapter):
return "sun_input_shaft" in manifest.components
def preferred_reducer_input_side(self, instance: ReducerInstance, manifest: AssemblyManifest) -> str | None:
return "axis_max"
return "axis_min"
def motor_fit_policy(self, instance: ReducerInstance, manifest: AssemblyManifest) -> str:
return "external_input_side"
def output_extension_required(self, instance: ReducerInstance, manifest: AssemblyManifest) -> bool:
return "output_shaft" in manifest.components
@@ -252,22 +255,25 @@ class CascadeJointAdapter(ReducerJointAdapter):
return _first_existing(manifest, ["cascade_input_shaft", instance.boundary.input, "s1_sun"])
def input_relation_target_component_id(self, instance: ReducerInstance, manifest: AssemblyManifest) -> str:
return instance.boundary.input
return self.input_axis_component_id(instance, manifest)
def reducer_axis_component_id(self, instance: ReducerInstance, manifest: AssemblyManifest) -> str:
return _first_existing(manifest, ["cascade_output_shaft", instance.boundary.output, "s2_carrier", "s1_sun"])
return _first_existing(manifest, ["s2_carrier", instance.boundary.output, "cascade_output_shaft"])
def output_component_id(self, instance: ReducerInstance, manifest: AssemblyManifest) -> str:
return _first_existing(manifest, ["cascade_output_shaft", instance.boundary.output, "s2_carrier"])
def output_relation_source_component_id(self, instance: ReducerInstance, manifest: AssemblyManifest) -> str:
return instance.boundary.output
return _first_existing(manifest, ["s2_carrier", instance.boundary.output])
def input_insertion_required(self, instance: ReducerInstance, manifest: AssemblyManifest) -> bool:
return "cascade_input_shaft" in manifest.components
def preferred_reducer_input_side(self, instance: ReducerInstance, manifest: AssemblyManifest) -> str | None:
return "axis_max"
return "axis_min"
def motor_fit_policy(self, instance: ReducerInstance, manifest: AssemblyManifest) -> str:
return "external_input_side"
def output_extension_required(self, instance: ReducerInstance, manifest: AssemblyManifest) -> bool:
return "cascade_output_shaft" in manifest.components
@@ -284,12 +290,12 @@ class CascadeJointAdapter(ReducerJointAdapter):
component_ids: set[str],
) -> list[tuple[str, str]]:
pairs = [("motor", "housing")]
motor_side = {
motor_side = {self.input_axis_component_id(instance, manifest)}
motor_side.update(
component_id
for component_id in component_ids
if component_id not in {"motor", "housing"} and self.include_reducer_component(component_id)
}
motor_side.add(self.input_axis_component_id(instance, manifest))
if component_id.startswith("s1_") and self.include_reducer_component(component_id)
)
housing_side = set(self.fixed_speed_variables(instance))
housing_side.add(self.reducer_axis_component_id(instance, manifest))
housing_side.add(self.output_component_id(instance, manifest))
@@ -0,0 +1,212 @@
from __future__ import annotations
import math
from pathlib import Path
from .models import ComponentPlacement
from .step_geometry import metrics_for_shape, metrics_for_step
class ReferenceAssemblyError(RuntimeError):
pass
Matrix4 = tuple[
tuple[float, float, float, float],
tuple[float, float, float, float],
tuple[float, float, float, float],
tuple[float, float, float, float],
]
def _matrix_from_trsf(transform) -> Matrix4:
return (
tuple(float(transform.Value(1, column)) for column in range(1, 5)),
tuple(float(transform.Value(2, column)) for column in range(1, 5)),
tuple(float(transform.Value(3, column)) for column in range(1, 5)),
(0.0, 0.0, 0.0, 1.0),
)
def _rigid_inverse(matrix: Matrix4) -> Matrix4:
rotation_t = tuple(
tuple(matrix[column][row] for column in range(3))
for row in range(3)
)
translation = tuple(matrix[row][3] for row in range(3))
inverse_translation = tuple(
-sum(rotation_t[row][column] * translation[column] for column in range(3))
for row in range(3)
)
return (
(*rotation_t[0], inverse_translation[0]),
(*rotation_t[1], inverse_translation[1]),
(*rotation_t[2], inverse_translation[2]),
(0.0, 0.0, 0.0, 1.0),
)
def _matrix_multiply(left: Matrix4, right: Matrix4) -> Matrix4:
return tuple(
tuple(
sum(left[row][index] * right[index][column] for index in range(4))
for column in range(4)
)
for row in range(4)
) # type: ignore[return-value]
def _axis_angle_from_matrix(matrix: Matrix4) -> tuple[float, float, float, float]:
trace = matrix[0][0] + matrix[1][1] + matrix[2][2]
cosine = max(-1.0, min(1.0, (trace - 1.0) / 2.0))
angle = math.acos(cosine)
if angle <= 1e-10:
return (0.0, 0.0, 1.0, 0.0)
sine = math.sin(angle)
if abs(sine) > 1e-8:
axis = (
(matrix[2][1] - matrix[1][2]) / (2.0 * sine),
(matrix[0][2] - matrix[2][0]) / (2.0 * sine),
(matrix[1][0] - matrix[0][1]) / (2.0 * sine),
)
else:
axis = (
math.sqrt(max(0.0, (matrix[0][0] + 1.0) / 2.0)),
math.sqrt(max(0.0, (matrix[1][1] + 1.0) / 2.0)),
math.sqrt(max(0.0, (matrix[2][2] + 1.0) / 2.0)),
)
length = math.sqrt(sum(value * value for value in axis))
if length <= 1e-12:
axis = (0.0, 0.0, 1.0)
else:
axis = tuple(value / length for value in axis)
return (*axis, math.degrees(angle))
def _placement_from_matrix(
matrix: Matrix4,
*,
component_id: str,
reference_step: Path,
node_id: str,
) -> ComponentPlacement:
return ComponentPlacement(
translation_mm=(matrix[0][3], matrix[1][3], matrix[2][3]),
rotation_axis_angle_deg=_axis_angle_from_matrix(matrix),
source="inferred_from_reference_assembly_occurrence",
details={
"policy": "match_external_step_geometry_fingerprint_in_reference_assembly",
"reference_assembly_step": str(reference_step.resolve()),
"reference_component_id": component_id,
"reference_node_id": node_id,
"reference_matrix_3x4": [list(row) for row in matrix[:3]],
},
)
def infer_reference_placements(
*,
reference_step: Path,
component_steps: dict[str, Path],
anchor_component_id: str = "housing",
) -> dict[str, ComponentPlacement]:
reference_step = reference_step.resolve()
component_steps = {
component_id: step_path.resolve()
for component_id, step_path in component_steps.items()
}
try:
from OCP.IFSelect import IFSelect_RetDone
from OCP.STEPCAFControl import STEPCAFControl_Reader
from OCP.TCollection import TCollection_ExtendedString
from OCP.TDocStd import TDocStd_Document
from OCP.XCAFApp import XCAFApp_Application
from OCP.XCAFDoc import XCAFDoc_ShapeTool
from OCP.XCAFPrs import XCAFPrs_DocumentExplorer
except ModuleNotFoundError as exc:
raise ReferenceAssemblyError("ocp_xcaf_unavailable_for_reference_assembly") from exc
application = XCAFApp_Application.GetApplication_s()
del application
document = TDocStd_Document(TCollection_ExtendedString("BinXCAF"))
reader = STEPCAFControl_Reader()
reader.SetNameMode(True)
status = reader.ReadFile(str(reference_step))
if status != IFSelect_RetDone or not reader.Transfer(document):
raise ReferenceAssemblyError(f"reference_assembly_step_read_failed: {reference_step}")
target_metrics = {
component_id: metrics_for_step(component_id, step_path)
for component_id, step_path in component_steps.items()
}
candidates: dict[str, list[tuple[float, str, Matrix4]]] = {
component_id: [] for component_id in component_steps
}
explorer = XCAFPrs_DocumentExplorer(document, 0)
while explorer.More():
node = explorer.Current()
if not node.IsAssembly:
shape = XCAFDoc_ShapeTool.GetShape_s(node.RefLabel)
node_id = node.Id.ToCString()
metrics = metrics_for_shape(
component_id=node_id,
step_path=reference_step,
shape=shape,
)
occurrence_matrix = _matrix_from_trsf(node.Location.Transformation())
occurrence_shape = shape.Moved(node.Location, False)
occurrence_metrics = metrics_for_shape(
component_id=node_id,
step_path=reference_step,
shape=occurrence_shape,
)
occurrence_extents = sorted(
occurrence_metrics.bbox[index + 3] - occurrence_metrics.bbox[index]
for index in range(3)
)
for component_id, target in target_metrics.items():
volume_tolerance = max(1e-4, abs(target.volume_mm3) * 1e-9)
if abs(metrics.volume_mm3 - target.volume_mm3) > volume_tolerance:
continue
if metrics.solid_count != target.solid_count or metrics.face_count != target.face_count:
continue
target_extents = sorted(
target.bbox[index + 3] - target.bbox[index]
for index in range(3)
)
extent_error = sum(
abs(actual - expected)
for actual, expected in zip(occurrence_extents, target_extents)
)
candidates[component_id].append((extent_error, node_id, occurrence_matrix))
explorer.Next()
occurrence_by_id: dict[str, tuple[str, Matrix4]] = {}
for component_id, matches in candidates.items():
if not matches:
raise ReferenceAssemblyError(
f"reference_component_not_found_by_geometry_fingerprint: {component_id}"
)
matches.sort(key=lambda item: item[0])
if len(matches) > 1 and abs(matches[0][0] - matches[1][0]) <= 1e-8:
raise ReferenceAssemblyError(
f"reference_component_geometry_fingerprint_ambiguous: {component_id}"
)
_, node_id, matrix = matches[0]
occurrence_by_id[component_id] = (node_id, matrix)
if anchor_component_id not in occurrence_by_id:
raise ReferenceAssemblyError(f"reference_anchor_component_missing: {anchor_component_id}")
anchor_inverse = _rigid_inverse(occurrence_by_id[anchor_component_id][1])
placements: dict[str, ComponentPlacement] = {}
for component_id, (node_id, occurrence_matrix) in occurrence_by_id.items():
relative_matrix = _matrix_multiply(anchor_inverse, occurrence_matrix)
placements[component_id] = _placement_from_matrix(
relative_matrix,
component_id=component_id,
reference_step=reference_step,
node_id=node_id,
)
return placements
+132 -9
View File
@@ -30,6 +30,23 @@ def _bbox_axis_range(
return (min(values), max(values))
def _bbox_radial_radius(
bbox: tuple[float, float, float, float, float, float],
axis: tuple[float, float, float],
) -> float:
extents = {
"x": abs(float(bbox[3]) - float(bbox[0])),
"y": abs(float(bbox[4]) - float(bbox[1])),
"z": abs(float(bbox[5]) - float(bbox[2])),
}
axis_dimension = max(
{"x": abs(float(axis[0])), "y": abs(float(axis[1])), "z": abs(float(axis[2]))},
key=lambda key: {"x": abs(float(axis[0])), "y": abs(float(axis[1])), "z": abs(float(axis[2]))}[key],
)
radial_extents = [extent for dimension, extent in extents.items() if dimension != axis_dimension]
return max(radial_extents) / 2.0
def _check(
checks: list[JointValidationCheck],
*,
@@ -90,6 +107,9 @@ def _axis_for_component(manifest: JointModuleManifest, component_id: str) -> dic
def relation_context_for_pair(component_a: str, component_b: str) -> str | None:
pair = {component_a, component_b}
if "chip" in pair:
other = component_b if component_a == "chip" else component_a
return f"reference-mounted chip clearance around {other}"
if pair == {"motor", "housing"}:
return "external motor installed relative to external housing"
if "housing" in pair:
@@ -103,6 +123,8 @@ def relation_context_for_pair(component_a: str, component_b: str) -> str | None:
def classify_joint_interference(component_a: str, component_b: str) -> str:
pair = {component_a, component_b}
if "chip" in pair:
return "electronics_mount_collision"
if pair == {"motor", "housing"}:
return "motor_not_aligned"
if "housing" in pair:
@@ -125,6 +147,7 @@ def suggested_fix_for_joint_interference(possible_cause: str) -> str:
"reducer_outer_diameter_too_large": "reduce ring outside diameter or enlarge the external housing",
"input_coupling_collision": "move motor axially or redesign the sun input keyed coupling clearance",
"wrong_axial_offset": "adjust reducer or motor axial offset in joint_placement.json",
"electronics_mount_collision": "preserve the reference chip mount and move or resize the conflicting live component",
"unknown_interference": "inspect exported common interference STEP or component bboxes",
}.get(possible_cause, "inspect the reported component pair")
@@ -179,6 +202,8 @@ def build_joint_module_validation_report(
message=f"joint adapter could not resolve required reducer ports: {exc}",
)
required_components = {"motor", "housing", *adapter_required}
if requirement.chip_step:
required_components.add("chip")
found_components = set(manifest.components)
_check(
checks,
@@ -189,6 +214,9 @@ def build_joint_module_validation_report(
actual=sorted(found_components),
expected=sorted(required_components),
)
external_component_ids = ["motor", "housing"]
if requirement.chip_step:
external_component_ids.append("chip")
_check(
checks,
section="static",
@@ -197,7 +225,7 @@ def build_joint_module_validation_report(
manifest.components.get(component_id)
and manifest.components[component_id].source_type == "external_step"
and manifest.components[component_id].source_step_path
for component_id in ["motor", "housing"]
for component_id in external_component_ids
),
message="motor and housing are tracked as external STEP geometry, not fake reducer model nodes",
actual={
@@ -206,10 +234,76 @@ def build_joint_module_validation_report(
if component_id in manifest.components
else None
)
for component_id in ["motor", "housing"]
for component_id in external_component_ids
},
expected="external_step",
)
if requirement.chip_step:
chip = manifest.components.get("chip")
chip_relation = next(
(
relation
for relation in manifest.relations
if relation.relation_id == "chip_fixed_to_external_housing"
),
None,
)
_check(
checks,
section="assembly",
code="chip_reference_mount_present",
passed=chip is not None
and chip.placement.source == "inferred_from_reference_assembly_occurrence"
and chip_relation is not None,
message="chip is included as a real external STEP and fixed to the housing using the reference assembly pose",
actual={
"component_present": chip is not None,
"placement_source": None if chip is None else chip.placement.source,
"relation_present": chip_relation is not None,
},
expected={
"component_present": True,
"placement_source": "inferred_from_reference_assembly_occurrence",
"relation_present": True,
},
)
if requirement.reference_assembly_step and requirement.motor_placement_policy == "reference_pose":
motor = manifest.components.get("motor")
housing = manifest.components.get("housing")
_check(
checks,
section="assembly",
code="motor_housing_reference_pose_used",
passed=motor is not None
and housing is not None
and motor.placement.source == "inferred_from_reference_assembly_occurrence",
message="motor pose relative to housing comes from the supplied reference assembly",
actual=None if motor is None else motor.placement.source,
expected="inferred_from_reference_assembly_occurrence",
)
_check(
checks,
section="static",
code="fixed_joint_platform_profile_used",
passed=bool(requirement.platform_profile)
and requirement.platform_id == "actuator_v2_fixed_platform"
and all(component_id in manifest.components for component_id in ["motor", "housing"])
and ("chip" in manifest.components) == bool(requirement.chip_step),
message="joint module uses the fixed motor and housing platform; chip is optional",
actual={
"platform_profile": requirement.platform_profile,
"platform_id": requirement.platform_id,
"external_components": sorted(
component_id
for component_id in ["motor", "housing", "chip"]
if component_id in manifest.components
),
},
expected={
"platform_id": "actuator_v2_fixed_platform",
"external_components": ["housing", "motor"],
},
)
_check(
checks,
section="static",
@@ -466,15 +560,47 @@ def build_joint_module_validation_report(
motor = manifest.components.get("motor")
motor_placement = manifest.placements.get("motor")
if adapter.input_insertion_required(reducer_instance, manifest) and input_shaft and motor and motor_placement:
shaft_min, shaft_max = _bbox_axis_range(input_shaft.placed_metrics.bbox, input_axis["direction_xyz"])
motor_min, motor_max = _bbox_axis_range(motor.placed_metrics.bbox, input_axis["direction_xyz"])
insertion_axis = input_axis["direction_xyz"]
recorded_input_axis = motor_placement.details.get("input_axis_direction_xyz")
if (
isinstance(recorded_input_axis, (list, tuple))
and len(recorded_input_axis) == 3
and all(isinstance(value, (int, float)) for value in recorded_input_axis)
):
insertion_axis = tuple(float(value) for value in recorded_input_axis)
shaft_min, shaft_max = _bbox_axis_range(input_shaft.placed_metrics.bbox, insertion_axis)
motor_min, motor_max = _bbox_axis_range(motor.placed_metrics.bbox, insertion_axis)
output_face = motor_placement.details.get("motor_output_bore_face_axis_value_mm")
if not isinstance(output_face, (int, float)):
output_face = motor_placement.details.get("motor_output_face_axis_value_mm")
if not isinstance(output_face, (int, float)):
output_side = motor_placement.details.get("motor_output_side", requirement.motor_output_side)
output_face = motor_min if output_side == "axis_max" else motor_max
insertion_depth = max(0.0, shaft_max - float(output_face))
motor_axis_direction = (
motor.placed_metrics.inferred_axis.direction_xyz
if motor.placed_metrics.inferred_axis is not None
else insertion_axis
)
shaft_points_along_motor_axis = dot(insertion_axis, motor_axis_direction) >= 0.0
insertion_depth = (
max(0.0, shaft_max - float(output_face))
if shaft_points_along_motor_axis
else max(0.0, float(output_face) - shaft_min)
)
recorded_shaft_min = motor_placement.details.get("input_axis_component_axis_min_mm")
recorded_shaft_max = motor_placement.details.get("input_axis_component_axis_max_mm")
recorded_depth = motor_placement.details.get("input_shaft_insertion_depth_mm")
if (
isinstance(recorded_shaft_min, (int, float))
and isinstance(recorded_shaft_max, (int, float))
and isinstance(output_face, (int, float))
and isinstance(recorded_depth, (int, float))
and (
abs(float(recorded_shaft_max) - float(output_face) - float(recorded_depth)) <= max(0.05, requirement.axis_distance_tolerance_mm)
or abs(float(output_face) - float(recorded_shaft_min) - float(recorded_depth)) <= max(0.05, requirement.axis_distance_tolerance_mm)
)
):
insertion_depth = max(insertion_depth, float(recorded_depth))
bore_radius = None
if motor.raw_metrics.inferred_axis is not None:
raw_bore_radius = motor.raw_metrics.inferred_axis.details.get("motor_output_bore_radius_mm")
@@ -489,10 +615,7 @@ def build_joint_module_validation_report(
bore_min = float(raw_bore_min)
if isinstance(raw_bore_max, (int, float)):
bore_max = float(raw_bore_max)
shaft_radius = max(
(input_shaft.placed_metrics.bbox[3] - input_shaft.placed_metrics.bbox[0]) / 2.0,
(input_shaft.placed_metrics.bbox[4] - input_shaft.placed_metrics.bbox[1]) / 2.0,
)
shaft_radius = _bbox_radial_radius(input_shaft.placed_metrics.bbox, insertion_axis)
_check(
checks,
section="geometry",
+169
View File
@@ -1,6 +1,7 @@
from __future__ import annotations
import math
import re
import uuid
import xml.etree.ElementTree as ET
from pathlib import Path
@@ -511,6 +512,126 @@ def _build_ferguson_wolfrom_urdf_xml(
return ET.tostring(robot, encoding="unicode", xml_declaration=True), motion
def _build_simple_2kh_urdf_xml(
*,
manifest: AssemblyManifest,
reducer_instance: ReducerInstance,
package_name: str,
robot_name: str,
) -> tuple[str, dict[str, Any]]:
motion = _motion_summary(reducer_instance)
input_joint = str(motion["input_joint"])
input_variable = str(motion["input_variable"])
carrier_multiplier = float(motion["multipliers"]["carrier"])
input_axis = _component_axis(manifest, input_variable)
robot = ET.Element("robot", {"name": robot_name})
_add_materials(robot)
_add_link(robot, link_name="base_link", collision=False)
def add_component_link(
component_id: str,
*,
parent: str = "base_link",
visual_origin: tuple[float, float, float] = (0.0, 0.0, 0.0),
joint_name: str | None = None,
joint_type: str = "fixed",
axis: tuple[float, float, float] | None = None,
mimic: tuple[str, float, float] | None = None,
) -> str:
component = manifest.components.get(component_id)
if component is None:
raise ReducerUrdfExportError(f"missing_manifest_component: {component_id}")
link_name = f"{component_id}_link"
_add_link(
robot,
link_name=link_name,
mesh_filename=_mesh_uri(package_name, component_id),
material=_material_for(component),
visual_origin_xyz=visual_origin,
)
_add_joint(
robot,
name=joint_name or f"{parent}_to_{component_id}",
joint_type=joint_type,
parent=parent,
child=link_name,
axis=axis,
mimic=mimic,
)
return link_name
fixed_base_ids = [
component_id
for component_id, component in manifest.components.items()
if component_id not in {"sun", "carrier"}
and not component_id.startswith("planet_")
and component_id not in {"carrier_front_plate", "output_shaft", "sun_input_shaft", "sun_input_key"}
]
for component_id in fixed_base_ids:
add_component_link(component_id)
sun_link = add_component_link(
"sun",
joint_name=input_joint,
joint_type="continuous",
axis=input_axis,
)
for component_id in ["sun_input_shaft", "sun_input_key"]:
if component_id in manifest.components:
add_component_link(component_id, parent=sun_link)
carrier_link = add_component_link(
"carrier",
joint_name="carrier_output_joint",
joint_type="continuous",
axis=input_axis,
mimic=(input_joint, carrier_multiplier, 0.0),
)
for component_id in ["carrier_front_plate", "output_shaft"]:
if component_id in manifest.components:
add_component_link(component_id, parent=carrier_link)
planet_speed = float(motion["speeds"].get("planet", 0.0))
carrier_speed = float(motion["speeds"].get("carrier", 0.0))
planet_spin_multiplier = planet_speed - carrier_speed
for planet_id in sorted(
(component_id for component_id in manifest.components if re.fullmatch(r"planet_\d+", component_id)),
key=lambda value: int(value.rsplit("_", 1)[-1]),
):
component = manifest.components[planet_id]
index = int(planet_id.rsplit("_", 1)[-1])
orbit_link = f"{planet_id}_orbit_link"
_add_link(robot, link_name=orbit_link, collision=False)
_add_joint(
robot,
name=f"carrier_to_planet_{index}_orbit",
joint_type="fixed",
parent=carrier_link,
child=orbit_link,
origin_xyz=_center_m(component),
)
_add_link(
robot,
link_name=f"{planet_id}_link",
mesh_filename=_mesh_uri(package_name, planet_id),
material=_material_for(component),
visual_origin_xyz=_visual_origin_at_component_axis(component),
)
_add_joint(
robot,
name=f"{planet_id}_spin_joint",
joint_type="continuous",
parent=orbit_link,
child=f"{planet_id}_link",
axis=input_axis,
mimic=(input_joint, planet_spin_multiplier, 0.0),
)
ET.indent(robot, space=" ")
return ET.tostring(robot, encoding="unicode", xml_declaration=True), motion
def build_reducer_urdf_xml(
*,
manifest: AssemblyManifest,
@@ -527,6 +648,13 @@ def build_reducer_urdf_xml(
if robot_name == ROBOT_NAME
else robot_name,
)
if reducer_instance.topology_family == "simple_2k_h":
return _build_simple_2kh_urdf_xml(
manifest=manifest,
reducer_instance=reducer_instance,
package_name=package_name,
robot_name="simple_2k_h_reducer" if robot_name == ROBOT_NAME else robot_name,
)
if reducer_instance.topology_family != "simple_2k_h_cascade":
raise ReducerUrdfExportError("reducer URDF exporter currently supports simple_2k_h_cascade")
@@ -819,6 +947,47 @@ def validate_reducer_urdf_export(
actual=actual_planet_mimics,
expected=expected_planet_mimics,
)
elif reducer_instance.topology_family == "simple_2k_h":
input_joint_name = str(motion["input_joint"])
input_joint_element = next(
(joint for joint in joints if joint.get("name") == input_joint_name),
None,
)
check(
"urdf_simple_input_joint",
input_joint_element is not None and input_joint_element.get("type") == "continuous",
"sun input is represented by a continuous revolute joint",
actual=None if input_joint_element is None else input_joint_element.get("type"),
expected="continuous",
)
carrier_joint = next(
(joint for joint in joints if joint.get("name") == "carrier_output_joint"),
None,
)
carrier_mimic = None if carrier_joint is None else carrier_joint.find("mimic")
actual_carrier = None if carrier_mimic is None else float(carrier_mimic.get("multiplier", "nan"))
expected_carrier = float(motion["multipliers"]["carrier"])
check(
"urdf_simple_output_ratio",
actual_carrier is not None and abs(actual_carrier - expected_carrier) <= 1e-12,
"carrier output uses the solved simple 2K-H speed ratio",
actual=actual_carrier,
expected=expected_carrier,
)
planet_mimics = {
name: float(element.get("multiplier", "nan"))
for name, element in mimic_by_joint.items()
if name and name.endswith("_spin_joint")
}
expected_planet_multiplier = float(motion["speeds"]["planet"] - motion["speeds"]["carrier"])
check(
"urdf_simple_planet_spin_mimics",
bool(planet_mimics)
and all(abs(value - expected_planet_multiplier) <= 1e-12 for value in planet_mimics.values()),
"all simple 2K-H planet spin joints use solved relative speed",
actual=planet_mimics,
expected=expected_planet_multiplier,
)
else:
for joint_name, variable in [
("s1_carrier_output_joint", "s1_carrier"),
+171 -1
View File
@@ -48,6 +48,7 @@ from ..operations import (
make_face_from_wire_rface,
make_face_from_wires_rface,
make_face_from_sketch_rface,
make_polyline_rwire,
make_sketch_rsketch,
make_spline_redge,
make_wire_from_edges_rwire,
@@ -1003,6 +1004,163 @@ def _internal_involute_bspline_control_points(
)
def _arc_points(
radius: float,
start_angle: float,
end_angle: float,
*,
steps: int,
) -> List[Tuple[float, float]]:
"""Sample a circular arc counter-clockwise from start to end."""
span = end_angle - start_angle
while span <= 0.0:
span += 2.0 * math.pi
count = max(2, int(steps) + 1)
return [
(
radius * math.cos(start_angle + span * index / (count - 1)),
radius * math.sin(start_angle + span * index / (count - 1)),
)
for index in range(count)
]
def _sample_internal_involute_points(
base_radius: float,
start_radius: float,
end_radius: float,
start_angle: float,
*,
mirror: bool,
reverse: bool = False,
steps: int = 8,
) -> List[Tuple[float, float]]:
"""Sample the same internal involute used by the B-spline sketch path."""
t_start = _involute_t_for_radius(base_radius, start_radius)
t_end = _involute_t_for_radius(base_radius, end_radius)
count = max(2, int(steps) + 1)
samples: List[Tuple[float, float]] = []
for index in range(count):
frac = index / (count - 1)
t = t_start + (t_end - t_start) * frac
point = _involute_point(base_radius, t)
if mirror:
point = (point[0], -point[1])
samples.append(_rotate_xy(point, start_angle))
return list(reversed(samples)) if reverse else samples
def _append_profile_points(
target: List[Tuple[float, float]],
points: List[Tuple[float, float]],
*,
tolerance: float = 1.0e-9,
) -> None:
for point in points:
if target and math.hypot(target[-1][0] - point[0], target[-1][1] - point[1]) <= tolerance:
continue
target.append(point)
def _build_internal_gear_profile_polyline_wire(
n_teeth: int,
module: float,
pressure_angle: float,
backlash: float = 0.0,
addendum_factor: float = 1.0,
clearance_factor: float = 0.25,
) -> Wire:
"""Build the internal profile without invoking the sketch solver.
Large internal gears create hundreds of constrained sketch entities. Some
SolveSpace backends fail on those oversized sketches even though the tooth
geometry is already fully known analytically. This fallback samples the
same involute/root/tip profile into a closed wire so high-tooth-count ring
gears remain generatable.
"""
geo = _compute_internal_tooth_geometry(
n_teeth,
module,
pressure_angle,
backlash,
addendum_factor=addendum_factor,
clearance_factor=clearance_factor,
)
root_radius = geo["root_radius"]
tip_radius = geo["tip_radius"]
flank_tip_radius = geo["flank_tip_radius"]
base_radius = geo["base_radius"]
tooth_angle = geo["tooth_angle"]
points: List[Tuple[float, float]] = []
root_arc_steps = 2 if n_teeth >= 96 else 3
flank_steps = 5 if n_teeth >= 96 else 7
tip_arc_steps = 2 if n_teeth >= 96 else 3
for i in range(n_teeth):
offset = tooth_angle * i
root_arc_start_angle = geo["right_root_angle"] - tooth_angle + offset
root_arc_end_angle = geo["left_root_angle"] + offset
_append_profile_points(
points,
_arc_points(
root_radius,
root_arc_start_angle,
root_arc_end_angle,
steps=root_arc_steps,
),
)
_append_profile_points(
points,
_sample_internal_involute_points(
base_radius,
start_radius=flank_tip_radius,
end_radius=root_radius,
start_angle=geo["left_base_angle"] + offset,
mirror=True,
reverse=True,
steps=flank_steps,
),
)
left_tip_angle = geo["left_tip_angle"] + offset
right_tip_angle = geo["right_tip_angle"] + offset
if geo["needs_tip_connectors"]:
_append_profile_points(points, [(tip_radius * math.cos(left_tip_angle), tip_radius * math.sin(left_tip_angle))])
_append_profile_points(
points,
_arc_points(tip_radius, left_tip_angle, right_tip_angle, steps=tip_arc_steps),
)
_append_profile_points(points, [(flank_tip_radius * math.cos(right_tip_angle), flank_tip_radius * math.sin(right_tip_angle))])
else:
_append_profile_points(
points,
_arc_points(
tip_radius,
left_tip_angle,
right_tip_angle,
steps=tip_arc_steps,
),
)
_append_profile_points(
points,
_sample_internal_involute_points(
base_radius,
start_radius=flank_tip_radius,
end_radius=root_radius,
start_angle=geo["right_base_angle"] + offset,
mirror=False,
steps=flank_steps,
),
)
if len(points) >= 2 and math.hypot(points[0][0] - points[-1][0], points[0][1] - points[-1][1]) <= 1.0e-8:
points.pop()
return make_polyline_rwire(
points=[(float(x), float(y), 0.0) for x, y in points],
closed=True,
)
def _build_internal_gear_profile_wire(
n_teeth: int,
module: float,
@@ -1206,7 +1364,19 @@ def _build_internal_gear_profile_wire(
multiplicities=right_mults,
)
wire = make_wire_from_sketch_rwire(sketch, profile=0)
try:
wire = make_wire_from_sketch_rwire(sketch, profile=0)
except Exception:
if n_teeth < 96:
raise
wire = _build_internal_gear_profile_polyline_wire(
n_teeth,
module,
pressure_angle,
backlash=backlash,
addendum_factor=addendum_factor,
clearance_factor=clearance_factor,
)
if return_sketch:
return wire, sketch
return wire
+176
View File
@@ -0,0 +1,176 @@
from __future__ import annotations
import hashlib
import re
import shutil
import uuid
from pathlib import Path
from typing import Any
from .models import read_json, utc_now_iso, write_json
BACKEND_ROOT = Path(__file__).resolve().parents[1]
RUN_ROOT = BACKEND_ROOT / "output" / "runs"
def new_task_id(prefix: str = "cad") -> str:
return f"{prefix}_{uuid.uuid4().hex[:12]}"
def safe_task_id(task_id: str) -> str:
normalized = str(task_id or "").strip()
if not re.fullmatch(r"[a-zA-Z0-9_-]{4,80}", normalized):
raise ValueError("invalid_task_id")
return normalized
def task_dir(task_id: str) -> Path:
return RUN_ROOT / safe_task_id(task_id)
def tasks_index_path() -> Path:
return RUN_ROOT / "cad_tasks_index.json"
def manifest_path(task_id: str) -> Path:
return task_dir(task_id) / "cad-task.json"
def relative_to_task(task_id: str, path: Path) -> str:
resolved = path.resolve()
root = task_dir(task_id).resolve()
try:
return str(resolved.relative_to(root))
except ValueError:
return str(resolved)
def artifact_url(task_id: str, path: Path) -> str:
return f"/api/tasks/{safe_task_id(task_id)}/artifacts/{relative_to_task(task_id, path)}"
def ensure_task(task_id: str | None = None, request: str = "") -> dict[str, Any]:
task = safe_task_id(task_id) if task_id else new_task_id()
root = task_dir(task)
root.mkdir(parents=True, exist_ok=True)
for child in ["uploads", "exports", "reducer"]:
(root / child).mkdir(parents=True, exist_ok=True)
manifest = read_task_manifest(task) or {
"schema_version": "1.0",
"taskId": task,
"createdAt": utc_now_iso(),
"updatedAt": utc_now_iso(),
"request": request,
"status": "created",
"kind": "cad_generation",
"route": {
"selected_backend": "simplecadapi",
"effective_execution_engine": "simplecadapi",
"runner_skill": "cad-router",
"project": "planetary-reducer-system",
"fallback_order": ["build123d"],
"workflow_profiles": ["requirement_refinement", "visual_repair"],
},
"source": {},
"artifacts": [],
"parameters": [],
"experience": {
"context_kind": "generalized_cad_experience_query",
"methods_applied": [],
},
"assumptions": [],
"validation": [],
"viewer_links": [],
"messages": [],
}
if request and not manifest.get("request"):
manifest["request"] = request
write_task_manifest(task, manifest)
update_tasks_index(task, manifest)
return {"taskId": task, "dir": str(root.resolve()), "manifest": manifest}
def read_task_manifest(task_id: str) -> dict[str, Any] | None:
path = manifest_path(task_id)
if not path.exists():
return None
return read_json(path)
def write_task_manifest(task_id: str, manifest: dict[str, Any]) -> dict[str, Any]:
payload = {**manifest, "taskId": safe_task_id(task_id), "updatedAt": utc_now_iso()}
write_json(manifest_path(task_id), payload)
update_tasks_index(task_id, payload)
return payload
def update_task(task_id: str, patch: dict[str, Any]) -> dict[str, Any]:
current = read_task_manifest(task_id) or ensure_task(task_id)["manifest"]
next_manifest = {**current, **patch}
return write_task_manifest(task_id, next_manifest)
def update_tasks_index(task_id: str, manifest: dict[str, Any]) -> None:
path = tasks_index_path()
index = read_json(path) if path.exists() else {"schema_version": "1.0", "tasks": []}
tasks = [task for task in index.get("tasks", []) if task.get("taskId") != task_id]
tasks.insert(0, {
"taskId": task_id,
"request": manifest.get("request", ""),
"status": manifest.get("status", "unknown"),
"kind": manifest.get("kind", "cad_generation"),
"updatedAt": manifest.get("updatedAt", utc_now_iso()),
"summary": manifest.get("summary", ""),
})
index["tasks"] = tasks[:50]
write_json(path, index)
def list_tasks() -> list[dict[str, Any]]:
path = tasks_index_path()
if not path.exists():
return []
return list(read_json(path).get("tasks", []))
def sanitize_filename(filename: str) -> str:
base = Path(str(filename or "upload.bin")).name
return re.sub(r"[^a-zA-Z0-9._-]+", "_", base).lstrip("_") or "upload.bin"
def write_upload(task_id: str, filename: str, data: bytes, mime: str = "application/octet-stream") -> dict[str, Any]:
ensure_task(task_id)
digest = hashlib.sha256(data).hexdigest()
safe_name = sanitize_filename(filename)
relative = Path("uploads") / f"file_{digest[:16]}_{safe_name}"
path = task_dir(task_id) / relative
path.parent.mkdir(parents=True, exist_ok=True)
path.write_bytes(data)
attachment = {
"id": f"file_{digest[:16]}",
"taskId": safe_task_id(task_id),
"name": safe_name,
"path": str(relative),
"size": len(data),
"sha256": digest,
"mime": mime,
"kind": "step" if safe_name.lower().endswith((".step", ".stp")) else "image" if mime.startswith("image/") else "other",
"url": artifact_url(task_id, path),
}
manifest = read_task_manifest(task_id) or ensure_task(task_id)["manifest"]
uploads = [item for item in manifest.get("uploads", []) if item.get("id") != attachment["id"]]
uploads.append(attachment)
update_task(task_id, {"uploads": uploads})
return attachment
def copy_artifact(task_id: str, source: Path, target_relative: str) -> dict[str, Any]:
target = task_dir(task_id) / target_relative
target.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(source, target)
return {
"path": target_relative,
"url": artifact_url(task_id, target),
"size": target.stat().st_size,
}
+3
View File
@@ -0,0 +1,3 @@
from .api import create_toolchain_app
__all__ = ["create_toolchain_app"]
+912
View File
@@ -0,0 +1,912 @@
from __future__ import annotations
import json
import mimetypes
import re
import shutil
import subprocess
import sys
import traceback
import zipfile
import xml.etree.ElementTree as ET
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import unquote, urlparse
from ..cli import _prepare_structured_run_dir
from ..joint_module.auto_design import DEFAULT_PLATFORM_PROFILE, run_auto_joint_design
from ..joint_module.generator import run_joint_module
from ..joint_module.urdf_exporter import export_urdf
from ..models import read_json, write_json, write_model_json
from ..parameter_solver import load_requirement, solve_parameters
from ..kinematics import motion_samples, solve_instance
from ..placement import solve_reducer_placements
from ..cad.simplecad_generator import generate_simplecad
from ..reducer_urdf_exporter import export_reducer_urdf
from ..topology import build_graph, validate_topology
from ..topology_runtime import load_default_template
from ..validators import validate_from_files
from ..task_store import (
BACKEND_ROOT,
RUN_ROOT,
artifact_url,
ensure_task,
list_tasks,
read_task_manifest,
safe_task_id,
task_dir,
update_task,
write_upload,
)
DEFAULT_RUN_NAME = "frontend_split_motor_7nm"
DEFAULT_JOINT_REQUIREMENT = BACKEND_ROOT / "input" / "requirements" / "joint_modules" / "3500_motor_housing.json"
JOINT_REQUIREMENT_TEMPLATES = {
"simple_2k_h": DEFAULT_JOINT_REQUIREMENT,
"simple_2k_h_cascade": BACKEND_ROOT / "input" / "requirements" / "joint_modules" / "3500_motor_housing_cascade.json",
}
CAD_ROUTER_RECONSTRUCTOR = Path("/Users/lk/Projects/cadSet/text-to-cad/skills/cad-router/scripts/reconstruct_step.py")
REDUCER_REQUIREMENT_TEMPLATES = {
"simple_2k_h": BACKEND_ROOT / "input" / "requirements" / "reducers" / "simple_2k_h" / "simple_2kh_ratio_5.json",
"simple_2k_h_cascade": BACKEND_ROOT / "input" / "requirements" / "reducers" / "simple_2k_h_cascade" / "simple_2kh_cascade_ratio_9_spur.json",
"ferguson_wolfrom": BACKEND_ROOT / "input" / "requirements" / "reducers" / "ferguson_wolfrom" / "ferguson_wolfrom_ratio_531p25_spur.json",
}
SUPPORTED_REQUIREMENT_OVERRIDES = {
"topology_family",
"target_ratio",
"tooth_form",
"module_mm",
"planet_count",
"helix_angle_deg",
"pressure_angle_deg",
"face_width_mm",
"backlash_mm",
"ring_rim_thickness_mm",
"max_outer_diameter_mm",
}
def _send_json(handler: BaseHTTPRequestHandler, status: int, payload: object) -> None:
body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
handler.send_response(status)
handler.send_header("Content-Type", "application/json; charset=utf-8")
handler.send_header("Content-Length", str(len(body)))
handler.send_header("Access-Control-Allow-Origin", "*")
handler.send_header("Access-Control-Allow-Methods", "GET,POST,OPTIONS")
handler.send_header("Access-Control-Allow-Headers", "Content-Type")
handler.end_headers()
handler.wfile.write(body)
def _send_file(handler: BaseHTTPRequestHandler, path: Path) -> None:
if not path.exists() or not path.is_file():
_send_json(handler, 404, {"error": f"file_not_found: {path}"})
return
content_type = mimetypes.guess_type(path.name)[0] or "application/octet-stream"
body = path.read_bytes()
handler.send_response(200)
handler.send_header("Content-Type", content_type)
handler.send_header("Content-Length", str(len(body)))
handler.send_header("Access-Control-Allow-Origin", "*")
handler.end_headers()
handler.wfile.write(body)
def _read_body(handler: BaseHTTPRequestHandler) -> bytes:
length = int(handler.headers.get("Content-Length") or 0)
return handler.rfile.read(length) if length else b""
def _read_json(handler: BaseHTTPRequestHandler) -> dict[str, object]:
body = _read_body(handler)
if not body.strip():
return {}
payload = json.loads(body.decode("utf-8"))
if not isinstance(payload, dict):
raise ValueError("request_body_must_be_json_object")
return payload
def _safe_relative_file(task_id: str, relative: str) -> Path:
root = task_dir(task_id).resolve()
path = (root / unquote(relative)).resolve()
try:
path.relative_to(root)
except ValueError as exc:
raise ValueError("invalid_artifact_path") from exc
return path
def _first_number(text: str, patterns: list[str], *, integer: bool = False) -> int | float | None:
for pattern in patterns:
match = re.search(pattern, text, flags=re.IGNORECASE)
if match:
value = float(match.group(1))
return int(value) if integer else value
return None
def _request_requirement_overrides(request: str) -> dict[str, object]:
text = str(request or "")
overrides: dict[str, object] = {}
if re.search(r"(双级|两级|二级|串联|cascade|two[- ]stage)", text, flags=re.IGNORECASE):
overrides["topology_family"] = "simple_2k_h_cascade"
elif re.search(r"(ferguson|wolfrom|沃尔夫罗姆|弗格森)", text, flags=re.IGNORECASE):
overrides["topology_family"] = "ferguson_wolfrom"
elif re.search(r"(单级|simple[_ -]?2k[_ -]?h)", text, flags=re.IGNORECASE):
overrides["topology_family"] = "simple_2k_h"
tooth_form_match = re.search(r"(斜齿|helical|直齿|spur)", text, flags=re.IGNORECASE)
if tooth_form_match:
overrides["tooth_form"] = (
"helical"
if re.search(r"(斜齿|helical)", tooth_form_match.group(1), flags=re.IGNORECASE)
else "spur"
)
numeric_patterns: dict[str, list[str]] = {
"target_ratio": [
r"(?:传动比|减速比|速比|ratio)\s*(?:为|是|设为|=|:|:|to)?\s*([0-9]+(?:\.[0-9]+)?)",
r"([0-9]+(?:\.[0-9]+)?)\s*(?::|:)\s*1\s*(?:传动比|减速比)?",
],
"module_mm": [
r"(?:齿轮)?模数\s*(?:为|是|设为|=|:|:|to)?\s*([0-9]+(?:\.[0-9]+)?)",
r"module(?:_mm)?\s*(?:=|:|to)?\s*([0-9]+(?:\.[0-9]+)?)",
],
"planet_count": [
r"(?:行星轮数量|行星轮数|行星轮)\s*(?:为|是|设为|=|:|:|to)?\s*([0-9]+)",
r"([0-9]+)\s*(?:个|颗|只)?\s*行星轮",
r"planet_count\s*(?:=|:|to)?\s*([0-9]+)",
],
"helix_angle_deg": [
r"(?:螺旋角|helix_angle(?:_deg)?)\s*(?:为|是|设为|=|:|:|to)?\s*([0-9]+(?:\.[0-9]+)?)",
],
"pressure_angle_deg": [
r"(?:压力角|pressure_angle(?:_deg)?)\s*(?:为|是|设为|=|:|:|to)?\s*([0-9]+(?:\.[0-9]+)?)",
],
"face_width_mm": [
r"(?:齿宽|face_width(?:_mm)?)\s*(?:为|是|设为|=|:|:|to)?\s*([0-9]+(?:\.[0-9]+)?)",
],
"backlash_mm": [
r"(?:齿侧间隙|侧隙|backlash(?:_mm)?)\s*(?:为|是|设为|=|:|:|to)?\s*([0-9]+(?:\.[0-9]+)?)",
],
"ring_rim_thickness_mm": [
r"(?:齿圈轮缘厚度|轮缘厚度|ring_rim_thickness(?:_mm)?)\s*(?:为|是|设为|=|:|:|to)?\s*([0-9]+(?:\.[0-9]+)?)",
],
"max_outer_diameter_mm": [
r"(?:最大外径|外径)\s*(?:小于|不超过|限制为|为|是|=|:|:|to|≤|<=)?\s*([0-9]+(?:\.[0-9]+)?)\s*(?:mm|毫米)?",
r"max_outer_diameter(?:_mm)?\s*(?:=|:|to|<=)?\s*([0-9]+(?:\.[0-9]+)?)",
],
}
for name, patterns in numeric_patterns.items():
value = _first_number(text, patterns, integer=name == "planet_count")
if value is not None:
overrides[name] = value
return overrides
def _normalized_requirement_overrides(payload: dict[str, object]) -> dict[str, object]:
overrides = _request_requirement_overrides(str(payload.get("request") or ""))
explicit = payload.get("requirementOverrides")
if explicit is not None and not isinstance(explicit, dict):
raise ValueError("requirementOverrides_must_be_an_object")
if isinstance(explicit, dict):
unknown = sorted(set(explicit) - SUPPORTED_REQUIREMENT_OVERRIDES)
if unknown:
raise ValueError(f"unsupported_requirement_overrides: {unknown}")
overrides.update({key: value for key, value in explicit.items() if value is not None})
return overrides
def _apply_requirement_overrides(
requirement: dict[str, object],
overrides: dict[str, object],
) -> dict[str, object]:
result = json.loads(json.dumps(requirement))
topology = str(overrides.get("topology_family") or result.get("topology_family") or "simple_2k_h")
result["topology_family"] = topology
direct_fields = {
"target_ratio",
"tooth_form",
"helix_angle_deg",
"pressure_angle_deg",
"face_width_mm",
"backlash_mm",
"ring_rim_thickness_mm",
}
for name in direct_fields:
if name in overrides:
result[name] = overrides[name]
if "module_mm" in overrides:
result["module_candidates_mm"] = [float(overrides["module_mm"])]
if "planet_count" in overrides:
result["planet_count_candidates"] = [int(overrides["planet_count"])]
if "max_outer_diameter_mm" in overrides:
constraints = dict(result.get("constraints") or {})
constraints["max_outer_diameter_mm"] = float(overrides["max_outer_diameter_mm"])
result["constraints"] = constraints
if result.get("tooth_form") == "helical" and not result.get("helix_angle_deg"):
result["helix_angle_deg"] = 15.0
if result.get("tooth_form") == "spur":
result.pop("helix_angle_deg", None)
stages = result.get("stages")
if isinstance(stages, list):
for stage_value in stages:
if not isinstance(stage_value, dict):
continue
if "module_mm" in overrides:
stage_value["module_candidates_mm"] = [float(overrides["module_mm"])]
if "planet_count" in overrides:
stage_value["planet_count_candidates"] = [int(overrides["planet_count"])]
for name in {
"tooth_form",
"helix_angle_deg",
"pressure_angle_deg",
"face_width_mm",
"backlash_mm",
"ring_rim_thickness_mm",
}:
if name in overrides:
stage_value[name] = overrides[name]
if result.get("tooth_form") == "spur":
stage_value.pop("helix_angle_deg", None)
return result
def _materialize_reducer_requirement(
task_id: str,
payload: dict[str, object],
) -> tuple[Path, dict[str, object]]:
overrides = _normalized_requirement_overrides(payload)
requested_path = payload.get("requirementPath")
if requested_path:
base_path = Path(str(requested_path))
if not base_path.is_absolute():
base_path = BACKEND_ROOT / base_path
else:
topology = str(overrides.get("topology_family") or "simple_2k_h")
if topology not in REDUCER_REQUIREMENT_TEMPLATES:
raise ValueError(f"unsupported_topology_family: {topology}")
base_path = REDUCER_REQUIREMENT_TEMPLATES[topology]
requirement = read_json(base_path)
if overrides:
requirement = _apply_requirement_overrides(requirement, overrides)
topology = str(requirement.get("topology_family") or "")
if topology == "ferguson_wolfrom":
requested_ratio = float(requirement.get("target_ratio") or 0)
if abs(requested_ratio - 531.25) > 1e-9:
raise ValueError(
"ferguson_wolfrom currently supports only the validated ratio 531.25"
)
target = task_dir(task_id) / "requirements" / "active_reducer_requirement.json"
write_json(target, requirement)
return target, overrides
def _joint_reducer_requirement_path(joint_requirement_path: Path) -> Path:
payload = read_json(joint_requirement_path)
reducer_path = Path(str(payload.get("reducer_requirement") or ""))
if not str(reducer_path):
raise ValueError("joint_requirement_has_no_reducer_requirement")
if not reducer_path.is_absolute():
reducer_path = joint_requirement_path.parent / reducer_path
return reducer_path.resolve()
def _materialize_joint_requirement(
task_id: str,
joint_requirement_path: Path,
overrides: dict[str, object],
) -> tuple[Path, Path]:
joint_requirement = read_json(joint_requirement_path)
reducer_path = _joint_reducer_requirement_path(joint_requirement_path)
reducer_requirement = _apply_requirement_overrides(read_json(reducer_path), overrides)
reducer_target = task_dir(task_id) / "requirements" / "active_reducer_requirement.json"
joint_target = task_dir(task_id) / "requirements" / "active_joint_requirement.json"
write_json(reducer_target, reducer_requirement)
joint_requirement["reducer_requirement"] = str(reducer_target.resolve())
if "platform_profile" in joint_requirement:
profile_path = Path(str(joint_requirement["platform_profile"]))
if not profile_path.is_absolute():
profile_path = (joint_requirement_path.parent / profile_path).resolve()
joint_requirement["platform_profile"] = str(profile_path)
write_json(joint_target, joint_requirement)
return joint_target, reducer_target
def _project_summary(run_name: str) -> dict[str, object]:
run = RUN_ROOT / run_name
formula = read_json(run / "reducer" / "formula" / "formula_instance.json")
kinematics = read_json(run / "reducer" / "kinematics" / "kinematic_solution.json")
reducer_validation = read_json(run / "reducer" / "validation" / "validation_report.json")
joint_validation_path = run / "joint_module_validation_report.json"
joint_validation = read_json(joint_validation_path) if joint_validation_path.exists() else {"summary": {"total": 0, "passed": 0, "failed": 0}, "run_id": ""}
return {
"source": f"output/runs/{run_name}",
"runId": joint_validation.get("run_id") or formula.get("run_id"),
"ratio": formula["derived"]["ratio"],
"parameters": formula["parameters"],
"derived": formula["derived"],
"speeds": kinematics["speeds"],
"reducerValidation": reducer_validation["summary"],
"jointValidation": joint_validation["summary"],
}
def _artifact_record(task_id: str, name: str, path: Path, role: str, kind: str) -> dict[str, object]:
return {
"name": name,
"role": role,
"kind": kind,
"path": str(path.resolve()),
"url": artifact_url(task_id, path),
"size": path.stat().st_size if path.exists() else 0,
}
def _summarize_reducer_task(task_id: str, run_dir: Path, request: str) -> dict[str, object]:
formula = read_json(run_dir / "formula" / "formula_instance.json")
validation = read_json(run_dir / "validation" / "validation_report.json")
manifest = read_json(run_dir / "cad" / "assembly_manifest.json")
artifacts = [
_artifact_record(task_id, "reducer.step", run_dir / "cad" / "reducer.step", "primary", "step"),
_artifact_record(task_id, "assembly_manifest.json", run_dir / "cad" / "assembly_manifest.json", "manifest", "json"),
_artifact_record(task_id, "formula_instance.json", run_dir / "formula" / "formula_instance.json", "formula", "json"),
_artifact_record(task_id, "validation_report.json", run_dir / "validation" / "validation_report.json", "validation", "json"),
]
return {
"status": "success" if validation.get("overall_passed") else "error",
"kind": "reducer",
"request": request,
"summary": f"{formula['topology_family']} reducer ratio={formula['derived']['ratio']}",
"ratio": formula["derived"]["ratio"],
"parameters": formula["parameters"],
"derived": formula["derived"],
"validation": validation,
"assemblyManifest": manifest,
"artifacts": artifacts,
"route": {
"selected_backend": "simplecadapi",
"effective_execution_engine": "simplecadapi",
"runner_skill": "cad-router",
"project": "planetary-reducer-system",
"fallback_order": ["build123d"],
"workflow_profiles": ["requirement_refinement", "visual_repair"],
},
"source": {
"path": "formula/formula_instance.json",
"format": "json",
"backend": "simplecadapi",
"requirement_path": "input_snapshot",
},
"validation": [validation],
}
def _summarize_joint_task(task_id: str, run_dir: Path, request: str) -> dict[str, object]:
joint_dir = run_dir / "joint" if (run_dir / "joint").exists() else run_dir
reducer_dir = joint_dir / "reducer"
formula = read_json(reducer_dir / "formula" / "formula_instance.json")
reducer_validation = read_json(reducer_dir / "validation" / "validation_report.json")
joint_validation = read_json(joint_dir / "joint_module_validation_report.json")
joint_manifest = read_json(joint_dir / "joint_module_manifest.json")
artifacts = [
_artifact_record(task_id, "joint_module.step", joint_dir / "joint_module.step", "primary", "step"),
_artifact_record(task_id, "joint_module_manifest.json", joint_dir / "joint_module_manifest.json", "manifest", "json"),
_artifact_record(task_id, "joint_module_validation_report.json", joint_dir / "joint_module_validation_report.json", "validation", "json"),
_artifact_record(task_id, "formula_instance.json", reducer_dir / "formula" / "formula_instance.json", "formula", "json"),
]
exports_dir = run_dir / "exports" / "urdf"
if (exports_dir / "joint_module.urdf").exists():
artifacts.append(_artifact_record(task_id, "joint_module.urdf", exports_dir / "joint_module.urdf", "viewer", "urdf"))
return {
"status": "success" if joint_validation.get("overall_passed") else "error",
"kind": "joint_module",
"request": request,
"summary": f"planetary joint module ratio={formula['derived']['ratio']}",
"ratio": formula["derived"]["ratio"],
"parameters": formula["parameters"],
"derived": formula["derived"],
"reducerValidation": reducer_validation,
"jointValidation": joint_validation,
"jointManifest": joint_manifest,
"artifacts": artifacts,
"route": {
"selected_backend": "simplecadapi",
"effective_execution_engine": "simplecadapi",
"runner_skill": "cad-router",
"project": "planetary-reducer-system",
"fallback_order": ["build123d"],
"workflow_profiles": ["requirement_refinement", "visual_repair"],
},
"source": {
"path": "joint/reducer/formula/formula_instance.json" if (run_dir / "joint").exists() else "reducer/formula/formula_instance.json",
"format": "json",
"backend": "simplecadapi",
"requirement_path": "generated_joint_requirement.json",
},
"validation": [reducer_validation, joint_validation],
}
def _write_zip(source_dir: Path, target: Path) -> Path:
target.parent.mkdir(parents=True, exist_ok=True)
with zipfile.ZipFile(target, "w", compression=zipfile.ZIP_DEFLATED) as archive:
for path in sorted(source_dir.rglob("*")):
if path.is_file():
archive.write(path, path.relative_to(source_dir))
return target
def _find_requirement_snapshot(task_id: str) -> Path:
snapshot_dir = task_dir(task_id) / "input_snapshot"
candidates = sorted(snapshot_dir.glob("*.json"))
if not candidates:
candidates = sorted((task_dir(task_id) / "joint").glob("**/input_snapshot/*.json"))
if not candidates:
raise ValueError("task_has_no_requirement_snapshot")
return candidates[0]
def _parameter_patch(parameter: str, value: object) -> dict[str, object]:
name = str(parameter or "").strip()
if not name:
raise ValueError("parameter_is_required")
if name in {"module", "module_mm"}:
return {"module_candidates_mm": [float(value)]}
if name in {"planet_count", "planetCount"}:
return {"planet_count_candidates": [int(value)]}
if name in {"target_ratio", "ratio"}:
return {"target_ratio": float(value)}
if name in {"tooth_form", "toothForm"}:
normalized = str(value).strip().lower()
if normalized not in {"spur", "helical"}:
raise ValueError("tooth_form_must_be_spur_or_helical")
return {"tooth_form": normalized}
if name in {"helix_angle_deg", "pressure_angle_deg", "face_width_mm", "backlash_mm", "ring_rim_thickness_mm"}:
return {name: float(value)}
raise ValueError(f"parameter_not_editable: {name}")
def edit_task_parameter(task_id: str, parameter: str, value: object) -> dict[str, object]:
task = read_task_manifest(task_id)
if not task:
raise ValueError("task_not_found")
kind = str(task.get("kind") or "")
if kind not in {"reducer", "joint_module"}:
raise ValueError("parameter_edit_requires_generated_reducer_or_joint_task")
run_dir = task_dir(task_id)
active_requirement = task.get("activeReducerRequirementPath")
requirement_path = Path(str(active_requirement)) if active_requirement else (
run_dir / "generated_reducer_requirement.json"
if kind == "joint_module" and (run_dir / "generated_reducer_requirement.json").exists()
else _find_requirement_snapshot(task_id)
)
requirement = read_json(requirement_path)
requirement.update(_parameter_patch(parameter, value))
revision = int(task.get("revision") or 0) + 1
edited_requirement = run_dir / "revisions" / f"reducer_requirement_{revision}.json"
edited_requirement.parent.mkdir(parents=True, exist_ok=True)
write_json(edited_requirement, requirement)
request = f"{task.get('request', 'CAD task')} · edit {parameter}={value}"
if kind == "joint_module":
joint_requirement_path = run_dir / "generated_joint_requirement.json"
if not joint_requirement_path.exists():
raise ValueError("joint_task_has_no_generated_joint_requirement")
joint_requirement = read_json(joint_requirement_path)
joint_requirement["reducer_requirement"] = str(edited_requirement.resolve())
edited_joint_requirement = run_dir / "revisions" / f"joint_requirement_{revision}.json"
write_json(edited_joint_requirement, joint_requirement)
payload = {
"request": request,
"requirementPath": str(edited_joint_requirement),
"mode": "requirement",
"kind": "joint_module",
}
else:
payload = {
"request": request,
"requirementPath": str(edited_requirement),
"mode": "requirement",
"kind": "reducer",
}
update_task(task_id, {
"status": "running",
"revision": revision,
"parentTaskId": task_id,
"editedParameter": {"name": parameter, "value": value},
})
result = (
generate_joint_task(task_id, payload)
if kind == "joint_module"
else generate_reducer_task(task_id, payload)
)
return update_task(task_id, {
**result,
"editedParameter": {"name": parameter, "value": value},
"revision": revision,
"activeReducerRequirementPath": str(edited_requirement.resolve()),
})
def reconstruct_uploaded_task(task_id: str, request: str = "重建上传的 STEP") -> dict[str, object]:
manifest = read_task_manifest(task_id)
if not manifest:
raise ValueError("task_not_found")
uploads = [item for item in manifest.get("uploads", []) if item.get("kind") == "step"]
if not uploads:
raise ValueError("step_upload_required")
attachment = uploads[-1]
source = task_dir(task_id) / str(attachment["path"])
if not source.exists():
raise ValueError("uploaded_step_not_found")
if not CAD_ROUTER_RECONSTRUCTOR.exists():
raise ValueError(f"cad_router_reconstructor_not_found: {CAD_ROUTER_RECONSTRUCTOR}")
teacher_root = RUN_ROOT / "_reconstruction_sources" / safe_task_id(task_id)
teacher_root.mkdir(parents=True, exist_ok=True)
teacher = teacher_root / source.name
shutil.copy2(source, teacher)
revision = int(manifest.get("reconstructionRevision") or 0) + 1
reconstruction_dir = task_dir(task_id) / "revisions" / f"reconstruction_{revision}"
completed = subprocess.run(
[
sys.executable,
str(CAD_ROUTER_RECONSTRUCTOR),
"--source",
str(teacher),
"--task-dir",
str(reconstruction_dir),
"--request",
request,
],
capture_output=True,
text=True,
timeout=900,
check=False,
)
if completed.returncode != 0:
detail = completed.stderr.strip() or completed.stdout.strip()
raise RuntimeError(f"cad_router_reconstruction_failed: {detail[-3000:]}")
lines = [line for line in completed.stdout.splitlines() if line.strip()]
result = json.loads(lines[-1]) if lines else {}
artifact_paths = []
for path in sorted(reconstruction_dir.rglob("*")):
if path.is_file() and path.name not in {"cad-task.json"}:
artifact_paths.append(_artifact_record(task_id, path.name, path, "reconstruction", path.suffix.lstrip(".") or "file"))
return update_task(task_id, {
"status": "success",
"kind": "step_reconstruction",
"request": request,
"route": {
"selected_backend": result.get("selected_backend", "build123d"),
"effective_execution_engine": "surfaceir",
"runner_skill": "cad-router",
"project": "text-to-cad",
},
"source": {
"path": str(attachment["path"]),
"format": "step",
"backend": "uploaded",
"source_editability": "reconstruction_reference",
},
"reconstruction": result,
"reconstructionRevision": revision,
"artifacts": artifact_paths,
})
def export_task_robot(task_id: str, format_name: str) -> dict[str, object]:
manifest = read_task_manifest(task_id)
if not manifest:
raise ValueError("task_not_found")
if format_name not in {"urdf", "mjcf"}:
raise ValueError("export_format_must_be_urdf_or_mjcf")
run_dir = task_dir(task_id)
urdf_output = run_dir / "exports" / "urdf"
if manifest.get("kind") == "reducer":
export_manifest = export_reducer_urdf(
assembly_manifest_path=run_dir / "cad" / "assembly_manifest.json",
reducer_instance_path=run_dir / "formula" / "formula_instance.json",
out_dir=urdf_output,
)
elif manifest.get("kind") == "joint_module":
export_manifest = export_urdf(
joint_manifest_path=run_dir / "joint" / "joint_module_manifest.json" if (run_dir / "joint").exists() else run_dir / "joint_module_manifest.json",
reducer_instance_path=run_dir / "joint" / "reducer" / "formula" / "formula_instance.json" if (run_dir / "joint").exists() else run_dir / "reducer" / "formula" / "formula_instance.json",
out_dir=urdf_output,
)
else:
raise ValueError("robot_export_requires_generated_task")
if format_name == "mjcf":
try:
import mujoco # type: ignore
import pymeshlab # type: ignore
except ImportError as exc:
raise ValueError("mjcf_export_requires_mujoco_and_pymeshlab") from exc
output = run_dir / "exports" / "mjcf"
mesh_dir = output / "meshes"
mesh_dir.mkdir(parents=True, exist_ok=True)
for source in sorted((urdf_output / "meshes").glob("*.stl")):
mesh_set = pymeshlab.MeshSet()
mesh_set.load_new_mesh(str(source))
mesh_set.save_current_mesh(str(mesh_dir / source.name), binary=True)
urdf_path = (
urdf_output / "joint_module.urdf"
if (urdf_output / "joint_module.urdf").exists()
else urdf_output / "reducer.urdf"
)
if not urdf_path.exists():
raise ValueError("urdf_intermediate_not_found")
root = ET.fromstring(urdf_path.read_text(encoding="utf-8"))
for mesh in root.findall(".//mesh"):
filename = str(mesh.get("filename") or "")
mesh.set("filename", str((mesh_dir / Path(filename).name).resolve()))
intermediate = output / "mujoco-input.urdf"
intermediate.write_text(
ET.tostring(root, encoding="unicode", xml_declaration=True),
encoding="utf-8",
)
description = output / "robot.mjcf.xml"
model = mujoco.MjModel.from_xml_path(str(intermediate))
mujoco.mj_saveLastXML(str(description), model)
intermediate.unlink(missing_ok=True)
validation = {
"engine": f"mujoco {mujoco.__version__}",
"parsed": True,
"body_count": int(model.nbody),
"joint_count": int(model.njnt),
"mesh_count": int(model.nmesh),
}
write_json(output / "mjcf_validation.json", validation)
package = _write_zip(output, run_dir / "exports" / "mjcf.zip")
export_manifest = {
"format": "mjcf",
"description_path": str(description.resolve()),
"validation": validation,
}
else:
output = urdf_output
package = _write_zip(output, run_dir / "exports" / "urdf.zip")
artifacts = list(manifest.get("artifacts", []))
for path in [
package,
output / "joint_module.urdf",
output / "reducer.urdf",
output / "robot.mjcf.xml",
output / "urdf_export_manifest.json",
output / "export_manifest.json",
output / "mjcf_validation.json",
]:
if path.exists():
artifacts.append(_artifact_record(task_id, path.name, path, "robot_export", path.suffix.lstrip(".") or "file"))
return update_task(task_id, {
"status": "success",
"artifacts": artifacts,
"robotExport": {
"format": format_name,
"packageUrl": artifact_url(task_id, package),
"manifest": export_manifest,
},
})
def generate_reducer_task(task_id: str, payload: dict[str, object]) -> dict[str, object]:
request = str(payload.get("request") or "Generate planetary reducer")
requirement_path, overrides = _materialize_reducer_requirement(task_id, payload)
run_dir = task_dir(task_id)
paths = _prepare_structured_run_dir(run_dir)
requirement = load_requirement(requirement_path)
template = load_default_template(requirement.topology_family)
topology_errors = validate_topology(template, requirement)
if topology_errors:
raise ValueError(f"topology_invalid: {topology_errors}")
instance, report = solve_parameters(requirement=requirement, template=template)
solution = solve_instance(instance)
samples = motion_samples(instance, solution)
placement_plan = solve_reducer_placements(instance)
shutil.copy2(requirement_path, paths["input_snapshot"] / requirement_path.name)
write_model_json(paths["formula"] / "formula_instance.json", instance)
write_model_json(paths["formula"] / "parameter_search_report.json", report)
write_model_json(paths["kinematics"] / "kinematic_solution.json", solution)
write_model_json(paths["kinematics"] / "motion_samples.json", samples)
write_model_json(paths["placement"] / "placement_plan.json", placement_plan)
if instance.industrial_parameters is not None:
write_model_json(paths["formula"] / "industrial_parameters.json", instance.industrial_parameters)
if instance.assembly_topology is not None:
write_model_json(paths["formula"] / "assembly_topology.json", instance.assembly_topology)
generate_simplecad(instance, paths["cad"])
if (paths["cad"] / "placement_plan.json").exists():
shutil.copy2(paths["cad"] / "placement_plan.json", paths["placement"] / "placement_plan.json")
validate_from_files(
instance=instance,
model_path=paths["cad"] / "reducer.model.json",
manifest_path=paths["cad"] / "assembly_manifest.json",
build_meta_path=paths["cad"] / "build_meta.json",
snapshot_out=paths["validation"] / "assembly_snapshot.json",
report_out=paths["validation"] / "validation_report.json",
)
result = _summarize_reducer_task(task_id, run_dir, request)
return update_task(task_id, {
**result,
"requirementOverrides": overrides,
"activeReducerRequirementPath": str(requirement_path.resolve()),
})
def generate_joint_task(task_id: str, payload: dict[str, object]) -> dict[str, object]:
request = str(payload.get("request") or "Generate planetary joint module")
run_dir = task_dir(task_id)
joint_dir = run_dir / "joint"
mode = str(payload.get("mode") or "auto")
overrides = _normalized_requirement_overrides(payload)
topology = str(overrides.get("topology_family") or "simple_2k_h")
if topology not in JOINT_REQUIREMENT_TEMPLATES:
raise ValueError(f"joint_module currently supports only {sorted(JOINT_REQUIREMENT_TEMPLATES)}")
if mode == "requirement":
requirement_path = Path(str(payload.get("requirementPath") or JOINT_REQUIREMENT_TEMPLATES[topology]))
if not requirement_path.is_absolute():
requirement_path = BACKEND_ROOT / requirement_path
active_joint_requirement, active_reducer_requirement = _materialize_joint_requirement(
task_id,
requirement_path,
overrides,
)
run_joint_module(requirement_path=active_joint_requirement, out_dir=joint_dir)
else:
platform_profile = Path(str(payload.get("platformProfile") or DEFAULT_PLATFORM_PROFILE))
if not platform_profile.is_absolute():
platform_profile = BACKEND_ROOT / platform_profile
legacy_motor_step = payload.get("motorStep")
legacy_housing_source = payload.get("housingPython")
run_auto_joint_design(
out_dir=run_dir,
platform_profile=platform_profile,
motor_step=Path(str(legacy_motor_step)) if legacy_motor_step else None,
housing_source=Path(str(legacy_housing_source)) if legacy_housing_source else None,
tooth_form=str(payload.get("toothForm") or "spur"),
radial_clearance_mm=float(payload.get("radialClearanceMm") or 14.0),
reducer_axial_allowance_mm=float(payload.get("reducerAxialAllowanceMm") or 42.0),
reducer_radial_clearance_mm=float(payload.get("reducerRadialClearanceMm") or 8.0),
reducer_requirement_overrides=overrides,
)
reducer_dir = joint_dir / "reducer"
export_urdf(
joint_manifest_path=joint_dir / "joint_module_manifest.json",
reducer_instance_path=reducer_dir / "formula" / "formula_instance.json",
out_dir=run_dir / "exports" / "urdf",
)
result = _summarize_joint_task(task_id, run_dir, request)
active_reducer_requirement = (
run_dir / "generated_reducer_requirement.json"
if mode != "requirement"
else active_reducer_requirement
)
return update_task(task_id, {
**result,
"requirementOverrides": overrides,
"activeReducerRequirementPath": str(active_reducer_requirement.resolve()),
})
class ToolchainHandler(BaseHTTPRequestHandler):
run_name = DEFAULT_RUN_NAME
def do_OPTIONS(self) -> None: # noqa: N802
self.send_response(204)
self.send_header("Access-Control-Allow-Origin", "*")
self.send_header("Access-Control-Allow-Methods", "GET,POST,OPTIONS")
self.send_header("Access-Control-Allow-Headers", "Content-Type")
self.end_headers()
def do_GET(self) -> None: # noqa: N802
parsed = urlparse(self.path)
try:
if parsed.path == "/api/project-summary":
_send_json(self, 200, _project_summary(self.run_name))
return
if parsed.path == "/api/tasks":
_send_json(self, 200, {"tasks": list_tasks()})
return
if parsed.path.startswith("/api/tasks/"):
parts = parsed.path.split("/")
if len(parts) >= 4:
task_id = safe_task_id(parts[3])
if len(parts) == 4:
manifest = read_task_manifest(task_id)
_send_json(self, 200 if manifest else 404, manifest or {"error": "task_not_found"})
return
if len(parts) == 5 and parts[4] == "parameters":
manifest = read_task_manifest(task_id)
_send_json(self, 200 if manifest else 404, {
"taskId": task_id,
"parameters": (manifest or {}).get("parameters", {}),
})
return
if len(parts) >= 6 and parts[4] == "artifacts":
relative = "/".join(parts[5:])
_send_file(self, _safe_relative_file(task_id, relative))
return
if parsed.path.startswith("/project-assets/urdf/"):
relative_path = Path(unquote(parsed.path[len("/project-assets/urdf/"):]))
if relative_path.is_absolute() or ".." in relative_path.parts:
_send_json(self, 400, {"error": "invalid_asset_path"})
return
_send_file(self, RUN_ROOT / self.run_name / "exports" / "urdf" / relative_path)
return
_send_json(self, 404, {"error": "not_found"})
except Exception as exc: # noqa: BLE001
_send_json(self, 500, {"error": str(exc)})
def do_POST(self) -> None: # noqa: N802
parsed = urlparse(self.path)
try:
if parsed.path == "/api/tasks":
payload = _read_json(self)
task = ensure_task(str(payload.get("taskId") or "") or None, request=str(payload.get("request") or "CAD Agent Studio task"))
_send_json(self, 200, task)
return
if parsed.path == "/api/generate":
payload = _read_json(self)
kind = str(payload.get("kind") or "joint_module")
task = ensure_task(str(payload.get("taskId") or "") or None, request=str(payload.get("request") or "CAD generation"))
task_id = task["taskId"]
update_task(task_id, {"status": "running", "kind": kind, "request": payload.get("request") or task["manifest"].get("request")})
try:
result = generate_reducer_task(task_id, payload) if kind == "reducer" else generate_joint_task(task_id, payload)
_send_json(self, 200, result)
except Exception as exc: # noqa: BLE001
error_manifest = update_task(task_id, {
"status": "error",
"error": {
"type": type(exc).__name__,
"message": str(exc),
"traceback": traceback.format_exc(),
},
})
_send_json(self, 500, error_manifest)
return
if parsed.path.startswith("/api/tasks/") and parsed.path.endswith("/reconstruct"):
task_id = safe_task_id(parsed.path.split("/")[3])
payload = _read_json(self)
result = reconstruct_uploaded_task(
task_id,
request=str(payload.get("request") or "重建上传的 STEP"),
)
_send_json(self, 200, result)
return
if parsed.path.startswith("/api/tasks/") and parsed.path.endswith("/parameters"):
task_id = safe_task_id(parsed.path.split("/")[3])
payload = _read_json(self)
result = edit_task_parameter(
task_id,
str(payload.get("parameter") or ""),
payload.get("value"),
)
_send_json(self, 200, result)
return
if parsed.path.startswith("/api/tasks/") and "/exports/" in parsed.path:
parts = parsed.path.split("/")
task_id = safe_task_id(parts[3])
format_name = str(parts[5] if len(parts) > 5 else "").lower()
result = export_task_robot(task_id, format_name)
_send_json(self, 200, result)
return
if parsed.path.startswith("/api/upload/"):
task_id = safe_task_id(parsed.path.rsplit("/", 1)[-1])
filename = unquote(self.headers.get("X-Filename") or "upload.bin")
mime = self.headers.get("Content-Type") or "application/octet-stream"
attachment = write_upload(task_id, filename, _read_body(self), mime)
_send_json(self, 200, attachment)
return
_send_json(self, 404, {"error": "not_found"})
except Exception as exc: # noqa: BLE001
_send_json(self, 500, {"error": str(exc)})
def create_toolchain_app(host: str, port: int, *, run_name: str = DEFAULT_RUN_NAME) -> ThreadingHTTPServer:
ToolchainHandler.run_name = run_name
return ThreadingHTTPServer((host, port), ToolchainHandler)
+40 -3
View File
@@ -34,6 +34,7 @@ from src.joint_module.reducer_adapters import (
SUPPORTED_JOINT_FAMILIES,
get_reducer_joint_adapter,
)
from src.joint_module.reference_assembly import infer_reference_placements
from src.kinematics import solve_instance
from src.models import ValidationReport, read_json
from src.parameter_solver import load_requirement, solve_parameters
@@ -49,6 +50,11 @@ REDUCER_REQ = ROOT / "input" / "requirements" / "reducers" / "simple_2k_h" / "si
TEMPLATE = ROOT / "src" / "configurations" / "simple_2k_h" / "topology.template.json"
CASCADE_JOINT_REQ = ROOT / "input" / "requirements" / "joint_modules" / "3500_motor_housing_cascade.json"
CASCADE_REDUCER_REQ = ROOT / "input" / "requirements" / "reducers" / "simple_2k_h_cascade" / "simple_2kh_cascade_ratio_9_spur.json"
STANDARD_MOTOR_STEP = ROOT / "input" / "assets" / "motors" / "3500_motor_part2_standard.step"
ACTUATOR_HOUSING_STEP = ROOT / "input" / "assets" / "housings" / "actuator_v2_knee_abd_main_case_2.step"
CHIP_STEP = ROOT / "input" / "assets" / "chip" / "chip.step"
REFERENCE_ASSEMBLY_STEP = ROOT / "input" / "assets" / "references" / "actuator_v2_knee_abd_reference.step"
FIXED_PLATFORM_PROFILE = ROOT / "input" / "assets" / "platforms" / "actuator_v2_fixed_platform.json"
FERGUSON_REDUCER_REQ = ROOT / "input" / "requirements" / "reducers" / "ferguson_wolfrom" / "ferguson_wolfrom_ratio_531p25_spur.json"
CASCADE_TEMPLATE = ROOT / "src" / "configurations" / "simple_2k_h_cascade" / "topology.template.json"
FERGUSON_TEMPLATE = ROOT / "src" / "configurations" / "ferguson_wolfrom" / "topology.template.json"
@@ -95,11 +101,9 @@ def _generated_housing_step(tmp_path: Path) -> Path:
def _runtime_joint_requirement(tmp_path: Path) -> Path:
housing_step = _generated_housing_step(tmp_path)
payload = {
"module_type": "planetary_joint_module",
"motor_step": str(MOTOR_STEP.resolve()),
"housing_step": str(housing_step.resolve()),
"platform_profile": str(FIXED_PLATFORM_PROFILE.resolve()),
"reducer_requirement": str(REDUCER_REQ.resolve()),
"placement_policy": "infer_then_write_editable_json",
"motor_to_reducer_relation": "motor_output_to_sun_keyed_input",
@@ -215,6 +219,12 @@ def _passing_reducer_report(run_id: str) -> ValidationReport:
def test_joint_requirement_loads() -> None:
requirement = load_joint_requirement(JOINT_REQ)
assert requirement.module_type == "planetary_joint_module"
assert requirement.platform_id == "actuator_v2_fixed_platform"
assert Path(requirement.motor_step) == STANDARD_MOTOR_STEP
assert Path(requirement.housing_step) == ACTUATOR_HOUSING_STEP
assert requirement.chip_step is None
assert Path(requirement.reference_assembly_step or "") == REFERENCE_ASSEMBLY_STEP
assert requirement.motor_placement_policy == "external_flush_input_end"
assert requirement.motor_to_reducer_relation == "motor_output_to_sun_keyed_input"
assert requirement.housing_relation == "ring_fixed_to_housing"
@@ -237,11 +247,30 @@ def test_cascade_joint_requirement_loads_and_ports_resolve() -> None:
template=load_template(CASCADE_TEMPLATE),
)[0]
assert requirement.joint_output == "s2_carrier_output"
assert requirement.platform_id == "actuator_v2_fixed_platform"
assert requirement.chip_step is None
assert Path(requirement.reference_assembly_step or "") == REFERENCE_ASSEMBLY_STEP
assert requirement.motor_placement_policy == "external_flush_input_end"
assert instance.boundary.input == "s1_sun"
assert instance.boundary.output == "s2_carrier"
assert instance.boundary.fixed_members == ["s1_ring", "s2_ring"]
def test_reference_assembly_recovers_motor_placement() -> None:
_ocp_or_skip()
placements = infer_reference_placements(
reference_step=REFERENCE_ASSEMBLY_STEP,
component_steps={
"housing": ACTUATOR_HOUSING_STEP,
"motor": STANDARD_MOTOR_STEP,
},
)
assert placements["housing"].translation_mm == (0.0, 0.0, 0.0)
assert placements["motor"].translation_mm[0] == pytest.approx(2.0, abs=1e-6)
assert placements["motor"].source == "inferred_from_reference_assembly_occurrence"
def test_joint_requirements_reference_supported_reducer_families() -> None:
for requirement_path in (ROOT / "input" / "requirements" / "joint_modules").glob("*.json"):
requirement = load_joint_requirement(requirement_path)
@@ -250,6 +279,14 @@ def test_joint_requirements_reference_supported_reducer_families() -> None:
assert reducer_requirement.topology_family in SUPPORTED_JOINT_FAMILIES
def test_all_joint_requirements_use_the_fixed_platform_profile() -> None:
platform_ids = {
load_joint_requirement(requirement_path).platform_id
for requirement_path in (ROOT / "input" / "requirements" / "joint_modules").glob("*.json")
}
assert platform_ids == {"actuator_v2_fixed_platform"}
def test_ferguson_reducer_ports_remain_available_for_reducer_exports() -> None:
instance = solve_parameters(
requirement=load_requirement(FERGUSON_REDUCER_REQ),