feat(cadfs): 补齐核心建模能力并建立代表性回归
- 新增 loft、双向切除、through-all/up-to-next 等 CADFS lowering 与 engine 支持 - 支持多种 reference plane、B-spline profile 和 circular pattern replay - 保留 transform 历史，并烘焙安全的单源平移/旋转变换 - 改进 selector 绑定、拓扑快照和 pattern 变换处理 - 建立 17 个代表样本的转换、重建与比较回归工具链 - 补充 schema、author guidance、运行时和几何回归测试
This commit is contained in:
@@ -0,0 +1,216 @@
|
||||
"""Deterministic, feature-covering CADFS regression pools.
|
||||
|
||||
The converted CADFS corpus is deliberately kept outside version control. This
|
||||
module turns the artifacts already produced in ``output/samples`` into a small,
|
||||
versionable manifest that records why every selected source sample is needed.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections import Counter
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from .reports import read_json, write_json
|
||||
|
||||
|
||||
REGRESSION_SCHEMA = "cadfs_to_cdsl.regression.v1"
|
||||
|
||||
|
||||
def _tag(kind: str, value: str) -> str:
|
||||
return f"{kind}:{value}"
|
||||
|
||||
|
||||
def _split_tag(tag: str) -> tuple[str, str]:
|
||||
kind, separator, value = tag.partition(":")
|
||||
if not separator:
|
||||
raise ValueError(f"Malformed regression coverage tag: {tag!r}")
|
||||
return kind, value
|
||||
|
||||
|
||||
def _sample_features(sample_dir: Path) -> dict[str, Any]:
|
||||
"""Read coverage signals from a converted sample without reparsing CADFS."""
|
||||
history = read_json(sample_dir / "history.json") if (sample_dir / "history.json").exists() else []
|
||||
candidate = read_json(sample_dir / "candidate.cdsl.json") if (sample_dir / "candidate.cdsl.json").exists() else {}
|
||||
diagnostics = read_json(sample_dir / "diagnostics.json") if (sample_dir / "diagnostics.json").exists() else []
|
||||
status = read_json(sample_dir / "status.json") if (sample_dir / "status.json").exists() else {}
|
||||
|
||||
source_operations: set[str] = set()
|
||||
sketch_entities: set[str] = set()
|
||||
for item in history:
|
||||
if not isinstance(item, dict):
|
||||
continue
|
||||
operation = item.get("operation")
|
||||
if isinstance(operation, str):
|
||||
source_operations.add(operation)
|
||||
for entity in item.get("entities") or ():
|
||||
if isinstance(entity, dict) and isinstance(entity.get("operation"), str):
|
||||
sketch_entities.add(str(entity["operation"]))
|
||||
|
||||
atomic_ids = {
|
||||
str(feature["atomic_id"])
|
||||
for feature in candidate.get("features") or ()
|
||||
if isinstance(feature, dict) and isinstance(feature.get("atomic_id"), str)
|
||||
}
|
||||
unsupported_capabilities = {
|
||||
str(item["capability"])
|
||||
for item in diagnostics
|
||||
if isinstance(item, dict)
|
||||
and item.get("code") == "unsupported_engine_capability"
|
||||
and isinstance(item.get("capability"), str)
|
||||
}
|
||||
unsupported_operations = {
|
||||
str(item["operation"])
|
||||
for item in diagnostics
|
||||
if isinstance(item, dict)
|
||||
and item.get("code") == "unsupported_operation"
|
||||
and isinstance(item.get("operation"), str)
|
||||
}
|
||||
return {
|
||||
"sample_id": sample_dir.name,
|
||||
"source_operations": sorted(source_operations),
|
||||
"sketch_entities": sorted(sketch_entities),
|
||||
"engine_atomic_ids": sorted(atomic_ids),
|
||||
"unsupported_capabilities": sorted(unsupported_capabilities),
|
||||
"unsupported_operations": sorted(unsupported_operations),
|
||||
"has_candidate": bool(candidate),
|
||||
"baseline": {
|
||||
key: status[key]
|
||||
for key in ("conversion_status", "rebuild_status", "comparison_decision", "status")
|
||||
if key in status
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def _tags(record: dict[str, Any]) -> set[str]:
|
||||
return {
|
||||
*(_tag("source_operation", item) for item in record["source_operations"]),
|
||||
*(_tag("sketch_entity", item) for item in record["sketch_entities"]),
|
||||
*(_tag("engine_atomic", item) for item in record["engine_atomic_ids"]),
|
||||
*(_tag("unsupported_capability", item) for item in record["unsupported_capabilities"]),
|
||||
*(_tag("unsupported_operation", item) for item in record["unsupported_operations"]),
|
||||
}
|
||||
|
||||
|
||||
def _greedy_cover(records: list[dict[str, Any]], wanted: set[str], selected: list[dict[str, Any]]) -> dict[str, list[str]]:
|
||||
"""Cover ``wanted`` with stable maximum-coverage selection.
|
||||
|
||||
The sample id breaks ties, so a corpus refresh is reviewable and never
|
||||
changes pool membership due to directory iteration order.
|
||||
"""
|
||||
coverage: dict[str, list[str]] = {}
|
||||
selected_ids = {record["sample_id"] for record in selected}
|
||||
remaining = set(wanted)
|
||||
while remaining:
|
||||
choices = [record for record in records if record["sample_id"] not in selected_ids]
|
||||
if not choices:
|
||||
break
|
||||
choice = min(
|
||||
choices,
|
||||
key=lambda record: (-len(_tags(record) & remaining), record["sample_id"]),
|
||||
)
|
||||
gained = _tags(choice) & remaining
|
||||
if not gained:
|
||||
break
|
||||
selected.append(choice)
|
||||
selected_ids.add(choice["sample_id"])
|
||||
for tag in sorted(gained):
|
||||
coverage.setdefault(tag, []).append(choice["sample_id"])
|
||||
remaining -= gained
|
||||
return coverage
|
||||
|
||||
|
||||
def build_regression_manifest(output: Path) -> dict[str, Any]:
|
||||
"""Build a compact pool covering every observed CADFS modeling signal."""
|
||||
records = [_sample_features(directory) for directory in sorted((output / "samples").glob("*")) if directory.is_dir()]
|
||||
if not records:
|
||||
raise FileNotFoundError(f"No converted CADFS samples found under {output / 'samples'}")
|
||||
|
||||
all_tags = set().union(*(_tags(record) for record in records))
|
||||
selected: list[dict[str, Any]] = []
|
||||
|
||||
# Give the engine pool a current successful baseline per atomic operation
|
||||
# whenever the corpus has one. Unsupported/failed atoms remain covered by
|
||||
# the full conversion pool and are never represented as passing builds.
|
||||
engine_records = [
|
||||
record for record in records
|
||||
if record["has_candidate"] and record["baseline"].get("rebuild_status") == "rebuilt"
|
||||
]
|
||||
engine_tags = {_tag("engine_atomic", atom) for record in engine_records for atom in record["engine_atomic_ids"]}
|
||||
_greedy_cover(engine_records, engine_tags, selected)
|
||||
# Samples chosen for the executable baseline can also cover source and
|
||||
# converter signals. Count those signals before the general pass so the
|
||||
# final fixture remains genuinely representative rather than redundant.
|
||||
coverage = {
|
||||
tag: sorted(record["sample_id"] for record in selected if tag in _tags(record))
|
||||
for tag in sorted(set().union(*(_tags(record) for record in selected)))
|
||||
}
|
||||
all_coverage = _greedy_cover(records, all_tags - set(coverage), selected)
|
||||
for tag, sample_ids in all_coverage.items():
|
||||
coverage[tag] = sorted(set(coverage.get(tag, []) + sample_ids))
|
||||
|
||||
selected_ids = {record["sample_id"] for record in selected}
|
||||
missing = sorted(all_tags - set(coverage))
|
||||
if missing:
|
||||
raise RuntimeError("Unable to cover CADFS regression features: " + ", ".join(missing))
|
||||
|
||||
entries = []
|
||||
for record in sorted(selected, key=lambda item: item["sample_id"]):
|
||||
tags = _tags(record)
|
||||
tiers = ["conversion"]
|
||||
if record["sample_id"] in {item["sample_id"] for item in engine_records}:
|
||||
tiers.append("engine")
|
||||
entries.append({
|
||||
**record,
|
||||
"tiers": tiers,
|
||||
"selection_reasons": sorted(tag for tag, sample_ids in coverage.items() if record["sample_id"] in sample_ids),
|
||||
})
|
||||
|
||||
inventory: dict[str, dict[str, int]] = {}
|
||||
for kind, _ in map(_split_tag, sorted(all_tags)):
|
||||
inventory.setdefault(kind, {})
|
||||
for tag in all_tags:
|
||||
kind, value = _split_tag(tag)
|
||||
inventory[kind][value] = sum(tag in _tags(record) for record in records)
|
||||
|
||||
return {
|
||||
"schema": REGRESSION_SCHEMA,
|
||||
"description": "Feature-covering representative CADFS regression pool. Engine samples have a prior successful rebuild; conversion samples retain unsupported-feature diagnostics.",
|
||||
"source_output": str(output),
|
||||
"source_sample_count": len(records),
|
||||
"selected_sample_count": len(entries),
|
||||
"engine_sample_count": sum("engine" in entry["tiers"] for entry in entries),
|
||||
"conversion_sample_count": len(entries),
|
||||
"engine_baseline_atomic_ids": sorted(tag.removeprefix("engine_atomic:") for tag in engine_tags),
|
||||
"engine_diagnostic_only_atomic_ids": sorted(
|
||||
tag.removeprefix("engine_atomic:") for tag in all_tags - engine_tags if tag.startswith("engine_atomic:")
|
||||
),
|
||||
"feature_inventory": {kind: dict(sorted(values.items())) for kind, values in sorted(inventory.items())},
|
||||
"coverage": coverage,
|
||||
"entries": entries,
|
||||
"selected_ids": sorted(selected_ids),
|
||||
}
|
||||
|
||||
|
||||
def write_regression_manifest(output: Path, manifest_path: Path) -> dict[str, Any]:
|
||||
manifest = build_regression_manifest(output)
|
||||
write_json(manifest_path, manifest)
|
||||
return manifest
|
||||
|
||||
|
||||
def regression_sample_ids(manifest_path: Path, tier: str) -> list[str]:
|
||||
manifest = read_json(manifest_path)
|
||||
if manifest.get("schema") != REGRESSION_SCHEMA:
|
||||
raise ValueError(f"Unsupported regression manifest schema: {manifest.get('schema')!r}")
|
||||
if tier not in {"engine", "conversion", "all"}:
|
||||
raise ValueError(f"Unknown regression tier: {tier!r}")
|
||||
entries = manifest.get("entries") or []
|
||||
selected = [item["sample_id"] for item in entries if tier == "all" or tier in (item.get("tiers") or ())]
|
||||
if not selected:
|
||||
raise ValueError(f"Regression manifest has no samples for tier {tier!r}")
|
||||
return sorted(selected)
|
||||
|
||||
|
||||
def summarize_regression(records: list[dict[str, Any]]) -> dict[str, Any]:
|
||||
statuses = Counter(str(record.get("status") or "unknown") for record in records)
|
||||
return {"sample_count": len(records), "statuses": dict(sorted(statuses.items()))}
|
||||
Reference in New Issue
Block a user