"""Research video contracts: continuous state and display-only sampling.""" import importlib.util import json from pathlib import Path from types import SimpleNamespace import numpy as np import pytest ROOT=Path(__file__).resolve().parents[2] spec=importlib.util.spec_from_file_location('continuous_viv_video_driver',ROOT/'research/continuous_viv/run.py') m=importlib.util.module_from_spec(spec) spec.loader.exec_module(m) from research.continuous_viv import render as r def test_physical_zero_and_independent_steady_control(): model=SimpleNamespace(predict=lambda *a,**k:pytest.fail('steady/OFF must not invoke PPO')) for scene in ('steady','karman'): for phase in ('off_before','off_after'): target,action=m.command(scene,phase,None,model) np.testing.assert_array_equal(target,[0,0,0]) np.testing.assert_array_equal(m.base.wall_target(action,2),target) target,action=m.command('steady','on',None,model) np.testing.assert_array_equal(target,np.array([0,-5.1,5.1])*.005) assert np.isnan(action).all() @pytest.mark.parametrize('bad',[[2,0,0],[np.nan,0,0],[0,0]]) def test_invalid_policy_output(bad): model=SimpleNamespace(predict=lambda *a,**k:(bad,None)) with pytest.raises(RuntimeError,match='invalid PPO'): m.command('karman','on',np.zeros(12),model) def test_continuous_switches_previous_observation_and_native_rhs(tmp_path): from CelerisLab.body.motion import MovingTables calls,forces,seen,frames=[],[],[],[] class Structure: y=v=0.;mass=2.;stiffness=3.;frequency=.1 def candidate(self,f): forces.append(f) return self.y+.001,self.v+.002 class Sim: stepper=SimpleNamespace(step_count=100) n=0 def set_body(self,p,**kw):calls.append((p,kw)) def run(self,n,**kw): assert n==1 and kw==dict(zero_obs=True,sync_obs=True) self.n+=1;self.stepper.step_count+=1 def moving_diagnostics(self): d=np.zeros((3,len(MovingTables.DIAG)),np.float32) d[:,MovingTables.DIAG.index('cme_y')]=self.n d[:,MovingTables.DIAG.index('wen_y')]=10*self.n d[:,MovingTables.DIAG.index('cover_jy')]=1000 return d def read_force(self,p):return [0.,float(self.n),0.] def read_sensor(self,p):return [.001,0.,0.] model=SimpleNamespace(predict=lambda obs,**kw:(seen.append(obs.copy()) or np.ones(3)*.1,None)) norm=dict(force_norm_fact=100.,sens_deviation=np.zeros(6),sens_norm_fact=np.ones(6)) sim,structure=Sim(),Structure() def render(sim,step,phase,eff,rows): if step%3==0:frames.append((step,sim.stepper.step_count,eff.copy())) m.evolve(sim,[1,2,3],[4,5,6],structure,7.,scene='karman',model=model,norm=norm, intervals=2,si=2,out=tmp_path,render=render,start=m.time.monotonic(),wall_cap=60) assert sim.n==12 and sim.stepper.step_count==112 assert forces==[7.,*list(33.*np.arange(1,12))] # no reset, SI averaging, or node impulse in RHS rows=np.load(tmp_path/'steps.npy') np.testing.assert_allclose(rows[:,1],.001*np.arange(1,13)) np.testing.assert_allclose(rows[:,2],.002*np.arange(1,13)) np.testing.assert_array_equal(rows[:4,15:],0) np.testing.assert_array_equal(rows[8:,15:],0) np.testing.assert_array_equal(rows[4,15:],rows[6,15:]) # per-SI EMA reset, not per-frame signals=json.loads((tmp_path/'signals.json').read_text()) np.testing.assert_array_equal(seen[0],np.asarray(signals[1]['observation'],np.float32)) np.testing.assert_array_equal(seen[1],np.asarray(signals[2]['observation'],np.float32)) assert len(seen)==2 and seen[0].any() report=json.loads((tmp_path/'report.json').read_text()) assert [x['absolute_clock'] for x in report['switches']]==[100,104,108] assert [x['previous_rhs'] for x in report['switches']]==[7.,132.,264.] np.testing.assert_allclose([x['y'] for x in report['switches']],[0,.004,.008]) assert [x[0] for x in frames]==[3,6,9,12] assert all(x[1]==100+x[0] for x in frames) assert len(calls)==36 and all(c[1]['vy']==pytest.approx(.001) for c in calls) def test_gear_schedule_and_complete_phase_endpoints(): n=1280000 schedule=r.video_schedule(n) times=list(schedule) assert times==sorted(set(times)) and times[-1]==3*n assert all(p*n in schedule for p in (1,2,3)) assert set(schedule.values())=={1,2,5,10} assert all(speed==1 for step,speed in schedule.items() if step%n>n//2) part=r.frame_schedule(n//2,800,True) assert [s for _,s in part[:15]]==[1]*15 assert [s for _,s in part[15:30]]==[2]*15 assert [s for _,s in part[-15:]]==[1]*15 assert part[-1][0]==n//2 def test_smoke_schedule_covers_all_switches(): schedule=r.video_schedule(32,True) assert list(schedule)==list(range(8,97,8)) assert set(schedule.values())=={1} def test_vorticity_sign_scale_and_solid_stencil_mask(): y,x=np.mgrid[:9,:11] ux=-.0001*y;uy=.0001*x flags=np.ones(x.shape,np.uint16) omega=r.vorticity(ux,uy,flags) np.testing.assert_allclose(omega[1:-1,1:-1],.0002*40/.005) assert np.isnan(omega[0]).all() flags[4,5]=2 omega=r.vorticity(ux,uy,flags) for yy,xx in ((4,5),(3,5),(5,5),(4,4),(4,6)): assert np.isnan(omega[yy,xx]) assert np.isfinite(omega[3,4]) def test_fixed_inset_coordinate_and_no_motion_gain(): a=np.array(r.pixel(30,0,r.INSET,r.INSET_LIMITS)) b=np.array(r.pixel(30,1,r.INSET,r.INSET_LIMITS)) c=np.array(r.pixel(31,0,r.INSET,r.INSET_LIMITS)) assert b[0]==a[0] and a[1]-b[1]==56 assert c[0]-a[0]==56 assert r.pixel(r.INSET_LIMITS[0],r.INSET_LIMITS[3],r.INSET,r.INSET_LIMITS)==r.INSET[:2] def test_config_real_loader_and_structure_unchanged(tmp_path): from CelerisLab.config import load_lbm_config p=tmp_path/'config.json';p.write_text(json.dumps(m.base.config(40))) cfg=load_lbm_config(str(p));cfg.validate() assert (cfg.nx,cfg.ny,cfg.velocity,cfg.viscosity)==(2559,1022,.005,.004) s=m.base.Structure(4.,2.5e-5,0.,40.) assert s.mass==pytest.approx(15079.644737231007) assert s.stiffness==pytest.approx(.00037207532016359776) assert s.damping==0 and s.frequency*40/.005==.2 def test_scene_timing_keeps_actual_legacy_intervals_and_same_physical_duration(): assert m.timing('steady',False)==(320,4000,2048000) assert m.timing('karman',False)==(400,3200,2048000) for scene in ('steady','karman'): n,si,warm=m.timing(scene,False) assert n*si*.005/40==160 assert m.timing(scene,True)==(1,32,32)