"""第 43 章：MMS-TTS 合成与 Whisper 识别；默认只核对随书记录。"""

from argparse import ArgumentParser
from hashlib import sha256
import json
from pathlib import Path
import platform
import re
from time import perf_counter
import wave


MODELS = {
    "tts": ("facebook/mms-tts-eng", "c71de0fe7204c83f1c10820a7d696d0b450048ba"),
    "asr": ("openai/whisper-tiny.en", "87c7102498dcde7456f24cfd30239ca606ed9063"),
}
TEXTS = [
    "A blue cup is on the table.",
    "The workshop opens at nine and closes at six.",
    "Please pick up order number three seven two.",
    "There are two red circles and one blue square.",
]
CONDITIONS = [("clean", None), ("snr20", 20), ("snr0", 0)]
DEFAULT = Path(__file__).parent / "data/speech/report.json"


def file_hash(path):
    h = sha256()
    with Path(path).open("rb") as stream:
        for chunk in iter(lambda: stream.read(1024 * 1024), b""):
            h.update(chunk)
    return h.hexdigest()


def words(text):
    # 明确的教学规范化：小写，字母/数字连续串；不改写数字、同义词。
    return re.findall(r"[a-z0-9]+", text.lower())


def edit_counts(reference, hypothesis):
    a, b = words(reference), words(hypothesis)
    cost = [[0] * (len(b) + 1) for _ in range(len(a) + 1)]
    for i in range(len(a) + 1):
        cost[i][0] = i
    for j in range(len(b) + 1):
        cost[0][j] = j
    for i in range(1, len(a) + 1):
        for j in range(1, len(b) + 1):
            cost[i][j] = min(cost[i-1][j-1] + (a[i-1] != b[j-1]),
                             cost[i-1][j] + 1, cost[i][j-1] + 1)
    i, j = len(a), len(b)
    counts = {"substitutions": 0, "deletions": 0, "insertions": 0, "reference_words": i}
    while i or j:
        if i and j and cost[i][j] == cost[i-1][j-1] + (a[i-1] != b[j-1]):
            counts["substitutions"] += int(a[i-1] != b[j-1])
            i, j = i - 1, j - 1
        elif i and cost[i][j] == cost[i-1][j] + 1:
            counts["deletions"] += 1
            i -= 1
        else:
            counts["insertions"] += 1
            j -= 1
    counts["errors"] = cost[-1][-1]
    counts["wer"] = counts["errors"] / len(a) if a else None
    return counts


def write_wav(path, values, sample_rate=16000):
    import numpy as np
    peak = float(np.abs(values).max(initial=0))
    gain = min(1.0, 0.95 / peak) if peak else 1.0
    pcm = np.rint(values * gain * 32767).astype("<i2")
    with wave.open(str(path), "wb") as output:
        output.setnchannels(1)
        output.setsampwidth(2)
        output.setframerate(sample_rate)
        output.writeframes(pcm.tobytes())
    return gain


def read_wav(path):
    import numpy as np
    with wave.open(str(path), "rb") as audio:
        assert audio.getnchannels() == 1 and audio.getsampwidth() == 2
        return np.frombuffer(audio.readframes(audio.getnframes()), dtype="<i2").astype("float32") / 32767, audio.getframerate()


def add_noise(clean, snr_db, seed):
    import numpy as np
    generator = np.random.default_rng(seed)
    noise = generator.normal(size=len(clean))
    scale = np.sqrt(np.mean(clean ** 2) / (10 ** (snr_db / 10) * np.mean(noise ** 2)))
    return noise * scale


