baram2584/FESADev
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add MITC4 drilling diagnostics output and comparison contract

Browse Source
This commit is contained in:
KOKO\Mimi
2026-10-20 09:45:58 +09:00
parent 3d5c296651
commit a73d3fc9ca
31 changed files with 1175 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/verification/compare_hht_dynamic.py
+36 -1
View File
@@ -197,6 +197,38 @@ def dataset(file: h5py.File, path: str, shape: tuple[int, ...]) -> h5py.Dataset:
return result
def drilling_diagnostics(file: h5py.File) -> dict[str, Any] | None:
"""Validate additive diagnostics without introducing a balance verdict."""
metadata = file["metadata"]
mode_present = "mitc4_drilling_formulation" in metadata.attrs
version_present = "drilling_diagnostics_version" in metadata.attrs
require(mode_present == version_present, "partial-drilling-markers", "Both drilling markers are required.")
if not mode_present:
return None
mode = string_attribute(metadata, "mitc4_drilling_formulation")
require(mode in ("direct-initial-director-spring", "legacy-corotated-twist"),
"unknown-drilling-formulation", mode)
numeric_attribute(metadata, "drilling_diagnostics_version", "<u8", 1)
frames = []
for frame in range(FRAME_COUNT):
root = f"steps/{STEP}/frames/{frame}/global"
values = {}
for name, size in (("equilibrium", 6), ("drilling_moment", 3), ("equilibrium_with_drilling", 6)):
stored = dataset(file, f"{root}/{name}", (size,))
require(stored.dtype == np.dtype("<f8"), "wrong-drilling-dataset-type", stored.name)
values[name] = stored[:]
require(np.isfinite(values[name]).all(), "nonfinite-drilling-value", stored.name)
expected = values["equilibrium"].copy()
expected[3:] -= values["drilling_moment"]
require(np.array_equal(expected, values["equilibrium_with_drilling"]),
"inconsistent-drilling-equilibrium", str(frame))
frames.append({"storage_step": STEP, "frame_index": frame,
"raw_equilibrium": values["equilibrium"].tolist(),
"drilling_moment": values["drilling_moment"].tolist(),
"equilibrium_with_drilling": values["equilibrium_with_drilling"].tolist()})
return {"formulation": mode, "version": 1, "frames": frames}
def read_results(path: Path, nodes: dict[int, tuple[float, ...]],
content_identity: str) -> tuple[list[np.ndarray], dict[str, Any]]:
"""Precheck every result node/frame/component, then project by source key."""
@@ -319,9 +351,10 @@ def read_results(path: Path, nodes: dict[int, tuple[float, ...]],
if frame == 0:
require((values == 0.0).all(), "nonzero-initial-displacement", "Frame zero must have q0=0.")
projected.append(values[selection])
drilling = drilling_diagnostics(file)
return projected, {"source_input_identity": source_identity, "source_node_count": NODE_COUNT,
"accepted_frame_count": FRAME_COUNT, "checked_result_scalars": NODE_COUNT * FRAME_COUNT * 6,
"projected_node_labels": list(NODE_LABELS)}
"projected_node_labels": list(NODE_LABELS), "drilling_diagnostics": drilling}
def evaluate_family(increment: int, quantity: str, result: list[float],
@@ -374,6 +407,7 @@ def evaluate_family(increment: int, quantity: str, result: list[float],
def compare(input_path: Path, reference_path: Path, results_path: Path) -> dict[str, Any]:
report: dict[str, Any] = {
"drilling_diagnostics": None,
"schema": "hht-reference-comparison-v1", "case": CASE, "policy": dict(POLICY),
"identities": {}, "precheck": {"passed": False}, "families": [],
"blocking_failures": [], "warnings": [], "blocking_passed": False, "verdict": "failed",
@@ -386,6 +420,7 @@ def compare(input_path: Path, reference_path: Path, results_path: Path) -> dict[
nodes, content_identity = input_inventory(input_path)
reference = read_reference(reference_path)
result, evidence = read_results(results_path, nodes, content_identity)
report["drilling_diagnostics"] = evidence.pop("drilling_diagnostics")
families = []
for frame in range(FRAME_COUNT):
for quantity, first in (("U", 0), ("UR", 3)):