184 lines
8.9 KiB
Python
184 lines
8.9 KiB
Python
"""Research interface contracts; no CUDA context or physical validation."""
|
|
import importlib.util
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
spec = importlib.util.spec_from_file_location('karman_viv_research_driver', ROOT/'research/karman_viv/run.py')
|
|
m = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(m)
|
|
|
|
|
|
def test_observation_unit_mapping_and_order():
|
|
norm = dict(force_norm_fact=2.,sens_deviation=np.arange(6)/10.,sens_norm_fact=np.ones(6)*3)
|
|
s=np.arange(6)*.001; f=np.arange(6)*.01
|
|
raw,obs,nclip=m.observation(s,f,2,norm)
|
|
np.testing.assert_array_equal(raw,np.asarray([*(s*156),*(f*2)],np.float32))
|
|
expected=np.hstack((raw[6:]/2,(raw[:6]-np.asarray(norm['sens_deviation'],np.float32))/3))
|
|
np.testing.assert_array_equal(obs,expected.astype(np.float32))
|
|
assert nclip==0
|
|
|
|
|
|
def test_same_physical_signal_d20_d40():
|
|
norm=dict(force_norm_fact=1.,sens_deviation=np.zeros(6),sens_norm_fact=np.ones(6))
|
|
a=m.observation(np.ones(6)*.001,np.arange(6)*.01,1,norm)
|
|
b=m.observation(np.ones(6)*.0005,np.arange(6)*.005,2,norm)
|
|
np.testing.assert_array_equal(a[0],b[0]); np.testing.assert_array_equal(a[1],b[1])
|
|
|
|
|
|
def test_zero_action_and_ccw_scale():
|
|
np.testing.assert_array_equal(m.wall_target(-m.BIAS/8,2),np.zeros(3))
|
|
np.testing.assert_allclose(m.wall_target(np.ones(3),2),[.04,.02,.06])
|
|
assert m.wall_target(np.ones(3),2)[0]/20 == .002
|
|
|
|
|
|
def test_ema_equal_physical_time_and_interval_reset():
|
|
target=np.array([.01,-.02,.03]); coarse=np.zeros(3); fine=np.zeros(3)
|
|
for _ in range(13): coarse=m.ema(coarse,target,1)
|
|
for _ in range(52): fine=m.ema(fine,target,2)
|
|
np.testing.assert_allclose(fine,coarse,rtol=2e-15,atol=1e-17)
|
|
np.testing.assert_allclose(m.ema(np.zeros(3),target,2),(1-.9**.25)*target)
|
|
|
|
|
|
def test_config_real_parser(tmp_path):
|
|
import json
|
|
from CelerisLab.config import load_lbm_config
|
|
path=tmp_path/'config.json';path.write_text(json.dumps(m.config(40)))
|
|
cfg=load_lbm_config(str(path));cfg.validate()
|
|
assert (cfg.nx,cfg.ny,cfg.velocity,cfg.viscosity)==(2559,1022,.005,.004)
|
|
|
|
|
|
def test_structure_physical_scaling_and_no_damping():
|
|
coarse=m.Structure(2,1e-4,0,20);fine=m.Structure(2,2.5e-5,0,40)
|
|
assert fine.mass==4*coarse.mass
|
|
assert fine.stiffness==coarse.stiffness/4
|
|
assert fine.damping==0
|
|
assert fine.candidate(0)==(0.,0.)
|
|
|
|
|
|
def test_observation_clipping_and_nonfinite():
|
|
norm=dict(force_norm_fact=1.,sens_deviation=np.zeros(6),sens_norm_fact=np.ones(6))
|
|
_,obs,n=m.observation(np.ones(6),np.ones(6),2,norm)
|
|
assert n==12 and np.array_equal(obs,np.ones(12,np.float32))
|
|
with pytest.raises(RuntimeError,match='nonfinite'):
|
|
m.observation(np.full(6,np.nan),np.zeros(6),2,norm)
|
|
|
|
|
|
@pytest.mark.parametrize('mass_ratio', [2.,4.])
|
|
def test_arm_step_order_and_native_rhs(tmp_path, mass_ratio):
|
|
import CelerisLab
|
|
from CelerisLab.body.motion import MovingTables
|
|
from contextlib import nullcontext
|
|
log=[];candidates=[];structure_args=[]
|
|
class Structure:
|
|
y=v=0.;mass=2.;stiffness=3.;frequency=.1
|
|
def __init__(self,*args): structure_args.append(args)
|
|
def candidate(self,f):
|
|
candidates.append(f)
|
|
return self.y+.001,self.v+.002
|
|
class Sim:
|
|
def __init__(self,*args): self.n=0
|
|
def __enter__(self): return self
|
|
def __exit__(self,*args): pass
|
|
def initialize(self): pass
|
|
def set_body(self,p,**kw): log.append(('set',p,kw))
|
|
def run(self,n,**kw): self.n+=1;log.append(('run',n,kw))
|
|
def moving_diagnostics(self):
|
|
a=np.zeros((3,len(MovingTables.DIAG)),np.float32)
|
|
a[:,MovingTables.DIAG.index('cme_y')]=self.n
|
|
a[:,MovingTables.DIAG.index('wen_y')]=10*self.n
|
|
a[:,MovingTables.DIAG.index('cover_jy')]=1000
|
|
return a
|
|
def read_force(self,p): return [0.,float(self.n),0.]
|
|
def read_sensor(self,p): return [.001,0.,0.]
|
|
seen=[]
|
|
model=SimpleNamespace(predict=lambda obs,**kw:(seen.append(obs.copy()) or np.ones(3)*.1,None))
|
|
norm=dict(force_norm_fact=1.,sens_deviation=np.zeros(6),sens_norm_fact=np.ones(6))
|
|
with patch.object(CelerisLab,'Simulation',Sim),patch.object(m,'Structure',Structure), \
|
|
patch.object(m,'add_scene',return_value=([1,2,3],[4,5,6])),patch.object(m,'install_seed',return_value={}), \
|
|
patch.object(m,'check_time'):
|
|
m.run_arm('controlled','cfg',2,2,2,None,7.,model,norm,tmp_path,0,1,mass_ratio=mass_ratio)
|
|
assert structure_args == [(mass_ratio,2.5e-5,0.,40.)]
|
|
assert candidates==[7.,33.,66.,99.] # previous native force, NOT SI mean or node impulse
|
|
assert not seen[0].any() and len(seen)==2
|
|
assert [e[0] for e in log]==['set','set','set','run']*4
|
|
for e in log:
|
|
if e[0]=='set': assert e[2]['vy']==pytest.approx(.001) and e[2]['omega']!=0
|
|
else: assert e[2]==dict(zero_obs=True,sync_obs=True)
|
|
rows=np.load(tmp_path/'controlled/steps.npy')
|
|
np.testing.assert_allclose(rows[:,1],[.001,.002,.003,.004])
|
|
assert rows.shape==(4,18)
|
|
np.testing.assert_array_equal(rows[0,15:],rows[2,15:]) # EMA resets at SI boundary
|
|
signals=np.load(tmp_path/'controlled/signals.npz')
|
|
np.testing.assert_array_equal(signals['forces'][:,1],[3.,7.]) # interval CME *2, not CME+Wen
|
|
|
|
|
|
def test_mass4_changes_inertia_and_stiffness_not_frequency_or_damping():
|
|
a=m.Structure(2,2.5e-5,0,40);b=m.Structure(4,2.5e-5,0,40)
|
|
assert b.mass==2*a.mass and b.stiffness==2*a.stiffness
|
|
assert b.frequency==a.frequency and b.damping==a.damping==0
|
|
np.testing.assert_allclose(b.candidate(.001),np.asarray(a.candidate(.001))/2,rtol=0,atol=0)
|
|
|
|
|
|
def make_release(path):
|
|
import json
|
|
(path/'zero').mkdir()
|
|
case=dict(source=m.RELEASE_SOURCE,model_sha256=m.MODEL_SHA,norm_sha256=m.NORM_SHA,
|
|
smoke=False,D=40,U=.005,nu=.004,warmup_each=2048000,target_intervals=150,SI=3200)
|
|
(path/'case.json').write_text(json.dumps(case))
|
|
(path/'config.json').write_text(json.dumps(m.config(40)))
|
|
seed=(np.arange(18,dtype=np.float32),np.arange(18,dtype=np.float32)+1,np.zeros(2,np.int32))
|
|
np.savez(path/'release.npz',current=seed[0],spare=seed[1],flags=seed[2],clock=4576000,force=.003)
|
|
np.save(path/'target.npy',np.zeros((150,6),np.float32))
|
|
transfer=dict(zip(('current','spare','flags'),map(m.array_sha,seed)),clock=4576000)
|
|
(path/'zero/report.json').write_text(json.dumps(dict(release_transfer=transfer)))
|
|
return seed
|
|
|
|
|
|
@pytest.mark.parametrize('damage', [None,'source','config','buffer','target','clock'])
|
|
def test_saved_release_checks_provenance_and_exact_inputs(tmp_path, damage):
|
|
import json
|
|
seed=make_release(tmp_path)
|
|
if damage in ('source','config'):
|
|
file=tmp_path/('case.json' if damage=='source' else 'config.json')
|
|
data=json.loads(file.read_text());data['source' if damage=='source' else 'extra']='changed'
|
|
file.write_text(json.dumps(data))
|
|
elif damage in ('buffer','clock'):
|
|
with np.load(tmp_path/'release.npz') as z: data={k:z[k] for k in z.files}
|
|
if damage=='buffer': data['current'][0]+=1
|
|
else: data['clock']=4576001
|
|
np.savez(tmp_path/'release.npz',**data)
|
|
elif damage=='target': np.save(tmp_path/'target.npy',np.zeros((149,6),np.float32))
|
|
if damage:
|
|
with pytest.raises(ValueError): m.saved_release(tmp_path)
|
|
else:
|
|
loaded,force=m.saved_release(tmp_path)
|
|
assert loaded[3]==4576000 and force==.003
|
|
for actual,expected in zip(loaded[:3],seed): np.testing.assert_array_equal(actual,expected)
|
|
|
|
|
|
def test_mass4_cli_uses_saved_release_without_warmup(tmp_path, monkeypatch):
|
|
import json
|
|
original=tmp_path/'original';original.mkdir();make_release(original)
|
|
norm=tmp_path/'norm.json';norm.write_text('{}');model=tmp_path/'model.zip'
|
|
out=tmp_path/'output';calls=[]
|
|
monkeypatch.setattr(m.sys,'argv',['run.py','--model',str(model),'--norm',str(norm),
|
|
'--out',str(out),'--mass-ratio','4','--release-dir',str(original)])
|
|
monkeypatch.setattr(m,'sha',lambda path:m.MODEL_SHA if path==model else m.NORM_SHA if path==norm else 'fixture-sha')
|
|
monkeypatch.setattr(m,'load_policy',lambda path:object())
|
|
monkeypatch.setattr(m.subprocess,'check_output',lambda *a,**k:'candidate\n')
|
|
monkeypatch.setattr(m,'precursor',lambda *a:pytest.fail('saved release must skip precursor'))
|
|
monkeypatch.setattr(m,'run_arm',lambda *a,**k:calls.append((a,k)))
|
|
m.main()
|
|
assert [a[0] for a,k in calls]==['zero','controlled']
|
|
assert all(a[2:5]==(2.,3200,400) and k=={'mass_ratio':4.} for a,k in calls)
|
|
receipt=json.loads((out/'case.json').read_text())
|
|
assert receipt['mass_ratio']==4 and receipt['structural_damping']==0 and receipt['warmup_executed']==0
|
|
assert receipt['exploratory'] and receipt['response_tU_D']==160
|
|
assert (out/'target.npy').read_bytes()==(original/'target.npy').read_bytes()
|
|
assert (out/'config.json').read_bytes()==(original/'config.json').read_bytes()
|