def validate(path=DEFAULT):
    report = json.loads(path.read_text())
    assert report["source_sha256"] == file_hash(__file__), "脚本与记录版本不同"
    assert report["models"] == {key: list(value) for key, value in MODELS.items()}
    assert report["texts"] == TEXTS and len(report["tts"]) == 4
    assert [row["id"] for row in report["asr"]] == [
        f"{i}-{name}" for i in range(4) for name, _ in CONDITIONS] + ["silence"]
    for row in report["asr"]:
        filename = path.parent / row["audio"]
        assert file_hash(filename) == row["audio_sha256"]
        with wave.open(str(filename), "rb") as audio:
            assert audio.getframerate() == 16000
            assert audio.getnchannels() == 1 and audio.getsampwidth() == 2
            assert audio.getnframes() == row["samples"]
        assert row["score"] == edit_counts(row["reference"], row["text"])
        assert row["eos"] == (row["token_ids"][-1] == report["eos_token_id"])
        assert row["feature_shape"] == [1, 80, 3000]
        assert row["seconds"] > 0
    return report


def run(folder):
    import numpy as np
    import torch
    import transformers
    from huggingface_hub import snapshot_download
    from transformers import AutoTokenizer, VitsModel, WhisperProcessor, WhisperForConditionalGeneration

    if folder.exists() and any(folder.iterdir()):
        raise ValueError("请使用空输出目录，保留已有实验结果")
    folder.mkdir(parents=True, exist_ok=True)
    torch.set_num_threads(4)
    torch.use_deterministic_algorithms(True)
    paths = {name: Path(snapshot_download(
        model, revision=revision,
        allow_patterns=["*.json", "*.txt", "model.safetensors", "README.md"]))
             for name, (model, revision) in MODELS.items()}
    tokenizer = AutoTokenizer.from_pretrained(paths["tts"], local_files_only=True)
    tts = VitsModel.from_pretrained(paths["tts"], torch_dtype=torch.float32,
                                    local_files_only=True).eval()
    sample_rate = tts.config.sampling_rate
    assert sample_rate == 16000
    report = {
        "models": {key: list(value) for key, value in MODELS.items()}, "texts": TEXTS,
        "source_sha256": file_hash(__file__), "tts": [], "asr": [],
        "environment": {"python": platform.python_version(), "platform": platform.platform(),
                        "torch": torch.__version__, "numpy": np.__version__,
                        "transformers": transformers.__version__,
                        "device": "cpu", "dtype": "float32", "threads": 4},
        "config": {"sample_rate": 16000, "max_new_tokens": 64, "do_sample": False,
                   "num_beams": 1, "return_timestamps": False,
                   "return_dict_in_generate": True, "vad": False,
                   "tts_seed_start": 2026, "noise_seed_start": 3030},
        "weights": {key: {"bytes": (path / "model.safetensors").stat().st_size,
                          "sha256": file_hash(path / "model.safetensors")}
                    for key, path in paths.items()},
        "tts_config": tts.config.to_dict(),
    }
    cases = []
    for i, text in enumerate(TEXTS):
        torch.manual_seed(2026 + i)
        inputs = tokenizer(text, return_tensors="pt")
        start = perf_counter()
        with torch.inference_mode():
            values = tts(**inputs).waveform[0].numpy()
        elapsed = perf_counter() - start
        gain = write_wav(folder / f"{i}-clean.wav", values, sample_rate)
        clean, _ = read_wav(folder / f"{i}-clean.wav")
        report["tts"].append({"id": i, "text": text, "seed": 2026 + i,
                              "seconds": elapsed, "samples": len(clean), "pcm_gain": gain,
                              "raw_sha256": sha256(values.tobytes()).hexdigest()})
        for name, snr in CONDITIONS:
            filename = f"{i}-{name}.wav"
            mix_gain = 1.0
            if snr is not None:
                noise = add_noise(clean, snr, 3030 + i)
                mix_gain = write_wav(folder / filename, clean + noise, sample_rate)
            cases.append({"id": f"{i}-{name}", "condition": name, "audio": filename,
                          "reference": text, "snr_db": snr, "mix_gain": mix_gain})
        print(f"TTS {i}：{len(clean) / sample_rate:.2f} 秒音频", flush=True)
    del tts
    write_wav(folder / "silence.wav", np.zeros(4 * sample_rate))
    cases.append({"id": "silence", "condition": "silence", "audio": "silence.wav",
                  "reference": "", "snr_db": None, "mix_gain": 1.0})
    processor = WhisperProcessor.from_pretrained(paths["asr"], local_files_only=True)
    asr = WhisperForConditionalGeneration.from_pretrained(
        paths["asr"], torch_dtype=torch.float32, attn_implementation="eager",
        local_files_only=True).eval()
    report["eos_token_id"] = asr.generation_config.eos_token_id
    report["asr_generation_config"] = asr.generation_config.to_dict()
    report["feature_config"] = processor.feature_extractor.to_dict()
    for case in cases:
        values, rate = read_wav(folder / case["audio"])
        start = perf_counter()
        inputs = processor(values, sampling_rate=rate, return_tensors="pt",
                           return_attention_mask=True)
        with torch.inference_mode():
            generated = asr.generate(**inputs, do_sample=False, num_beams=1,
                                     max_new_tokens=64, return_timestamps=False,
                                     return_dict_in_generate=True)
            # 普通返回值会移除 EOS；完整返回值保留开始/结束标记供核对。
            ids = generated.sequences[0]
        text = processor.decode(ids, skip_special_tokens=True)
        elapsed = perf_counter() - start
        row = {**case, "samples": len(values), "duration": len(values) / rate,
               "audio_sha256": file_hash(folder / case["audio"]),
               "feature_shape": list(inputs.input_features.shape),
               "token_ids": ids.tolist(), "text": text,
               "eos": bool(ids[-1] == report["eos_token_id"]),
               "seconds": elapsed, "rtf": elapsed / (len(values) / rate),
               "score": edit_counts(case["reference"], text)}
        report["asr"].append(row)
        (folder / "report.json").write_text(json.dumps(
            report, ensure_ascii=False, indent=2) + "\n")
        print(f"ASR {case['id']}：{text}", flush=True)
    validate(folder / "report.json")
    return report


