"""Native catalog regression against frozen, independent Python results.""" import json import subprocess import unittest import numpy as np from app.simulation.native_codegen.extended import catalog_contracts, compile_extended_program from app.simulation.native_codegen.compiler import compile_native_program from app.simulation.native_codegen.build import build_native, toolchain from app.simulation.systems.network import SimulationNetwork from app.simulation.components.amesim.media.mediums import AmesimHeliumPengRobinsonMedium from tests.native_reference import reference_data, reference_network class Circuit: def __init__(self, medium=None): self.medium=medium or AmesimHeliumPengRobinsonMedium() self.net=SimulationNetwork('native_catalog') def add(self, kind, name, **params): cls=catalog_contracts()[kind] values={p.name:p.default for p in cls.PARAMETERS};values.update(params) c=cls.create(name=name,medium=self.medium,parameters=values) self.net.add_component(c) return c def connect(self,a,ap,b,bp): self.net.connect(a.name,ap,b.name,bp) def chamber(self,name,**params): return self.add('amesim_pnch023',name,**params) def seal(self): connected={e.key for edge in self.net.connections for e in edge.endpoints} for c in list(self.net.components.values()): for p in c.active_port_definitions: if p.domain=='pneumatic' and (c.name,p.name) not in connected: cap=self.add('amesim_pnpl01','cap_'+str(len(self.net.components))) self.connect(c,p.name,cap,'port_1') return self.net class NativeCatalogTests(unittest.TestCase): @classmethod def setUpClass(cls): try: toolchain() except (OSError, RuntimeError, subprocess.SubprocessError) as exc: raise unittest.SkipTest(f'Native toolchain unavailable: {exc}') def test_frozen_python_reference(self): data = reference_data() for index, case in enumerate(data['cases']): with self.subTest(network=index): net = reference_network(case) # Exercise the optimized compiler as well as the general catalog. programs = [compile_extended_program(net)] if index >= len(data['cases']) - 3: programs.append(compile_native_program(net)) for program in programs: build = build_native(program) state_order = [case['stateKeys'].index(key) for key in program.state_keys] output_order = [case['outputKeys'].index(v.key) for v in program.variables] initial = json.loads(subprocess.run([str(build.executable), '--init'], capture_output=True, text=True, check=True, timeout=15).stdout) np.testing.assert_allclose(initial, [case['initialState'][i] for i in state_order], rtol=data['rtol'], atol=data['atol']) inputs = [' '.join(format(x, '.17g') for x in (probe['time'], *[probe['state'][i] for i in state_order])) for probe in case['probes']] result = subprocess.run([str(build.executable), '--probe'], input='\n'.join(inputs)+'\n', capture_output=True, text=True, check=True, timeout=30) rows = [json.loads(line) for line in result.stdout.splitlines()] self.assertEqual(len(rows), len(inputs)) for row, probe in zip(rows, case['probes']): self.assertTrue(row['success']) np.testing.assert_allclose(row['rhs'], [probe['rhs'][i] for i in state_order], rtol=data['rtol'], atol=data['atol'], err_msg=f'network {index} RHS') np.testing.assert_allclose(row['outputs'], [probe['outputs'][i] for i in output_order], rtol=data['rtol'], atol=data['atol'], err_msg=f'network {index} outputs') def test_entire_registered_catalog_is_covered(self): from app.simulation.native_codegen.contracts import SUPPORTED_VERSIONS self.assertEqual(len(catalog_contracts()), 27) self.assertEqual({key: cls.MODEL_VERSION for key, cls in catalog_contracts().items()}, SUPPORTED_VERSIONS) def test_four_rigidly_connected_masses_obey_newtons_law(self): b = Circuit() node = b.add('amesim_lmechn1', 'node', v1=3) for i, name in enumerate((p.name for p in node.active_port_definitions), start=1): mass = b.add('amesim_mecmas21', 'm'+str(i), mass=i, stoptype=4, useFriction=1) force = b.add('amesim_forc', 'f'+str(i)) signal = b.add('amesim_step0', 's'+str(i), initial=i*i, final=i*i, time=1) b.connect(node, name, mass, 'port_1') b.connect(mass, 'port_2', force, 'port_2') b.connect(force, 'res', signal, 'out') build = build_native(compile_extended_program(b.net)) initial = json.loads(subprocess.run([str(build.executable), '--init'], capture_output=True, text=True, check=True).stdout) self.assertEqual(len(initial), 2) row = json.loads(subprocess.run([str(build.executable), '--probe'], input='0 '+' '.join(map(str, initial))+'\n', capture_output=True, text=True, check=True).stdout) self.assertTrue(row['success']) np.testing.assert_allclose(row['rhs'], [3, 0], atol=1e-12) def test_restitution_and_algebraic_models_execute_both_solvers(self): import tempfile from pathlib import Path from app.simulation.config import SolveIVPConfig from app.simulation.native_codegen.runner import execute_native b=Circuit();m=b.add('amesim_mecmas21','mass',mass=1,stoptype=3,xmin=0,xmax=1,x0=.9,v0=1,restcoeff=.5,restdvel=.001,useFriction=1) for name in ('port_1','port_2'): free=b.add('amesim_f000','free_'+name);b.connect(m,name,free,'port_1') build=build_native(compile_extended_program(b.net)) signal=Circuit();signal.add('amesim_step0','step',time=.1,initial=2,final=4) algebraic=build_native(compile_extended_program(signal.net)) with tempfile.TemporaryDirectory() as tmp: for method in ('RK45','BDF'): cfg=SolveIVPConfig(method=method,t_stop=.4,max_step=.02,rtol=1e-7) result=execute_native(build,cfg,.02,run_dir=Path(tmp)/method) self.assertTrue(result['success']) np.testing.assert_allclose(result['finalState'],[-.5,.85],atol=2e-6,rtol=1e-6) result=execute_native(algebraic,cfg,.02,run_dir=Path(tmp)/(method+'-signal')) self.assertTrue(result['success']) self.assertEqual(result['series']['step.y'][-1],4)