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