def show(report):
    print("4 句合成语音；干净、20 dB、0 dB 各 4 段，另有 1 段静音。")
    for name, _ in CONDITIONS:
        rows = [r["score"] for r in report["asr"] if r["condition"] == name]
        errors = sum(r["errors"] for r in rows)
        total = sum(r["reference_words"] for r in rows)
        print(f"{name}：往返词错误 {errors}/{total}，WER={errors / total:.3f}")
    silence = report["asr"][-1]
    print(f"静音输出：{silence['text']!r}；插入词数={silence['score']['insertions']}")
    print("默认模式只核对保存记录；这不是人声基准，也没有测量主观音质。")


if __name__ == "__main__":
    parser = ArgumentParser(description=__doc__)
    parser.add_argument("--run", action="store_true")
    parser.add_argument("--out", type=Path)
    parser.add_argument("--record", type=Path, default=DEFAULT)
    args = parser.parse_args()
    if args.run and args.out is None:
        parser.error("--run 需要 --out 指定空输出目录")
    show(run(args.out) if args.run else validate(args.record))

"""第 43 章：采样、短时频谱、视频顺序和延迟队列；无模型下载。"""

from argparse import ArgumentParser
from hashlib import sha256
import json
from pathlib import Path
import platform

import numpy as np
from sklearn.linear_model import LogisticRegression


def tone_signal(rate=16000):
    t = np.arange(rate) / rate
    frequency = np.where(t < 0.5, 440.0, 880.0)
    return 0.6 * np.sin(2 * np.pi * frequency * t)


def spectrum(values, window=400, hop=160):
    frames = np.lib.stride_tricks.sliding_window_view(values, window)[::hop]
    taper = np.hanning(window)
    return np.abs(np.fft.rfft(frames * taper, axis=1)).T ** 2


