866 lines
29 KiB
Python
866 lines
29 KiB
Python
from __future__ import annotations
|
|
|
|
import ast
|
|
from collections import Counter
|
|
from collections.abc import Mapping
|
|
from dataclasses import dataclass
|
|
import hashlib
|
|
import json
|
|
import math
|
|
import operator
|
|
from pathlib import Path
|
|
import re
|
|
import tarfile
|
|
import unittest
|
|
import xml.etree.ElementTree as ET
|
|
|
|
from app.simulation.paths import HISTORICAL_MQL8_AME_PATH, HISTORICAL_MQL8_PROJECT_PATH, HISTORICAL_MQL8_XML_PATH
|
|
from app.simulation.registry import get_component_model_spec
|
|
from app.simulation.components.amesim.semantics import amesim_choice_to_public
|
|
from app.simulation.reporting.amesim_results import load_test_mql_amesim_results
|
|
|
|
|
|
REPOSITORY_ROOT = Path(__file__).resolve().parents[1]
|
|
AME_PATH = HISTORICAL_MQL8_AME_PATH
|
|
JSON_PATH = HISTORICAL_MQL8_PROJECT_PATH
|
|
XML_PATH = HISTORICAL_MQL8_XML_PATH
|
|
MANIFEST_PATH = (
|
|
REPOSITORY_ROOT
|
|
/ "tests/baselines/simulation/test_mql_8/manifest.json"
|
|
)
|
|
|
|
EXPECTED_AME_SHA256 = (
|
|
"cbc3aadd4569a49b3a63e5d66d4143ec16126c0f950df73fb637e07673c20fbb"
|
|
)
|
|
EXPECTED_AME_BYTES = 21_708_800
|
|
EXPECTED_MODEL_TYPE_COUNT = 20
|
|
EXPECTED_NODE_COUNT = 157
|
|
EXPECTED_PARAMETER_COUNT = 1_092
|
|
EXPECTED_CONNECTION_COUNT = 178
|
|
EXPECTED_RESULT_POINT_COUNT = 1_002
|
|
AMESIM_REFERENCE_PRESSURE_PA = 101_300.0
|
|
|
|
_BINARY_OPERATORS = {
|
|
ast.Add: operator.add,
|
|
ast.Sub: operator.sub,
|
|
ast.Mult: operator.mul,
|
|
ast.Div: operator.truediv,
|
|
ast.Pow: operator.pow,
|
|
}
|
|
_UNARY_OPERATORS = {ast.UAdd: operator.pos, ast.USub: operator.neg}
|
|
|
|
# Public component names for AMESim state/output values that are promoted to
|
|
# explicit initial-value parameters. Unlisted names map one-to-one.
|
|
_SOURCE_FIELD_BY_PARAMETER = {
|
|
"amesim_mecmas21": {"v0": "v1", "x0": "x1"},
|
|
"amesim_pnch012": {"p0": "press", "T0": "temp"},
|
|
"amesim_pnch023": {"p0": "press", "T0": "temp"},
|
|
"amesim_pnl0001": {"p0": "p2", "T0": "t2"},
|
|
"amesim_pnl0002": {"p0": "pctr", "T0": "tctr"},
|
|
"amesim_pnl0003": {
|
|
"p1_0": "p1",
|
|
"T1_0": "t1",
|
|
"p2_0": "p2",
|
|
"T2_0": "t2",
|
|
},
|
|
"amesim_step0": {"initial": "out0", "final": "out1", "time": "t0"},
|
|
}
|
|
_FIXED_PUBLIC_DEFAULTS = {
|
|
"amesim_pnch012": {
|
|
"vol1": 0.0,
|
|
"vol2": 0.0,
|
|
"vol3": 0.0,
|
|
"vol4": 0.0,
|
|
"dvol1": 0.0,
|
|
"dvol2": 0.0,
|
|
"dvol3": 0.0,
|
|
"dvol4": 0.0,
|
|
},
|
|
"amesim_pnvo001": {"opening0": 1.0},
|
|
}
|
|
_GAUGE_PRESSURE_FIELDS = {"press", "p1", "p2", "pctr"}
|
|
_SOURCE_TO_SI_SCALE = {
|
|
("mm", "m"): 1.0e-3,
|
|
("mm**2", "m2"): 1.0e-6,
|
|
("L", "m3"): 1.0e-3,
|
|
("N/mm", "N/m"): 1.0e3,
|
|
("N/(mm/s)", "N/(m/s)"): 1.0e3,
|
|
}
|
|
_EQUIVALENT_UNIT_PAIRS = {
|
|
("", ""),
|
|
("null", ""),
|
|
("degree", ""),
|
|
("J/m**2/K/s", "W/(m2*K)"),
|
|
("N/m**2", "Pa"),
|
|
("m**2", "m2"),
|
|
("N/(m/s)**2", "N/(m/s)^2"),
|
|
}
|
|
_SELECTABLE_UNITS_BY_QUANTITY = {
|
|
"area": {"m2", "cm2", "mm2"},
|
|
"heat_transfer_coefficient": {"W/(m2*K)"},
|
|
"pressure": {"Pa", "kPa", "MPa", "bar"},
|
|
"volume": {"m3", "L", "mL"},
|
|
"temperature": {"K", "degC"},
|
|
"length": {"m", "cm", "mm"},
|
|
}
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class AmesimValue:
|
|
expression: str
|
|
unit: str
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class AmesimContract:
|
|
simulation_values: tuple[float, ...]
|
|
node_types: Mapping[str, str]
|
|
parameters: Mapping[str, Mapping[str, float]]
|
|
edges: tuple[tuple[str, str], ...]
|
|
component_count: int
|
|
modeled_line_count: int
|
|
direct_line_count: int
|
|
contact_count: int
|
|
|
|
|
|
def _element_text(body: str, name: str) -> str | None:
|
|
match = re.search(rf"<{name}>(.*?)</{name}>", body, flags=re.DOTALL)
|
|
return match.group(1).strip() if match else None
|
|
|
|
|
|
def _required_element_text(body: str, name: str) -> str:
|
|
value = _element_text(body, name)
|
|
if value is None:
|
|
raise AssertionError(f"AME circuit element is missing <{name}>.")
|
|
return value
|
|
|
|
|
|
def _blocks(body: str, name: str) -> list[str]:
|
|
return re.findall(
|
|
rf"<{name}>\s*(.*?)\s*</{name}>", body, flags=re.DOTALL
|
|
)
|
|
|
|
|
|
def _evaluate_expression(expression: str, variables: Mapping[str, float]) -> float:
|
|
parsed = ast.parse(expression.strip().replace("^", "**"), mode="eval")
|
|
|
|
def evaluate(node: ast.AST) -> float:
|
|
if (
|
|
isinstance(node, ast.Constant)
|
|
and isinstance(node.value, (int, float))
|
|
and not isinstance(node.value, bool)
|
|
):
|
|
return float(node.value)
|
|
if isinstance(node, ast.Name):
|
|
if node.id not in variables:
|
|
raise ValueError(f"unknown AME expression variable {node.id!r}")
|
|
return float(variables[node.id])
|
|
if isinstance(node, ast.BinOp) and type(node.op) in _BINARY_OPERATORS:
|
|
return float(
|
|
_BINARY_OPERATORS[type(node.op)](
|
|
evaluate(node.left), evaluate(node.right)
|
|
)
|
|
)
|
|
if isinstance(node, ast.UnaryOp) and type(node.op) in _UNARY_OPERATORS:
|
|
return float(_UNARY_OPERATORS[type(node.op)](evaluate(node.operand)))
|
|
raise ValueError(f"unsupported expression node {type(node).__name__}")
|
|
|
|
value = evaluate(parsed.body)
|
|
if not math.isfinite(value):
|
|
raise ValueError(f"expression is not finite: {expression!r}")
|
|
return value
|
|
|
|
|
|
def _public_model_type(submodel: str) -> str:
|
|
return (
|
|
"amesim_helium_medium"
|
|
if submodel == "PNGD00"
|
|
else f"amesim_{submodel.lower()}"
|
|
)
|
|
|
|
|
|
def _value_fields(entity_body: str) -> dict[str, AmesimValue]:
|
|
fields: dict[str, AmesimValue] = {}
|
|
for kind in ("RPARAM", "IPARAM", "IVAR", "EVAR"):
|
|
for field_body in _blocks(entity_body, kind):
|
|
name = _element_text(field_body, "VARNAME")
|
|
value = _element_text(field_body, "VALUE")
|
|
if name is not None and value is not None:
|
|
fields[name] = AmesimValue(
|
|
value, _element_text(field_body, "UNITS") or ""
|
|
)
|
|
return fields
|
|
|
|
|
|
def _resolve_globals(cir_text: str) -> dict[str, float]:
|
|
remaining = {
|
|
_required_element_text(body, "GLOB_PARAM_NAME"): _required_element_text(
|
|
body, "VALUE"
|
|
)
|
|
for body in _blocks(cir_text, "GLOBALPARAM")
|
|
}
|
|
resolved: dict[str, float] = {}
|
|
while remaining:
|
|
progressed = False
|
|
for name, expression in tuple(remaining.items()):
|
|
try:
|
|
resolved[name] = _evaluate_expression(expression, resolved)
|
|
except ValueError:
|
|
continue
|
|
del remaining[name]
|
|
progressed = True
|
|
if not progressed:
|
|
raise AssertionError(
|
|
"Unable to resolve AME globals: " + ", ".join(sorted(remaining))
|
|
)
|
|
return resolved
|
|
|
|
|
|
def _source_value_in_public_units(
|
|
source: AmesimValue,
|
|
*,
|
|
target_unit: str,
|
|
source_name: str,
|
|
globals_: Mapping[str, float],
|
|
) -> float:
|
|
value = _evaluate_expression(source.expression, globals_)
|
|
if source_name in _GAUGE_PRESSURE_FIELDS:
|
|
if (source.unit, target_unit) != ("Pa", "Pa"):
|
|
raise AssertionError(
|
|
f"Unexpected gauge-pressure units for {source_name}: "
|
|
f"{source.unit!r} -> {target_unit!r}."
|
|
)
|
|
return value + AMESIM_REFERENCE_PRESSURE_PA
|
|
|
|
pair = (source.unit, target_unit)
|
|
if pair in _SOURCE_TO_SI_SCALE:
|
|
return value * _SOURCE_TO_SI_SCALE[pair]
|
|
if source.unit == target_unit or pair in _EQUIVALENT_UNIT_PAIRS:
|
|
return value
|
|
raise AssertionError(
|
|
f"No AME-to-public conversion for {source_name}: "
|
|
f"{source.unit!r} -> {target_unit!r}."
|
|
)
|
|
|
|
|
|
def _expected_parameters(
|
|
entity_body: str,
|
|
model_type: str,
|
|
globals_: Mapping[str, float],
|
|
*, legacy_raw_choices: bool = False,
|
|
) -> dict[str, float]:
|
|
spec = get_component_model_spec(model_type)
|
|
fields = _value_fields(entity_body)
|
|
expected = {
|
|
parameter.name: float(parameter.default) for parameter in spec.parameters
|
|
}
|
|
source_names = _SOURCE_FIELD_BY_PARAMETER.get(model_type, {})
|
|
fixed_defaults = _FIXED_PUBLIC_DEFAULTS.get(model_type, {})
|
|
unbound: list[str] = []
|
|
for parameter in spec.parameters:
|
|
source_name = source_names.get(parameter.name, parameter.name)
|
|
if source_name in fields:
|
|
expected[parameter.name] = _source_value_in_public_units(
|
|
fields[source_name],
|
|
target_unit=parameter.unit,
|
|
source_name=source_name,
|
|
globals_=globals_,
|
|
)
|
|
if not legacy_raw_choices:
|
|
expected[parameter.name] = amesim_choice_to_public(
|
|
model_type, parameter.name, expected[parameter.name])
|
|
elif parameter.name in fixed_defaults:
|
|
expected[parameter.name] = fixed_defaults[parameter.name]
|
|
if float(parameter.default) != expected[parameter.name]:
|
|
raise AssertionError(
|
|
f"{model_type}.{parameter.name} changed its fixed public "
|
|
"default without an AME source-field mapping."
|
|
)
|
|
elif not (
|
|
(model_type == "amesim_forc" and parameter.name == "direction")
|
|
or (
|
|
model_type == "amesim_helium_medium"
|
|
and parameter.name == "property_model"
|
|
)
|
|
):
|
|
unbound.append(parameter.name)
|
|
|
|
if unbound:
|
|
raise AssertionError(
|
|
f"AME parameter mapping is incomplete for {model_type}: "
|
|
+ ", ".join(unbound)
|
|
)
|
|
|
|
if model_type == "amesim_forc":
|
|
geometry = _required_element_text(entity_body, "COMP_GEOMETRY")
|
|
if geometry not in {"2", "8"}:
|
|
raise AssertionError(f"Unsupported FORC geometry {geometry!r}.")
|
|
expected["direction"] = 1.0 if geometry == "2" else -1.0
|
|
|
|
if model_type == "amesim_helium_medium":
|
|
codes = tuple(
|
|
_evaluate_expression(fields[name].expression, globals_)
|
|
for name in ("fluidType", "eosType", "gasSetting")
|
|
)
|
|
if codes != (12.0, 6.0, 1.0):
|
|
raise AssertionError(
|
|
"PNGD00 no longer selects helium/Peng-Robinson."
|
|
)
|
|
expected["property_model"] = 0.0
|
|
return expected
|
|
|
|
|
|
def _load_ame_contract(path: Path) -> AmesimContract:
|
|
with tarfile.open(path) as archive:
|
|
cir_file = archive.extractfile("test_mql_.cir")
|
|
sim_file = archive.extractfile("test_mql_.sim")
|
|
if cir_file is None or sim_file is None:
|
|
raise AssertionError(
|
|
"AME archive is missing test_mql_.cir or test_mql_.sim."
|
|
)
|
|
cir_text = cir_file.read().decode("latin1")
|
|
simulation_values = tuple(
|
|
float(value)
|
|
for value in sim_file.read().decode("ascii").splitlines()[0].split()
|
|
)
|
|
|
|
globals_ = _resolve_globals(cir_text)
|
|
component_bodies = _blocks(cir_text, "COMP")
|
|
line_bodies = _blocks(cir_text, "LINE")
|
|
node_types: dict[str, str] = {}
|
|
parameters: dict[str, Mapping[str, float]] = {}
|
|
for index, body in enumerate(component_bodies):
|
|
key = f"component:{index}"
|
|
model_type = _public_model_type(
|
|
_required_element_text(body, "SUB_NAME")
|
|
)
|
|
node_types[key] = model_type
|
|
# This loader preserves the August frozen baseline, including its old
|
|
# raw UD00 encoding. Current AME audits use the semantic default above.
|
|
parameters[key] = _expected_parameters(body, model_type, globals_, legacy_raw_choices=True)
|
|
|
|
# PORT_CONNECT=1 is an AMESim contact; line-backed ports use value 2 and
|
|
# are reconstructed from LINES_LIST below.
|
|
directed_contacts: set[
|
|
tuple[tuple[int, int], tuple[int, int]]
|
|
] = set()
|
|
contacts: set[tuple[tuple[int, int], tuple[int, int]]] = set()
|
|
for component_index, body in enumerate(component_bodies):
|
|
for port_index, port_body in enumerate(_blocks(body, "COMP_PORT")):
|
|
if _required_element_text(port_body, "PORT_CONNECT") != "1":
|
|
continue
|
|
for connection_body in _blocks(port_body, "CONNECT"):
|
|
target_index = int(
|
|
_required_element_text(
|
|
connection_body, "CONNECT_ENTITY_NUM"
|
|
)
|
|
)
|
|
target_port = int(
|
|
_required_element_text(
|
|
connection_body, "CONNECT_ENTITY_PORT"
|
|
)
|
|
)
|
|
endpoint = (component_index, port_index)
|
|
target = (target_index, target_port)
|
|
directed_contacts.add((endpoint, target))
|
|
contacts.add(tuple(sorted((endpoint, target))))
|
|
|
|
for endpoint, target in directed_contacts:
|
|
if (target, endpoint) not in directed_contacts:
|
|
raise AssertionError(
|
|
"AME component contact is not reciprocal: "
|
|
f"{endpoint} -> {target}."
|
|
)
|
|
edges = [
|
|
(f"component:{first[0]}", f"component:{second[0]}")
|
|
for first, second in contacts
|
|
]
|
|
|
|
direct_line_count = 0
|
|
modeled_line_count = 0
|
|
for line_index, body in enumerate(line_bodies):
|
|
if (
|
|
_required_element_text(body, "LINE_START_TYPE") != "0"
|
|
or _required_element_text(body, "LINE_END_TYPE") != "0"
|
|
):
|
|
raise AssertionError(
|
|
"test_mql contains a non-component line endpoint."
|
|
)
|
|
start = (
|
|
"component:"
|
|
+ _required_element_text(body, "LINE_START_ENTITY")
|
|
)
|
|
end = (
|
|
"component:"
|
|
+ _required_element_text(body, "LINE_END_ENTITY")
|
|
)
|
|
if start not in node_types or end not in node_types:
|
|
raise AssertionError(
|
|
f"AME line references an unknown endpoint: {start}, {end}."
|
|
)
|
|
submodel = _required_element_text(body, "SUB_NAME")
|
|
if submodel == "DIRECT":
|
|
direct_line_count += 1
|
|
edges.append((start, end))
|
|
continue
|
|
|
|
modeled_line_count += 1
|
|
key = f"line:{line_index}"
|
|
model_type = _public_model_type(submodel)
|
|
node_types[key] = model_type
|
|
parameters[key] = _expected_parameters(
|
|
body, model_type, globals_, legacy_raw_choices=True
|
|
)
|
|
edges.extend(((start, key), (key, end)))
|
|
|
|
return AmesimContract(
|
|
simulation_values=simulation_values,
|
|
node_types=node_types,
|
|
parameters=parameters,
|
|
edges=tuple(edges),
|
|
component_count=len(component_bodies),
|
|
modeled_line_count=modeled_line_count,
|
|
direct_line_count=direct_line_count,
|
|
contact_count=len(contacts),
|
|
)
|
|
|
|
|
|
def _numeric_project_value(value: object) -> float:
|
|
if isinstance(value, bool):
|
|
raise AssertionError(
|
|
"Boolean is not a valid project parameter value."
|
|
)
|
|
if isinstance(value, (int, float)):
|
|
result = float(value)
|
|
elif isinstance(value, str):
|
|
result = _evaluate_expression(value, {})
|
|
else:
|
|
raise AssertionError(
|
|
f"Unsupported project parameter value {value!r}."
|
|
)
|
|
if not math.isfinite(result):
|
|
raise AssertionError(
|
|
f"Project parameter value is not finite: {value!r}."
|
|
)
|
|
return result
|
|
|
|
|
|
def _parameter_fingerprint(
|
|
values: Mapping[str, object],
|
|
) -> tuple[tuple[str, str], ...]:
|
|
# Fifteen significant digits collapse only normal IEEE-754 expression
|
|
# noise (for example 100000.00000000001 versus 100000).
|
|
return tuple(
|
|
sorted(
|
|
(name, format(_numeric_project_value(value), ".15g"))
|
|
for name, value in values.items()
|
|
)
|
|
)
|
|
|
|
|
|
def _topology_fingerprints(
|
|
node_types: Mapping[str, str],
|
|
edges: tuple[tuple[str, str], ...],
|
|
*,
|
|
rounds: int = 12,
|
|
) -> tuple[Counter[str], ...]:
|
|
adjacency = {node_id: [] for node_id in node_types}
|
|
for first, second in edges:
|
|
if first not in adjacency or second not in adjacency:
|
|
raise AssertionError(
|
|
f"Unknown topology endpoint: {first}, {second}."
|
|
)
|
|
adjacency[first].append(second)
|
|
adjacency[second].append(first)
|
|
|
|
colors = {
|
|
node_id: hashlib.sha256(model_type.encode()).hexdigest()
|
|
for node_id, model_type in node_types.items()
|
|
}
|
|
fingerprints: list[Counter[str]] = []
|
|
for _ in range(rounds):
|
|
colors = {
|
|
node_id: hashlib.sha256(
|
|
repr(
|
|
(
|
|
node_types[node_id],
|
|
colors[node_id],
|
|
sorted(
|
|
colors[neighbor]
|
|
for neighbor in adjacency[node_id]
|
|
),
|
|
)
|
|
).encode()
|
|
).hexdigest()
|
|
for node_id in node_types
|
|
}
|
|
fingerprints.append(Counter(colors.values()))
|
|
return tuple(fingerprints)
|
|
|
|
|
|
class TestMqlAmeAuthorityContractTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls) -> None:
|
|
cls.ame_payload = AME_PATH.read_bytes()
|
|
cls.ame = _load_ame_contract(AME_PATH)
|
|
cls.project = json.loads(JSON_PATH.read_text(encoding="utf-8"))
|
|
cls.xml_root = ET.parse(XML_PATH).getroot()
|
|
cls.manifest = json.loads(
|
|
MANIFEST_PATH.read_text(encoding="utf-8")
|
|
)
|
|
|
|
def test_manifest_and_results_bind_the_exact_ame_archive(self) -> None:
|
|
self.assertEqual(len(self.ame_payload), EXPECTED_AME_BYTES)
|
|
self.assertEqual(
|
|
hashlib.sha256(self.ame_payload).hexdigest(),
|
|
EXPECTED_AME_SHA256,
|
|
)
|
|
|
|
reference = self.manifest["source"]["referenceArchive"]
|
|
self.assertEqual(
|
|
reference,
|
|
{
|
|
"path": "tests/data/AmesimModels/test_mql.ame",
|
|
"sha256": EXPECTED_AME_SHA256,
|
|
"bytes": EXPECTED_AME_BYTES,
|
|
"role": "authoritativePhysicalBaseline",
|
|
"resultStartTime": 0.0,
|
|
"resultStopTime": 10.0,
|
|
"resultSampleStep": 0.01,
|
|
"resultPointCount": EXPECTED_RESULT_POINT_COUNT,
|
|
},
|
|
)
|
|
self.assertEqual(
|
|
(REPOSITORY_ROOT / reference["path"]).resolve(),
|
|
AME_PATH.resolve(),
|
|
)
|
|
|
|
# .sim: start, stop, print interval, ..., maximum integration step.
|
|
self.assertGreaterEqual(len(self.ame.simulation_values), 6)
|
|
self.assertEqual(self.ame.simulation_values[0], 0.0)
|
|
self.assertEqual(self.ame.simulation_values[1], 10.0)
|
|
self.assertEqual(self.ame.simulation_values[2], 0.01)
|
|
self.assertEqual(self.ame.simulation_values[5], 0.001)
|
|
|
|
results = load_test_mql_amesim_results(AME_PATH)
|
|
self.assertEqual(
|
|
len(results.times), EXPECTED_RESULT_POINT_COUNT
|
|
)
|
|
self.assertEqual(
|
|
results.times[0], reference["resultStartTime"]
|
|
)
|
|
self.assertEqual(
|
|
results.times[-1], reference["resultStopTime"]
|
|
)
|
|
self.assertTrue(
|
|
all(
|
|
left < right
|
|
for left, right in zip(
|
|
results.times, results.times[1:]
|
|
)
|
|
)
|
|
)
|
|
step = reference["resultSampleStep"]
|
|
self.assertTrue(
|
|
all(
|
|
math.isclose(
|
|
time,
|
|
round(time / step) * step,
|
|
rel_tol=0.0,
|
|
abs_tol=1.0e-10,
|
|
)
|
|
for time in results.times
|
|
)
|
|
)
|
|
|
|
def test_json_inventory_parameters_and_topology_derive_from_ame(
|
|
self,
|
|
) -> None:
|
|
nodes = self.project["nodes"]
|
|
edges = self.project["edges"]
|
|
json_node_types = {
|
|
node["id"]: node["data"]["modelType"] for node in nodes
|
|
}
|
|
json_edges = tuple(
|
|
(edge["source"], edge["target"]) for edge in edges
|
|
)
|
|
|
|
self.assertEqual(self.ame.component_count, 117)
|
|
self.assertEqual(self.ame.modeled_line_count, 40)
|
|
self.assertEqual(self.ame.direct_line_count, 44)
|
|
self.assertEqual(self.ame.contact_count, 54)
|
|
self.assertEqual(
|
|
len(self.ame.node_types), EXPECTED_NODE_COUNT
|
|
)
|
|
self.assertEqual(
|
|
len(self.ame.edges), EXPECTED_CONNECTION_COUNT
|
|
)
|
|
self.assertEqual(
|
|
len(set(self.ame.node_types.values())),
|
|
EXPECTED_MODEL_TYPE_COUNT,
|
|
)
|
|
|
|
self.assertEqual(len(nodes), EXPECTED_NODE_COUNT)
|
|
self.assertEqual(len(edges), EXPECTED_CONNECTION_COUNT)
|
|
self.assertEqual(
|
|
len(set(json_node_types.values())),
|
|
EXPECTED_MODEL_TYPE_COUNT,
|
|
)
|
|
self.assertEqual(
|
|
Counter(self.ame.node_types.values()),
|
|
Counter(json_node_types.values()),
|
|
)
|
|
|
|
expected_parameter_count = sum(
|
|
len(values) for values in self.ame.parameters.values()
|
|
)
|
|
actual_parameter_count = sum(
|
|
len(node["data"]["parameters"]) for node in nodes
|
|
)
|
|
self.assertEqual(
|
|
expected_parameter_count, EXPECTED_PARAMETER_COUNT
|
|
)
|
|
self.assertEqual(
|
|
actual_parameter_count, EXPECTED_PARAMETER_COUNT
|
|
)
|
|
|
|
expected_by_type: dict[
|
|
str, Counter[tuple[tuple[str, str], ...]]
|
|
] = {}
|
|
for node_id, model_type in self.ame.node_types.items():
|
|
expected_by_type.setdefault(model_type, Counter()).update(
|
|
[
|
|
_parameter_fingerprint(
|
|
self.ame.parameters[node_id]
|
|
)
|
|
]
|
|
)
|
|
actual_by_type: dict[
|
|
str, Counter[tuple[tuple[str, str], ...]]
|
|
] = {}
|
|
for node in nodes:
|
|
model_type = node["data"]["modelType"]
|
|
actual_by_type.setdefault(model_type, Counter()).update(
|
|
[
|
|
_parameter_fingerprint(
|
|
node["data"]["parameters"]
|
|
)
|
|
]
|
|
)
|
|
self.assertEqual(actual_by_type, expected_by_type)
|
|
|
|
expected_edge_types = Counter(
|
|
tuple(
|
|
sorted(
|
|
(
|
|
self.ame.node_types[first],
|
|
self.ame.node_types[second],
|
|
)
|
|
)
|
|
)
|
|
for first, second in self.ame.edges
|
|
)
|
|
actual_edge_types = Counter(
|
|
tuple(
|
|
sorted(
|
|
(
|
|
json_node_types[first],
|
|
json_node_types[second],
|
|
)
|
|
)
|
|
)
|
|
for first, second in json_edges
|
|
)
|
|
self.assertEqual(actual_edge_types, expected_edge_types)
|
|
self.assertEqual(
|
|
_topology_fingerprints(
|
|
json_node_types, json_edges
|
|
),
|
|
_topology_fingerprints(
|
|
self.ame.node_types, self.ame.edges
|
|
),
|
|
)
|
|
|
|
def test_json_units_and_component_contracts_are_complete(
|
|
self,
|
|
) -> None:
|
|
for node in self.project["nodes"]:
|
|
with self.subTest(component=node["id"]):
|
|
data = node["data"]
|
|
self.assertEqual(
|
|
data["componentType"], data["modelType"]
|
|
)
|
|
spec = get_component_model_spec(data["modelType"])
|
|
self.assertEqual(
|
|
data["modelVersion"], spec.model_version
|
|
)
|
|
self.assertEqual(
|
|
set(data["parameters"]),
|
|
{
|
|
parameter.name
|
|
for parameter in spec.parameters
|
|
},
|
|
)
|
|
self.assertEqual(
|
|
{port["name"] for port in data["ports"]},
|
|
{port.name for port in spec.ports},
|
|
)
|
|
|
|
units = data.get("parameterUnits", {})
|
|
self.assertEqual(
|
|
set(units),
|
|
{
|
|
parameter.name
|
|
for parameter in spec.parameters
|
|
if parameter.unit
|
|
},
|
|
)
|
|
for parameter in spec.parameters:
|
|
if not parameter.unit:
|
|
continue
|
|
selected = units[parameter.name]
|
|
allowed = _SELECTABLE_UNITS_BY_QUANTITY.get(
|
|
parameter.quantity, {parameter.unit}
|
|
)
|
|
self.assertIn(selected, allowed, parameter.name)
|
|
|
|
def test_json_and_xml_match_by_id_parameter_endpoint_and_time(
|
|
self,
|
|
) -> None:
|
|
xml_components = self.xml_root.findall(
|
|
"./Components/Component"
|
|
)
|
|
xml_connections = self.xml_root.findall(
|
|
"./Connections/Connection"
|
|
)
|
|
xml_component_by_id = {
|
|
component.attrib["id"]: component
|
|
for component in xml_components
|
|
}
|
|
xml_connection_by_id = {
|
|
connection.attrib["id"]: connection
|
|
for connection in xml_connections
|
|
}
|
|
json_node_by_id = {
|
|
node["id"]: node for node in self.project["nodes"]
|
|
}
|
|
json_edge_by_id = {
|
|
edge["id"]: edge for edge in self.project["edges"]
|
|
}
|
|
|
|
self.assertEqual(
|
|
len(xml_component_by_id), len(xml_components)
|
|
)
|
|
self.assertEqual(
|
|
len(xml_connection_by_id), len(xml_connections)
|
|
)
|
|
self.assertEqual(
|
|
set(xml_component_by_id), set(json_node_by_id)
|
|
)
|
|
self.assertEqual(
|
|
set(xml_connection_by_id), set(json_edge_by_id)
|
|
)
|
|
|
|
for node_id, node in json_node_by_id.items():
|
|
with self.subTest(component=node_id):
|
|
component = xml_component_by_id[node_id]
|
|
data = node["data"]
|
|
self.assertEqual(
|
|
component.attrib["type"], data["modelType"]
|
|
)
|
|
self.assertEqual(
|
|
component.attrib["modelVersion"],
|
|
data["modelVersion"],
|
|
)
|
|
xml_parameters = {
|
|
parameter.attrib["name"]: float(
|
|
parameter.attrib["value"]
|
|
)
|
|
for parameter in component.findall("./Parameter")
|
|
}
|
|
json_parameters = {
|
|
name: _numeric_project_value(value)
|
|
for name, value in data["parameters"].items()
|
|
}
|
|
self.assertEqual(
|
|
set(xml_parameters), set(json_parameters)
|
|
)
|
|
for name, value in json_parameters.items():
|
|
self.assertTrue(
|
|
math.isclose(
|
|
xml_parameters[name],
|
|
value,
|
|
rel_tol=1.0e-13,
|
|
abs_tol=1.0e-13,
|
|
),
|
|
(
|
|
f"{node_id}.{name}: "
|
|
f"XML={xml_parameters[name]!r}, "
|
|
f"JSON={value!r}"
|
|
),
|
|
)
|
|
|
|
for edge_id, edge in json_edge_by_id.items():
|
|
with self.subTest(connection=edge_id):
|
|
endpoints = {
|
|
(
|
|
endpoint.attrib["component"],
|
|
endpoint.attrib["port"],
|
|
)
|
|
for endpoint in xml_connection_by_id[
|
|
edge_id
|
|
].findall("./Endpoint")
|
|
}
|
|
self.assertEqual(
|
|
endpoints,
|
|
{
|
|
(
|
|
edge["source"],
|
|
edge["sourceHandle"],
|
|
),
|
|
(
|
|
edge["target"],
|
|
edge["targetHandle"],
|
|
),
|
|
},
|
|
)
|
|
|
|
xml_simulation = self.xml_root.find("./Simulation")
|
|
self.assertIsNotNone(xml_simulation)
|
|
assert xml_simulation is not None
|
|
project_simulation = self.project["simulation"]
|
|
expected_simulation = {
|
|
"tStart": float(project_simulation["t_start"]),
|
|
"tStop": float(project_simulation["t_stop"]),
|
|
"sampleStep": float(project_simulation["step"]),
|
|
"maxStep": float(project_simulation["max_step"]),
|
|
}
|
|
for name, expected in expected_simulation.items():
|
|
self.assertEqual(
|
|
float(xml_simulation.attrib[name]), expected
|
|
)
|
|
self.assertEqual(
|
|
xml_simulation.attrib["method"],
|
|
project_simulation["method"],
|
|
)
|
|
self.assertEqual(
|
|
project_simulation["t_start"],
|
|
self.ame.simulation_values[0],
|
|
)
|
|
self.assertEqual(
|
|
project_simulation["t_stop"],
|
|
self.ame.simulation_values[1],
|
|
)
|
|
self.assertEqual(
|
|
project_simulation["step"],
|
|
self.ame.simulation_values[2],
|
|
)
|
|
self.assertEqual(
|
|
project_simulation["max_step"],
|
|
self.ame.simulation_values[5],
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|