附录｜TTS 机制实验：代码地图、运行入口与完整源码

"""CPU teaching lab: mechanisms, not pretrained TTS or an audio-quality benchmark."""
from pathlib import Path
import json
import torch
import torch.nn.functional as F

def length_regulate(hidden, durations):
    """One unpadded item: [N,H], integer [N] -> [sum(d),H]."""
    if durations.dtype != torch.long or (durations < 0).any():
        raise ValueError('durations must be non-negative int64')
    if len(hidden) != len(durations): raise ValueError('length mismatch')
    return torch.repeat_interleave(hidden, durations, dim=0)

def monotonic_path(score):
    """[frames,phones]; stay or advance by one; every phone gets >=1 frame."""
    frames, phones = score.shape
    if frames < phones: raise ValueError('not enough frames')
    dp = torch.full_like(score, -torch.inf)
    move = torch.zeros_like(score, dtype=torch.bool)
    dp[0,0] = score[0,0]
    for t in range(1, frames):
        for n in range(min(t+1, phones)):
            stay = dp[t-1,n]
            advance = dp[t-1,n-1] if n else score.new_tensor(-torch.inf)
            move[t,n] = advance > stay
            dp[t,n] = score[t,n] + torch.maximum(stay,advance)
    n = phones-1
    path = torch.zeros_like(score)
    for t in range(frames-1,-1,-1):
        path[t,n] = 1
        n -= int(move[t,n])
    return path, dp[-1,-1]

def residual_quantize(z, codebooks):
    """[T,D], [K,V,D] -> IDs [T,K], reconstructed [T,D], residual [T,D]."""
    residual = z.clone()
    reconstruction = torch.zeros_like(z)
    ids = []
    for book in codebooks:
        distance = (residual[:,None,:] - book[None,:,:]).square().sum(-1)
        index = distance.argmin(-1)
        selected = book[index]
        ids.append(index)
        reconstruction += selected
        residual -= selected
    return torch.stack(ids,dim=-1), reconstruction, residual

def prefix_mask(text_length, audio_length):
    """True means BLOCKED. Text bidirectional; audio causal over all text."""
    n, t = text_length, audio_length
    mask = torch.zeros(n+t,n+t,dtype=torch.bool)
    mask[:n,n:] = True
    mask[n:,n:] = torch.ones(t,t,dtype=torch.bool).triu(1)
    return mask

def flow_target(noise, target, time):
    return (1-time)*noise + time*target, target-noise

def masked_mse(prediction, target, generated_mask):
    """Mask [B,T] includes generated valid frames only; tensor [B,T,D]."""
    weight = generated_mask.unsqueeze(-1).expand_as(prediction)
    if not weight.any(): raise ValueError('empty generated region')
    return (prediction-target).square()[weight].mean()

def euler(initial, velocity, steps):
    if steps < 1: raise ValueError('steps must be positive')
    state = initial.clone()
    for j in range(steps): state = state + velocity(state,j/steps)/steps
    return state

def period_view(waveform, period):
    """[B,1,L] -> [B,1,ceil(L/p),p]; constant pad is teaching-only."""
    if period < 1: raise ValueError('period must be positive')
    pad = (-waveform.shape[-1]) % period
    padded = F.pad(waveform,(0,pad))
    return padded.reshape(*padded.shape[:-1],-1,period)

def delay_pattern(tokens):
    """[T,K] -> [T+K-1,K]; -1 denotes BOS/EOS padding in this demo."""
    t,k=tokens.shape
    grid=torch.full((t+k-1,k),-1,dtype=tokens.dtype)
    for j in range(k):grid[j:j+t,j]=tokens[:,j]
    return grid

def run_checks():
    torch.manual_seed(2026)
    out={'scope':'Teaching tensor invariants; no speech training or MOS measurement.',
         'torch':torch.__version__,'device':'cpu'}
    h=torch.tensor([[1.,10.],[2.,20.],[3.,30.]],requires_grad=True)
    expanded=length_regulate(h,torch.tensor([2,1,3]))
    expanded.sum().backward()
    assert expanded.shape==(6,2)
    assert h.grad[:,0].tolist()==[2,1,3]
    try:
        length_regulate(h,torch.tensor([2,-1,3]))
    except ValueError:
        pass
    else:
        raise AssertionError('negative duration was accepted')
    out['duration']={'expanded':expanded.detach().tolist(),'gradient':h.grad[:,0].tolist()}
    scores=torch.tensor([[3.,-4.,-9.],[2.,1.,-8.],[-4.,3.,-1.],[-6.,1.,4.],[-7.,0.,5.]])
    path,best=monotonic_path(scores)
    assert path.argmax(-1).tolist()==[0,0,1,2,2]
    assert path.sum(1).eq(1).all() and path.sum(0).gt(0).all()
    assert best.item()==17
    # Cross-check dynamic programming against exhaustive legal duration compositions.
    brute=[]
    for a in range(1,4):
        for b in range(1,5-a):
            c=5-a-b
            if c>0:brute.append(scores[range(5),[0]*a+[1]*b+[2]*c].sum().item())
    assert best.item()==max(brute)
    out['mas']={'path':path.argmax(-1).tolist(),'durations':path.sum(0).tolist(),'score':best.item()}
    z=torch.tensor([[2.2,1.1]])
    books=torch.tensor([[[0.,0.],[2.,0.],[0.,2.]],[[0.,0.],[.25,1.],[1.,.5]]])
    ids,q,r=residual_quantize(z,books)
    assert ids.tolist()==[[1,1]] and torch.allclose(q+r,z)
    out['rvq']={'ids':ids.tolist(),'reconstructed':q.tolist(),'residual':r.tolist()}
    mask=prefix_mask(3,4)
    qk=torch.randn(7,7)
    weights=qk.masked_fill(mask,-torch.inf).softmax(-1)
    assert weights[mask].eq(0).all()
    values=torch.randn(7,3); altered=values.clone();altered[6]+=100
    assert torch.equal((weights@values)[:6],(weights@altered)[:6])
    out['prefix']={'shape':list(mask.shape),'blocked_probability':float(weights[mask].sum())}
    x0=torch.tensor([0.,2.]);x1=torch.tensor([4.,0.])
    xt,velocity=flow_target(x0,x1,.25)
    final=euler(x0,lambda x,t:velocity,4)
    assert torch.allclose(final,x1)
    pred=torch.zeros(1,4,2,requires_grad=True); target=torch.ones_like(pred)
    loss=masked_mse(pred,target,torch.tensor([[False,True,True,False]]));loss.backward()
    assert pred.grad[:,[0,3]].eq(0).all() and pred.grad[:,[1,2]].ne(0).all()
    out['flow']={'at_quarter':xt.tolist(),'velocity':velocity.tolist(),'euler_final':final.tolist(),'masked_mse':loss.item()}
    folded=period_view(torch.arange(10.).reshape(1,1,10),3)
    assert folded.shape==(1,1,4,3)
    out['period']={'rows':folded[0,0].tolist()}
    tokens=torch.arange(12).reshape(4,3);grid=delay_pattern(tokens)
    assert all(torch.equal(grid[j:j+4,j],tokens[:,j]) for j in range(3))
    out['delay']={'grid':grid.tolist(),'steps':len(grid)}
    return out

if __name__=='__main__':
    result=run_checks()
    (Path(__file__).parent/'results').mkdir(exist_ok=True)
    (Path(__file__).parent/'results'/'mechanism-checks.json').write_text(json.dumps(result,indent=2),encoding='utf8')
    print(json.dumps(result,indent=2))