181 lines
11 KiB
Python
181 lines
11 KiB
Python
"""Native catalog regression against frozen, independent Python results."""
|
|
import json
|
|
import math
|
|
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, NativeCapabilityError
|
|
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
|
|
|
|
|
|
def has_revised_pipe_law(case):
|
|
"""The frozen flow values include capped pipe iterates, not checked roots.
|
|
|
|
All PNL media now share the residual-checked scalar solver. Keep the frozen
|
|
thermodynamic oracle; the pipe suite checks the actual resistance equation
|
|
against independent bisection as well as system-level Amesim curves.
|
|
"""
|
|
return any(c['type'].startswith('amesim_pnl') for c in case['components'])
|
|
|
|
|
|
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)
|
|
invalid_friction = any(c['type']=='amesim_mecmas21' and c['parameters']['useFriction']==2
|
|
and c['parameters']['stoptype']!=3 and c['parameters']['fcoul']>c['parameters']['fstick']
|
|
for c in case['components'])
|
|
if invalid_friction:
|
|
# Three historical Python fixtures had fcoul=2, fstick=0.
|
|
# Amesim itself rejects that active configuration. Preserve
|
|
# the old files and verify explicit rejection in both paths.
|
|
for compiler in (compile_native_program,compile_extended_program):
|
|
with self.assertRaisesRegex(NativeCapabilityError,'fcoul'):
|
|
compiler(net)
|
|
continue
|
|
# 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'])
|
|
if not has_revised_pipe_law(case):
|
|
expected_rhs=list(probe['rhs']);expected_outputs=list(probe['outputs'])
|
|
# The frozen Python model incorrectly applied dry
|
|
# friction in restitution mode. Keep its bytes, but
|
|
# use Newton's law as the oracle for those two fields.
|
|
old=dict(zip(case['outputKeys'],probe['outputs']))
|
|
for component in case['components']:
|
|
if component['type']!='amesim_mecmas21':continue
|
|
params=component['parameters'];name=component['name']
|
|
if params['stoptype']!=3 or params['useFriction']!=2 or params['fcoul']==0:continue
|
|
velocity=old[name+'.v'];mass=params['mass']
|
|
self.assertTrue(params['xmin']<old[name+'.x']<params['xmax'])
|
|
acceleration=(old[name+'.port_1.f']+old[name+'.port_2.f']
|
|
-params['rvisc']*velocity-params['wind']*velocity*abs(velocity))/mass
|
|
acceleration+=9.80665*math.sin(math.radians(params['theta']))
|
|
expected_rhs[case['stateKeys'].index(name+'.v')]=acceleration
|
|
expected_outputs[case['outputKeys'].index(name+'.a')]=acceleration
|
|
expected_outputs[case['outputKeys'].index(name+'.Ffric')]=0
|
|
np.testing.assert_allclose(row['rhs'], [expected_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'], [expected_outputs[i] for i in output_order], rtol=data['rtol'], atol=data['atol'], err_msg=f'network {index} outputs')
|
|
else:
|
|
# Preserve the independent thermodynamic oracle. Flow
|
|
# values intentionally changed; test their balance here
|
|
# and the resistance law/Amesim curves in the pipe suite.
|
|
fields = {'m','U','p','T','rho','u','h',
|
|
'm1','U1','p1','T1','rho1','u1','h1',
|
|
'm2','U2','p2','T2','rho2','u2','h2'}
|
|
keep = [j for j,v in enumerate(program.variables)
|
|
if v.key.count('.') == 1 and v.key.split('.')[1] in fields]
|
|
np.testing.assert_allclose([row['outputs'][j] for j in keep],
|
|
[probe['outputs'][output_order[j]] for j in keep],
|
|
rtol=data['rtol'], atol=data['atol'])
|
|
self.assertTrue(np.isfinite(row['rhs']).all())
|
|
self.assertTrue(np.isfinite(row['outputs']).all())
|
|
for prefix in ('m', 'U'):
|
|
indices = [j for j,k in enumerate(program.state_keys)
|
|
if k.rsplit('.',1)[1].startswith(prefix)]
|
|
actual = sum(row['rhs'][j] for j in indices)
|
|
expected = sum(probe['rhs'][state_order[j]] for j in indices)
|
|
scale = sum(abs(row['rhs'][j]) for j in indices)
|
|
self.assertLessEqual(abs(actual-expected), 1e-9+scale*1e-10)
|
|
outputs = dict(zip((v.key for v in program.variables), row['outputs']))
|
|
for left,right in case['connections']:
|
|
keys = ['.'.join(e)+'.m_flow' for e in (left,right)]
|
|
if all(k in outputs for k in keys):
|
|
self.assertAlmostEqual(sum(outputs[k] for k in keys),0,places=10)
|
|
|
|
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)
|