92 lines
4.7 KiB
Python
92 lines
4.7 KiB
Python
"""CPU-only fixed-window observation tests; no GPU or simulated acceptance."""
|
|
from contextlib import redirect_stdout
|
|
from io import StringIO
|
|
from pathlib import Path
|
|
from tempfile import TemporaryDirectory
|
|
from types import SimpleNamespace, ModuleType
|
|
from unittest.mock import patch
|
|
import json
|
|
import sys
|
|
import unittest
|
|
import numpy as np
|
|
sys.path.insert(0,str(Path(__file__).resolve().parent))
|
|
import diagnose_uniform_srt as d
|
|
|
|
|
|
def test_raw_bits_copy_and_prefix():
|
|
flags=np.ones((12,20),np.uint16);p=dict(dt=.25,mean_speed=.01,diameter=80)
|
|
h=d.InletHistory(flags,p)
|
|
raw=np.arange(9*12*20,dtype=np.uint32).reshape(9,12,20).view(np.float32)
|
|
raw[0,0,0]=np.float32(-0.)
|
|
expected=raw[:,:,:4].view(np.uint32).copy()
|
|
h.record(dict(raw=raw,step=0)); raw.fill(7); flags.fill(0)
|
|
h.record(dict(raw=raw,step=1))
|
|
with TemporaryDirectory() as temp:
|
|
receipt=h.save(Path(temp),1);z=np.load(Path(temp)/'inlet_history.npz')
|
|
np.testing.assert_array_equal(z['raw_shifted'][0].view(np.uint32),expected)
|
|
assert np.all(z['flags']==1)
|
|
np.testing.assert_array_equal(z['steps'],[0,1])
|
|
np.testing.assert_array_equal(z['time_s'],[0,.25])
|
|
np.testing.assert_array_equal(z['tstar'],[0,.01/80])
|
|
assert not receipt['complete_for_requested_window'] and receipt['diagnostic_terminal_step']==1
|
|
|
|
|
|
def test_fixed_window_bound_and_late_stop():
|
|
flags=np.ones((330,1761),np.uint16);p=dict(dt=.1,mean_speed=.01,diameter=80)
|
|
h=d.InletHistory(flags,p);raw=np.zeros((9,330,4),np.float32)
|
|
for step in range(321):h.record(dict(raw=raw,step=step))
|
|
assert len(h.frames)==197
|
|
with TemporaryDirectory() as temp:
|
|
r=h.save(Path(temp),320)
|
|
assert r['complete_for_requested_window'] and r['last_step']==196 and r['diagnostic_terminal_step']==320
|
|
assert r['bytes']<d.HISTORY_LIMIT_BYTES
|
|
with unittest.TestCase().assertRaises(ValueError):d.InletHistory(flags,p,limit=100)
|
|
with unittest.TestCase().assertRaises(ValueError):
|
|
d.InletHistory(flags,p).record(dict(raw=np.zeros((9,331,4),np.float32),step=0))
|
|
|
|
|
|
def test_off_by_default():
|
|
import subprocess
|
|
with TemporaryDirectory() as temp:
|
|
subprocess.run([sys.executable,str(Path(d.__file__)),'--output',temp],check=True,capture_output=True)
|
|
r=json.loads((Path(temp)/'preflight.json').read_text())
|
|
assert r['observation_mode']=='first_health_pair' and r['history_requested_steps'] is None
|
|
assert not (Path(temp)/'inlet_history.npz').exists()
|
|
# Off-path must not inspect or modify observation state.
|
|
d.record_history(None,object())
|
|
|
|
|
|
def test_execution_identical_and_failure_included():
|
|
class Sim:
|
|
def __init__(self,*args,**kwargs):self.stepper=SimpleNamespace(step_count=0);self.calls=[]
|
|
def __enter__(self):instances.append(self);return self
|
|
def __exit__(self,*args):pass
|
|
def initialize(self):self.calls.append('initialize')
|
|
def get_flags(self):return np.ones((12,20),np.uint16)
|
|
def run(self,n):self.calls.append(('run',n));self.stepper.step_count+=n
|
|
module=ModuleType('CelerisLab.simulation');module.Simulation=Sim
|
|
def snapshot(sim,p):
|
|
raw=np.full((9,12,20),sim.stepper.step_count,np.float32)
|
|
return dict(raw=raw,g=raw.copy(),macro={},step=sim.stepper.step_count)
|
|
def diagnose(g,m,f,flags,p,step):return dict(physical_failure=step==3)
|
|
p=dict(ny=12,nx=20,dt=.1,mean_speed=.01,diameter=80);instances=[];details=[]
|
|
def save(output,prev,current,flags,report):details.append((prev['step'],current['step'],report));return 0
|
|
with patch.dict(sys.modules,{'CelerisLab.simulation':module}),patch.object(d,'snapshot',snapshot),patch.object(d,'diagnose',diagnose),patch.object(d,'save_states',save),redirect_stdout(StringIO()):
|
|
for enabled in [False,True]:
|
|
with TemporaryDirectory() as temp:
|
|
a=SimpleNamespace(wall_seconds=100,output=Path(temp),identity={},record_inlet_history=enabled)
|
|
assert d.execute(a,{},dict(objects=[]),p)==2
|
|
if enabled:
|
|
z=np.load(Path(temp)/'inlet_history.npz')
|
|
np.testing.assert_array_equal(z['steps'],[0,1,2,3])
|
|
assert np.all(z['raw_shifted'][-1]==3)
|
|
assert details[-1][2]['inlet_history']['diagnostic_terminal_step']==3
|
|
assert instances[0].calls==instances[1].calls==['initialize',('run',1),('run',1),('run',1)]
|
|
assert [(p,c) for p,c,_ in details]==[(2,3),(2,3)]
|
|
assert details[0][2]['status']==details[1][2]['status']=='first_physical_failure'
|
|
|
|
if __name__=='__main__':
|
|
result=unittest.TextTestRunner(verbosity=2).run(unittest.TestSuite(
|
|
unittest.FunctionTestCase(fn) for name,fn in sorted(globals().items()) if name.startswith('test_')))
|
|
raise SystemExit(not result.wasSuccessful())
|