from __future__ import annotations import math from collections.abc import Callable, Iterable, Mapping, Sequence from pathlib import Path from .relation_formulas import rewrite_relation_formula_ids def validate_scdm_edit_result( edit_result: Mapping[str, object], *, before_signature: Mapping[str, object] | None = None, before_cache: Mapping[str, object] | None = None, after_cache: Mapping[str, object] | None = None, capability_key: str = "", expected_target: object = None, edited_object_id: str = "", tolerance: float = 1.0e-6, brep_validator: Callable[[Path], Mapping[str, object]] | None = None, ) -> dict[str, object]: if edit_result.get("ok") is not True: return { "ok": False, "reason": str(edit_result.get("reason") or "edit-failed"), "message": str(edit_result.get("message") or "SCDM edit did not succeed."), "editResult": dict(edit_result), } output_step = Path(str(edit_result.get("output_step") or edit_result.get("outputStep") or "")).expanduser() if not output_step.is_file(): return { "ok": False, "reason": "missing-output-step", "message": f"SCDM result STEP does not exist: {output_step}", "editResult": dict(edit_result), } if brep_validator is not None: brep = dict(brep_validator(output_step)) if brep.get("ok") is not True: return { "ok": False, "reason": str(brep.get("reason") or "brep-invalid"), "message": str(brep.get("message") or "OCCT rejected the result STEP."), "brep": brep, "editResult": dict(edit_result), } else: brep = {"ok": None, "reason": "not-run", "message": "B-Rep validation callback was not provided."} summary_check: dict[str, object] = {"ok": None, "reason": "not-run", "message": "SCDM summary check needs the old and new caches."} topology_check: dict[str, object] = {"ok": None, "reason": "not-run", "message": "Object drift check needs the old and new SCDM caches."} if before_cache and after_cache: summary_check = check_scdm_summary_delta(before_cache, after_cache, capability_key=capability_key) if summary_check.get("ok") is False: return { "ok": False, "reason": str(summary_check.get("reason") or "summary-drift"), "message": str(summary_check.get("message") or "SCDM result changed the model summary too much."), "summaryCheck": summary_check, "brep": brep, "editResult": dict(edit_result), } topology_check = check_scdm_unedited_objects( before_cache, after_cache, edited_object_id=edited_object_id, edited_signature=before_signature, capability_key=capability_key, ) if topology_check.get("ok") is not True: return { "ok": False, "reason": str(topology_check.get("reason") or "unexpected-object-drift"), "message": str(topology_check.get("message") or "SCDM result changed unrelated recognized objects."), "topologyCheck": topology_check, "brep": brep, "editResult": dict(edit_result), } matched: dict[str, object] | None = None if before_signature and after_cache: match = match_scdm_object_by_signature(before_signature, after_cache, capability_key=capability_key) status = str(match.get("status") or "") if status != "unique": return { "ok": False, "reason": f"object-match-{status or 'failed'}", "message": str(match.get("message") or "Edited object could not be uniquely matched in the new SCDM cache."), "match": match, "brep": brep, "editResult": dict(edit_result), } candidate = match.get("object") if isinstance(candidate, Mapping): matched = dict(candidate) if expected_target is not None and matched is not None: check = check_scdm_target(matched, capability_key=capability_key, expected_target=expected_target, tolerance=tolerance) if check.get("ok") is not True: return { "ok": False, "reason": str(check.get("reason") or "target-check-failed"), "message": str(check.get("message") or "SCDM result did not reach the target value."), "targetCheck": check, "matchedObject": matched, "brep": brep, "editResult": dict(edit_result), } else: check = {"ok": None, "reason": "not-run", "message": "Target check needs a matched object and an expected target."} return { "ok": True, "reason": "ok", "message": "SCDM edit result passed the available validation checks.", "output_step": str(output_step), "matchedObject": matched, "targetCheck": check, "summaryCheck": summary_check, "topologyCheck": topology_check, "brep": brep, "editResult": dict(edit_result), } def match_scdm_object_by_signature( before_signature: Mapping[str, object], after_cache: Mapping[str, object], *, capability_key: str = "", min_score: float = 5.0, unique_margin: float = 0.75, ) -> dict[str, object]: candidates = [] objects = after_cache.get("objects") if not isinstance(objects, list): return {"status": "none", "message": "New SCDM cache does not contain objects.", "candidates": []} for raw_object in objects: if not isinstance(raw_object, Mapping): continue signature = raw_object.get("geometrySignature") if not isinstance(signature, Mapping): continue score = _signature_score(before_signature, signature, capability_key=capability_key) if score <= 0: continue candidates.append({"score": score, "object": dict(raw_object), "geometrySignature": dict(signature)}) candidates.sort(key=lambda item: float(item.get("score") or 0.0), reverse=True) if not candidates or float(candidates[0].get("score") or 0.0) < min_score: return {"status": "none", "message": "No matching SCDM object reached the confidence threshold.", "candidates": candidates[:5]} if len(candidates) > 1: top = float(candidates[0].get("score") or 0.0) second = float(candidates[1].get("score") or 0.0) if top - second < unique_margin: return {"status": "multiple", "message": "More than one SCDM object matches the old signature.", "candidates": candidates[:5]} best = candidates[0] return { "status": "unique", "message": "Matched one SCDM object.", "score": best.get("score"), "object": best.get("object"), "candidates": candidates[:5], } def build_scdm_id_mapping( before_cache: Mapping[str, object], after_cache: Mapping[str, object], *, capability_key: str = "", ) -> dict[str, object]: face_id_map: dict[int, int] = {} edge_id_map: dict[int, int] = {} object_id_map: dict[str, str] = {} unmatched: list[str] = [] ambiguous: list[str] = [] before_objects = before_cache.get("objects") if not isinstance(before_objects, list): before_objects = [] for raw_object in before_objects: if not isinstance(raw_object, Mapping): continue before_signature = raw_object.get("geometrySignature") if not isinstance(before_signature, Mapping): continue object_id = str(raw_object.get("objectId") or "") match = match_scdm_object_by_signature(before_signature, after_cache, capability_key=capability_key) status = str(match.get("status") or "") if status != "unique": if status == "multiple": ambiguous.append(object_id) else: unmatched.append(object_id) continue new_object = match.get("object") if not isinstance(new_object, Mapping): unmatched.append(object_id) continue new_signature = new_object.get("geometrySignature") if not isinstance(new_signature, Mapping): unmatched.append(object_id) continue new_object_id = str(new_object.get("objectId") or "") if object_id and new_object_id: object_id_map[object_id] = new_object_id _extend_single_or_zipped_id_map(face_id_map, _int_values(before_signature.get("faceIds")), _int_values(new_signature.get("faceIds"))) _extend_single_or_zipped_id_map(edge_id_map, _int_values(before_signature.get("edgeIds")), _int_values(new_signature.get("edgeIds"))) return { "ok": not unmatched and not ambiguous, "objectIdMap": object_id_map, "faceIdMap": face_id_map, "edgeIdMap": edge_id_map, "unmatched": unmatched, "ambiguous": ambiguous, } def check_scdm_unedited_objects( before_cache: Mapping[str, object], after_cache: Mapping[str, object], *, edited_object_id: str = "", edited_signature: Mapping[str, object] | None = None, capability_key: str = "", max_report: int = 5, ) -> dict[str, object]: before_objects = before_cache.get("objects") if not isinstance(before_objects, list): return {"ok": False, "reason": "missing-before-cache", "message": "Old SCDM cache does not contain objects.", "checked": 0} after_objects = after_cache.get("objects") if not isinstance(after_objects, list): return {"ok": False, "reason": "missing-after-cache", "message": "New SCDM cache does not contain objects.", "checked": 0} unmatched: list[dict[str, object]] = [] ambiguous: list[dict[str, object]] = [] checked = 0 for raw_object in before_objects: if not isinstance(raw_object, Mapping): continue object_id = str(raw_object.get("objectId") or "") signature = raw_object.get("geometrySignature") if not isinstance(signature, Mapping) or not _signature_has_enough_identity(signature): continue if object_id and edited_object_id and object_id == edited_object_id: continue if edited_signature and _same_signature_subject(signature, edited_signature): continue checked += 1 match = match_scdm_object_by_signature(signature, after_cache, capability_key=capability_key) status = str(match.get("status") or "") if status == "unique": matched_object = match.get("object") matched_signature = matched_object.get("geometrySignature") if isinstance(matched_object, Mapping) else None if isinstance(matched_signature, Mapping) and _unchanged_signature_still_matches(signature, matched_signature): continue status = "none" row = { "objectId": object_id, "objectType": raw_object.get("objectType"), "status": status or "none", "message": match.get("message"), } if status == "multiple": ambiguous.append(row) else: unmatched.append(row) if unmatched or ambiguous: parts = [] if unmatched: parts.append(f"{len(unmatched)} recognized object(s) disappeared or changed too much") if ambiguous: parts.append(f"{len(ambiguous)} recognized object(s) became ambiguous") return { "ok": False, "reason": "unexpected-object-drift", "message": "; ".join(parts) + ".", "checked": checked, "unmatched": unmatched[:max_report], "ambiguous": ambiguous[:max_report], } return { "ok": True, "reason": "ok", "message": "Unedited recognized objects still match after the SCDM edit.", "checked": checked, "unmatched": [], "ambiguous": [], } def rewrite_scdm_relation_formula_ids( text: str, mapping: Mapping[str, object], ) -> str: return rewrite_relation_formula_ids( text, _int_map(mapping.get("faceIdMap")), _int_map(mapping.get("edgeIdMap")), ) def check_scdm_target( raw_object: Mapping[str, object], *, capability_key: str, expected_target: object, tolerance: float = 1.0e-6, ) -> dict[str, object]: if capability_key == "hole.diameter": actual = _number(_capability_value(raw_object, capability_key) or _geometry_value(raw_object, "diameter")) if actual is None: radius = _number(_geometry_value(raw_object, "radius")) actual = radius * 2.0 if radius is not None else None expected = _number(expected_target) return _number_check(actual, expected, "hole.diameter", tolerance) if capability_key == "hole.position": actual_vector = _vector(_capability_value(raw_object, capability_key) or _geometry_value(raw_object, "center")) expected_vector = _vector(expected_target) return _vector_check(actual_vector, expected_vector, "hole.position", tolerance) if capability_key == "face.offset": actual = _number(_capability_value(raw_object, capability_key) or _geometry_value(raw_object, "offset") or _geometry_value(raw_object, "planeOffset")) expected = _number(expected_target) return _number_check(actual, expected, "face.offset", tolerance) if capability_key == "feature.fill": return {"ok": True, "reason": "not-applicable", "message": "feature.fill is checked by object disappearance in the caller."} return {"ok": None, "reason": "unsupported-post-check", "message": f"No target checker is registered for {capability_key}."} def check_scdm_summary_delta( before_cache: Mapping[str, object], after_cache: Mapping[str, object], *, capability_key: str = "", relative_tolerance: float = 0.25, absolute_tolerance: int = 12, ) -> dict[str, object]: if capability_key in {"feature.fill", "feature.delete_round_or_chamfer"}: return {"ok": None, "reason": "skipped-command-feature", "message": "Command features are expected to change Face/Edge counts."} before_summary = _raw_summary(before_cache) after_summary = _raw_summary(after_cache) if not before_summary or not after_summary: return {"ok": None, "reason": "missing-summary", "message": "SCDM raw summary was not available in both caches."} body_before = _int_or_none(before_summary.get("bodyCount")) body_after = _int_or_none(after_summary.get("bodyCount")) if body_before is not None and body_after is not None and body_before != body_after: return { "ok": False, "reason": "summary-drift", "message": f"SCDM result changed body count unexpectedly: {body_before} -> {body_after}.", "before": dict(before_summary), "after": dict(after_summary), } for key, label in (("faceCount", "Face"), ("edgeCount", "Edge"), ("objectCount", "对象")): before_value = _int_or_none(before_summary.get(key)) after_value = _int_or_none(after_summary.get(key)) if before_value is None or after_value is None: continue delta = abs(after_value - before_value) limit = max(int(absolute_tolerance), int(math.ceil(abs(before_value) * float(relative_tolerance)))) if delta > limit: return { "ok": False, "reason": "summary-drift", "message": f"SCDM result changed {label} count too much: {before_value} -> {after_value}, limit {limit}.", "before": dict(before_summary), "after": dict(after_summary), "metric": key, "delta": delta, "limit": limit, } return { "ok": True, "reason": "ok", "message": "SCDM model summary stayed within the allowed range.", "before": dict(before_summary), "after": dict(after_summary), } def _signature_score(before: Mapping[str, object], after: Mapping[str, object], *, capability_key: str) -> float: score = 0.0 before_type = str(before.get("objectType") or "").lower() after_type = str(after.get("objectType") or "").lower() if before_type and before_type == after_type: score += 4.0 elif {before_type, after_type} <= {"hole", "cylindrical_hole", ""}: score += 2.0 before_surface = str(before.get("surfaceType") or "").lower() after_surface = str(after.get("surfaceType") or "").lower() if before_surface and before_surface == after_surface: score += 1.0 before_faces = set(_int_values(before.get("faceIds"))) after_faces = set(_int_values(after.get("faceIds"))) if before_faces and after_faces: overlap = len(before_faces & after_faces) if overlap: score += 1.0 + min(overlap, 3) * 0.25 before_edges = set(_int_values(before.get("edgeIds"))) after_edges = set(_int_values(after.get("edgeIds"))) if before_edges and after_edges and before_edges & after_edges: score += 0.5 if capability_key != "hole.position": center_score = _vector_distance_score(_vector(before.get("center")), _vector(after.get("center"))) score += center_score axis_score = _axis_score(_vector(before.get("axis")), _vector(after.get("axis"))) score += axis_score if capability_key != "hole.diameter": score += _number_similarity_score(_diameter_from_signature(before), _diameter_from_signature(after)) return score def _signature_has_enough_identity(signature: Mapping[str, object]) -> bool: if _vector(signature.get("center")) and _vector(signature.get("axis")): return True if _vector(signature.get("center")) and str(signature.get("surfaceType") or ""): return True if _int_values(signature.get("faceIds")) or _int_values(signature.get("edgeIds")): return True return False def _same_signature_subject(left: Mapping[str, object], right: Mapping[str, object]) -> bool: left_faces = set(_int_values(left.get("faceIds"))) right_faces = set(_int_values(right.get("faceIds"))) if left_faces and right_faces and left_faces == right_faces: return True left_edges = set(_int_values(left.get("edgeIds"))) right_edges = set(_int_values(right.get("edgeIds"))) if left_edges and right_edges and left_edges == right_edges: return True left_center = _vector(left.get("center")) right_center = _vector(right.get("center")) if len(left_center) == 3 and len(right_center) == 3 and _vector_error(left_center, right_center) <= 1.0e-8: left_axis = _vector(left.get("axis")) right_axis = _vector(right.get("axis")) if len(left_axis) == 3 and len(right_axis) == 3 and _axis_score(left_axis, right_axis) >= 3.0: return True return False def _unchanged_signature_still_matches(before: Mapping[str, object], after: Mapping[str, object]) -> bool: before_center = _vector(before.get("center")) after_center = _vector(after.get("center")) if before_center and after_center and _vector_error(before_center, after_center) > _vector_tolerance(before_center, after_center): return False before_axis = _vector(before.get("axis")) after_axis = _vector(after.get("axis")) if before_axis and after_axis and _axis_score(before_axis, after_axis) < 3.0: return False before_diameter = _diameter_from_signature(before) after_diameter = _diameter_from_signature(after) if before_diameter is not None and after_diameter is not None: tolerance = max(abs(before_diameter), abs(after_diameter), 1.0) * 1.0e-5 if abs(float(before_diameter) - float(after_diameter)) > tolerance: return False before_offset = _number(before.get("planeOffset")) after_offset = _number(after.get("planeOffset")) if before_offset is not None and after_offset is not None: tolerance = max(abs(before_offset), abs(after_offset), 1.0) * 1.0e-5 if abs(float(before_offset) - float(after_offset)) > tolerance: return False return True def _vector_tolerance(left: Sequence[float], right: Sequence[float]) -> float: scale = 1.0 values = list(left) + list(right) if values: scale = max(scale, max(abs(float(item)) for item in values)) return max(scale * 1.0e-5, 1.0e-7) def _extend_single_or_zipped_id_map(target: dict[int, int], old_ids: Sequence[int], new_ids: Sequence[int]) -> None: old_unique = sorted(set(old_ids)) new_unique = sorted(set(new_ids)) if len(old_unique) == 1 and len(new_unique) == 1: target[int(old_unique[0])] = int(new_unique[0]) elif len(old_unique) == len(new_unique) and len(old_unique) > 1: for old_id, new_id in zip(old_unique, new_unique): target[int(old_id)] = int(new_id) def _capability_value(raw_object: Mapping[str, object], capability_key: str) -> object: capabilities = raw_object.get("capabilities") if not isinstance(capabilities, list): return None for capability in capabilities: if isinstance(capability, Mapping) and capability.get("key") == capability_key: return capability.get("currentValue") return None def _geometry_value(raw_object: Mapping[str, object], key: str) -> object: signature = raw_object.get("geometrySignature") if isinstance(signature, Mapping) and key in signature: return signature.get(key) geometry = raw_object.get("geometry") if isinstance(geometry, Mapping) and key in geometry: return geometry.get(key) return None def _number_check(actual: float | None, expected: float | None, label: str, tolerance: float) -> dict[str, object]: if actual is None or expected is None: return {"ok": False, "reason": "target-value-missing", "message": f"{label} target check does not have comparable values.", "actual": actual, "expected": expected} error = abs(actual - expected) return { "ok": error <= tolerance, "reason": "ok" if error <= tolerance else "target-mismatch", "message": "Target value matched." if error <= tolerance else f"{label} actual={actual:g}, expected={expected:g}, error={error:g}.", "actual": actual, "expected": expected, "error": error, "tolerance": tolerance, } def _vector_check(actual: Sequence[float], expected: Sequence[float], label: str, tolerance: float) -> dict[str, object]: if len(actual) != 3 or len(expected) != 3: return {"ok": False, "reason": "target-value-missing", "message": f"{label} target check does not have comparable vectors.", "actual": list(actual), "expected": list(expected)} error = math.sqrt(sum((float(actual[index]) - float(expected[index])) ** 2 for index in range(3))) return { "ok": error <= tolerance, "reason": "ok" if error <= tolerance else "target-mismatch", "message": "Target vector matched." if error <= tolerance else f"{label} vector error={error:g}.", "actual": list(actual), "expected": list(expected), "error": error, "tolerance": tolerance, } def _vector_error(actual: Sequence[float], expected: Sequence[float]) -> float: if len(actual) != 3 or len(expected) != 3: return math.inf return math.sqrt(sum((float(actual[index]) - float(expected[index])) ** 2 for index in range(3))) def _vector_distance_score(before: Sequence[float], after: Sequence[float]) -> float: if len(before) != 3 or len(after) != 3: return 0.0 distance = math.sqrt(sum((float(before[index]) - float(after[index])) ** 2 for index in range(3))) if distance <= 1.0e-5: return 4.0 if distance <= 1.0e-3: return 3.0 if distance <= 1.0e-1: return 1.5 return 0.0 def _axis_score(before: Sequence[float], after: Sequence[float]) -> float: if len(before) != 3 or len(after) != 3: return 0.0 before_len = math.sqrt(sum(float(item) * float(item) for item in before)) after_len = math.sqrt(sum(float(item) * float(item) for item in after)) if before_len <= 1.0e-12 or after_len <= 1.0e-12: return 0.0 dot = abs(sum(float(before[index]) * float(after[index]) for index in range(3)) / (before_len * after_len)) if dot >= 0.999: return 3.0 if dot >= 0.99: return 2.0 return 0.0 def _number_similarity_score(before: float | None, after: float | None) -> float: if before is None or after is None: return 0.0 error = abs(float(before) - float(after)) scale = max(abs(float(before)), abs(float(after)), 1.0) relative = error / scale if relative <= 1.0e-5: return 3.0 if relative <= 1.0e-3: return 2.0 if relative <= 5.0e-2: return 0.75 return 0.0 def _diameter_from_signature(signature: Mapping[str, object]) -> float | None: diameter = _number(signature.get("diameter")) if diameter is not None: return diameter radius = _number(signature.get("radius")) return radius * 2.0 if radius is not None else None def _number(value: object) -> float | None: try: return float(value) except (TypeError, ValueError): return None def _vector(value: object) -> list[float]: if isinstance(value, (str, bytes)) or value is None: return [] try: values = list(value) # type: ignore[arg-type] except TypeError: return [] if len(values) != 3: return [] try: return [float(item) for item in values] except (TypeError, ValueError): return [] def _int_values(value: object) -> list[int]: if isinstance(value, (str, bytes)) or value is None: return [] try: values = list(value) # type: ignore[arg-type] except TypeError: return [] result: list[int] = [] for item in values: try: result.append(int(item)) except (TypeError, ValueError): continue return result def _raw_summary(cache: Mapping[str, object]) -> Mapping[str, object]: diagnostics = cache.get("diagnostics") if not isinstance(diagnostics, Mapping): return {} summary = diagnostics.get("raw_summary") or diagnostics.get("summary") return summary if isinstance(summary, Mapping) else {} def _int_or_none(value: object) -> int | None: try: return int(value) except (TypeError, ValueError): return None def _int_map(value: object) -> dict[int, int]: if not isinstance(value, Mapping): return {} result: dict[int, int] = {} for key, item in value.items(): try: result[int(key)] = int(item) except (TypeError, ValueError): continue return result __all__ = [ "build_scdm_id_mapping", "check_scdm_summary_delta", "check_scdm_unedited_objects", "check_scdm_target", "match_scdm_object_by_signature", "rewrite_scdm_relation_formula_ids", "validate_scdm_edit_result", ]