"""Physical regressions for the corrected native pipe and error-control path.""" import json import math from pathlib import Path import subprocess import tempfile import unittest import numpy as np from app.main import simulation_event_stream from app.simulation.core.medium import IdealGasMedium from app.simulation.native_codegen.build import build_native, toolchain from app.simulation.native_codegen.extended import compile_extended_program from tests.test_native_catalog import Circuit ROOT = Path(__file__).resolve().parents[1] class NativePipePhysicsTests(unittest.TestCase): @classmethod def setUpClass(cls): try: cls.compiler = toolchain()[0] except (OSError, RuntimeError, subprocess.SubprocessError) as exc: raise unittest.SkipTest(f'Native toolchain unavailable: {exc}') def test_compressible_pipe_law_is_symmetric_and_regular_near_equilibrium(self): harness = r''' #include #include #include "kernels.h" #define CHECK(x) do { if(!(x)) { fprintf(stderr,"line %d\n",__LINE__);return 1; } } while(0) int main(void) { NativeMedium medium={0,287,1005,300,0,1.8e-5,300,110.4}; for(int helium=0;helium<2;helium++) { medium.real_helium=helium; if(helium) { medium.R=2077.26439404998;medium.cp=5193.1609851249505; } double last=0; const double dp[]={0,1e-5,1e-4,1e-3,.01,.1,1,100,1e4,1e5}; for(unsigned i=0;i=last && reversed==-q); CHECK(q==native_pipe_flow(&medium,2e5+dp[i],2e5,300,.01,1,1e-5,1)); if(i==0) CHECK(q==0); if(dp[i]<=.01) CHECK(q<=1e-7); last=q; } for(double T=200;T<=1000;T+=100) { double mu=native_viscosity(&medium,T,0); CHECK(mu>0 && mu==native_viscosity(&medium,T,1)); if(!helium && T==300) CHECK(fabs(mu-1.8e-5)<1e-18); double d[4],q=.0001; native_pipe_diagnostics(&medium,q,2e5,T,.01,1,1e-5,1,d); CHECK(fabs(d[0]*mu-4*q/(3.14159265358979323846*.01))<1e-12); } } return 0; } ''' with tempfile.TemporaryDirectory(prefix='native-pipe-physics-') as tmp: directory=Path(tmp); source=directory/'check.c'; exe=directory/'check.exe' source.write_text(harness) build=subprocess.run([self.compiler,'-std=c11','-O3','-Wall','-Wextra','-Werror', '-I',str(ROOT/'native/include'),str(source),str(ROOT/'native/components/kernels.c'), '-lm','-o',str(exe)],capture_output=True,text=True,timeout=60) self.assertEqual(build.returncode,0,build.stderr) run=subprocess.run([str(exe)],capture_output=True,text=True,timeout=15) self.assertEqual(run.returncode,0,run.stderr) def test_scalar_pipe_solver_checks_residual_and_bracket_fallback(self): harness = r''' #include #define CHECK(x) do { if(!(x)){fprintf(stderr,"line %d\n",__LINE__);return 1;} } while(0) static double reference(double K,double rr) { double lo=0,hi=fmax(sqrt(K/.02),1); while(hi*hi*pipe_friction(hi,rr)K)hi=mid;else lo=mid; } return .5*(lo+hi); } int main(void) { NativePipeSolve s;int fallbacks=0; for(int i=0;i<=150;i++)for(int r=0;r<6;r++)for(int scale=0;scale<3;scale++) { double re=pow(10,-6+.1*i),rr=r==0?0:pow(10,-6+r); double K=re*re*pipe_friction(re,rr),q_per_re=pow(10,-12+6*scale); double actual=native_pipe_resistance(K,rr,q_per_re,&s),expected=reference(K,rr); CHECK(s.converged && isfinite(actual) && s.relative_residual<=1e-9); CHECK(s.iterations<20 && fabs(actual-expected)<=1e-12+expected*1e-9); CHECK(fabs(actual*actual*pipe_friction(actual,rr)/K-1)<=1e-9); fallbacks+=s.bisections; } CHECK(fallbacks>0); CHECK(native_pipe_resistance(0,0,1,&s)==0 && s.converged); CHECK(isnan(native_pipe_resistance(-1,0,1,&s)) && !s.converged); CHECK(isnan(native_pipe_resistance(NAN,0,1,&s)) && !s.converged); CHECK(isnan(native_pipe_resistance(1,0,0,&s)) && !s.converged); /* Direct laminar handling preserves the distinct PNL00R branch. */ NativeMedium medium={0,287,1005,300,0,1.8e-5,300,110.4}; double p=2e5,T=300,d=.01,L=1,cm,v; medium_valve(NULL,&medium,p,p-1,T,&cm,&v); double expected=pow(PI*d*d/4*p*cm,2)/(16*PI*native_viscosity(&medium,T,0)*L*T); CHECK(4*expected/(PI*d*native_viscosity(&medium,T,0))<1000); CHECK(native_pipe_flow(&medium,p,p-1,T,d,L,1e-5,0)==expected); return 0; } ''' with tempfile.TemporaryDirectory(prefix='native-pipe-root-') as tmp: directory=Path(tmp);source=directory/'check.c';exe=directory/'check.exe' source.write_text((ROOT/'native/components/kernels.c').read_text()+harness) build=subprocess.run([self.compiler,'-std=c11','-O3','-Wall','-Wextra','-Werror', '-ffp-contract=off','-fno-fast-math','-static-libgcc','-I',str(ROOT/'native/include'), str(source),'-lm','-o',str(exe)],capture_output=True,text=True,timeout=60) self.assertEqual(build.returncode,0,build.stderr) run=subprocess.run([str(exe)],capture_output=True,text=True,timeout=30) self.assertEqual(run.returncode,0,run.stderr) def test_pnl0001_uses_upstream_temperature_in_both_directions(self): for reverse in (False,True): with self.subTest(reverse=reverse): circuit=Circuit(IdealGasMedium()) left=circuit.chamber('left',p0=1e5 if reverse else 2e5,T0=300,kth=0) pipe=circuit.add('amesim_pnl0001','pipe',p0=2e5 if reverse else 1e5, T0=300,diam=.01,le=1,rr=1e-5,kth=0) circuit.connect(left,'port_1',pipe,'port_1') program=compile_extended_program(circuit.seal()); build=build_native(program) initial=json.loads(subprocess.run([str(build.executable),'--init'], capture_output=True,text=True,check=True,timeout=15).stdout) states=[initial] # For ideal gas, at fixed U and volume, halving mass doubles # temperature while preserving pressure. Vary each side alone. for name in ('left','pipe'): state=initial.copy(); state[program.state_keys.index(name+'.m')]/=2 states.append(state) inputs=''.join('0 '+' '.join(map(str,state))+'\n' for state in states) run=subprocess.run([str(build.executable),'--probe'],input=inputs, capture_output=True,text=True,check=True,timeout=15) rows=[json.loads(line) for line in run.stdout.splitlines()] outputs=[dict(zip((v.key for v in program.variables),row['outputs'])) for row in rows] q=[out['pipe.port_1.m_flow'] for out in outputs] self.assertTrue(all(row['success'] for row in rows)) self.assertTrue(all(flow<0 if reverse else flow>0 for flow in q)) upstream=2 if reverse else 1; downstream=1 if reverse else 2 self.assertLess(abs(q[upstream]),abs(q[0])*.9) self.assertAlmostEqual(q[downstream],q[0],places=12) self.assertAlmostEqual(outputs[downstream]['pipe.re'],outputs[0]['pipe.re'],places=8) for row in rows: for field in ('m','U'): rates=[value for key,value in zip(program.state_keys,row['rhs']) if key.endswith('.'+field)] self.assertLess(abs(sum(rates)),1e-10+sum(map(abs,rates))*1e-12) def test_web_stream_completes_mql4_and_matches_amesim_reference(self): reference=json.loads((ROOT/'tests/data/test-mql-4-amesim-reference.json').read_text()) xml=(ROOT/'tests/data/test-mql-4-corrected.xml').read_bytes() events=[json.loads(line) for line in simulation_event_stream(xml)] self.assertFalse([e for e in events if e['event']=='error']) self.assertTrue(any(e['event']=='progress' for e in events)) result=next(e['result'] for e in events if e['event']=='result') self.assertTrue(result['success'],result['message']) self.assertEqual(result['simulatedUntil'],10) self.assertEqual(result['diagnostics']['backend'],'native-c') self.assertEqual(result['diagnostics']['integration']['rtol'],1e-7) self.assertEqual(result['diagnostics']['integration']['method'],'BDF') self.assertEqual(result['diagnostics']['stateCount'],64) # Bound the formerly stalled tiny-step failure by work, not machine time. self.assertLess(result['diagnostics']['native']['nfev'],60000) series=result['series']; times=series['time'] self.assertTrue(all(math.isfinite(v) for values in series.values() for v in values)) limits={'pressure':250,'temperature':.015,'displacement':2e-6, 'velocity':1e-5,'mass_flow':3e-5} for key,ref in reference['series'].items(): actual=np.interp(reference['times'],times,series[key]) error=float(np.max(np.abs(actual-ref['values']))) self.assertLessEqual(error,limits[ref['quantity']],(key,error)) masses=[v for k,v in series.items() if k.rsplit('.',1)[-1] in ('m','m1','m2')] total=np.sum(masses,axis=0) self.assertLess(float(np.max(abs(total-total[0]))),1e-10) if __name__=='__main__': unittest.main()