김경종
435 lines
18 KiB
Python
435 lines
18 KiB
Python
import csv
|
|
import contextlib
|
|
import hashlib
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
|
|
def load_compare_extracted_csv():
|
|
module_path = Path(__file__).resolve().parent / "compare_extracted_csv.py"
|
|
spec = importlib.util.spec_from_file_location("compare_extracted_csv", module_path)
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def write_json(path: Path, payload: dict):
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(json.dumps(payload, indent=2), encoding="utf-8")
|
|
|
|
|
|
def write_csv(path: Path, rows: list[dict[str, str]]):
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
with path.open("w", newline="", encoding="utf-8") as handle:
|
|
writer = csv.DictWriter(handle, fieldnames=list(rows[0]))
|
|
writer.writeheader()
|
|
writer.writerows(rows)
|
|
|
|
|
|
def write_text(path: Path, text: str):
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(text, encoding="utf-8")
|
|
|
|
|
|
def metadata_payload() -> dict:
|
|
return {
|
|
"schema_version": "abaqus-user-subroutine-artifact-v1",
|
|
"feature_id": "umat",
|
|
"model_id": "single-element",
|
|
"artifact_status": "ready-for-comparison",
|
|
"abaqus": {"version": "2024", "precision": "double"},
|
|
"compiler": {"vendor": "Intel oneAPI", "name": "ifx", "version": "2024"},
|
|
"subroutine": {"entry_points": ["UMAT"], "source_files": []},
|
|
"input_file": "model.inp",
|
|
"outputs": {
|
|
"tails": {
|
|
"msg": "job.msg.tail.txt",
|
|
"dat": "job.dat.tail.txt",
|
|
"log": "job.log.tail.txt",
|
|
"sta": "job.sta.tail.txt",
|
|
},
|
|
"csv": {"stresses": "extracted/stresses.csv"},
|
|
},
|
|
"extraction": {
|
|
"source_odb": "job.odb",
|
|
"tool": "Abaqus Python",
|
|
"extracted_at": "2026-06-10T00:00:00+09:00",
|
|
"csv_directory": "extracted",
|
|
},
|
|
"comparisons": {
|
|
"stresses": {
|
|
"reference_csv": "extracted/stresses.csv",
|
|
"actual_csv": "extracted/stresses.csv",
|
|
"required_columns": [
|
|
"step",
|
|
"frame",
|
|
"instance",
|
|
"element_label",
|
|
"integration_point",
|
|
"section_point",
|
|
"output_position",
|
|
"component",
|
|
"coordinate_system",
|
|
"unit",
|
|
"value",
|
|
],
|
|
"key_columns": [
|
|
"step",
|
|
"frame",
|
|
"instance",
|
|
"element_label",
|
|
"integration_point",
|
|
"section_point",
|
|
"output_position",
|
|
"component",
|
|
],
|
|
"value_column": "value",
|
|
"unit_column": "unit",
|
|
"coordinate_system_column": "coordinate_system",
|
|
"tolerance": {"absolute": 1.0e-8, "relative": 1.0e-6, "relative_floor": 1.0e-12},
|
|
}
|
|
},
|
|
}
|
|
|
|
|
|
def stress_rows(value: str = "100.0") -> list[dict[str, str]]:
|
|
return [
|
|
{
|
|
"step": "Step-1",
|
|
"frame": "1",
|
|
"instance": "PART-1-1",
|
|
"element_label": "1",
|
|
"integration_point": "1",
|
|
"section_point": "",
|
|
"output_position": "INTEGRATION_POINT",
|
|
"component": "S11",
|
|
"coordinate_system": "GLOBAL",
|
|
"unit": "MPa",
|
|
"value": value,
|
|
}
|
|
]
|
|
|
|
|
|
def make_metadata_valid_for_artifact_validation(root: Path, model_dir: Path, payload: dict) -> dict:
|
|
source = root / "src" / "fortran" / "abaqus" / "UMAT.for"
|
|
source.parent.mkdir(parents=True, exist_ok=True)
|
|
source.write_text(" subroutine umat()\n end\n", encoding="utf-8")
|
|
source_hash = hashlib.sha256(source.read_bytes()).hexdigest()
|
|
payload["subroutine"]["source_files"] = [
|
|
{
|
|
"path": "src/fortran/abaqus/UMAT.for",
|
|
"language": "Fortran",
|
|
"sha256": source_hash,
|
|
}
|
|
]
|
|
for name in [
|
|
"model.inp",
|
|
"job.msg.tail.txt",
|
|
"job.dat.tail.txt",
|
|
"job.log.tail.txt",
|
|
"job.sta.tail.txt",
|
|
]:
|
|
path = model_dir / name
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text("ok\n", encoding="utf-8")
|
|
return payload
|
|
|
|
|
|
def compare_stresses(reference_rows: list[dict[str, str]], actual_rows: list[dict[str, str]]) -> dict:
|
|
compare = load_compare_extracted_csv()
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
reference = root / "references" / "umat" / "single-element"
|
|
actual = root / "external-results" / "umat" / "single-element"
|
|
write_json(reference / "metadata.json", metadata_payload())
|
|
write_csv(reference / "extracted" / "stresses.csv", reference_rows)
|
|
write_csv(actual / "extracted" / "stresses.csv", actual_rows)
|
|
|
|
return compare.compare_metadata(
|
|
reference / "metadata.json",
|
|
actual,
|
|
quantities=["stresses"],
|
|
validate_artifacts=False,
|
|
)
|
|
|
|
|
|
def call_main_silently(compare, argv: list[str]) -> int:
|
|
with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()):
|
|
return compare.main(argv)
|
|
|
|
|
|
class CompareExtractedCsvTests(unittest.TestCase):
|
|
def test_quantity_passes_when_schema_keys_units_and_values_match_within_tolerance(self):
|
|
compare = load_compare_extracted_csv()
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
reference = root / "references" / "umat" / "single-element"
|
|
actual = root / "external-results" / "umat" / "single-element"
|
|
write_json(reference / "metadata.json", metadata_payload())
|
|
write_csv(reference / "extracted" / "stresses.csv", stress_rows("100.0"))
|
|
write_csv(actual / "extracted" / "stresses.csv", stress_rows("100.00000001"))
|
|
|
|
report = compare.compare_metadata(
|
|
reference / "metadata.json",
|
|
actual,
|
|
quantities=["stresses"],
|
|
validate_artifacts=False,
|
|
)
|
|
|
|
self.assertEqual(report["overall_result"], "pass")
|
|
self.assertEqual(report["quantities"][0]["result"], "pass")
|
|
self.assertEqual(report["quantities"][0]["classification"], "N/A")
|
|
self.assertEqual(report["quantities"][0]["compared_rows"], 1)
|
|
|
|
def test_missing_actual_csv_is_missing_generated_output(self):
|
|
compare = load_compare_extracted_csv()
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
reference = root / "references" / "umat" / "single-element"
|
|
actual = root / "external-results" / "umat" / "single-element"
|
|
write_json(reference / "metadata.json", metadata_payload())
|
|
write_csv(reference / "extracted" / "stresses.csv", stress_rows("100.0"))
|
|
|
|
report = compare.compare_metadata(
|
|
reference / "metadata.json",
|
|
actual,
|
|
quantities=["stresses"],
|
|
validate_artifacts=False,
|
|
)
|
|
|
|
self.assertEqual(report["overall_result"], "fail")
|
|
self.assertEqual(report["quantities"][0]["classification"], "missing-generated-output")
|
|
|
|
def test_missing_required_column_is_schema_mismatch(self):
|
|
compare = load_compare_extracted_csv()
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
reference = root / "references" / "umat" / "single-element"
|
|
actual = root / "external-results" / "umat" / "single-element"
|
|
write_json(reference / "metadata.json", metadata_payload())
|
|
row = stress_rows("100.0")[0]
|
|
write_csv(reference / "extracted" / "stresses.csv", [row])
|
|
actual_row = dict(row)
|
|
actual_row.pop("coordinate_system")
|
|
write_csv(actual / "extracted" / "stresses.csv", [actual_row])
|
|
|
|
report = compare.compare_metadata(
|
|
reference / "metadata.json",
|
|
actual,
|
|
quantities=["stresses"],
|
|
validate_artifacts=False,
|
|
)
|
|
|
|
self.assertEqual(report["overall_result"], "fail")
|
|
self.assertEqual(report["quantities"][0]["classification"], "schema-mismatch")
|
|
|
|
def test_missing_quantity_contract_is_upstream_contract(self):
|
|
compare = load_compare_extracted_csv()
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
reference = root / "references" / "umat" / "single-element"
|
|
actual = root / "external-results" / "umat" / "single-element"
|
|
payload = metadata_payload()
|
|
payload["comparisons"].pop("stresses")
|
|
write_json(reference / "metadata.json", payload)
|
|
|
|
report = compare.compare_metadata(
|
|
reference / "metadata.json",
|
|
actual,
|
|
quantities=["stresses"],
|
|
validate_artifacts=False,
|
|
)
|
|
|
|
self.assertEqual(report["overall_result"], "fail")
|
|
self.assertEqual(report["quantities"][0]["classification"], "upstream-contract")
|
|
|
|
def test_missing_comparisons_block_is_upstream_contract(self):
|
|
compare = load_compare_extracted_csv()
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
reference = root / "references" / "umat" / "single-element"
|
|
actual = root / "external-results" / "umat" / "single-element"
|
|
payload = metadata_payload()
|
|
payload.pop("comparisons")
|
|
write_json(reference / "metadata.json", payload)
|
|
|
|
report = compare.compare_metadata(reference / "metadata.json", actual, validate_artifacts=False)
|
|
|
|
self.assertEqual(report["overall_result"], "fail")
|
|
self.assertEqual(report["quantities"][0]["classification"], "upstream-contract")
|
|
|
|
def test_incomplete_quantity_contract_is_upstream_contract(self):
|
|
compare = load_compare_extracted_csv()
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
reference = root / "references" / "umat" / "single-element"
|
|
actual = root / "external-results" / "umat" / "single-element"
|
|
payload = metadata_payload()
|
|
payload["comparisons"]["stresses"].pop("value_column")
|
|
write_json(reference / "metadata.json", payload)
|
|
|
|
report = compare.compare_metadata(
|
|
reference / "metadata.json",
|
|
actual,
|
|
quantities=["stresses"],
|
|
validate_artifacts=False,
|
|
)
|
|
|
|
self.assertEqual(report["overall_result"], "fail")
|
|
self.assertEqual(report["quantities"][0]["classification"], "upstream-contract")
|
|
|
|
def test_duplicate_header_is_schema_mismatch(self):
|
|
compare = load_compare_extracted_csv()
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
reference = root / "references" / "umat" / "single-element"
|
|
actual = root / "external-results" / "umat" / "single-element"
|
|
write_json(reference / "metadata.json", metadata_payload())
|
|
write_csv(reference / "extracted" / "stresses.csv", stress_rows("100.0"))
|
|
header = ",".join(list(stress_rows("100.0")[0]) + ["value"])
|
|
row = ",".join(stress_rows("100.0")[0].values()) + ",100.0"
|
|
write_text(actual / "extracted" / "stresses.csv", f"{header}\n{row}\n")
|
|
|
|
report = compare.compare_metadata(
|
|
reference / "metadata.json",
|
|
actual,
|
|
quantities=["stresses"],
|
|
validate_artifacts=False,
|
|
)
|
|
|
|
self.assertEqual(report["overall_result"], "fail")
|
|
self.assertEqual(report["quantities"][0]["classification"], "schema-mismatch")
|
|
|
|
def test_changed_row_key_is_id_mismatch(self):
|
|
actual_row = dict(stress_rows("100.0")[0])
|
|
actual_row["element_label"] = "2"
|
|
|
|
report = compare_stresses(stress_rows("100.0"), [actual_row])
|
|
|
|
self.assertEqual(report["overall_result"], "fail")
|
|
self.assertEqual(report["quantities"][0]["classification"], "id-mismatch")
|
|
self.assertEqual(report["quantities"][0]["missing_rows"], 1)
|
|
self.assertEqual(report["quantities"][0]["extra_rows"], 1)
|
|
|
|
def test_unit_mismatch_is_unit_or_coordinate_mismatch(self):
|
|
actual_row = dict(stress_rows("100.0")[0])
|
|
actual_row["unit"] = "Pa"
|
|
|
|
report = compare_stresses(stress_rows("100.0"), [actual_row])
|
|
|
|
self.assertEqual(report["overall_result"], "fail")
|
|
self.assertEqual(report["quantities"][0]["classification"], "unit-or-coordinate-mismatch")
|
|
|
|
def test_coordinate_system_mismatch_is_unit_or_coordinate_mismatch(self):
|
|
actual_row = dict(stress_rows("100.0")[0])
|
|
actual_row["coordinate_system"] = "LOCAL-1"
|
|
|
|
report = compare_stresses(stress_rows("100.0"), [actual_row])
|
|
|
|
self.assertEqual(report["overall_result"], "fail")
|
|
self.assertEqual(report["quantities"][0]["classification"], "unit-or-coordinate-mismatch")
|
|
|
|
def test_nonfinite_value_is_nonfinite_result(self):
|
|
report = compare_stresses(stress_rows("100.0"), stress_rows("nan"))
|
|
|
|
self.assertEqual(report["overall_result"], "fail")
|
|
self.assertEqual(report["quantities"][0]["classification"], "nonfinite-result")
|
|
|
|
def test_value_outside_tolerance_is_tolerance_failure(self):
|
|
report = compare_stresses(stress_rows("100.0"), stress_rows("101.0"))
|
|
|
|
self.assertEqual(report["overall_result"], "fail")
|
|
self.assertEqual(report["quantities"][0]["classification"], "tolerance-failure")
|
|
self.assertEqual(report["quantities"][0]["max_abs_error"], 1.0)
|
|
self.assertEqual(report["quantities"][0]["result"], "fail")
|
|
|
|
def test_cli_writes_json_report_and_returns_zero_when_all_quantities_pass(self):
|
|
compare = load_compare_extracted_csv()
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
reference = root / "references" / "umat" / "single-element"
|
|
actual = root / "external-results" / "umat" / "single-element"
|
|
report_json = root / "build" / "reference-verification" / "umat-single-element.json"
|
|
payload = make_metadata_valid_for_artifact_validation(root, reference, metadata_payload())
|
|
write_json(reference / "metadata.json", payload)
|
|
write_csv(reference / "extracted" / "stresses.csv", stress_rows("100.0"))
|
|
write_csv(actual / "extracted" / "stresses.csv", stress_rows("100.00000001"))
|
|
|
|
exit_code = call_main_silently(
|
|
compare,
|
|
[
|
|
"--metadata",
|
|
str(reference / "metadata.json"),
|
|
"--actual-root",
|
|
str(actual),
|
|
"--quantity",
|
|
"stresses",
|
|
"--report-json",
|
|
str(report_json),
|
|
],
|
|
)
|
|
|
|
report = json.loads(report_json.read_text(encoding="utf-8"))
|
|
|
|
self.assertEqual(exit_code, 0)
|
|
self.assertEqual(report["overall_result"], "pass")
|
|
|
|
def test_cli_writes_json_report_and_returns_one_when_a_quantity_fails(self):
|
|
compare = load_compare_extracted_csv()
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
reference = root / "references" / "umat" / "single-element"
|
|
actual = root / "external-results" / "umat" / "single-element"
|
|
report_json = root / "build" / "reference-verification" / "umat-single-element.json"
|
|
payload = make_metadata_valid_for_artifact_validation(root, reference, metadata_payload())
|
|
write_json(reference / "metadata.json", payload)
|
|
write_csv(reference / "extracted" / "stresses.csv", stress_rows("100.0"))
|
|
write_csv(actual / "extracted" / "stresses.csv", stress_rows("101.0"))
|
|
|
|
exit_code = call_main_silently(
|
|
compare,
|
|
[
|
|
"--metadata",
|
|
str(reference / "metadata.json"),
|
|
"--actual-root",
|
|
str(actual),
|
|
"--quantity",
|
|
"stresses",
|
|
"--report-json",
|
|
str(report_json),
|
|
],
|
|
)
|
|
|
|
report = json.loads(report_json.read_text(encoding="utf-8"))
|
|
|
|
self.assertEqual(exit_code, 1)
|
|
self.assertEqual(report["overall_result"], "fail")
|
|
|
|
def test_cli_returns_two_for_invalid_arguments(self):
|
|
compare = load_compare_extracted_csv()
|
|
|
|
self.assertEqual(call_main_silently(compare, []), 2)
|
|
|
|
def test_default_comparison_validates_reference_artifact_metadata(self):
|
|
compare = load_compare_extracted_csv()
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
reference = root / "references" / "umat" / "single-element"
|
|
actual = root / "external-results" / "umat" / "single-element"
|
|
write_json(reference / "metadata.json", metadata_payload())
|
|
write_csv(reference / "extracted" / "stresses.csv", stress_rows("100.0"))
|
|
write_csv(actual / "extracted" / "stresses.csv", stress_rows("100.0"))
|
|
|
|
report = compare.compare_metadata(reference / "metadata.json", actual)
|
|
|
|
self.assertEqual(report["overall_result"], "fail")
|
|
self.assertEqual(report["quantities"][0]["classification"], "missing-reference-artifact")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|