89 lines
3.9 KiB
Python
89 lines
3.9 KiB
Python
"""Predeclared synthetic measurement tests; no flow outcome is used for tuning."""
|
|
from pathlib import Path
|
|
import sys
|
|
import unittest
|
|
import numpy as np
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
import periodic
|
|
import periodic_v2 as v2
|
|
|
|
|
|
def signal(periods=None, noise=0., offset=0., amplitude=1., irregular=False, harmonic=0.):
|
|
periods = np.full(20, 3.3) if periods is None else np.asarray(periods)
|
|
boundaries = 250.37 + np.r_[0., np.cumsum(periods)]
|
|
t = np.arange(boundaries[0]-.1, boundaries[-1]+.1, .02)
|
|
if irregular: t = t[np.arange(len(t)) % 7 != 0]
|
|
phase = np.interp(t, boundaries, np.arange(len(boundaries))) * 2*np.pi
|
|
lift = amplitude*(np.sin(phase) + harmonic*np.sin(2*phase)) + offset
|
|
# Fixed seed and fixed peak-local mask; not fitted to any flow dataset.
|
|
lift += noise*np.random.default_rng(1729).uniform(-1,1,len(t))*(abs(np.sin(phase))>.9)
|
|
return [dict(tstar=x+.01, tstar_force=x, cl=l,
|
|
cd=3.2+.03*np.cos(2*p), dp=2.45+.05*np.cos(p),
|
|
mass=1., flux_in=1., flux_out=1.) for x,l,p in zip(t,lift,phase)]
|
|
|
|
|
|
def test_constant_period_peak_noise_and_mean_offset():
|
|
for noise, offset in [(0.,0.),(.0001,0.),(.001,0.),(.001,.08)]:
|
|
result=v2.assess(signal(noise=noise,offset=offset))
|
|
assert result['periodic'], result['checks']
|
|
periods=np.array([c['period'] for c in result['cycles'][-5:]])
|
|
assert max(abs(periods-3.3))<1e-5
|
|
|
|
|
|
def test_noncommensurate_asymmetric_waveform():
|
|
result=v2.assess(signal(periods=[3.313]*20,harmonic=.1,offset=.08,noise=.001))
|
|
assert result['periodic'], result['checks']
|
|
periods=np.array([c['period'] for c in result['cycles'][-5:]])
|
|
assert max(abs(periods-3.313))<5e-5
|
|
|
|
|
|
def test_true_drift_is_rejected():
|
|
result=v2.assess(signal(periods=np.arange(20)*.01+3.2))
|
|
assert not result['periodic'] and not result['checks']['period']
|
|
measured=np.array([c['period'] for c in result['cycles'][-5:]])
|
|
assert np.ptp(measured)/np.mean(measured)>.005
|
|
|
|
|
|
def test_insufficient_cycles_and_amplitude():
|
|
assert not v2.assess(signal(periods=[3.3]*4))['periodic']
|
|
result=v2.assess(signal(amplitude=.05))
|
|
assert not result['periodic'] and not result['checks']['shedding']
|
|
|
|
|
|
def test_exact_zero_plateau_and_crossing_noise():
|
|
assert v2.ascending_crossings([0,1,2,3,4,5],[-1,0,0,1,-1,0])==[1.,5.]
|
|
# All crossings retained, even a short sign reversal: no hidden debounce.
|
|
assert len(v2.ascending_crossings([0,1,2,3,4,5],[-1,.001,-.001,.001,1,2]))==2
|
|
assert v2.ascending_crossings([0,1,2],[1,.1,2])==[]
|
|
|
|
|
|
def test_extra_and_missing_cycles_not_reconnected():
|
|
for last in [[3.3,.2,3.1,3.3],[3.3,6.6,3.3,3.3]]:
|
|
result=v2.assess(signal(periods=[3.3]*12+last))
|
|
assert not result['periodic'] and not result['checks']['consecutive']
|
|
|
|
|
|
def test_invalid_inputs_and_irregular_sampling():
|
|
for t,y in [([0,0],[-1,1]),([1,0],[-1,1]),([0,1],[-1,float('nan')])]:
|
|
with unittest.TestCase().assertRaises(ValueError):v2.ascending_crossings(t,y)
|
|
rows=signal();rows[1]['dp']=float('inf')
|
|
with unittest.TestCase().assertRaises(ValueError):v2.assess(rows)
|
|
assert v2.assess(signal(irregular=True))['periodic']
|
|
|
|
|
|
def test_coverage_and_unchanged_gates():
|
|
rows=signal()
|
|
cycles=v2.extract(rows,min_time=250.)
|
|
# Move end-sample coverage late; force remains unchanged. Partial cycles excluded.
|
|
delayed=[dict(r,tstar=r['tstar']+10.) for r in rows]
|
|
other=v2.extract(delayed,min_time=250.)
|
|
assert len(other)<len(cycles) and all(c['start']>=delayed[0]['tstar'] for c in other)
|
|
assert v2.GATES is periodic.GATES and v2.REFERENCE_GATES is periodic.REFERENCE_GATES
|
|
assert v2.compare is periodic.compare
|
|
|
|
|
|
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())
|