File size: 6,004 Bytes
10ef792 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 | """Text-specification conformance probes, not an independent blind replication.
Small deterministic arrays are software fixtures, not HAR evidence. References
below are written from the manuscript's stated operations and compared with the
frozen implementation; standard classifiers are not reimplemented.
"""
from pathlib import Path
import importlib.util
import itertools
import sys
import numpy as np
import pytest
ROOT = Path(__file__).resolve().parents[1]
CPU = ROOT/'src/wisp/cpu'
def load(name, file):
spec = importlib.util.spec_from_file_location(name, CPU/file)
mod = importlib.util.module_from_spec(spec)
sys.modules[name] = mod
spec.loader.exec_module(mod)
return mod
F=load('r7_frozen_representations','features.py')
H=load('r7_frozen_hmm','hmm.py')
def zscore(x):
x=np.asarray(x,dtype=float)
mu=np.nanmean(x,axis=1,keepdims=True)
sd=np.nanstd(x,axis=1,keepdims=True)
return np.nan_to_num((x-mu)/np.where(sd<1e-6,1,sd),nan=0.,posinf=0.,neginf=0.)
@pytest.mark.parametrize('seed',[42,43,44])
def test_window_preparation(seed):
x=np.random.default_rng(seed).normal(size=(4,35,3)); x[0,:,0]=2; x[1,3,1]=np.nan
np.testing.assert_allclose(zscore(x),F._safe_zscore_per_window(x),rtol=0,atol=0)
@pytest.mark.parametrize('bins',[4,6,8])
def test_relative_states_ties_and_flattening(bins):
x=np.array([[[0.],[0.],[1.],[2.],[1.],[0.]],[[3.],[3.],[3.],[3.],[3.],[3.]]])
expected=[]
for window in x:
v=window[:,0]; q=np.quantile(v,np.linspace(0,1,bins+1)[1:-1]); a=(v[:,None]>q).sum(1)
hist=np.bincount(a,minlength=bins)/len(a); counts=np.zeros((bins,bins))
for before,after in zip(a[:-1],a[1:]): counts[before,after]+=1
row=counts.sum(1,keepdims=True); Q=counts/np.where(row==0,1,row)
ent=-(Q*np.log(Q+1e-12)).sum()
expected.append(np.r_[hist,Q.ravel(),ent])
np.testing.assert_allclose(expected,F._transition_features(x,bins),atol=1e-14)
@pytest.mark.parametrize('seed',[42,43,44])
def test_random_convolution_complete_sampler(seed):
x=np.random.default_rng(9).normal(size=(3,39,3));xn=zscore(x);rng=np.random.default_rng(seed); bank=[];cols=[]
for _ in range(5):
L=int(rng.choice([7,9,11,15]));d=int(rng.choice([1,2,4]));
if (L-1)*d+1>39:d=max(1,38//(L-1))
nc=int(rng.integers(1,4)); channels=np.sort(rng.choice(3,size=nc,replace=False))
w=rng.normal(0,1,size=(L,nc));w-=w.mean(0,keepdims=True);w/=np.linalg.norm(w)+1e-12;b=float(rng.normal(0,.25))
out=[]
for i in range(len(x)):
responses=[b+np.sum(xn[i,t+np.arange(L)*d][:,channels]*w) for t in range(39-(L-1)*d)]
responses=np.array(responses);out.append([(responses>0).mean(),responses.max(),responses.mean(),responses.std()])
cols.append(np.asarray(out));bank.append((L,d,channels,w,b))
impl=F.RandomConvSketch(n_kernels=5,random_state=seed).fit(x)
for row,target in zip(impl.kernels_,bank):
assert row['length']==target[0] and row['dilation']==target[1]
np.testing.assert_equal(row['channels'],target[2]);np.testing.assert_allclose(row['weights'],target[3]);assert row['bias']==target[4]
np.testing.assert_allclose(impl.transform(x),np.concatenate(cols,axis=1),atol=1e-12)
@pytest.mark.parametrize('T',[31,64])
def test_interval_endpoints_and_summary_order(T):
x=np.random.default_rng(8).normal(size=(2,T,3));rng=np.random.default_rng(71);intervals=[]
for parts in [2,4,8]:
width=max(2,int(np.ceil(T/parts)))
intervals.extend((start,min(T,start+width)) for start in range(0,T,width) if min(T,start+width)-start>=2)
for _ in range(7):
lo=int(rng.integers(0,max(T-2,1)));length=int(rng.integers(2,max(3,T-lo)+1));hi=min(T,lo+length)
if hi-lo>=2: intervals.append((lo,hi))
intervals=list(dict.fromkeys(intervals));impl=F.IntervalDistributionSketch(n_random_intervals=7,random_state=71).fit(x)
assert intervals==impl.intervals_
xn=zscore(x);outputs=[]
for lo,hi in intervals:
v=xn[:,lo:hi];t=np.arange(hi-lo,dtype=float);t-=t.mean()
slope=np.sum((v-v.mean(1,keepdims=True))*t[None,:,None],axis=1)/(t@t or 1.)
q=np.quantile(v,[.1,.25,.5,.75,.9],axis=1).transpose(1,2,0).reshape(len(v),-1)
outputs.append(np.concatenate([v.mean(1),v.std(1),np.ptp(v,axis=1),(v*v).mean(1),slope,q],axis=1))
np.testing.assert_allclose(impl.transform(x),np.concatenate(outputs,axis=1),atol=1e-12)
def test_shapelet_match_position_denominator():
x=np.array([[[1.],[2.],[4.],[1.],[5.],[9.],[2.],[1.]]]);s=F.RandomShapeletSketch(n_shapelets=1,zscore=False)
prototype=np.array([1.,5.,9.]);prototype=(prototype-prototype.mean())/prototype.std()
s.shapelets_=[{'values':prototype,'channel':0,'dilation':1}]
y=s.transform(x);assert y[0,0]<1e-12;assert y[0,1]==pytest.approx(3/5)
def test_hmm_counts_and_group_boundaries():
y=np.array([1,0,1,2,2,1]);groups=np.array(['a','a','a','b','b','b']);order=np.array([2,0,1,1,0,2]);K=4
A=np.ones((K,K));np.fill_diagonal(A,3)
for g in np.unique(groups):
ids=np.where(groups==g)[0];ids=ids[np.argsort(order[ids],kind='stable')]
for a,b in zip(y[ids[:-1]],y[ids[1:]]):A[a,b]+=1
pi=np.bincount(y,minlength=K)+1
s=H.TransitionSmoother().fit(y,groups=groups,time_index=order,n_classes=K)
np.testing.assert_allclose(s.transition_,A/A.sum(1,keepdims=True));np.testing.assert_allclose(s.prior_,pi/pi.sum())
def test_viterbi_matches_exhaustive_path():
s=H.TransitionSmoother().fit(np.array([0,0,1,1,0]),groups=np.zeros(5),n_classes=2)
P=np.array([[.9,.1],[.4,.6],[.1,.9],[.7,.3]])
paths=list(itertools.product(range(2),repeat=4));scores=[]
for path in paths:
score=np.log(s.prior_[path[0]])+sum(np.log(P[t,k]) for t,k in enumerate(path))
score+=sum(np.log(s.transition_[a,b]) for a,b in zip(path[:-1],path[1:]));scores.append(score)
np.testing.assert_equal(s.predict_from_proba(P,groups=np.zeros(4)),paths[np.argmax(scores)])
|