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.registry import get_component_model_spec from app.simulation.reporting.amesim_results import load_test_mql_amesim_results REPOSITORY_ROOT = Path(__file__).resolve().parents[1] AME_PATH = REPOSITORY_ROOT / "AmesimModels" / "test_mql.ame" JSON_PATH = REPOSITORY_ROOT / "tests/baselines/simulation/test_mql_8/sources/test-mql-8.json" XML_PATH = REPOSITORY_ROOT / "tests/baselines/simulation/test_mql_8/sources/test-mql-8.xml" 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}>(.*?)", 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*", 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], ) -> 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_, ) 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 parameters[key] = _expected_parameters(body, model_type, globals_) # 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_ ) 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": "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()