def moving_video(rng):
    # 一个发亮的圆点；背景恒为零，不使用外部视频。
    start, end = rng.uniform(5, 10), rng.uniform(22, 27)
    yy = rng.uniform(8, 24)
    progress = np.linspace(0, 1, 8) ** rng.uniform(0.7, 1.5)
    cx = start + (end - start) * progress
    y, x = np.mgrid[:32, :32]
    return np.exp(-((x[None] - cx[:, None, None]) ** 2
                    + (y[None] - yy) ** 2) / (2 * 1.5 ** 2))


def centroids(frames):
    yy, xx = np.mgrid[:frames.shape[-2], :frames.shape[-1]]
    mass = frames.sum(axis=(-2, -1))
    if np.any(mass <= 0):
        raise ValueError("空白帧不能计算亮度质心")
    return np.stack([(frames * xx).sum(axis=(-2, -1)) / mass,
                     (frames * yy).sum(axis=(-2, -1)) / mass], axis=-1) / 31


def video_data(split):
    size, seed = {"train": (80, 2026), "validation": (20, 2027), "test": (40, 2028)}[split]
    rng = np.random.default_rng(seed)
    videos, labels, ids = [], [], []
    for i in range(size):
        forward = moving_video(rng)
        for direction, frames in [(1, forward), (0, forward[::-1].copy())]:
            videos.append(frames)
            labels.append(direction)
            ids.append(f"{split}-{i:03d}-{'right' if direction else 'left'}")
    pixels = np.stack(videos)
    return pixels, centroids(pixels), np.array(labels), ids


def features(points, kind):
    if kind == "mean":
        # 舍去不影响位置含义的浮点累加尾差，使正反序列得到完全相同特征。
        return np.round(points.mean(axis=1), 12)
    return np.concatenate([points[:, 0], points[:, -1]], axis=1)


def queue_schedule(process_ms, count=10, interval_ms=200):
    finished = 0
    rows = []
    for i in range(1, count + 1):
        arrived = i * interval_ms
        started = max(arrived, finished)
        finished = started + process_ms
        rows.append({"chunk": i, "arrived_ms": arrived, "started_ms": started,
                     "finished_ms": finished, "delay_ms": finished - arrived})
    return rows


def run():
    rate = 1000
    t = np.arange(20) / rate
    low = np.cos(2 * np.pi * 300 * t)
    high = np.cos(2 * np.pi * 700 * t)
    power = spectrum(tone_signal())
    frequencies = np.fft.rfftfreq(400, 1 / 16000)
    data = {split: video_data(split) for split in ["train", "validation", "test"]}
    models = {}
    for kind in ["mean", "ordered"]:
        model = LogisticRegression(C=1, max_iter=1000).fit(
            features(data["train"][1], kind), data["train"][2])
        row = {}
        for split, (_, points, labels, ids) in data.items():
            predicted = model.predict(features(points, kind))
            row[split] = {"accuracy": float(np.mean(predicted == labels)),
                          "predicted": predicted.tolist(), "labels": labels.tolist(), "ids": ids}
        row["weights"], row["bias"] = model.coef_.tolist(), model.intercept_.tolist()
        models[kind] = row
    signal = np.zeros(20)
    signal[[3, 8, 15]] = 1
    delayed = np.concatenate([np.zeros(2), signal[:-2]])
    lag = int(np.argmax(np.correlate(delayed, signal, mode="full")) - (len(signal) - 1))
    return {
        "source_sha256": sha256(Path(__file__).read_bytes()).hexdigest(),
        "environment": {"python": platform.python_version(), "numpy": np.__version__},
        "alias_max_error": float(np.max(np.abs(low - high))),
        "spectrum": {"shape": list(power.shape), "window": 400, "hop": 160,
                     "first_peak_hz": float(frequencies[power[:, :40].mean(1).argmax()]),
                     "second_peak_hz": float(frequencies[power[:, -40:].mean(1).argmax()])},
        "videos": models,
        "data_sha256": {name: sha256(v[0].tobytes()).hexdigest() for name, v in data.items()},
        "sample_video": data["test"][0][:2].tolist(),
        "sample_points": data["test"][1][:2].tolist(),
        "pair_mean_difference": float(np.max(np.abs(
            features(data["test"][1][::2], "mean") - features(data["test"][1][1::2], "mean")))),
        "alignment": {"video": signal.tolist(), "audio": delayed.tolist(),
                      "grid_seconds": 0.1, "audio_lag_steps": lag},
        "queues": {str(ms): queue_schedule(ms) for ms in [120, 260]},
    }


