TTS contamination detector: HuBERT-Large features and a linear head

175bd80 verified 20 days ago

20.8 kB

# Copyright 2026 Spero TTS authors
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""TTS contamination detector: flags synthetic (TTS) speech in web-scraped speech training data.
A frozen HuBERT-Large reads each clip; the frame mean and standard deviation of eight hidden states are pooled into
16,384 numbers, and one linear logistic model turns them into the clip's logit of being synthetic. Decisions are
meant to be made per video (or per voice), from the duration-weighted mean clip probability over at least three
clips, not per clip.
    python detector.py clip1.wav clip2.flac clip3.flac            # per-clip p and the group decision
    python detector.py --model ./tts-contamination-detector *.flac
    from detector import TTSDetector, video_decision
    det = TTSDetector.from_pretrained("<user>/tts-contamination-detector")
    logits = det.score_files(paths)
    decision = video_decision(durations, logits)
Dependencies: numpy, scipy (the resampler), soundfile (decoding), torch and transformers (HuBERT); huggingface_hub
(installed with transformers) only to download from the Hub. The resampler is scipy's polyphase filter with the
same Kaiser FIR the training pipeline used, so features match the pipeline's; torchaudio or soxr would not.
"""

from __future__ import annotations

import argparse
import hashlib
import json
import math
import sys
from collections.abc import Sequence
from dataclasses import dataclass
from math import gcd
from pathlib import Path
from typing import Any

import numpy as np
import numpy.typing as npt

__all__ = ["TTSDetector", "VideoDecision", "front_end", "load_audio", "probability", "video_decision"]

SAMPLE_RATE = 16000
CONFIG_NAME = "config.json"
WEIGHTS_NAME = "model.safetensors"


# ---------------------------------------------------------------------------------------------------------------------
# Audio front end (a port of spero.audio.io.as_mono / resample and spero.data.synthdetect.front_end)


def as_mono(x: Any) -> npt.NDArray[Any]:
    """Float32 mono from (samples,) or (samples, channels); channels are averaged, integer PCM scaled to [-1, 1)."""
    raw = np.asarray(x)
    if np.issubdtype(raw.dtype, np.integer):
        info = np.iinfo(raw.dtype)
        half = (float(info.max) - float(info.min) + 1.0) / 2.0
        a = ((raw.astype(np.float64) - (float(info.min) + half)) / half).astype(np.float32)
    else:
        a = raw.astype(np.float32, copy=False)
    if a.ndim == 2:
        a = a.mean(axis=1, dtype=np.float32)
    if a.ndim != 1:
        raise ValueError(f"expected (samples,) or (samples, channels), got shape {a.shape}")
    return np.ascontiguousarray(a, dtype=np.float32)


def load_audio(path: str | Path) -> tuple[npt.NDArray[Any], int]:
    """(float32 mono, native sample rate) through libsndfile: WAV, FLAC, Ogg Vorbis/Opus and, with libsndfile 1.1 or
    newer, MP3."""
    import soundfile as sf

    array, sr = sf.read(str(path), dtype="float32", always_2d=True)
    return as_mono(array), int(sr)


_FILTERS: dict[tuple[int, int], npt.NDArray[Any]] = {}


def resample(x: npt.NDArray[Any], sr_from: int, sr_to: int) -> npt.NDArray[Any]:
    """Polyphase resampling with a Kaiser (beta 8.6) FIR, cutoff 0.97 of the lower Nyquist, in float64."""
    from scipy.signal import firwin, resample_poly

    a = as_mono(x)
    if sr_from <= 0 or sr_to <= 0:
        raise ValueError("sample rates must be positive")
    if sr_from == sr_to or a.size == 0:
        return a.copy()
    g = gcd(sr_from, sr_to)
    up, down = sr_to // g, sr_from // g
    taps = _FILTERS.get((up, down))
    if taps is None:
        max_rate = max(up, down)
        taps = np.asarray(firwin(2 * 32 * max_rate + 1, 0.97 / max_rate, window=("kaiser", 8.6)), dtype=np.float64)
        _FILTERS[(up, down)] = taps
    y = resample_poly(a.astype(np.float64), up, down, window=taps)
    return np.asarray(y, dtype=np.float32)


def trim_edges(
    x: npt.NDArray[Any], sr: int, below_peak_db: float, keep_before: float = 0.1, keep_after: float = 0.2
) -> npt.NDArray[Any]:
    """Cut leading and trailing 20 ms frames quieter than the loudest frame minus `below_peak_db`, keeping
    `keep_before` seconds before the first and `keep_after` after the last loud frame."""
    hop = max(1, int(0.02 * sr))
    n = x.size // hop
    if n < 3:
        return x
    frames = x[: n * hop].astype(np.float64).reshape(n, hop)
    db = 10.0 * np.log10(np.mean(frames * frames, axis=1) + 1e-12)
    loud = np.flatnonzero(db >= db.max() - below_peak_db)
    start = max(0, loud[0] * hop - int(keep_before * sr))
    end = min(x.size, (loud[-1] + 1) * hop + int(keep_after * sr))
    return np.ascontiguousarray(x[start:end])


def centre_crop(x: npt.NDArray[Any], n: int) -> npt.NDArray[Any]:
    if x.size <= n:
        return x
    start = (x.size - n) // 2
    return np.ascontiguousarray(x[start : start + n])


def front_end(
    x: Any,
    sr: int,
    *,
    max_seconds: float = 10.0,
    trim_db: float = 40.0,
    keep_before: float = 0.1,
    keep_after: float = 0.2,
) -> npt.NDArray[Any]:
    """Mono, edges trimmed, the middle `max_seconds` (cut at the native rate), then 16 kHz."""
    a = trim_edges(as_mono(x), sr, trim_db, keep_before, keep_after)
    a = centre_crop(a, round(max_seconds * sr))
    return resample(a, sr, SAMPLE_RATE) if sr != SAMPLE_RATE else a


# ---------------------------------------------------------------------------------------------------------------------
# Decisions


def probability(logits: Any) -> npt.NDArray[Any]:
    """Clip probabilities of being synthetic from logits (NaN stays NaN)."""
    return np.asarray(1.0 / (1.0 + np.exp(-np.asarray(logits, dtype=np.float64))), dtype=np.float64)


@dataclass(frozen=True)
class VideoDecision:
    synthetic: bool
    prob: float
    """Duration-weighted mean clip probability (NaN when no clip is scored)."""
    clips: int
    """Scored clips counted."""
    seconds: float
    """Their summed weights (durations floored at 0.1 s)."""


def video_decision(
    durations: npt.ArrayLike, logits: npt.ArrayLike, threshold: float = 0.32, min_clips: int = 3
) -> VideoDecision:
    """One video's (or one voice's) decision: synthetic when at least `min_clips` scored clips have a
    duration-weighted mean probability at or above `threshold`. Durations are floored at 0.1 s; clips whose logit is
    NaN (not scored) are left out. Durations are the clips' own lengths, before the detector's trim and crop."""
    d = np.asarray(durations, dtype=np.float64)
    z = np.asarray(logits, dtype=np.float64)
    if d.shape != z.shape:
        raise ValueError("durations and logits must have the same length")
    keep = np.isfinite(z)
    if not keep.any():
        return VideoDecision(False, math.nan, 0, 0.0)
    w = np.maximum(d[keep], 0.1)
    prob = float(np.dot(w, probability(z[keep])) / w.sum())
    clips = int(keep.sum())
    return VideoDecision(clips >= max(1, min_clips) and prob >= threshold, prob, clips, float(w.sum()))


# ---------------------------------------------------------------------------------------------------------------------
# The detector


def _batches_by_seconds(lengths: Sequence[int], seconds: float) -> list[list[int]]:
    """Indices sorted by length, cut into batches of at most `seconds` of padded 16 kHz audio."""
    order = sorted(range(len(lengths)), key=lambda i: lengths[i])
    out: list[list[int]] = []
    batch: list[int] = []
    for i in order:
        longest = max([lengths[j] for j in batch] + [lengths[i]])
        if batch and longest * (len(batch) + 1) > seconds * SAMPLE_RATE:
            out.append(batch)
            batch = []
        batch.append(i)
    if batch:
        out.append(batch)
    return out


class TTSDetector:
    """Frozen HuBERT-Large, mean/std pooling of the configured hidden states, and a linear logistic head.
    logit = clip(weight . (x - mean) / scale + bias, -12, 12), where x is the [layers x (mean, std) x 1024] pool
    (cast to float16 and back, as the training pipeline stored it)."""

    def __init__(
        self,
        config: dict[str, Any],
        tensors: dict[str, npt.NDArray[Any]],
        *,
        device: str | None = None,
        local_files_only: bool = False,
        attn_implementation: str | None = None,
    ) -> None:
        import torch
        from transformers import HubertModel

        self.config = config
        enc, head, fe = config["encoder"], config["head"], config["front_end"]
        self.layers = tuple(int(v) for v in enc["layers"])
        self.head_layers = tuple(int(v) for v in head["layers"])
        self.mean = np.ascontiguousarray(tensors["mean"], dtype=np.float32)
        self.scale = np.ascontiguousarray(tensors["scale"], dtype=np.float32)
        self.weight = np.ascontiguousarray(tensors["weight"], dtype=np.float32)
        self.bias = float(np.asarray(tensors["bias"], dtype=np.float64).reshape(-1)[0])
        self.logit_clip = float(head["logit_clip"])
        self.max_seconds = float(fe["max_seconds"])
        self.trim_db = float(fe["trim_db"])
        self.keep_before = float(fe["keep_before_s"])
        self.keep_after = float(fe["keep_after_s"])
        self.min_seconds = float(fe["min_seconds"])
        self.batch_seconds = float(config.get("batch_seconds", 320.0))
        dim = len(self.head_layers) * 2 * int(enc["hidden_size"])
        if not (self.weight.size == self.mean.size == self.scale.size == dim):
            raise ValueError(f"head tensors do not have {dim} features")

        self.torch = torch
        if device is None:
            device = "cuda" if torch.cuda.is_available() else "cpu"
        self.device = torch.device(device)
        if self.device.type == "cuda":
            # The pipeline ran HuBERT in float32 with TF32 matmuls and convolutions (a process-wide torch setting).
            tf32 = enc.get("precision", "tf32") == "tf32"
            torch.backends.cuda.matmul.allow_tf32 = tf32
            torch.backends.cudnn.allow_tf32 = tf32
        kwargs: dict[str, Any] = {"revision": enc["revision"], "local_files_only": local_files_only}
        if attn_implementation:
            kwargs["attn_implementation"] = attn_implementation
        model: Any = HubertModel.from_pretrained(enc["hf_id"], **kwargs)
        self.model = model.float().to(self.device).eval()
        cfg = self.model.config
        self.conv = list(zip(cfg.conv_kernel, cfg.conv_stride, strict=True))
        if max(self.layers) > int(cfg.num_hidden_layers) or min(self.layers) < 0:
            raise ValueError(f"layers must lie in 0..{cfg.num_hidden_layers}: {self.layers}")

    # -- loading -------------------------------------------------------------------------------------------------------

    @classmethod
    def from_pretrained(
        cls,
        path_or_repo: str | Path,
        *,
        revision: str | None = None,
        device: str | None = None,
        local_files_only: bool = False,
        attn_implementation: str | None = None,
    ) -> TTSDetector:
        """Load from a local folder holding config.json and model.safetensors, or from a Hugging Face model repo."""
        from safetensors.numpy import load_file

        local = Path(path_or_repo)
        if (local / CONFIG_NAME).is_file():
            config_path, weights_path = local / CONFIG_NAME, local / WEIGHTS_NAME
        else:
            from huggingface_hub import hf_hub_download

            repo = str(path_or_repo)
            config_path = Path(hf_hub_download(repo, CONFIG_NAME, revision=revision, local_files_only=local_files_only))
            weights_path = Path(
                hf_hub_download(repo, WEIGHTS_NAME, revision=revision, local_files_only=local_files_only)
            )
        config = json.loads(config_path.read_text())
        pinned = config.get("head", {}).get("safetensors_sha256")
        if pinned:
            digest = hashlib.sha256(weights_path.read_bytes()).hexdigest()
            if digest != pinned:
                raise ValueError(f"{weights_path}: sha256 {digest[:12]} is not the config's {pinned[:12]}")
        return cls(
            config,
            load_file(str(weights_path)),
            device=device,
            local_files_only=local_files_only,
            attn_implementation=attn_implementation,
        )

    # -- features ------------------------------------------------------------------------------------------------------

    def preprocess(self, audio: Any, sr: int) -> npt.NDArray[Any]:
        """The 16 kHz waveform the encoder reads."""
        return front_end(
            audio,
            sr,
            max_seconds=self.max_seconds,
            trim_db=self.trim_db,
            keep_before=self.keep_before,
            keep_after=self.keep_after,
        )

    def _frames(self, samples: int) -> int:
        n = samples
        for kernel, stride in self.conv:
            n = (n - kernel) // stride + 1
        return max(n, 1)

    def embed(self, waves: Sequence[npt.NDArray[Any]]) -> npt.NDArray[Any]:
        """One padded batch of 16 kHz waves to pooled features [clips, layers, 2 * hidden] in float16."""
        torch = self.torch
        lengths = [max(int(w.size), 400) for w in waves]
        batch = np.zeros((len(waves), max(lengths)), dtype=np.float32)
        mask = np.zeros(batch.shape, dtype=np.int64)
        for k, wave in enumerate(waves):
            a = wave.astype(np.float64)
            if a.size:
                batch[k, : a.size] = (a - a.mean()) / math.sqrt(a.var() + 1e-7)
            mask[k, : lengths[k]] = 1
        values = torch.from_numpy(batch).to(self.device, torch.float32)
        attention = torch.from_numpy(mask).to(self.device)
        valid = torch.tensor([self._frames(n) for n in lengths], device=self.device)
        with torch.inference_mode():
            states = self.model(values, attention_mask=attention, output_hidden_states=True).hidden_states
            t = states[0].shape[1]
            frame_mask = (torch.arange(t, device=self.device)[None, :] < valid[:, None]).float()[:, :, None]
            count = frame_mask.sum(dim=1).clamp(min=1.0)
            pooled = []
            for k in self.layers:
                h = states[k].float()
                mean = (h * frame_mask).sum(dim=1) / count
                var = (((h - mean[:, None, :]) ** 2) * frame_mask).sum(dim=1) / count
                pooled.append(torch.cat([mean, var.clamp(min=0).sqrt()], dim=-1))
            result = torch.stack(pooled, dim=1).clamp(-65000, 65000).to(torch.float16).cpu().numpy()
        return np.asarray(result, dtype=np.float16)

    def head(self, pooled: npt.NDArray[Any]) -> npt.NDArray[Any]:
        """Pooled features [clips, layers, 2 * hidden] to clipped logits (float64)."""
        index = [self.layers.index(k) for k in self.head_layers]
        x = np.ascontiguousarray(pooled[:, index, :].reshape(pooled.shape[0], -1), dtype=np.float32)
        z = ((x - self.mean) / self.scale) @ self.weight + self.bias
        return np.clip(np.asarray(z, dtype=np.float64), -self.logit_clip, self.logit_clip)

    # -- scoring -------------------------------------------------------------------------------------------------------

    def score_waves(self, waves: Sequence[npt.NDArray[Any]], batch_seconds: float | None = None) -> npt.NDArray[Any]:
        """Logits for already preprocessed 16 kHz waves; NaN for a wave shorter than `min_seconds`."""
        out = np.full(len(waves), np.nan, dtype=np.float64)
        ok = [i for i, w in enumerate(waves) if w.size >= int(self.min_seconds * SAMPLE_RATE)]
        lengths = [int(waves[i].size) for i in ok]
        for batch in _batches_by_seconds(lengths, batch_seconds or self.batch_seconds):
            idx = [ok[j] for j in batch]
            out[idx] = self.head(self.embed([waves[i] for i in idx]))
        return out

    def score_audio(self, audio: Sequence[tuple[Any, int]], batch_seconds: float | None = None) -> npt.NDArray[Any]:
        """Logits for (samples, sample rate) pairs at any rate."""
        return self.score_waves([self.preprocess(a, sr) for a, sr in audio], batch_seconds)

    def score_files(self, paths: Sequence[str | Path], batch_seconds: float | None = None) -> npt.NDArray[Any]:
        """Logits for audio files; NaN for a file that cannot be decoded or is too short after trimming."""
        waves: list[npt.NDArray[Any]] = []
        for p in paths:
            try:
                audio, sr = load_audio(p)
                waves.append(self.preprocess(audio, sr))
            except Exception as error:  # a broken file is reported, not fatal
                print(f"warning: {p}: {error}", file=sys.stderr)
                waves.append(np.zeros(0, dtype=np.float32))
        return self.score_waves(waves, batch_seconds)

    def probability(self, logits: Any) -> npt.NDArray[Any]:
        return probability(logits)

    def video_decision(self, durations: npt.ArrayLike, logits: npt.ArrayLike) -> VideoDecision:
        d = self.config["decision"]
        return video_decision(durations, logits, float(d["video_threshold"]), int(d["video_min_clips"]))


# ---------------------------------------------------------------------------------------------------------------------
# CLI


def _duration(path: str) -> float:
    try:
        import soundfile as sf

        info = sf.info(path)
        return float(info.frames) / float(info.samplerate)
    except Exception:
        return math.nan


def _finite(v: float) -> float | None:
    return None if math.isnan(v) else v


def main(argv: Sequence[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        description="Score clips of one video or voice for synthetic (TTS) speech; prints per-clip p and the group "
        "decision (duration-weighted mean p over at least --min-clips clips, against --threshold)."
    )
    parser.add_argument("files", nargs="+", help="audio clips of one video or one voice")
    parser.add_argument("--model", default=str(Path(__file__).resolve().parent), help="local folder or Hub repo id")
    parser.add_argument("--device", default=None, help="cuda, cuda:N or cpu (default: cuda if available)")
    parser.add_argument("--threshold", type=float, default=None, help="default: the config's (0.32)")
    parser.add_argument("--min-clips", type=int, default=None, help="default: the config's (3)")
    parser.add_argument("--json", action="store_true", help="print JSON instead of a table")
    a = parser.parse_args(argv)
    det = TTSDetector.from_pretrained(a.model, device=a.device)
    threshold = a.threshold if a.threshold is not None else float(det.config["decision"]["video_threshold"])
    min_clips = a.min_clips if a.min_clips is not None else int(det.config["decision"]["video_min_clips"])
    logits = det.score_files(a.files)
    durations = [_duration(f) for f in a.files]
    probs = probability(logits)
    dec = video_decision([0.0 if math.isnan(d) else d for d in durations], logits, threshold, min_clips)
    if a.json:
        print(
            json.dumps(
                {
                    "clips": [
                        {"file": f, "seconds": _finite(d), "logit": _finite(z), "p": _finite(p)}
                        for f, d, z, p in zip(a.files, durations, logits.tolist(), probs.tolist(), strict=True)
                    ],
                    "group": {
                        **{k: _finite(v) if isinstance(v, float) else v for k, v in dec.__dict__.items()},
                        "threshold": threshold,
                        "min_clips": min_clips,
                    },
                },
                indent=2,
            )
        )
        return 0
    for f, d, p in zip(a.files, durations, probs.tolist(), strict=True):
        shown = "   n/a" if math.isnan(p) else f"{p:6.3f}"
        print(f"p {shown}  {d:7.2f} s  {f}")
    if dec.clips < max(1, min_clips):
        verdict = f"no decision: {dec.clips} scored clips, need {min_clips}"
    else:
        verdict = "SYNTHETIC" if dec.synthetic else "not flagged"
    mean = "n/a" if math.isnan(dec.prob) else f"{dec.prob:.3f}"
    print(f"group: mean p {mean} over {dec.clips} clips (threshold {threshold}): {verdict}")
    return 0


if __name__ == "__main__":
    sys.exit(main())

	# Copyright 2026 Spero TTS authors
	#
	# Licensed under the Apache License, Version 2.0 (the "License");
	# you may not use this file except in compliance with the License.
	# You may obtain a copy of the License at
	#
	# http://www.apache.org/licenses/LICENSE-2.0
	#
	# Unless required by applicable law or agreed to in writing, software
	# distributed under the License is distributed on an "AS IS" BASIS,
	# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	# See the License for the specific language governing permissions and
	# limitations under the License.
	"""TTS contamination detector: flags synthetic (TTS) speech in web-scraped speech training data.

	A frozen HuBERT-Large reads each clip; the frame mean and standard deviation of eight hidden states are pooled into
	16,384 numbers, and one linear logistic model turns them into the clip's logit of being synthetic. Decisions are
	meant to be made per video (or per voice), from the duration-weighted mean clip probability over at least three
	clips, not per clip.

	python detector.py clip1.wav clip2.flac clip3.flac # per-clip p and the group decision
	python detector.py --model ./tts-contamination-detector *.flac

	from detector import TTSDetector, video_decision
	det = TTSDetector.from_pretrained("<user>/tts-contamination-detector")
	logits = det.score_files(paths)
	decision = video_decision(durations, logits)

	Dependencies: numpy, scipy (the resampler), soundfile (decoding), torch and transformers (HuBERT); huggingface_hub
	(installed with transformers) only to download from the Hub. The resampler is scipy's polyphase filter with the
	same Kaiser FIR the training pipeline used, so features match the pipeline's; torchaudio or soxr would not.
	"""

	from __future__ import annotations

	import argparse
	import hashlib
	import json
	import math
	import sys
	from collections.abc import Sequence
	from dataclasses import dataclass
	from math import gcd
	from pathlib import Path
	from typing import Any

	import numpy as np
	import numpy.typing as npt

	__all__ = ["TTSDetector", "VideoDecision", "front_end", "load_audio", "probability", "video_decision"]

	SAMPLE_RATE = 16000
	CONFIG_NAME = "config.json"
	WEIGHTS_NAME = "model.safetensors"


	# ---------------------------------------------------------------------------------------------------------------------
	# Audio front end (a port of spero.audio.io.as_mono / resample and spero.data.synthdetect.front_end)


	def as_mono(x: Any) -> npt.NDArray[Any]:
	"""Float32 mono from (samples,) or (samples, channels); channels are averaged, integer PCM scaled to [-1, 1)."""
	raw = np.asarray(x)
	if np.issubdtype(raw.dtype, np.integer):
	info = np.iinfo(raw.dtype)
	half = (float(info.max) - float(info.min) + 1.0) / 2.0
	a = ((raw.astype(np.float64) - (float(info.min) + half)) / half).astype(np.float32)
	else:
	a = raw.astype(np.float32, copy=False)
	if a.ndim == 2:
	a = a.mean(axis=1, dtype=np.float32)
	if a.ndim != 1:
	raise ValueError(f"expected (samples,) or (samples, channels), got shape {a.shape}")
	return np.ascontiguousarray(a, dtype=np.float32)


	def load_audio(path: str \| Path) -> tuple[npt.NDArray[Any], int]:
	"""(float32 mono, native sample rate) through libsndfile: WAV, FLAC, Ogg Vorbis/Opus and, with libsndfile 1.1 or
	newer, MP3."""
	import soundfile as sf

	array, sr = sf.read(str(path), dtype="float32", always_2d=True)
	return as_mono(array), int(sr)


	_FILTERS: dict[tuple[int, int], npt.NDArray[Any]] = {}


	def resample(x: npt.NDArray[Any], sr_from: int, sr_to: int) -> npt.NDArray[Any]:
	"""Polyphase resampling with a Kaiser (beta 8.6) FIR, cutoff 0.97 of the lower Nyquist, in float64."""
	from scipy.signal import firwin, resample_poly

	a = as_mono(x)
	if sr_from <= 0 or sr_to <= 0:
	raise ValueError("sample rates must be positive")
	if sr_from == sr_to or a.size == 0:
	return a.copy()
	g = gcd(sr_from, sr_to)
	up, down = sr_to // g, sr_from // g
	taps = _FILTERS.get((up, down))
	if taps is None:
	max_rate = max(up, down)
	taps = np.asarray(firwin(2 * 32 * max_rate + 1, 0.97 / max_rate, window=("kaiser", 8.6)), dtype=np.float64)
	_FILTERS[(up, down)] = taps
	y = resample_poly(a.astype(np.float64), up, down, window=taps)
	return np.asarray(y, dtype=np.float32)


	def trim_edges(
	x: npt.NDArray[Any], sr: int, below_peak_db: float, keep_before: float = 0.1, keep_after: float = 0.2
	) -> npt.NDArray[Any]:
	"""Cut leading and trailing 20 ms frames quieter than the loudest frame minus `below_peak_db`, keeping
	`keep_before` seconds before the first and `keep_after` after the last loud frame."""
	hop = max(1, int(0.02 * sr))
	n = x.size // hop
	if n < 3:
	return x
	frames = x[: n * hop].astype(np.float64).reshape(n, hop)
	db = 10.0 * np.log10(np.mean(frames * frames, axis=1) + 1e-12)
	loud = np.flatnonzero(db >= db.max() - below_peak_db)
	start = max(0, loud[0] * hop - int(keep_before * sr))
	end = min(x.size, (loud[-1] + 1) * hop + int(keep_after * sr))
	return np.ascontiguousarray(x[start:end])


	def centre_crop(x: npt.NDArray[Any], n: int) -> npt.NDArray[Any]:
	if x.size <= n:
	return x
	start = (x.size - n) // 2
	return np.ascontiguousarray(x[start : start + n])


	def front_end(
	x: Any,
	sr: int,
	*,
	max_seconds: float = 10.0,
	trim_db: float = 40.0,
	keep_before: float = 0.1,
	keep_after: float = 0.2,
	) -> npt.NDArray[Any]:
	"""Mono, edges trimmed, the middle `max_seconds` (cut at the native rate), then 16 kHz."""
	a = trim_edges(as_mono(x), sr, trim_db, keep_before, keep_after)
	a = centre_crop(a, round(max_seconds * sr))
	return resample(a, sr, SAMPLE_RATE) if sr != SAMPLE_RATE else a


	# ---------------------------------------------------------------------------------------------------------------------
	# Decisions


	def probability(logits: Any) -> npt.NDArray[Any]:
	"""Clip probabilities of being synthetic from logits (NaN stays NaN)."""
	return np.asarray(1.0 / (1.0 + np.exp(-np.asarray(logits, dtype=np.float64))), dtype=np.float64)


	@dataclass(frozen=True)
	class VideoDecision:
	synthetic: bool
	prob: float
	"""Duration-weighted mean clip probability (NaN when no clip is scored)."""
	clips: int
	"""Scored clips counted."""
	seconds: float
	"""Their summed weights (durations floored at 0.1 s)."""


	def video_decision(
	durations: npt.ArrayLike, logits: npt.ArrayLike, threshold: float = 0.32, min_clips: int = 3
	) -> VideoDecision:
	"""One video's (or one voice's) decision: synthetic when at least `min_clips` scored clips have a
	duration-weighted mean probability at or above `threshold`. Durations are floored at 0.1 s; clips whose logit is
	NaN (not scored) are left out. Durations are the clips' own lengths, before the detector's trim and crop."""
	d = np.asarray(durations, dtype=np.float64)
	z = np.asarray(logits, dtype=np.float64)
	if d.shape != z.shape:
	raise ValueError("durations and logits must have the same length")
	keep = np.isfinite(z)
	if not keep.any():
	return VideoDecision(False, math.nan, 0, 0.0)
	w = np.maximum(d[keep], 0.1)
	prob = float(np.dot(w, probability(z[keep])) / w.sum())
	clips = int(keep.sum())
	return VideoDecision(clips >= max(1, min_clips) and prob >= threshold, prob, clips, float(w.sum()))


	# ---------------------------------------------------------------------------------------------------------------------
	# The detector


	def _batches_by_seconds(lengths: Sequence[int], seconds: float) -> list[list[int]]:
	"""Indices sorted by length, cut into batches of at most `seconds` of padded 16 kHz audio."""
	order = sorted(range(len(lengths)), key=lambda i: lengths[i])
	out: list[list[int]] = []
	batch: list[int] = []
	for i in order:
	longest = max([lengths[j] for j in batch] + [lengths[i]])
	if batch and longest * (len(batch) + 1) > seconds * SAMPLE_RATE:
	out.append(batch)
	batch = []
	batch.append(i)
	if batch:
	out.append(batch)
	return out


	class TTSDetector:
	"""Frozen HuBERT-Large, mean/std pooling of the configured hidden states, and a linear logistic head.

	logit = clip(weight . (x - mean) / scale + bias, -12, 12), where x is the [layers x (mean, std) x 1024] pool
	(cast to float16 and back, as the training pipeline stored it)."""

	def __init__(
	self,
	config: dict[str, Any],
	tensors: dict[str, npt.NDArray[Any]],
	*,
	device: str \| None = None,
	local_files_only: bool = False,
	attn_implementation: str \| None = None,
	) -> None:
	import torch
	from transformers import HubertModel

	self.config = config
	enc, head, fe = config["encoder"], config["head"], config["front_end"]
	self.layers = tuple(int(v) for v in enc["layers"])
	self.head_layers = tuple(int(v) for v in head["layers"])
	self.mean = np.ascontiguousarray(tensors["mean"], dtype=np.float32)
	self.scale = np.ascontiguousarray(tensors["scale"], dtype=np.float32)
	self.weight = np.ascontiguousarray(tensors["weight"], dtype=np.float32)
	self.bias = float(np.asarray(tensors["bias"], dtype=np.float64).reshape(-1)[0])
	self.logit_clip = float(head["logit_clip"])
	self.max_seconds = float(fe["max_seconds"])
	self.trim_db = float(fe["trim_db"])
	self.keep_before = float(fe["keep_before_s"])
	self.keep_after = float(fe["keep_after_s"])
	self.min_seconds = float(fe["min_seconds"])
	self.batch_seconds = float(config.get("batch_seconds", 320.0))
	dim = len(self.head_layers) * 2 * int(enc["hidden_size"])
	if not (self.weight.size == self.mean.size == self.scale.size == dim):
	raise ValueError(f"head tensors do not have {dim} features")

	self.torch = torch
	if device is None:
	device = "cuda" if torch.cuda.is_available() else "cpu"
	self.device = torch.device(device)
	if self.device.type == "cuda":
	# The pipeline ran HuBERT in float32 with TF32 matmuls and convolutions (a process-wide torch setting).
	tf32 = enc.get("precision", "tf32") == "tf32"
	torch.backends.cuda.matmul.allow_tf32 = tf32
	torch.backends.cudnn.allow_tf32 = tf32
	kwargs: dict[str, Any] = {"revision": enc["revision"], "local_files_only": local_files_only}
	if attn_implementation:
	kwargs["attn_implementation"] = attn_implementation
	model: Any = HubertModel.from_pretrained(enc["hf_id"], **kwargs)
	self.model = model.float().to(self.device).eval()
	cfg = self.model.config
	self.conv = list(zip(cfg.conv_kernel, cfg.conv_stride, strict=True))
	if max(self.layers) > int(cfg.num_hidden_layers) or min(self.layers) < 0:
	raise ValueError(f"layers must lie in 0..{cfg.num_hidden_layers}: {self.layers}")

	# -- loading -------------------------------------------------------------------------------------------------------

	@classmethod
	def from_pretrained(
	cls,
	path_or_repo: str \| Path,
	*,
	revision: str \| None = None,
	device: str \| None = None,
	local_files_only: bool = False,
	attn_implementation: str \| None = None,
	) -> TTSDetector:
	"""Load from a local folder holding config.json and model.safetensors, or from a Hugging Face model repo."""
	from safetensors.numpy import load_file

	local = Path(path_or_repo)
	if (local / CONFIG_NAME).is_file():
	config_path, weights_path = local / CONFIG_NAME, local / WEIGHTS_NAME
	else:
	from huggingface_hub import hf_hub_download

	repo = str(path_or_repo)
	config_path = Path(hf_hub_download(repo, CONFIG_NAME, revision=revision, local_files_only=local_files_only))
	weights_path = Path(
	hf_hub_download(repo, WEIGHTS_NAME, revision=revision, local_files_only=local_files_only)
	)
	config = json.loads(config_path.read_text())
	pinned = config.get("head", {}).get("safetensors_sha256")
	if pinned:
	digest = hashlib.sha256(weights_path.read_bytes()).hexdigest()
	if digest != pinned:
	raise ValueError(f"{weights_path}: sha256 {digest[:12]} is not the config's {pinned[:12]}")
	return cls(
	config,
	load_file(str(weights_path)),
	device=device,
	local_files_only=local_files_only,
	attn_implementation=attn_implementation,
	)

	# -- features ------------------------------------------------------------------------------------------------------

	def preprocess(self, audio: Any, sr: int) -> npt.NDArray[Any]:
	"""The 16 kHz waveform the encoder reads."""
	return front_end(
	audio,
	sr,
	max_seconds=self.max_seconds,
	trim_db=self.trim_db,
	keep_before=self.keep_before,
	keep_after=self.keep_after,
	)

	def _frames(self, samples: int) -> int:
	n = samples
	for kernel, stride in self.conv:
	n = (n - kernel) // stride + 1
	return max(n, 1)

	def embed(self, waves: Sequence[npt.NDArray[Any]]) -> npt.NDArray[Any]:
	"""One padded batch of 16 kHz waves to pooled features [clips, layers, 2 * hidden] in float16."""
	torch = self.torch
	lengths = [max(int(w.size), 400) for w in waves]
	batch = np.zeros((len(waves), max(lengths)), dtype=np.float32)
	mask = np.zeros(batch.shape, dtype=np.int64)
	for k, wave in enumerate(waves):
	a = wave.astype(np.float64)
	if a.size:
	batch[k, : a.size] = (a - a.mean()) / math.sqrt(a.var() + 1e-7)
	mask[k, : lengths[k]] = 1
	values = torch.from_numpy(batch).to(self.device, torch.float32)
	attention = torch.from_numpy(mask).to(self.device)
	valid = torch.tensor([self._frames(n) for n in lengths], device=self.device)
	with torch.inference_mode():
	states = self.model(values, attention_mask=attention, output_hidden_states=True).hidden_states
	t = states[0].shape[1]
	frame_mask = (torch.arange(t, device=self.device)[None, :] < valid[:, None]).float()[:, :, None]
	count = frame_mask.sum(dim=1).clamp(min=1.0)
	pooled = []
	for k in self.layers:
	h = states[k].float()
	mean = (h * frame_mask).sum(dim=1) / count
	var = (((h - mean[:, None, :]) ** 2) * frame_mask).sum(dim=1) / count
	pooled.append(torch.cat([mean, var.clamp(min=0).sqrt()], dim=-1))
	result = torch.stack(pooled, dim=1).clamp(-65000, 65000).to(torch.float16).cpu().numpy()
	return np.asarray(result, dtype=np.float16)

	def head(self, pooled: npt.NDArray[Any]) -> npt.NDArray[Any]:
	"""Pooled features [clips, layers, 2 * hidden] to clipped logits (float64)."""
	index = [self.layers.index(k) for k in self.head_layers]
	x = np.ascontiguousarray(pooled[:, index, :].reshape(pooled.shape[0], -1), dtype=np.float32)
	z = ((x - self.mean) / self.scale) @ self.weight + self.bias
	return np.clip(np.asarray(z, dtype=np.float64), -self.logit_clip, self.logit_clip)

	# -- scoring -------------------------------------------------------------------------------------------------------

	def score_waves(self, waves: Sequence[npt.NDArray[Any]], batch_seconds: float \| None = None) -> npt.NDArray[Any]:
	"""Logits for already preprocessed 16 kHz waves; NaN for a wave shorter than `min_seconds`."""
	out = np.full(len(waves), np.nan, dtype=np.float64)
	ok = [i for i, w in enumerate(waves) if w.size >= int(self.min_seconds * SAMPLE_RATE)]
	lengths = [int(waves[i].size) for i in ok]
	for batch in _batches_by_seconds(lengths, batch_seconds or self.batch_seconds):
	idx = [ok[j] for j in batch]
	out[idx] = self.head(self.embed([waves[i] for i in idx]))
	return out

	def score_audio(self, audio: Sequence[tuple[Any, int]], batch_seconds: float \| None = None) -> npt.NDArray[Any]:
	"""Logits for (samples, sample rate) pairs at any rate."""
	return self.score_waves([self.preprocess(a, sr) for a, sr in audio], batch_seconds)

	def score_files(self, paths: Sequence[str \| Path], batch_seconds: float \| None = None) -> npt.NDArray[Any]:
	"""Logits for audio files; NaN for a file that cannot be decoded or is too short after trimming."""
	waves: list[npt.NDArray[Any]] = []
	for p in paths:
	try:
	audio, sr = load_audio(p)
	waves.append(self.preprocess(audio, sr))
	except Exception as error: # a broken file is reported, not fatal
	print(f"warning: {p}: {error}", file=sys.stderr)
	waves.append(np.zeros(0, dtype=np.float32))
	return self.score_waves(waves, batch_seconds)

	def probability(self, logits: Any) -> npt.NDArray[Any]:
	return probability(logits)

	def video_decision(self, durations: npt.ArrayLike, logits: npt.ArrayLike) -> VideoDecision:
	d = self.config["decision"]
	return video_decision(durations, logits, float(d["video_threshold"]), int(d["video_min_clips"]))


	# ---------------------------------------------------------------------------------------------------------------------
	# CLI


	def _duration(path: str) -> float:
	try:
	import soundfile as sf

	info = sf.info(path)
	return float(info.frames) / float(info.samplerate)
	except Exception:
	return math.nan


	def _finite(v: float) -> float \| None:
	return None if math.isnan(v) else v


	def main(argv: Sequence[str] \| None = None) -> int:
	parser = argparse.ArgumentParser(
	description="Score clips of one video or voice for synthetic (TTS) speech; prints per-clip p and the group "
	"decision (duration-weighted mean p over at least --min-clips clips, against --threshold)."
	)
	parser.add_argument("files", nargs="+", help="audio clips of one video or one voice")
	parser.add_argument("--model", default=str(Path(__file__).resolve().parent), help="local folder or Hub repo id")
	parser.add_argument("--device", default=None, help="cuda, cuda:N or cpu (default: cuda if available)")
	parser.add_argument("--threshold", type=float, default=None, help="default: the config's (0.32)")
	parser.add_argument("--min-clips", type=int, default=None, help="default: the config's (3)")
	parser.add_argument("--json", action="store_true", help="print JSON instead of a table")
	a = parser.parse_args(argv)
	det = TTSDetector.from_pretrained(a.model, device=a.device)
	threshold = a.threshold if a.threshold is not None else float(det.config["decision"]["video_threshold"])
	min_clips = a.min_clips if a.min_clips is not None else int(det.config["decision"]["video_min_clips"])
	logits = det.score_files(a.files)
	durations = [_duration(f) for f in a.files]
	probs = probability(logits)
	dec = video_decision([0.0 if math.isnan(d) else d for d in durations], logits, threshold, min_clips)
	if a.json:
	print(
	json.dumps(
	{
	"clips": [
	{"file": f, "seconds": _finite(d), "logit": _finite(z), "p": _finite(p)}
	for f, d, z, p in zip(a.files, durations, logits.tolist(), probs.tolist(), strict=True)
	],
	"group": {
	**{k: _finite(v) if isinstance(v, float) else v for k, v in dec.__dict__.items()},
	"threshold": threshold,
	"min_clips": min_clips,
	},
	},
	indent=2,
	)
	)
	return 0
	for f, d, p in zip(a.files, durations, probs.tolist(), strict=True):
	shown = " n/a" if math.isnan(p) else f"{p:6.3f}"
	print(f"p {shown} {d:7.2f} s {f}")
	if dec.clips < max(1, min_clips):
	verdict = f"no decision: {dec.clips} scored clips, need {min_clips}"
	else:
	verdict = "SYNTHETIC" if dec.synthetic else "not flagged"
	mean = "n/a" if math.isnan(dec.prob) else f"{dec.prob:.3f}"
	print(f"group: mean p {mean} over {dec.clips} clips (threshold {threshold}): {verdict}")
	return 0


	if __name__ == "__main__":
	sys.exit(main())