File size: 4,145 Bytes
f8a1702 | 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 | """Deterministic contract tests; real audio validation is documented separately."""
from types import SimpleNamespace
from pathlib import Path
import numpy as np
import pytest
import torch
from streampa.pipeline import StreamPASession, load_scorer
from streampa import frontend as fe
class ScheduleSession(StreamPASession):
def _step(self,commit_samples,end_samples,is_final):
row=dict(commit=commit_samples,end=end_samples,final=is_final,samples=self.audio[:end_samples].copy())
self.last_result=row
return row
def session():
return ScheduleSession(SimpleNamespace(assets={'max_audio_sec':30,'chunk_sec':.64,'right_context_sec':.16}))
def test_received_boundary_and_short_eof():
s=session()
assert s.append(np.zeros(12799,np.float32))==[]
r=s.append(np.ones(1,np.float32))[0]
assert r['commit']==10240 and r['end']==12800 and not r['final']
s.append(np.ones(100,np.float32))
r=s.finish()
assert r['end']==12900 and r['commit']==12900 and r['final']
def test_reset_and_new_session_do_not_share_state():
a=session(); b=session()
a.state=torch.ones(1,64); a.commit_state={'committed_slots':[0]}
a.append(np.ones(100)); a.reset()
assert a.state is None and a.commit_state is None and len(a.audio)==0
assert b.state is None and len(b.audio)==0
def test_suffix_is_not_in_current_input():
prefix=np.arange(12800,dtype=np.float32)
a=session(); b=session()
x=a.append(np.concatenate([prefix,np.ones(10000)]))[0]
y=b.append(np.concatenate([prefix,-np.ones(10000)]))[0]
assert np.array_equal(x['samples'],y['samples'])
def test_closed_empty_and_invalid_input():
s=session(); assert s.finish()['end']==0
with pytest.raises(RuntimeError): s.append(np.zeros(1))
s.reset()
with pytest.raises(ValueError): s.append(np.array([np.nan]))
with pytest.raises(ValueError): s.append(np.zeros((2,2)))
with pytest.raises(ValueError): s.append(np.zeros(480001))
assert len(s.audio)==0
def test_normalization_and_gap_mass():
assert fe.normalize_phone('AH0')=='AH'
assert fe.normalize_word('Hello!')=='HELLO'
post=np.array([[.6,.1,.3],[.2,.6,.2]],np.float32)
pcn={'cn_post':post,'eps_index':2}
fe.validate_pcn(pcn)
stats,_=fe.pcn_stats(post,[0,1],[],2)
assert np.allclose(stats[:,0],[.3,.2])
with pytest.raises(ValueError): fe.validate_pcn({'cn_post':post*0,'eps_index':2})
def test_partial_commit_and_legacy_persistence():
pcn={'top_phone_ids':[1,2],'slot_times':[(0,.3),(.3,.8)]}
ids=np.array([0,1],np.int32)
times=[{'word':'A','start':0,'end':.3},{'word':'B','start':.3,'end':.8}]
cum,new,_,_,state=fe.build_stateful_commit_masks(None,pcn,ids,times,.64,.8,False,commit_rule='legacy')
assert new.tolist()==[1.,0.] and cum.tolist()==[1.,0.]
cum,new,_,_,state=fe.build_stateful_commit_masks(state,pcn,ids,times,.8,.8,True,commit_rule='legacy')
assert new.tolist()==[0.,1.] and cum.tolist()==[1.,1.]
_,new,_,_,_=fe.build_stateful_commit_masks(state,pcn,ids,times,1,1,True,commit_rule='legacy')
assert new.sum()==0
def test_no_new_words_keeps_state_but_visible_path_remains_active():
torch.set_num_threads(2); torch.manual_seed(1337)
model,c=load_scorer(Path(__file__).resolve().parents[1])
n=c['seq_len']; d=c['phone_dim']
batch=dict(cn_post=torch.softmax(torch.randn(1,n,d),-1),cn_stats=torch.randn(1,n,5),
acoustic_post=torch.softmax(torch.randn(1,n,d),-1),acoustic_stats=torch.randn(1,n,4),
prosody=torch.randn(1,c['prosody_dim']),slot_prosody=torch.randn(1,n,c['slot_prosody_dim']),
visible_len=torch.tensor([4]),cumulative_commit_mask=torch.zeros(1,n),new_commit_mask=torch.zeros(1,n),
word_ids=torch.arange(n)[None])
state=torch.randn(1,1,64)
with torch.inference_mode():
first=model(**batch,prev_state=state)
batch['cn_stats'][:,:4]+=2
second=model(**batch,prev_state=state)
assert torch.equal(first['next_state'],state) and torch.equal(second['next_state'],state)
assert first['new_word_mask'].sum()==0
assert not torch.equal(first['utt_scores'],second['utt_scores'])
|