tests/test_method_conformance.py

"""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)])