Buckets:
| """Extract F0 (pyin) + RMS from WAVs, tokenize text, compute durations.""" | |
| import argparse | |
| import json | |
| import os | |
| import numpy as np | |
| import torch | |
| import librosa | |
| # Character vocabulary: PAD=0, UNK=1, space=2, a-z=3-28, 0-9=29-38, punct 39+ | |
| VOCAB = {chr(0): 0} # PAD | |
| VOCAB['<UNK>'] = 1 | |
| VOCAB[' '] = 2 | |
| for i, c in enumerate('abcdefghijklmnopqrstuvwxyz'): | |
| VOCAB[c] = 3 + i | |
| for i, c in enumerate('0123456789'): | |
| VOCAB[c] = 29 + i | |
| PUNCT = ".,;:!?'-\"()/" | |
| for i, c in enumerate(PUNCT): | |
| VOCAB[c] = 39 + i | |
| VOCAB_SIZE = max(VOCAB.values()) + 1 | |
| def tokenize(text): | |
| """Lowercase text → char ID list.""" | |
| text = text.lower() | |
| return [VOCAB.get(c, VOCAB['<UNK>']) for c in text] | |
| def proportional_durations(n_chars, n_frames): | |
| """Split n_frames proportionally across n_chars.""" | |
| if n_chars == 0: | |
| return [] | |
| base = n_frames // n_chars | |
| remainder = n_frames % n_chars | |
| durations = [base + (1 if i < remainder else 0) for i in range(n_chars)] | |
| return durations | |
| def extract_one(wav_path, text, sr=24000, hop_length=2400): | |
| """Extract features for a single sample.""" | |
| y, _ = librosa.load(wav_path, sr=sr, mono=True) | |
| # F0 via pyin | |
| f0, voiced_flag, _ = librosa.pyin( | |
| y, fmin=50, fmax=600, sr=sr, hop_length=hop_length | |
| ) | |
| # Clip to 300 Hz to suppress octave jumps; mark >300 as unvoiced | |
| too_high = ~np.isnan(f0) & (f0 > 300) | |
| voiced_flag[too_high] = False | |
| f0 = np.where(np.isnan(f0), 0.0, np.clip(f0, 50, 300)) | |
| # RMS | |
| rms = librosa.feature.rms(y=y, hop_length=hop_length, frame_length=hop_length)[0] | |
| # Align lengths | |
| n_frames = min(len(f0), len(rms)) | |
| f0 = f0[:n_frames] | |
| voiced_flag = voiced_flag[:n_frames] | |
| rms = rms[:n_frames] | |
| # Log space | |
| voiced_mask = voiced_flag.astype(bool) | |
| log_f0 = np.zeros_like(f0) | |
| log_f0[voiced_mask] = np.log(f0[voiced_mask]) | |
| log_rms = np.log(rms + 1e-8) | |
| # Tokenize | |
| char_ids = tokenize(text) | |
| # Duration alignment | |
| durations = proportional_durations(len(char_ids), n_frames) | |
| return { | |
| 'char_ids': np.array(char_ids, dtype=np.int64), | |
| 'durations': np.array(durations, dtype=np.int64), | |
| 'log_f0': log_f0.astype(np.float32), | |
| 'log_rms': log_rms.astype(np.float32), | |
| 'voiced_mask': voiced_mask, | |
| 'n_frames': n_frames, | |
| 'text': text, | |
| } | |
| def main(): | |
| parser = argparse.ArgumentParser() | |
| parser.add_argument('--audio_dir', required=True) | |
| parser.add_argument('--transcripts', required=True) | |
| parser.add_argument('--output', default='features.pt') | |
| args = parser.parse_args() | |
| with open(args.transcripts) as f: | |
| transcripts = json.load(f) | |
| # Filter to sample_*.wav keys only | |
| keys = sorted([k for k in transcripts if k.startswith('sample_') and k.endswith('.wav')]) | |
| print(f"Processing {len(keys)} samples...") | |
| samples = [] | |
| all_voiced_f0 = [] | |
| all_log_rms = [] | |
| for i, key in enumerate(keys): | |
| wav_path = os.path.join(args.audio_dir, key) | |
| if not os.path.exists(wav_path): | |
| print(f" SKIP {key}: file not found") | |
| continue | |
| feat = extract_one(wav_path, transcripts[key]) | |
| samples.append(feat) | |
| # Collect for normalization | |
| if feat['voiced_mask'].any(): | |
| all_voiced_f0.append(feat['log_f0'][feat['voiced_mask']]) | |
| all_log_rms.append(feat['log_rms']) | |
| if (i + 1) % 200 == 0: | |
| print(f" {i+1}/{len(keys)} done") | |
| # Global normalization stats | |
| all_voiced_f0 = np.concatenate(all_voiced_f0) | |
| all_log_rms = np.concatenate(all_log_rms) | |
| norm_stats = { | |
| 'f0_mean': float(np.mean(all_voiced_f0)), | |
| 'f0_std': float(np.std(all_voiced_f0)), | |
| 'rms_mean': float(np.mean(all_log_rms)), | |
| 'rms_std': float(np.std(all_log_rms)), | |
| } | |
| print(f"Norm stats: {norm_stats}") | |
| # Z-score normalize | |
| for s in samples: | |
| voiced = s['voiced_mask'] | |
| s['log_f0'][voiced] = (s['log_f0'][voiced] - norm_stats['f0_mean']) / norm_stats['f0_std'] | |
| s['log_rms'] = (s['log_rms'] - norm_stats['rms_mean']) / norm_stats['rms_std'] | |
| # Print stats | |
| voiced_ratios = [s['voiced_mask'].mean() for s in samples] | |
| frame_counts = [s['n_frames'] for s in samples] | |
| print(f"Samples: {len(samples)}") | |
| print(f"Voiced ratio: {np.mean(voiced_ratios):.3f} ± {np.std(voiced_ratios):.3f}") | |
| print(f"Frame counts: {np.mean(frame_counts):.1f} ± {np.std(frame_counts):.1f} " | |
| f"(min={np.min(frame_counts)}, max={np.max(frame_counts)})") | |
| # Convert to tensors for saving | |
| for s in samples: | |
| s['char_ids'] = torch.from_numpy(s['char_ids']) | |
| s['durations'] = torch.from_numpy(s['durations']) | |
| s['log_f0'] = torch.from_numpy(s['log_f0']) | |
| s['log_rms'] = torch.from_numpy(s['log_rms']) | |
| s['voiced_mask'] = torch.from_numpy(s['voiced_mask']) | |
| torch.save({'samples': samples, 'norm_stats': norm_stats, 'vocab_size': VOCAB_SIZE}, args.output) | |
| print(f"Saved to {args.output}") | |
| if __name__ == '__main__': | |
| main() | |
Xet Storage Details
- Size:
- 5.08 kB
- Xet hash:
- 77fe1d2bd675c956c43e0f6de03ce438554adf711cc2b568e56060b3637dbab2
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.