def show(report):
    print("1 kHz 采样：300 Hz 与 700 Hz 余弦的样本相同（数值误差 < 1e-12）。")
    spec = report["spectrum"]
    print(f"短时频谱：{spec['shape']}；前后主频={spec['first_peak_hz']:.0f}/{spec['second_peak_hz']:.0f} Hz")
    print("视频训练/验证/测试：160 / 40 / 80；每对正反序列在同一划分。")
    for kind in ["mean", "ordered"]:
        row = report["videos"][kind]
        print(f"{kind}：验证准确率={row['validation']['accuracy']:.3f}，"
              f"测试准确率={row['test']['accuracy']:.3f}")
    print(f"正反视频的平均位置特征差={report['pair_mean_difference']:.1f}")
    print(f"模拟音频比视频晚 {report['alignment']['audio_lag_steps'] * .1:.1f} 秒。")
    for key, rows in report["queues"].items():
        print(f"假设每块处理 {key} ms：第 10 块完成延迟={rows[-1]['delay_ms']} ms")
    print("视频只使用亮点质心；队列和同步是教学模拟，不是真实视频大模型或在线性能。")


if __name__ == "__main__":
    parser = ArgumentParser(description=__doc__)
    parser.add_argument("--out", type=Path)
    args = parser.parse_args()
    result = run()
    if args.out:
        args.out.parent.mkdir(parents=True, exist_ok=True)
        args.out.write_text(json.dumps(result, ensure_ascii=False, indent=2) + "\n")
    show(result)

1. 时长是 $48000/16000=3$ 秒。只改声明后会播放六秒，速度与音高一起下降；这没有完成正确重采样。
2. 在整数采样位置，两条余弦的值相同。观测中已没有区分两者的信息；模型可以基于先验猜测，却不能保证恢复真实来源。
3. 有 98 个时间位置。窗口更长通常能提供更细的频率划分，却会把更长时间内的变化混在一起；时间分辨与频率分辨需要取舍。
4. 一次替换、一次插入，共 $2/3$。参考为空时 WER 分母为零，应另报无依据输出或插入词数，不能当成 0 分错误率。
5. 数字的英文词与阿拉伯数字被当前规则计为不同词。订单任务应预先规定数字规范化，并检查规范化后的订单号字段；不能在看到结果后只替有利样例改规则。
6. 0 dB 表示两者平均功率相等；20 dB 表示信号功率为噪声的 100 倍。这里功率由整段采样的平方均值估计。
7. 平均对排列不敏感。“亮点一直停在左边”与“先向右移动、再回左边”可以有相同首尾，却有不同过程。
8. 第五块的延迟是 $260+4\times60=500$ ms。这里每块对应 200 ms 音频，处理 RTF 为 $260/200=1.3$，持续到达时积压增长。
9. 可能先等两秒音频全部收齐才开始算，也可能还需等后续生成和播放。要记录采集开始/结束、第一份可用识别结果、第一段生成音频与实际播放开始等时刻。
10. 先写明新增输入、种子和评分规则，使用新目录保留全部输出，报告相对于原配置改变了什么。旧失败也应留在报告中，不能用一次新的成功覆盖它。