404 lines
16 KiB
Python
404 lines
16 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from copy import deepcopy
|
|
from typing import Any
|
|
|
|
from jsonschema import Draft202012Validator
|
|
|
|
from app.services.engine_service import feature_atomic_contract
|
|
|
|
|
|
class CanonicalFragmentError(ValueError):
|
|
def __init__(self, message: str, *, path: str = "fragment", code: str = "CDSL_SCHEMA_INVALID") -> None:
|
|
super().__init__(message)
|
|
self.path = path
|
|
self.code = code
|
|
|
|
|
|
def _point(size: int) -> dict[str, Any]:
|
|
return {
|
|
"type": "array",
|
|
"items": {"type": "number"},
|
|
"minItems": size,
|
|
"maxItems": size,
|
|
}
|
|
|
|
|
|
def _workplane_schema() -> dict[str, Any]:
|
|
point3 = _point(3)
|
|
return {
|
|
"type": "object",
|
|
"properties": {
|
|
"origin_mm": deepcopy(point3),
|
|
"x_dir": deepcopy(point3),
|
|
"normal": deepcopy(point3),
|
|
},
|
|
"required": ["origin_mm", "x_dir", "normal"],
|
|
"additionalProperties": False,
|
|
}
|
|
|
|
|
|
def _analytic_segment_schema() -> dict[str, Any]:
|
|
point2 = _point(2)
|
|
return {
|
|
"oneOf": [
|
|
{
|
|
"type": "object",
|
|
"properties": {
|
|
"type": {"const": "line"},
|
|
"start": deepcopy(point2),
|
|
"end": deepcopy(point2),
|
|
},
|
|
"required": ["type", "start", "end"],
|
|
"additionalProperties": False,
|
|
},
|
|
{
|
|
"type": "object",
|
|
"properties": {
|
|
"type": {"const": "arc"},
|
|
"start": deepcopy(point2),
|
|
"end": deepcopy(point2),
|
|
"center": deepcopy(point2),
|
|
"radius_mm": {"type": "number", "exclusiveMinimum": 0},
|
|
"clockwise": {"type": "boolean"},
|
|
},
|
|
"required": ["type", "start", "end", "center", "radius_mm"],
|
|
"additionalProperties": False,
|
|
},
|
|
{
|
|
"type": "object",
|
|
"properties": {
|
|
"type": {"const": "circle"},
|
|
"center": deepcopy(point2),
|
|
"radius_mm": {"type": "number", "exclusiveMinimum": 0},
|
|
},
|
|
"required": ["type", "center", "radius_mm"],
|
|
"additionalProperties": False,
|
|
},
|
|
]
|
|
}
|
|
|
|
|
|
def canonical_profile_schema() -> dict[str, Any]:
|
|
point2 = _point(2)
|
|
segment = _analytic_segment_schema()
|
|
contour = {
|
|
"type": "object",
|
|
"properties": {
|
|
"role": {"enum": ["outer", "inner"]},
|
|
"closed": {"const": True},
|
|
"segments": {"type": "array", "minItems": 1, "items": segment},
|
|
},
|
|
"required": ["role", "closed", "segments"],
|
|
"additionalProperties": False,
|
|
}
|
|
return {
|
|
"oneOf": [
|
|
{
|
|
"type": "object",
|
|
"properties": {
|
|
"type": {"const": "circle"},
|
|
"center": deepcopy(point2),
|
|
"radius_mm": {"type": "number", "exclusiveMinimum": 0},
|
|
},
|
|
"required": ["type", "radius_mm"],
|
|
"additionalProperties": False,
|
|
},
|
|
{
|
|
"type": "object",
|
|
"properties": {
|
|
"type": {"const": "polygon"},
|
|
"vertices": {"type": "array", "minItems": 3, "items": deepcopy(point2)},
|
|
},
|
|
"required": ["type", "vertices"],
|
|
"additionalProperties": False,
|
|
},
|
|
{
|
|
"type": "object",
|
|
"properties": {
|
|
"type": {"const": "analytic_contours"},
|
|
"contours": {"type": "array", "minItems": 1, "maxItems": 2, "items": contour},
|
|
},
|
|
"required": ["type", "contours"],
|
|
"additionalProperties": False,
|
|
},
|
|
]
|
|
}
|
|
|
|
|
|
def _end_condition_schema() -> dict[str, Any]:
|
|
return {
|
|
"type": "object",
|
|
"properties": {
|
|
"type": {"type": "string", "minLength": 1},
|
|
"solidworks_code": {"type": "integer"},
|
|
},
|
|
"required": ["type", "solidworks_code"],
|
|
"additionalProperties": False,
|
|
}
|
|
|
|
|
|
def _axis_schema() -> dict[str, Any]:
|
|
point3 = _point(3)
|
|
return {
|
|
"type": "object",
|
|
"properties": {"origin_mm": deepcopy(point3), "direction": deepcopy(point3)},
|
|
"required": ["origin_mm", "direction"],
|
|
"additionalProperties": False,
|
|
}
|
|
|
|
|
|
def _hole_positions_schema() -> dict[str, Any]:
|
|
return {
|
|
"type": "array",
|
|
"minItems": 1,
|
|
"items": {
|
|
"type": "object",
|
|
"properties": {"mm": _point(3)},
|
|
"required": ["mm"],
|
|
"additionalProperties": False,
|
|
},
|
|
}
|
|
|
|
|
|
def _parameter_schema(atomic_id: str, contract: dict[str, Any]) -> dict[str, Any]:
|
|
positive = {"type": "number", "exclusiveMinimum": 0}
|
|
number = {"type": "number"}
|
|
integer = {"type": "integer", "minimum": 1}
|
|
point3 = _point(3)
|
|
properties: dict[str, Any] = {}
|
|
required = [name for name in contract["required_params"] if name not in {"host_face", "source_feature_ids", "mirror_plane"}]
|
|
|
|
if atomic_id.startswith("extrude_"):
|
|
properties = {
|
|
"distance_mm": deepcopy(number),
|
|
"reverse_distance_mm": deepcopy(number),
|
|
"reverse": {"type": "boolean"},
|
|
"end_condition": _end_condition_schema(),
|
|
"reverse_end_condition": _end_condition_schema(),
|
|
}
|
|
elif atomic_id.startswith("revolve_"):
|
|
properties = {
|
|
"angle_deg": {"type": "number", "exclusiveMinimum": 0, "maximum": 360},
|
|
"axis": _axis_schema(),
|
|
"reverse": {"type": "boolean"},
|
|
}
|
|
elif atomic_id.startswith("hole_"):
|
|
properties = {
|
|
"diameter_mm": deepcopy(positive),
|
|
"depth_mm": deepcopy(positive),
|
|
"positions": _hole_positions_schema(),
|
|
"drill_angle_rad": {"type": "number", "exclusiveMinimum": 0, "maximum": 3.141592653589793},
|
|
"countersink_diameter_mm": deepcopy(positive),
|
|
"countersink_angle_rad": {"type": "number", "exclusiveMinimum": 0, "maximum": 3.141592653589793},
|
|
"counterbore_diameter_mm": deepcopy(positive),
|
|
"counterbore_depth_mm": deepcopy(positive),
|
|
}
|
|
elif atomic_id == "sphere_add":
|
|
properties = {"radius_mm": deepcopy(positive), "center_mm": deepcopy(point3)}
|
|
elif atomic_id == "fillet":
|
|
properties = {"radius_mm": deepcopy(positive), "tangent_propagation": {"type": "boolean"}}
|
|
elif atomic_id == "chamfer":
|
|
properties = {
|
|
"distance_mm": deepcopy(positive),
|
|
"distance_2_mm": deepcopy(positive),
|
|
"angle_rad": {"type": "number", "exclusiveMinimum": 0, "maximum": 3.141592653589793},
|
|
}
|
|
elif atomic_id == "pattern_linear":
|
|
properties = {
|
|
"direction_1": deepcopy(point3),
|
|
"spacing_1_mm": deepcopy(positive),
|
|
"pattern_count_1": deepcopy(integer),
|
|
"direction_2": deepcopy(point3),
|
|
"spacing_2_mm": deepcopy(positive),
|
|
"pattern_count_2": deepcopy(integer),
|
|
}
|
|
elif atomic_id == "pattern_mirror":
|
|
properties = {}
|
|
elif atomic_id == "reference_axis":
|
|
properties = {"axis": _axis_schema()}
|
|
elif atomic_id == "reference_plane":
|
|
properties = {"plane": _workplane_schema()}
|
|
elif atomic_id == "hole_wizard":
|
|
required = ["hole_type", "diameter_mm", "depth_mm"]
|
|
properties = {
|
|
"hole_type": {"type": "string", "minLength": 1},
|
|
"diameter_mm": deepcopy(positive),
|
|
"depth_mm": deepcopy(positive),
|
|
"positions": _hole_positions_schema(),
|
|
"thread": {"type": "object"},
|
|
"countersink": {"type": "object"},
|
|
"counterbore": {"type": "object"},
|
|
}
|
|
else:
|
|
for name in [*contract["required_params"], *contract["optional_params"]]:
|
|
if name not in {"host_face", "source_feature_ids", "mirror_plane"}:
|
|
properties[name] = {}
|
|
|
|
allowed = set(required) | {
|
|
name for name in contract["optional_params"]
|
|
if name not in {"host_face", "source_feature_ids", "mirror_plane"}
|
|
}
|
|
properties = {name: value for name, value in properties.items() if name in allowed}
|
|
return {
|
|
"type": "object",
|
|
"properties": properties,
|
|
"required": required,
|
|
"additionalProperties": False,
|
|
}
|
|
|
|
|
|
def _selector_limits(atomic_id: str, contract: dict[str, Any]) -> tuple[int, int] | None:
|
|
if atomic_id.startswith("hole_") or atomic_id == "hole_wizard":
|
|
return 1, 1
|
|
slot = contract.get("selector_slot")
|
|
if not isinstance(slot, dict):
|
|
return None
|
|
return int(slot.get("min_items") or 1), int(slot.get("max_items") or 1)
|
|
|
|
|
|
def build_operation_fragment_schema(
|
|
engine: Any,
|
|
atomic_id: str,
|
|
*,
|
|
correction: dict[str, Any] | None = None,
|
|
allow_multi_feature: bool = False,
|
|
) -> dict[str, Any]:
|
|
del correction
|
|
if allow_multi_feature:
|
|
raise ValueError("Canonical autonomous authoring currently permits exactly one feature")
|
|
contract = feature_atomic_contract(engine, atomic_id)
|
|
feature_properties: dict[str, Any] = {
|
|
"atomic_id": {"const": atomic_id},
|
|
"params": _parameter_schema(atomic_id, contract),
|
|
}
|
|
feature_required = ["atomic_id", "params"]
|
|
selector_limits = _selector_limits(atomic_id, contract)
|
|
if selector_limits is not None:
|
|
minimum, maximum = selector_limits
|
|
feature_properties["selector_tokens"] = {
|
|
"type": "array",
|
|
"items": {"type": "string", "minLength": 1},
|
|
"minItems": minimum,
|
|
"maxItems": maximum,
|
|
"uniqueItems": True,
|
|
}
|
|
feature_required.append("selector_tokens")
|
|
feature_schema = {
|
|
"type": "object",
|
|
"properties": feature_properties,
|
|
"required": feature_required,
|
|
"additionalProperties": False,
|
|
}
|
|
properties: dict[str, Any] = {"feature": feature_schema}
|
|
required = ["feature"]
|
|
if contract["requires_sketch"]:
|
|
properties["sketch"] = {
|
|
"type": "object",
|
|
"properties": {"workplane": _workplane_schema(), "profile": canonical_profile_schema()},
|
|
"required": ["workplane", "profile"],
|
|
"additionalProperties": False,
|
|
}
|
|
required.insert(0, "sketch")
|
|
return {
|
|
"type": "object",
|
|
"properties": properties,
|
|
"required": required,
|
|
"additionalProperties": False,
|
|
}
|
|
|
|
|
|
def canonical_fragment_example(engine: Any, atomic_id: str) -> dict[str, Any]:
|
|
contract = feature_atomic_contract(engine, atomic_id)
|
|
feature: dict[str, Any] = {"atomic_id": atomic_id, "params": {}}
|
|
params = feature["params"]
|
|
if atomic_id.startswith("extrude_"):
|
|
params["distance_mm"] = 10
|
|
if "reverse_distance_mm" in contract["required_params"]:
|
|
params["reverse_distance_mm"] = 5
|
|
elif atomic_id.startswith("revolve_"):
|
|
params.update({"angle_deg": 360, "axis": {"origin_mm": [0, 0, 0], "direction": [0, 0, 1]}})
|
|
elif atomic_id.startswith("hole_"):
|
|
params.update({"diameter_mm": 10, "depth_mm": 20, "positions": [{"mm": [0, 0, 0]}]})
|
|
if atomic_id == "hole_countersink":
|
|
params.update({"countersink_diameter_mm": 16, "countersink_angle_rad": 1.5707963267948966})
|
|
if atomic_id == "hole_counterbore":
|
|
params.update({"counterbore_diameter_mm": 16, "counterbore_depth_mm": 4})
|
|
feature["selector_tokens"] = ["face-token-from-current-topology"]
|
|
elif atomic_id == "hole_wizard":
|
|
params.update({"hole_type": "simple", "diameter_mm": 10, "depth_mm": 20})
|
|
feature["selector_tokens"] = ["face-token-from-current-topology"]
|
|
elif atomic_id == "sphere_add":
|
|
params.update({"radius_mm": 10, "center_mm": [0, 0, 0]})
|
|
elif atomic_id == "fillet":
|
|
params["radius_mm"] = 0.5
|
|
feature["selector_tokens"] = ["edge-token-from-current-topology"]
|
|
elif atomic_id == "chamfer":
|
|
params["distance_mm"] = 0.5
|
|
feature["selector_tokens"] = ["edge-token-from-current-topology"]
|
|
elif atomic_id == "pattern_linear":
|
|
params.update({"direction_1": [1, 0, 0], "spacing_1_mm": 10, "pattern_count_1": 2})
|
|
elif atomic_id == "pattern_mirror":
|
|
feature["selector_tokens"] = ["plane-token-from-current-topology"]
|
|
elif atomic_id == "reference_axis":
|
|
params["axis"] = {"origin_mm": [0, 0, 0], "direction": [0, 0, 1]}
|
|
elif atomic_id == "reference_plane":
|
|
params["plane"] = {"origin_mm": [0, 0, 0], "x_dir": [1, 0, 0], "normal": [0, 0, 1]}
|
|
|
|
fragment: dict[str, Any] = {"feature": feature}
|
|
if contract["requires_sketch"]:
|
|
fragment["sketch"] = {
|
|
"workplane": {"origin_mm": [0, 0, 0], "x_dir": [1, 0, 0], "normal": [0, 0, 1]},
|
|
"profile": {"type": "polygon", "vertices": [[-10, -5], [10, -5], [10, 5], [-10, 5]]},
|
|
}
|
|
fragment = {"sketch": fragment["sketch"], "feature": feature}
|
|
return fragment
|
|
|
|
|
|
def operation_contract_hash(atomic_id: str, schema: dict[str, Any]) -> str:
|
|
encoded = json.dumps({"atomic_id": atomic_id, "schema": schema}, sort_keys=True, separators=(",", ":"))
|
|
return hashlib.sha256(encoded.encode("utf-8")).hexdigest()
|
|
|
|
|
|
def validate_canonical_fragment(engine: Any, fragment: dict[str, Any], *, expected_atomic_id: str = "") -> str:
|
|
feature = fragment.get("feature") if isinstance(fragment, dict) else None
|
|
atomic_id = str((feature or {}).get("atomic_id") or "") if isinstance(feature, dict) else ""
|
|
if expected_atomic_id and atomic_id and atomic_id != expected_atomic_id:
|
|
raise CanonicalFragmentError(
|
|
f"OPERATION_CONTRACT_MISMATCH: expected {expected_atomic_id}, received {atomic_id}",
|
|
path="fragment.feature.atomic_id",
|
|
code="OPERATION_CONTRACT_MISMATCH",
|
|
)
|
|
if not atomic_id:
|
|
raise CanonicalFragmentError(
|
|
"CDSL_SCHEMA_INVALID: canonical fragment requires fragment.feature.atomic_id",
|
|
path="fragment.feature.atomic_id",
|
|
)
|
|
schema = build_operation_fragment_schema(engine, atomic_id)
|
|
errors = sorted(Draft202012Validator(schema).iter_errors(fragment), key=lambda error: (list(error.absolute_path), error.message))
|
|
if errors:
|
|
error = errors[0]
|
|
path = "fragment" + "".join(
|
|
f"[{item}]" if isinstance(item, int) else f".{item}"
|
|
for item in error.absolute_path
|
|
)
|
|
raise CanonicalFragmentError(f"CDSL_SCHEMA_INVALID at {path}: {error.message}", path=path)
|
|
profile = ((fragment.get("sketch") or {}).get("profile") or {}) if isinstance(fragment, dict) else {}
|
|
if isinstance(profile, dict) and profile.get("type") == "analytic_contours":
|
|
roles = [str(item.get("role") or "") for item in profile.get("contours") or () if isinstance(item, dict)]
|
|
if roles.count("outer") != 1 or roles.count("inner") > 1:
|
|
raise CanonicalFragmentError(
|
|
"CDSL_PROFILE_MULTIPLE_OUTERS: analytic_contours requires exactly one outer contour and at most one inner contour",
|
|
path="fragment.sketch.profile.contours",
|
|
code="CDSL_PROFILE_MULTIPLE_OUTERS",
|
|
)
|
|
if roles.count("outer") == 0 and roles.count("inner"):
|
|
raise CanonicalFragmentError(
|
|
"CDSL_PROFILE_INNER_WITHOUT_OUTER: analytic_contours cannot contain an inner contour without an outer contour",
|
|
path="fragment.sketch.profile.contours",
|
|
code="CDSL_PROFILE_INNER_WITHOUT_OUTER",
|
|
)
|
|
return atomic_id
|