Jerry
594 lines
22 KiB
Python
594 lines
22 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import json
|
|
import math
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
|
|
SCRIPT = (
|
|
Path(__file__).parents[1]
|
|
/ "skills"
|
|
/ "cad-experience-builder"
|
|
/ "scripts"
|
|
/ "cad_experience.py"
|
|
)
|
|
SPEC = importlib.util.spec_from_file_location("cad_experience", SCRIPT)
|
|
MODULE = importlib.util.module_from_spec(SPEC)
|
|
assert SPEC.loader
|
|
SPEC.loader.exec_module(MODULE)
|
|
|
|
STEP_SCRIPT = (
|
|
Path(__file__).parents[1]
|
|
/ "skills"
|
|
/ "cad-experience-builder"
|
|
/ "scripts"
|
|
/ "step_to_case.py"
|
|
)
|
|
STEP_SPEC = importlib.util.spec_from_file_location("step_to_case", STEP_SCRIPT)
|
|
STEP_MODULE = importlib.util.module_from_spec(STEP_SPEC)
|
|
assert STEP_SPEC.loader
|
|
STEP_SPEC.loader.exec_module(STEP_MODULE)
|
|
|
|
|
|
def make_case(index: int) -> dict:
|
|
return {
|
|
"provenance": {"source_sha256": f"case_{index}"},
|
|
"design_ir": {
|
|
"part_family": "flanged_hub_adapter",
|
|
"features": [
|
|
{"type": "base_flange", "center": [91.0, 52.0]},
|
|
{"type": "hollow_sleeve", "surface_ids": ["face_1"]},
|
|
{"type": "counterbore"},
|
|
],
|
|
"constraints": [{"id": "coaxial_stack", "type": "coaxial"}],
|
|
"parameters": {
|
|
"base_outer_diameter": 91.0,
|
|
"counterbore_spacing": 52.0,
|
|
},
|
|
"normalized_observations": [
|
|
{
|
|
"name": "sleeve_od_to_base_od",
|
|
"value": 0.3 + index * 0.001,
|
|
"numerator_role": "sleeve_outer_diameter",
|
|
"denominator_role": "base_outer_diameter",
|
|
}
|
|
],
|
|
"reconstruction_evidence": {
|
|
"parameter_roles": [
|
|
"overall_long_span",
|
|
"overall_middle_span",
|
|
"overall_short_span",
|
|
],
|
|
"datum_roles": ["part_center", "primary_axis"],
|
|
"feature_roles": ["base_flange", "hollow_sleeve", "counterbore"],
|
|
"relation_roles": ["coaxial_stack"],
|
|
"canonical_stages": [
|
|
{
|
|
"id": "establish_reference_frame",
|
|
"operation": "define_datums",
|
|
"feature_roles": [],
|
|
"reference_roles": ["part_center", "primary_axis"],
|
|
},
|
|
{
|
|
"id": "construct_primary_envelope",
|
|
"operation": "revolve_profile",
|
|
"feature_roles": ["base_flange", "hollow_sleeve"],
|
|
"reference_roles": ["primary_axis"],
|
|
},
|
|
],
|
|
"private_cardinality_evidence": {
|
|
"surface_type_classes": {
|
|
"plane": "repeated",
|
|
"cylinder": "dense",
|
|
},
|
|
"feature_role_count_class": "repeated",
|
|
},
|
|
"validation_roles": ["closed_solid", "coaxial_stack"],
|
|
},
|
|
},
|
|
"experience": {
|
|
"rules": [
|
|
{
|
|
"id": "bound_counterbore_to_host",
|
|
"scope": ["flanged_hub_adapter"],
|
|
"statement": "Terminate a counterbore at its host boundary.",
|
|
"repair": "Bind the cutter extent to the host feature.",
|
|
}
|
|
],
|
|
"validation_targets": [
|
|
{
|
|
"id": "host_boundary",
|
|
"check": "A counterbore does not cross an unrelated adjacent feature.",
|
|
}
|
|
],
|
|
},
|
|
}
|
|
|
|
|
|
def positive_validation(*experience_ids: str) -> dict:
|
|
return {
|
|
"validation_kind": "heldout_experience_ab",
|
|
"method_results": [
|
|
{
|
|
"experience_id": experience_id,
|
|
"independent_case_count": 3,
|
|
"edit_contract_pass_rate": 1.0,
|
|
"baseline_score_delta": 0.5,
|
|
"passed": True,
|
|
}
|
|
for experience_id in experience_ids
|
|
],
|
|
}
|
|
|
|
|
|
class CadExperienceTest(unittest.TestCase):
|
|
def test_default_data_paths_use_shallow_local_layout(self):
|
|
pipeline_root = Path(__file__).parents[1].resolve()
|
|
self.assertEqual(pipeline_root / "input", MODULE.input_root())
|
|
self.assertEqual(pipeline_root / "runs", MODULE.runs_root())
|
|
self.assertEqual(
|
|
pipeline_root / "output" / "library.json",
|
|
MODULE.default_library_path(),
|
|
)
|
|
|
|
def reviewed_proposal(self) -> dict:
|
|
return {
|
|
"id": "constraint.flanged_hub_adapter.shared_axis",
|
|
"kind": "constraint",
|
|
"scope": ["flanged_hub_adapter"],
|
|
"evidence_query": {
|
|
"required_features": ["base_flange", "hollow_sleeve"],
|
|
"required_relations": ["coaxial_stack"],
|
|
},
|
|
"guidance": "Preserve a shared axis across participating feature roles.",
|
|
"semantic_rationale": "The relationship transfers across dimensional variants.",
|
|
}
|
|
|
|
def test_semantic_summary_derives_roles_without_copying_dimensions(self):
|
|
surfaces = [
|
|
{
|
|
"type": "cylinder",
|
|
"axis": [0.0, 1.0, 0.0],
|
|
"location": [0.0, 0.0, 0.0],
|
|
"radius": radius,
|
|
"area": 100.0,
|
|
}
|
|
for radius in (5.0, 10.0, 20.0)
|
|
]
|
|
surfaces.extend(
|
|
{
|
|
"type": "cylinder",
|
|
"axis": [0.0, 1.0, 0.0],
|
|
"location": [offset, 0.0, 0.0],
|
|
"radius": 2.0,
|
|
"area": 20.0,
|
|
}
|
|
for offset in (-15.0, 15.0, 16.0)
|
|
)
|
|
surfaces.append(
|
|
{
|
|
"type": "cone",
|
|
"axis": [0.0, 1.0, 0.0],
|
|
"location": [0.0, 0.0, 0.0],
|
|
"reference_radius": 10.0,
|
|
"semi_angle_radians": 0.5,
|
|
"area": 50.0,
|
|
}
|
|
)
|
|
family, features, constraints, summary = STEP_MODULE._semantic_summary(
|
|
surfaces, [0.0, 0.0, 0.0], [40.0, 20.0, 40.0]
|
|
)
|
|
self.assertEqual("flanged_rotational_part", family)
|
|
self.assertIn(
|
|
"coaxial_cylindrical_stack", {item["type"] for item in features}
|
|
)
|
|
self.assertIn("coaxial_stack", {item["id"] for item in constraints})
|
|
self.assertEqual("y", summary["dominant_axis"])
|
|
self.assertTrue(summary["normalized_observations"])
|
|
reconstruction = STEP_MODULE._reconstruction_evidence(
|
|
surfaces, features, constraints, summary
|
|
)
|
|
self.assertIn("overall_long_span", reconstruction["parameter_roles"])
|
|
self.assertTrue(reconstruction["canonical_stages"])
|
|
self.assertEqual(
|
|
"canonical_reconstruction_plan_not_recovered_history",
|
|
reconstruction["interpretation"],
|
|
)
|
|
parameters = STEP_MODULE._semantic_parameter_candidates(
|
|
surfaces,
|
|
[0.0, 0.0, 0.0],
|
|
[40.0, 20.0, 40.0],
|
|
features,
|
|
summary,
|
|
)
|
|
names = {item["name"] for item in parameters}
|
|
self.assertIn("repeated_feature_count", names)
|
|
self.assertIn("repeated_feature_diameter", names)
|
|
self.assertTrue(
|
|
all(
|
|
item["epistemic_status"] == "inferred_from_final_brep"
|
|
for item in parameters
|
|
)
|
|
)
|
|
|
|
def test_non_solid_teacher_is_rejected_from_induction(self):
|
|
payload = make_case(1)
|
|
payload["geometry_evidence"] = {
|
|
"valid": False,
|
|
"reconstruction_eligible": False,
|
|
"solid_count": 0,
|
|
"volume": 0.0,
|
|
"rejection_reasons": ["no_closed_solid", "zero_volume"],
|
|
}
|
|
normalized = MODULE.normalize_case(payload)
|
|
self.assertFalse(normalized["reconstruction_eligible"])
|
|
self.assertIn("no_closed_solid", normalized["rejection_reasons"])
|
|
|
|
def test_spur_gear_flanks_are_not_mislabeled_as_hole_pattern(self):
|
|
surfaces = [
|
|
{
|
|
"type": "cylinder",
|
|
"axis": [1.0, 0.0, 0.0],
|
|
"location": [0.0, 0.0, 0.0],
|
|
"radius": radius,
|
|
"area": 100.0,
|
|
}
|
|
for radius in (3.0, 8.8)
|
|
]
|
|
for index in range(40):
|
|
angle = 2.0 * math.pi * index / 40.0
|
|
surfaces.append(
|
|
{
|
|
"type": "cylinder",
|
|
"axis": [1.0, 0.0, 0.0],
|
|
"location": [
|
|
0.0,
|
|
7.5 * math.cos(angle),
|
|
7.5 * math.sin(angle),
|
|
],
|
|
"radius": 3.2,
|
|
"area": 10.0,
|
|
}
|
|
)
|
|
surfaces.extend(
|
|
{"type": "plane", "area": 250.0} for _ in range(2)
|
|
)
|
|
|
|
family, features, constraints, summary = STEP_MODULE._semantic_summary(
|
|
surfaces, [4.0, 0.0, 0.0], [8.0, 17.6, 17.6]
|
|
)
|
|
feature_roles = {item["type"] for item in features}
|
|
self.assertEqual("external_spur_gear", family)
|
|
self.assertEqual(20, summary["inferred_tooth_count"])
|
|
self.assertIn("external_spur_gear", feature_roles)
|
|
self.assertIn("equal_angular_tooth_pattern", feature_roles)
|
|
self.assertNotIn("repeated_axial_hole_pattern", feature_roles)
|
|
self.assertIn(
|
|
"equal_angular_tooth_pattern",
|
|
{item["id"] for item in constraints},
|
|
)
|
|
|
|
parameters = STEP_MODULE._semantic_parameter_candidates(
|
|
surfaces,
|
|
[4.0, 0.0, 0.0],
|
|
[8.0, 17.6, 17.6],
|
|
features,
|
|
summary,
|
|
)
|
|
values = {item["name"]: item["value"] for item in parameters}
|
|
self.assertEqual(20, values["tooth_count"])
|
|
self.assertAlmostEqual(0.8, values["gear_module"], places=6)
|
|
self.assertEqual(8.0, values["gear_width"])
|
|
self.assertNotIn("repeated_feature_count", values)
|
|
|
|
def test_sha256_identity_deduplicates_renamed_sources(self):
|
|
digest = "0" * 64
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
root = Path(temporary)
|
|
for index in range(2):
|
|
payload = make_case(index)
|
|
payload["provenance"]["source_sha256"] = digest
|
|
(root / f"renamed-{index}.json").write_text(json.dumps(payload))
|
|
cases, stats = MODULE.load_cases(root)
|
|
self.assertEqual(1, len(cases))
|
|
self.assertEqual(1, stats["duplicate_case_count"])
|
|
|
|
def test_conditional_feature_context_does_not_dilute_valid_method(self):
|
|
proposal = self.reviewed_proposal()
|
|
proposal["evidence_query"]["context_features"] = [
|
|
"base_flange",
|
|
"hollow_sleeve",
|
|
]
|
|
target = [MODULE.normalize_case(make_case(index)) for index in range(20)]
|
|
unrelated = []
|
|
for index in range(80):
|
|
payload = make_case(index + 100)
|
|
payload["design_ir"]["features"] = [{"type": "base_flange"}]
|
|
payload["design_ir"]["constraints"] = []
|
|
unrelated.append(MODULE.normalize_case(payload))
|
|
library = MODULE.build_reviewed_library(
|
|
target + unrelated,
|
|
[MODULE.normalize_proposals({
|
|
"draft_kind": "llm_generalized_experience_proposals",
|
|
"proposals": [proposal],
|
|
})[0]],
|
|
20,
|
|
0.8,
|
|
{"duplicate_case_count": 0, "rejected_case_count": 0},
|
|
positive_validation(proposal["id"]),
|
|
)
|
|
self.assertEqual(1, len(library["experiences"]))
|
|
|
|
def test_single_case_never_promotes(self):
|
|
case = MODULE.normalize_case(make_case(1))
|
|
library = MODULE.induce(
|
|
[case],
|
|
2,
|
|
0.8,
|
|
{"duplicate_case_count": 0, "rejected_case_count": 0},
|
|
)
|
|
self.assertEqual(library["status"], "collecting_evidence")
|
|
self.assertEqual(library["experiences"], [])
|
|
self.assertTrue(library["candidate_experiences"])
|
|
self.assertTrue(
|
|
all(
|
|
item["promotion_state"] == "candidate"
|
|
for item in library["candidate_experiences"]
|
|
)
|
|
)
|
|
self.assertTrue(
|
|
all(
|
|
item["consumer_policy"]
|
|
== "visible_for_review_but_not_available_to_cad_router"
|
|
for item in library["candidate_experiences"]
|
|
)
|
|
)
|
|
|
|
def test_batch_promotes_methods_without_instance_values(self):
|
|
cases = [MODULE.normalize_case(make_case(index)) for index in range(20)]
|
|
library = MODULE.induce(
|
|
cases,
|
|
20,
|
|
0.8,
|
|
{"duplicate_case_count": 0, "rejected_case_count": 0},
|
|
)
|
|
self.assertEqual(MODULE.audit_library(library), [])
|
|
rendered = json.dumps(library)
|
|
self.assertNotIn('"parameters"', rendered)
|
|
self.assertNotIn("91.0", rendered)
|
|
self.assertNotIn("52.0", rendered)
|
|
self.assertNotIn('"center"', rendered)
|
|
ids = {item["id"] for item in library["experiences"]}
|
|
self.assertIn("rule.bound_counterbore_to_host", ids)
|
|
self.assertIn("constraint.flanged_hub_adapter.coaxial_stack", ids)
|
|
self.assertIn("distribution.sleeve_od_to_base_od", ids)
|
|
|
|
def test_audit_allows_aggregate_validation_counts_only(self):
|
|
library = {
|
|
"schema_version": "2.0",
|
|
"library_kind": "generalized_cad_experience",
|
|
"experiences": [
|
|
{
|
|
"id": "edit_strategy.test",
|
|
"support": 20,
|
|
"validation": {
|
|
"heldout_pair_count": 20,
|
|
"treatment_pass_count": 20,
|
|
"regressed_pair_count": 0,
|
|
"pass_rate_delta": 0.25,
|
|
},
|
|
}
|
|
],
|
|
"candidate_experiences": [],
|
|
}
|
|
self.assertEqual([], MODULE.audit_library(library))
|
|
library["experiences"][0]["validation"]["radius"] = 3.0
|
|
errors = MODULE.audit_library(library)
|
|
self.assertTrue(any("forbidden instance key" in error for error in errors))
|
|
|
|
def test_family_support_is_not_diluted_by_unrelated_families(self):
|
|
target = [MODULE.normalize_case(make_case(index)) for index in range(20)]
|
|
unrelated = []
|
|
for index in range(100):
|
|
payload = make_case(index + 1000)
|
|
payload["design_ir"]["part_family"] = "gear"
|
|
payload["design_ir"]["features"] = [
|
|
{"type": "gear_blank"},
|
|
{"type": "gear_teeth"},
|
|
]
|
|
unrelated.append(MODULE.normalize_case(payload))
|
|
library = MODULE.induce(
|
|
target + unrelated,
|
|
20,
|
|
0.8,
|
|
{"duplicate_case_count": 0, "rejected_case_count": 0},
|
|
)
|
|
ids = {item["id"] for item in library["experiences"]}
|
|
self.assertIn(
|
|
"motif.flanged_hub_adapter.base_flange+hollow_sleeve", ids
|
|
)
|
|
|
|
def test_query_returns_backend_neutral_context(self):
|
|
cases = [MODULE.normalize_case(make_case(index)) for index in range(20)]
|
|
library = MODULE.induce(
|
|
cases,
|
|
20,
|
|
0.8,
|
|
{"duplicate_case_count": 0, "rejected_case_count": 0},
|
|
)
|
|
result = MODULE.query_library(
|
|
library,
|
|
"flanged_hub_adapter",
|
|
{"base_flange", "hollow_sleeve", "counterbore"},
|
|
)
|
|
self.assertTrue(result["experiences"])
|
|
self.assertFalse(result["policy"]["contains_instance_parameters"])
|
|
|
|
def test_llm_proposal_is_candidate_until_evidence_threshold(self):
|
|
proposal = self.reviewed_proposal()
|
|
one_case = MODULE.normalize_case(make_case(1))
|
|
candidate_library = MODULE.build_reviewed_library(
|
|
[one_case],
|
|
[proposal],
|
|
20,
|
|
0.8,
|
|
{"duplicate_case_count": 0, "rejected_case_count": 0},
|
|
)
|
|
self.assertEqual([], candidate_library["experiences"])
|
|
self.assertEqual(1, len(candidate_library["candidate_experiences"]))
|
|
self.assertFalse(candidate_library["policy"]["router_consumable"])
|
|
|
|
cases = [MODULE.normalize_case(make_case(index)) for index in range(20)]
|
|
promoted_library = MODULE.build_reviewed_library(
|
|
cases,
|
|
[proposal],
|
|
20,
|
|
0.8,
|
|
{"duplicate_case_count": 0, "rejected_case_count": 0},
|
|
positive_validation(proposal["id"]),
|
|
)
|
|
self.assertEqual(1, len(promoted_library["experiences"]))
|
|
self.assertEqual([], promoted_library["candidate_experiences"])
|
|
|
|
def test_parameter_naming_requires_positive_heldout_validation(self):
|
|
cases = []
|
|
for index in range(20):
|
|
payload = make_case(index)
|
|
payload["design_ir"]["semantic_parameter_candidates"] = [
|
|
{
|
|
"name": "bolt_circle_diameter",
|
|
"unit": "mm",
|
|
"confidence": 0.9,
|
|
"semantic_aliases": ["hole_spacing"],
|
|
"affects_feature_roles": ["mounting_pattern"],
|
|
"epistemic_status": "inferred_from_final_brep",
|
|
"edit_contract": {
|
|
"expected_change": "pattern_radius_changes"
|
|
},
|
|
}
|
|
]
|
|
cases.append(MODULE.normalize_case(payload))
|
|
raw = {
|
|
"id": "parameter.flange.bolt_circle_diameter",
|
|
"kind": "parameter_naming",
|
|
"scope": ["flanged_hub_adapter"],
|
|
"evidence_query": {
|
|
"required_features": ["base_flange"],
|
|
"required_relations": [],
|
|
"required_parameters": ["bolt_circle_diameter"],
|
|
},
|
|
"guidance": "Expose the circular mounting pattern diameter as an editable semantic parameter.",
|
|
"semantic_rationale": "The named role preserves edit intent across dimensional variants.",
|
|
"canonical_parameter": "bolt_circle_diameter",
|
|
"semantic_aliases": ["hole_spacing"],
|
|
}
|
|
proposal = MODULE.normalize_proposals(
|
|
{
|
|
"draft_kind": "llm_generalized_experience_proposals",
|
|
"proposals": [raw],
|
|
}
|
|
)[0]
|
|
unvalidated = MODULE.build_reviewed_library(
|
|
cases,
|
|
[proposal],
|
|
20,
|
|
0.8,
|
|
{"duplicate_case_count": 0, "rejected_case_count": 0},
|
|
)
|
|
self.assertEqual([], unvalidated["experiences"])
|
|
self.assertEqual(
|
|
"awaiting_positive_heldout_ab",
|
|
unvalidated["candidate_experiences"][0]["validation_state"],
|
|
)
|
|
validated = MODULE.build_reviewed_library(
|
|
cases,
|
|
[proposal],
|
|
20,
|
|
0.8,
|
|
{"duplicate_case_count": 0, "rejected_case_count": 0},
|
|
positive_validation(raw["id"]),
|
|
)
|
|
self.assertEqual("parameter_naming", validated["experiences"][0]["kind"])
|
|
|
|
def test_numeric_instance_answer_is_rejected_from_llm_draft(self):
|
|
proposal = self.reviewed_proposal()
|
|
proposal["guidance"] = "Always use a diameter of 91 millimeters."
|
|
with self.assertRaisesRegex(ValueError, "instance numbers"):
|
|
MODULE.normalize_proposals(
|
|
{
|
|
"draft_kind": "llm_generalized_experience_proposals",
|
|
"proposals": [proposal],
|
|
}
|
|
)
|
|
|
|
def test_reconstruction_grammar_promotes_without_instance_values(self):
|
|
proposal = {
|
|
"id": "reconstruction.flanged_hub_adapter.primary",
|
|
"kind": "reconstruction_grammar",
|
|
"scope": ["flanged_hub_adapter"],
|
|
"evidence_query": {
|
|
"required_features": ["base_flange", "hollow_sleeve"],
|
|
"required_relations": ["coaxial_stack"],
|
|
"context_features": ["base_flange", "hollow_sleeve"],
|
|
},
|
|
"guidance": "Establish shared datums before composing the primary envelope.",
|
|
"semantic_rationale": "The symbolic stages transfer across dimensional variants.",
|
|
"reconstruction_grammar": {
|
|
"parameter_roles": [
|
|
"overall_long_span",
|
|
"overall_middle_span",
|
|
"overall_short_span",
|
|
],
|
|
"datum_roles": ["part_center", "primary_axis"],
|
|
"feature_roles": ["base_flange", "hollow_sleeve"],
|
|
"relation_roles": ["coaxial_stack"],
|
|
"canonical_stages": [
|
|
{
|
|
"id": "establish_reference_frame",
|
|
"operation": "define_datums",
|
|
"feature_roles": [],
|
|
"reference_roles": ["part_center", "primary_axis"],
|
|
},
|
|
{
|
|
"id": "construct_primary_envelope",
|
|
"operation": "revolve_profile",
|
|
"feature_roles": ["base_flange", "hollow_sleeve"],
|
|
"reference_roles": ["primary_axis"],
|
|
},
|
|
],
|
|
"cardinality_classes": ["plane_repeated", "cylinder_dense"],
|
|
"validation_roles": ["closed_solid", "coaxial_stack"],
|
|
},
|
|
}
|
|
normalized = MODULE.normalize_proposals(
|
|
{
|
|
"draft_kind": "llm_generalized_experience_proposals",
|
|
"proposals": [proposal],
|
|
}
|
|
)
|
|
cases = [MODULE.normalize_case(make_case(index)) for index in range(20)]
|
|
library = MODULE.build_reviewed_library(
|
|
cases,
|
|
normalized,
|
|
20,
|
|
0.8,
|
|
{"duplicate_case_count": 0, "rejected_case_count": 0},
|
|
positive_validation(proposal["id"]),
|
|
)
|
|
self.assertEqual([], MODULE.audit_library(library))
|
|
self.assertEqual("reconstruction_grammar", library["experiences"][0]["kind"])
|
|
rendered = json.dumps(library)
|
|
self.assertNotIn("91.0", rendered)
|
|
self.assertNotIn('"parameters"', rendered)
|
|
|
|
def test_direct_distill_without_model_draft_is_blocked(self):
|
|
with self.assertRaisesRegex(SystemExit, "statistical distillation is disabled"):
|
|
MODULE.main(["distill"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|