from __future__ import annotations import json import re import unicodedata from dataclasses import dataclass from pathlib import Path from typing import Any _WORD = re.compile(r"[a-z0-9]+(?:[-'][a-z0-9]+)*") _REPLACEMENT_INTENT = ( "replace the whole part", "replace entire part", "replace the part", "replace this part", "start over as", "replace with a", "替换整个零件", "替换整个部件", "替换零件", "重新生成一个", "改成一个新的", ) def _normalize(value: str) -> str: return unicodedata.normalize("NFKC", str(value or "")).casefold().strip() def _matches(text: str, phrase: str) -> bool: normalized_text = _normalize(text) normalized_phrase = _normalize(phrase) if not normalized_phrase: return False if any("\u4e00" <= char <= "\u9fff" for char in normalized_phrase): return normalized_phrase in normalized_text words = _WORD.findall(normalized_phrase) if not words: return normalized_phrase in normalized_text return bool(re.search(r"(? int: normalized = _normalize(phrase) chinese = [char for char in normalized if "\u4e00" <= char <= "\u9fff"] return len(chinese) if chinese else len(_WORD.findall(normalized)) @dataclass(frozen=True) class PartSkill: id: str kind: str title: str priority: int triggers: tuple[str, ...] exclude: tuple[str, ...] bridge: str source: str related: tuple[str, ...] capability_translation_rules: tuple[str, ...] @classmethod def from_mapping(cls, value: dict[str, Any]) -> "PartSkill": return cls( id=str(value["id"]), kind=str(value["kind"]), title=str(value.get("title") or value["id"]), priority=int(value.get("priority") or 0), triggers=tuple(str(item) for item in value.get("triggers") or []), exclude=tuple(str(item) for item in value.get("exclude") or []), bridge=str(value["bridge"]), source=str(value["source"]), related=tuple(str(item) for item in value.get("related") or []), capability_translation_rules=tuple( str(item) for item in value.get("capability_translation_rules") or [] ), ) class PartSkillLibrary: """Deterministic local selector for CDSL-specific part planning guidance.""" def __init__(self, root: Path) -> None: self.root = Path(root) payload = json.loads((self.root / "catalog.json").read_text(encoding="utf-8")) self.max_planning = int(payload.get("max_planning") or 1) self.max_support = int(payload.get("max_support") or 3) self.skills = tuple(PartSkill.from_mapping(item) for item in payload.get("skills") or []) self.by_id = {skill.id: skill for skill in self.skills} if len(self.by_id) != len(self.skills): raise ValueError("Part skill catalog contains duplicate ids") for skill in self.skills: for relative in (skill.bridge, skill.source): if not (self.root / relative).is_file(): raise ValueError(f"Part skill {skill.id} references missing file: {relative}") def _matched(self, skill: PartSkill, request: str) -> tuple[int, tuple[str, ...]]: if any(_matches(request, phrase) for phrase in skill.exclude): return 0, () triggers = tuple(trigger for trigger in skill.triggers if _matches(request, trigger)) if not triggers: return 0, () # Longer phrases are more specific than generic words such as "shaft" # or "hole". Priority breaks ties between equally specific skills. score = sum(max(1, _specificity(trigger)) * 10 for trigger in triggers) return score + skill.priority, triggers def _record(self, skill: PartSkill, *, selection: str, matched: tuple[str, ...] = ()) -> dict[str, Any]: return { "id": skill.id, "kind": skill.kind, "category": skill.kind, "title": skill.title, "source": skill.source, "bridge": skill.bridge, "capability_translation_rules": list(skill.capability_translation_rules), "selection": selection, "matched_triggers": list(matched), } def select(self, request: str, inherited_ids: list[str] | tuple[str, ...] = ()) -> dict[str, Any]: inherited = [self.by_id[item] for item in inherited_ids if item in self.by_id] scores = [(self._matched(skill, request), skill) for skill in self.skills] planning_matches = sorted( ((score, triggers, skill) for (score, triggers), skill in scores if skill.kind == "planning" and score), key=lambda item: (-item[0], -item[2].priority, item[2].id), ) inherited_planning = [skill for skill in inherited if skill.kind == "planning"] planning: list[PartSkill] = inherited_planning[: self.max_planning] conflict: dict[str, Any] | None = None replacement_requested = any(_matches(request, phrase) for phrase in _REPLACEMENT_INTENT) if not planning and planning_matches: planning = [planning_matches[0][2]] elif planning and planning_matches and planning_matches[0][2].id not in {skill.id for skill in planning}: if replacement_requested: planning = [planning_matches[0][2]] else: conflict = { "current": planning[0].id, "matched": planning_matches[0][2].id, "message": "A different primary part family matched this revision request; preserve the current family unless replacement is explicit.", } selected_ids = {skill.id for skill in planning} inherited_planning_ids = {skill.id for skill in inherited_planning} records = [ self._record( skill, selection="inherited" if skill.id in inherited_planning_ids else "selected", matched=next((triggers for score, triggers, candidate in planning_matches if candidate.id == skill.id), ()), ) for skill in planning ] support: list[tuple[int, int, int, str, PartSkill, tuple[str, ...]]] = [] related_ids = {related for skill in planning for related in skill.related} direct_support: list[tuple[int, PartSkill, tuple[str, ...]]] = [] for (score, triggers), skill in scores: if skill.kind == "planning" or not score: continue direct_support.append((score, skill, triggers)) bonus = 100 if skill.id in related_ids else 0 # Direct request matches always win over inherited context, even # when the current planning family's catalog does not name them. support.append((3, bonus + score, skill.priority, skill.id, skill, triggers)) direct_related_ids = { related for _, skill, _ in direct_support for related in skill.related if related in self.by_id and self.by_id[related].kind != "planning" } known_support_ids = {item[3] for item in support} for related_id in sorted(direct_related_ids - known_support_ids): skill = self.by_id[related_id] planning_bonus = 25 if skill.id in related_ids else 0 support.append((2, 50 + planning_bonus, skill.priority, skill.id, skill, ())) known_support_ids.add(skill.id) for skill in inherited: if skill.kind != "planning" and skill.id not in known_support_ids: # Keep prior guidance only after direct request matches and # their catalog-declared supporting rules. support.append((1, 0, skill.priority, skill.id, skill, ())) known_support_ids.add(skill.id) support.sort(key=lambda item: (-item[0], -item[1], -item[2], item[3])) for _, _, _, _, skill, triggers in support: if len(records) - len(planning) >= self.max_support or skill.id in selected_ids: continue selected_ids.add(skill.id) inherited_marker = "inherited" if skill in inherited else "selected" records.append(self._record(skill, selection=inherited_marker, matched=triggers)) return { "schema_version": "1.0", "request": str(request or ""), "skills": records, "skill_ids": [record["id"] for record in records], "planning_ids": [record["id"] for record in records if record["kind"] == "planning"], "support_ids": [record["id"] for record in records if record["kind"] != "planning"], "inherited_skill_ids": [skill.id for skill in inherited], "replacement_requested": replacement_requested, "conflict": conflict, } def inherited_from_task(self, task: dict[str, Any] | None) -> list[str]: revision_id = str((task or {}).get("current_revision") or "") revisions = (task or {}).get("revisions") or [] revision = next((item for item in revisions if item.get("revision_id") == revision_id), None) revision_ids = [str(item) for item in (revision or {}).get("part_skill_ids") or [] if str(item) in self.by_id] if revision_ids: return revision_ids # A valid DesignIntent creates a task before the first runtime build. # Its canonical backend selection must therefore be inheritable too. intent_id = str((task or {}).get("current_design_intent_id") or "") intent = next( (item for item in (task or {}).get("design_intents") or [] if str(item.get("intent_id") or "") == intent_id), None, ) return [str(item) for item in (intent or {}).get("part_skill_ids") or [] if str(item) in self.by_id] def render_context(self, selection: dict[str, Any]) -> str: records = selection.get("skills") or [] if not records: return "No part-family skill matched this request. Use the CDSL schema and library only." sections = [ "Selected CDSL part-skill guidance (planning context only):", "Part skill guidance never overrides the user request or the authoritative CDSL schema/runtime.", "Translate the guidance into valid CDSL; do not emit build123d source or invent atomics.", ] for record in records: path = self.root / str(record["bridge"]) sections.append(f"\n[{record['id']}] ({record['selection']})\n{path.read_text(encoding='utf-8').strip()}") if selection.get("conflict"): sections.append("\nPrimary-family conflict: preserve the current task family. Ask one concise clarification question and do not generate until the user explicitly requests whole-part replacement or confirms the current family.") return "\n".join(sections) def audit(self, selection: dict[str, Any], cdsl: dict[str, Any], assumptions: list[str] | None = None) -> dict[str, Any]: features = [item for item in cdsl.get("features") or [] if isinstance(item, dict)] feature_ids = [str(item.get("id")) for item in features if item.get("id")] atomics = [str(item.get("atomic_id")) for item in features] sketches = [item for item in (cdsl.get("geometry") or {}).get("sketches") or [] if isinstance(item, dict)] profiles = [str((item.get("profile") or {}).get("type")) for item in sketches] evidence = { "feature_ids": feature_ids, "atomic_ids": sorted(set(atomics)), "profile_types": sorted(set(profiles)), "features": [ { "id": str(item.get("id") or ""), "atomic_id": str(item.get("atomic_id") or ""), "sketch_id": str(item.get("sketch_id") or ""), "params": item.get("params") if isinstance(item.get("params"), dict) else {}, } for item in features ], "profiles": [ { "id": str(item.get("id") or ""), "type": str((item.get("profile") or {}).get("type") or ""), "profile": item.get("profile") if isinstance(item.get("profile"), dict) else {}, } for item in sketches ], } translations: list[dict[str, Any]] = [] for record in selection.get("skills") or []: skill_id = str(record.get("id")) status = "exact" reason = "The selected structural guidance is represented by the submitted CDSL plan." if skill_id == "functional/flange-bolt-circle": if "circles" in profiles and "extrude_cut_blind" in atomics: status, reason = "expanded", "Circular bolt layout is represented by explicit circle geometry because no circular-pattern atomic exists." else: status, reason = "blocked", "No explicit circular bolt layout evidence was found in the CDSL." elif skill_id == "atomic/threaded-hole-creation": status, reason = "approximated", "The current runtime preserves a cylindrical bore but does not generate helical thread topology." elif skill_id == "atomic/fillet-chamfer-last": finishing = [item for item in features if item.get("atomic_id") in {"fillet", "chamfer"}] if not finishing: status, reason = "omitted", "No stable finishing selector was submitted; edge treatment was omitted." elif skill_id == "atomic/pattern-holes-from-datum": if "circles" in profiles or "circle_grid" in profiles: status, reason = "expanded", "The layout is represented by explicit profile circles." elif not any(item in {"pattern_linear", "pattern_mirror"} for item in atomics): status, reason = "blocked", "No supported pattern or explicit circle layout was found." translation: dict[str, Any] = {"skill_id": skill_id, "status": status, "reason": reason, "evidence": evidence} if skill_id == "functional/flange-bolt-circle" and status == "expanded": translation["translation"] = "circular_pattern_to_explicit_circles" elif skill_id == "atomic/threaded-hole-creation": translation["translation"] = "thread_geometry_omitted" elif skill_id == "atomic/fillet-chamfer-last" and status == "omitted": translation["translation"] = "selector_unavailable" translations.append(translation) return { "schema_version": "1.0", "request": selection.get("request", ""), "structural_intent": selection.get("request", ""), "skill_ids": list(selection.get("skill_ids") or []), "skills": list(selection.get("skills") or []), "inherited_skill_ids": list(selection.get("inherited_skill_ids") or []), "conflict": selection.get("conflict"), "assumptions": [str(item) for item in assumptions or []], "evidence": evidence, "capability_translations": translations, }