| """VeriSpoof inference: detection, recording condition, generator identification, source verification and |
| reasoning. |
| |
| Closed answers are scored by the log-likelihood of each candidate answer sentence (detection: |
| score = log p(spoof answer) - log p(bona fide answer), > 0 means spoof); reasoning is generated greedily. |
| Audio: 16 kHz mono; the first 8 s are used, shorter recordings are repeated to fill the window. |
| """ |
| import json |
| import os |
| import re |
|
|
| import numpy as np |
| import torch |
|
|
| from verispoof.model import AUDIO, Network |
|
|
| SR = 16000 |
|
|
| AUTH_QUESTION = "Is this recording genuine human speech or spoofed speech? Options: bona fide; spoof." |
| AUTH_ANSWERS = ("This recording is bona fide: it is genuine human speech.", |
| "This recording is a spoof: it is synthetic or converted speech, not genuine human speech.") |
| PAIR = { |
| "generator": ("Do Audio A and Audio B come from the same generator?", |
| ("Yes, both recordings were produced by the same speech generation system.", |
| "No, the two recordings were produced by different speech generation systems.")), |
| "acoustic_model": ("Were Audio A and Audio B generated with the same acoustic model?", |
| ("Yes, both recordings were generated with the same acoustic model.", |
| "No, the two recordings use different acoustic models.")), |
| "vocoder": ("Do Audio A and Audio B share the same vocoder?", |
| ("Yes, both recordings were synthesized with the same vocoder.", |
| "No, the two recordings use different vocoders.")), |
| } |
| CONDITION = { |
| "noise": ("What is the background noise level of this recording?", |
| ["clean", "mild", "moderate", "severe", "extreme"], |
| lambda o: "The recording is clean, with no audible noise." if o == "clean" |
| else f"The recording has {o} background noise."), |
| "reverb": ("How much room reverberation does this recording have?", |
| ["minimal", "slight", "moderate", "heavy", "extreme"], |
| lambda o: "The recording has no noticeable reverberation." if o == "minimal" |
| else f"The recording has {o} reverberation."), |
| "codec": ("Was this recording compressed by a codec, and if so which kind?", |
| ["none", "MP3", "AAC", "Opus", "narrowband telephone", "wideband telephone", "neural codec", |
| "MP3 + neural codec"], |
| lambda o: "The recording shows no codec compression." if o == "none" |
| else f"The recording shows {o} coding." if o.endswith("telephone") |
| else "The recording shows MP3 and neural-codec compression in cascade." if o == "MP3 + neural codec" |
| else f"The recording shows {o} compression."), |
| } |
| GENERATOR_QUESTION = "Which generator was used to create this speech?" |
| EXPLAIN_QUESTION = "Analyse this recording step by step and decide whether it is genuine human speech or spoofed." |
| COMPARE_QUESTION = ("Compare the source profiles of the two recordings and decide whether they were produced by the " |
| "same generation system.") |
| REASONING_FIELDS = ("OBSERVATIONS", "CHANNEL_STATUS", "LOCALIZATION", "SOURCE_PROFILE", "FINGERPRINT", |
| "PROFILE_COMPATIBILITY", "DECISION") |
|
|
|
|
| def load_audio(x, sr=None): |
| """A path, or a 1-D array with its sample rate -> mono float32 at 16 kHz.""" |
| if isinstance(x, (str, os.PathLike)): |
| import soundfile as sf |
| try: |
| x, sr = sf.read(str(x), dtype="float32", always_2d=True) |
| x = x[:, 0] |
| except sf.LibsndfileError: |
| import librosa |
| x, sr = librosa.load(str(x), sr=None, mono=False) |
| x = x if x.ndim == 1 else x[0] |
| else: |
| x = np.asarray(x, dtype=np.float32) |
| if sr is None: |
| raise ValueError("give the sample rate of an array input") |
| if x.ndim > 1: |
| x = x[0] if x.shape[0] < x.shape[1] else x[:, 0] |
| if sr != SR: |
| import librosa |
| x = librosa.resample(x, orig_sr=sr, target_sr=SR, res_type="soxr_hq") |
| return x.astype(np.float32) |
|
|
|
|
| def fit_window(x, S): |
| """First S samples; shorter recordings are tiled; an empty one is digital silence.""" |
| if len(x) >= S: |
| return x[:S] |
| if not len(x): |
| return np.zeros(S, np.float32) |
| return np.tile(x, S // len(x) + 1)[:S] |
|
|
|
|
| def parse_reasoning(text): |
| """Reasoning text -> its lines by field name, the decision and any localized spans (s).""" |
| f = {} |
| for line in text.split("\n"): |
| for k in REASONING_FIELDS: |
| if line.startswith(k + ":"): |
| f[k] = line[len(k) + 1:].strip() |
| d = f.get("DECISION", "") |
| short = text.strip() |
| if "come from the same generation system" in d or short.startswith("Yes, both recordings were produced by the same"): |
| dec = "same" |
| elif "come from different generation systems" in d or short.startswith("No, the two recordings were produced by"): |
| dec = "different" |
| else: |
| dec = "bonafide" if d.startswith("The recording is bona fide") else \ |
| "spoof" if d.startswith("The recording is a spoof") else None |
| spans = [[float(a), float(b)] for a, b in re.findall(r"(\d+\.\d+)-(\d+\.\d+) s", f.get("LOCALIZATION", ""))] |
| return {"lines": f, "decision": dec, "spans": spans} |
|
|
|
|
| class VeriSpoof: |
| def __init__(self, net, device="cuda"): |
| self.net, self.device = net, device |
|
|
| @classmethod |
| def from_pretrained(cls, path, device="cuda", revision=None): |
| """path: a local directory or a Hugging Face repository id.""" |
| if not os.path.isdir(path): |
| from huggingface_hub import snapshot_download |
| path = snapshot_download(path, revision=revision) |
| cfg = json.load(open(f"{path}/config.json")) |
| return cls(Network(path, cfg, device=device).to(device).eval(), device) |
|
|
| |
| def _windows(self, audios, sr=None): |
| w = [fit_window(load_audio(a, sr), self.net.window) for a in audios] |
| return torch.from_numpy(np.stack(w)) |
|
|
| @staticmethod |
| def _batch(items, n): |
| return [items[i:i + n] for i in range(0, len(items), n)] |
|
|
| def _left_pad(self, P): |
| Lp = max(map(len, P)) |
| ids = torch.full((len(P), Lp), self.net.pad_id, dtype=torch.long) |
| att = torch.zeros((len(P), Lp), dtype=torch.long) |
| for i, p in enumerate(P): |
| ids[i, Lp - len(p):], att[i, Lp - len(p):] = torch.tensor(p), 1 |
| return ids.to(self.device), att.to(self.device) |
|
|
| |
| @torch.no_grad() |
| def _score(self, prompts, wav, answers): |
| """prompts: B token-id lists; wav: (N_audio, S) in prompt order; answers: per prompt, a list of candidate |
| sentences -> list of arrays of total log-likelihoods.""" |
| lm = self.net.causal_lm() |
| ids, att = self._left_pad(prompts) |
| pos = (att.cumsum(-1) - 1).clamp(min=0) |
| emb = self.net.embed(ids, wav.to(self.device)) |
| with torch.autocast("cuda", dtype=torch.bfloat16, enabled=self.net.amp): |
| out = lm.model(inputs_embeds=emb, attention_mask=att, position_ids=pos, use_cache=True) |
| first = lm.lm_head(out.last_hidden_state[:, -1]).float().log_softmax(-1) |
| rows = [(b, self.net.text_ids(c + "<|im_end|>")) for b, cs in enumerate(answers) for c in cs] |
| Lc = max(len(c) for _, c in rows) |
| cand = torch.full((len(rows), Lc), self.net.pad_id, dtype=torch.long, device=self.device) |
| cm = torch.zeros((len(rows), Lc), dtype=torch.long, device=self.device) |
| for r, (_, c) in enumerate(rows): |
| cand[r, :len(c)], cm[r, :len(c)] = torch.tensor(c), 1 |
| owner = torch.tensor([b for b, _ in rows], device=self.device) |
| lp = first[owner].gather(1, cand[:, :1]).squeeze(1) |
| if Lc > 1: |
| cache = out.past_key_values |
| cache.reorder_cache(owner) |
| att2 = torch.cat([att[owner], cm], 1) |
| pos2 = att.sum(1)[owner, None] + torch.arange(Lc, device=self.device) |
| with torch.autocast("cuda", dtype=torch.bfloat16, enabled=self.net.amp): |
| h = lm.model(input_ids=cand, attention_mask=att2, position_ids=pos2, past_key_values=cache, |
| use_cache=False).last_hidden_state |
| logp = lm.lm_head(h[:, :-1]).float().log_softmax(-1) |
| lp = lp + (logp.gather(2, cand[:, 1:, None]).squeeze(2) * cm[:, 1:]).sum(1) |
| lp = lp.cpu().numpy() |
| out_, o = [], 0 |
| for cs in answers: |
| out_.append(lp[o:o + len(cs)]) |
| o += len(cs) |
| return out_ |
|
|
| |
| @torch.no_grad() |
| def _generate(self, prompts, wav, max_new_tokens): |
| lm = self.net.causal_lm() |
| ids, att = self._left_pad(prompts) |
| pos = (att.cumsum(-1) - 1).clamp(min=0) |
| emb = self.net.embed(ids, wav.to(self.device)) |
| eos, pad = self.net.end_id, self.net.pad_id |
| with torch.autocast("cuda", dtype=torch.bfloat16, enabled=self.net.amp): |
| out = lm.model(inputs_embeds=emb, attention_mask=att, position_ids=pos, use_cache=True) |
| nxt = lm.lm_head(out.last_hidden_state[:, -1]).argmax(-1) |
| h_prompt = out.last_hidden_state |
| cache, cur = out.past_key_values, att.sum(1) |
| gen, done = [nxt], nxt == eos |
| for _ in range(max_new_tokens - 1): |
| if done.all(): |
| break |
| att = torch.cat([att, torch.ones_like(att[:, :1])], 1) |
| with torch.autocast("cuda", dtype=torch.bfloat16, enabled=self.net.amp): |
| out = lm.model(input_ids=nxt[:, None], attention_mask=att, position_ids=cur[:, None], |
| past_key_values=cache, use_cache=True) |
| nxt = lm.lm_head(out.last_hidden_state[:, -1]).argmax(-1) |
| cache, cur = out.past_key_values, cur + 1 |
| nxt = torch.where(done, torch.full_like(nxt, pad), nxt) |
| gen.append(nxt) |
| done |= nxt == eos |
| G = torch.stack(gen, 1).cpu().tolist() |
| texts = [self.net.text_tok.decode(g[:g.index(eos)] if eos in g else g) for g in G] |
| return texts, h_prompt, ids |
|
|
| |
| def detect(self, audios, sr=None, batch_size=16): |
| """One recording or a list -> {'label', 'score'} (or a list of them). score = log p(spoof answer) - |
| log p(bona fide answer); the paper's EERs are computed on this score, 'label' uses threshold 0.""" |
| single = not isinstance(audios, (list, tuple)) |
| audios = [audios] if single else list(audios) |
| p = self.net.prompt_ids(f"{AUDIO}\n{AUTH_QUESTION}") |
| out = [] |
| for chunk in self._batch(audios, batch_size): |
| for s in self._score([p] * len(chunk), self._windows(chunk, sr), [AUTH_ANSWERS] * len(chunk)): |
| score = float(s[1] - s[0]) |
| out.append({"label": "spoof" if score > 0 else "bonafide", "score": score}) |
| return out[0] if single else out |
|
|
| def choose(self, audio, question, options, answers, sr=None): |
| """Closed question about one recording: the option whose answer sentence is most likely.""" |
| prompt = self.net.prompt_ids(f"{AUDIO}\n{question} Options: {'; '.join(options)}.") |
| s = self._score([prompt], self._windows([audio], sr), [answers])[0] |
| return {"answer": options[int(np.argmax(s))], "loglik": dict(zip(options, map(float, s)))} |
|
|
| def generator(self, audio, candidates, sr=None): |
| """Which of the candidate generation systems produced the recording (closed set you supply).""" |
| return self.choose(audio, GENERATOR_QUESTION, list(candidates), |
| [f"This recording was generated by {c}." for c in candidates], sr) |
|
|
| def condition(self, audio, attr="codec", sr=None): |
| """Recording condition: attr in noise, reverb, codec.""" |
| q, opts, sent = CONDITION[attr] |
| return self.choose(audio, q, opts, [sent(o) for o in opts], sr) |
|
|
| def same_source(self, audio_a, audio_b, level="generator", sr=None): |
| """Do two spoofed recordings share the generator (or acoustic model, or vocoder)? score = log p(yes) - |
| log p(no), > 0 means the same.""" |
| q, (yes, no) = PAIR[level] |
| prompt = self.net.prompt_ids(f"Audio A: {AUDIO}\nAudio B: {AUDIO}\n{q}") |
| s = self._score([prompt], self._windows([audio_a, audio_b], sr), [[yes, no]])[0] |
| score = float(s[0] - s[1]) |
| return {"same": score > 0, "score": score} |
|
|
| @torch.no_grad() |
| def explain(self, audio, sr=None, max_new_tokens=320): |
| """Grounded detection reasoning (greedy) plus the per-token head: probability that each 0.16 s token of the |
| heard window is synthesized.""" |
| p = self.net.prompt_ids(f"{AUDIO}\n{EXPLAIN_QUESTION}") |
| texts, h, ids = self._generate([p], self._windows([audio], sr), max_new_tokens) |
| z = self.net.token_logits(h, ids)[0].sigmoid().cpu().numpy() |
| r = parse_reasoning(texts[0]) |
| r.update(text=texts[0], token_prob=z.tolist(), token_seconds=self.net.seconds_per_token()) |
| return r |
|
|
| def compare(self, audio_a, audio_b, sr=None, max_new_tokens=320): |
| """Source-verification reasoning for two recordings (greedy).""" |
| p = self.net.prompt_ids(f"Audio A: {AUDIO}\nAudio B: {AUDIO}\n{COMPARE_QUESTION}") |
| texts, _, _ = self._generate([p], self._windows([audio_a, audio_b], sr), max_new_tokens) |
| r = parse_reasoning(texts[0]) |
| r["text"] = texts[0] |
| return r |
|
|
