Spaces:

vipan-kumar
/

Audio_Deepfake_Detection

Sleeping

Audio_Deepfake_Detection / backend /app /utils /sample_generator.py

Initial commit: Audio Deepfake Detector with 8 detectors trained on jay15k

e6a1f55 5 months ago

5.25 kB

"""Bootstrap sample audio for the demo's Sample Library.
Strategy (per slot):
  1. If file already present and non-empty, leave it alone.
  2. For "real" slots: download a small public-domain LibriSpeech clip from
     HuggingFace's CDN (stable URLs used in their docs for years). Trim,
     resample to 16 kHz mono, save as PCM_16 WAV.
  3. If download fails, fall back to the high-quality source-filter
     synthesizer in ``speech_synth.py`` — much better than the previous
     formant-resonator approach.
  4. For "fake" slots: always synthesise (these are *meant* to be TTS-like).
Result: the Sample Library has actually-real speech for the real slots when
the user has internet, gracefully degrading to plausible synthesis offline.
"""
from __future__ import annotations

import io
from pathlib import Path
from typing import Optional

import numpy as np
import soundfile as sf

from app.features.audio_preprocessor import preprocess_audio
from app.logging_setup import get_logger
from app.utils.speech_synth import (
    synth_real_speech,
    synth_fake_neural_tts,
    synth_fake_tts_commercial,
    synth_fake_voice_clone,
)

logger = get_logger(__name__)

SR = 16000

# Stable, long-lived public URLs for real LibriSpeech samples. These are the
# clips HuggingFace ships in their wav2vec2 docs and have been stable for
# years. Both are CC-BY 4.0 (LibriSpeech license).
_REAL_DOWNLOAD_CANDIDATES: list[tuple[str, str]] = [
    ("real_news_excerpt.wav",   "https://cdn-media.huggingface.co/speech_samples/sample1.flac"),
    ("real_conversation.wav",   "https://cdn-media.huggingface.co/speech_samples/sample2.flac"),
    # Third real slot uses the same source but the second half — we trim
    # different segments client-side so we get audibly distinct clips.
    ("real_lecture.wav",        "https://cdn-media.huggingface.co/speech_samples/sample1.flac"),
]


def _try_download_speech(url: str, max_seconds: float, segment: str) -> Optional[np.ndarray]:
    """Download → decode → trim → resample to 16 kHz mono. Returns None on failure.
    ``segment`` selects which slice to keep:
        'head'   -> first max_seconds
        'mid'    -> middle max_seconds
        'tail'   -> last max_seconds
    """
    try:
        import httpx

        with httpx.Client(timeout=15.0, follow_redirects=True) as client:
            r = client.get(url)
            if r.status_code != 200 or not r.content:
                logger.warning("download %s -> HTTP %s", url, r.status_code)
                return None
            data = r.content
    except Exception as exc:  # noqa: BLE001
        logger.warning("download %s failed: %s", url, exc)
        return None

    try:
        loaded = preprocess_audio(data, target_sr=SR, max_seconds=60.0)
    except Exception as exc:  # noqa: BLE001
        logger.warning("decode of %s failed: %s", url, exc)
        return None

    wav = loaded.waveform.squeeze().detach().cpu().numpy().astype(np.float32)
    target_len = int(max_seconds * SR)
    if wav.size <= target_len:
        return wav

    if segment == "head":
        return wav[:target_len]
    if segment == "tail":
        return wav[-target_len:]
    # 'mid' — pick a deterministic offset
    start = (wav.size - target_len) // 2
    return wav[start : start + target_len]


# ---------------------------------------------------------------------------
# Public API
# ---------------------------------------------------------------------------

def ensure_sample_audios(target_dir: Path) -> int:
    """Create any missing sample WAV files. Returns number of new files written."""
    target_dir.mkdir(parents=True, exist_ok=True)
    rng = np.random.default_rng(42)
    created = 0

    # --- REAL slots: try download first --------------------------------- #
    real_segments = ["head", "head", "tail"]
    for (filename, url), seg in zip(_REAL_DOWNLOAD_CANDIDATES, real_segments):
        path = target_dir / filename
        if path.exists() and path.stat().st_size > 4096:
            continue
        wav = _try_download_speech(url, max_seconds=4.5, segment=seg)
        if wav is None or wav.size < SR:
            logger.info("Falling back to synthesised real speech for %s", filename)
            wav = synth_real_speech(seconds=4.5, rng=rng)
        else:
            logger.info("Downloaded real speech for %s (%.1fs)", filename, wav.size / SR)
        sf.write(str(path), wav, SR, subtype="PCM_16")
        created += 1

    # --- FAKE slots: always synthesise ---------------------------------- #
    fake_specs = [
        ("fake_tts_commercial.wav", lambda r: synth_fake_tts_commercial(4.5, r)),
        ("fake_voice_clone.wav",    lambda r: synth_fake_voice_clone(4.5, r)),
        ("fake_neural_tts.wav",     lambda r: synth_fake_neural_tts(4.5, r)),
    ]
    for filename, gen in fake_specs:
        path = target_dir / filename
        if path.exists() and path.stat().st_size > 4096:
            continue
        wav = gen(rng)
        sf.write(str(path), wav, SR, subtype="PCM_16")
        logger.info("Generated fake sample: %s", filename)
        created += 1

    if created:
        logger.info("Sample audio bootstrap complete: %d new files in %s", created, target_dir)
    return created

	"""Bootstrap sample audio for the demo's Sample Library.

	Strategy (per slot):
	1. If file already present and non-empty, leave it alone.
	2. For "real" slots: download a small public-domain LibriSpeech clip from
	HuggingFace's CDN (stable URLs used in their docs for years). Trim,
	resample to 16 kHz mono, save as PCM_16 WAV.
	3. If download fails, fall back to the high-quality source-filter
	synthesizer in ``speech_synth.py`` — much better than the previous
	formant-resonator approach.
	4. For "fake" slots: always synthesise (these are meant to be TTS-like).

	Result: the Sample Library has actually-real speech for the real slots when
	the user has internet, gracefully degrading to plausible synthesis offline.
	"""
	from __future__ import annotations

	import io
	from pathlib import Path
	from typing import Optional

	import numpy as np
	import soundfile as sf

	from app.features.audio_preprocessor import preprocess_audio
	from app.logging_setup import get_logger
	from app.utils.speech_synth import (
	synth_real_speech,
	synth_fake_neural_tts,
	synth_fake_tts_commercial,
	synth_fake_voice_clone,
	)

	logger = get_logger(__name__)

	SR = 16000

	# Stable, long-lived public URLs for real LibriSpeech samples. These are the
	# clips HuggingFace ships in their wav2vec2 docs and have been stable for
	# years. Both are CC-BY 4.0 (LibriSpeech license).
	_REAL_DOWNLOAD_CANDIDATES: list[tuple[str, str]] = [
	("real_news_excerpt.wav", "https://cdn-media.huggingface.co/speech_samples/sample1.flac"),
	("real_conversation.wav", "https://cdn-media.huggingface.co/speech_samples/sample2.flac"),
	# Third real slot uses the same source but the second half — we trim
	# different segments client-side so we get audibly distinct clips.
	("real_lecture.wav", "https://cdn-media.huggingface.co/speech_samples/sample1.flac"),
	]


	def _try_download_speech(url: str, max_seconds: float, segment: str) -> Optional[np.ndarray]:
	"""Download → decode → trim → resample to 16 kHz mono. Returns None on failure.

	``segment`` selects which slice to keep:
	'head' -> first max_seconds
	'mid' -> middle max_seconds
	'tail' -> last max_seconds
	"""
	try:
	import httpx

	with httpx.Client(timeout=15.0, follow_redirects=True) as client:
	r = client.get(url)
	if r.status_code != 200 or not r.content:
	logger.warning("download %s -> HTTP %s", url, r.status_code)
	return None
	data = r.content
	except Exception as exc: # noqa: BLE001
	logger.warning("download %s failed: %s", url, exc)
	return None

	try:
	loaded = preprocess_audio(data, target_sr=SR, max_seconds=60.0)
	except Exception as exc: # noqa: BLE001
	logger.warning("decode of %s failed: %s", url, exc)
	return None

	wav = loaded.waveform.squeeze().detach().cpu().numpy().astype(np.float32)
	target_len = int(max_seconds * SR)
	if wav.size <= target_len:
	return wav

	if segment == "head":
	return wav[:target_len]
	if segment == "tail":
	return wav[-target_len:]
	# 'mid' — pick a deterministic offset
	start = (wav.size - target_len) // 2
	return wav[start : start + target_len]


	# ---------------------------------------------------------------------------
	# Public API
	# ---------------------------------------------------------------------------

	def ensure_sample_audios(target_dir: Path) -> int:
	"""Create any missing sample WAV files. Returns number of new files written."""
	target_dir.mkdir(parents=True, exist_ok=True)
	rng = np.random.default_rng(42)
	created = 0

	# --- REAL slots: try download first --------------------------------- #
	real_segments = ["head", "head", "tail"]
	for (filename, url), seg in zip(_REAL_DOWNLOAD_CANDIDATES, real_segments):
	path = target_dir / filename
	if path.exists() and path.stat().st_size > 4096:
	continue
	wav = _try_download_speech(url, max_seconds=4.5, segment=seg)
	if wav is None or wav.size < SR:
	logger.info("Falling back to synthesised real speech for %s", filename)
	wav = synth_real_speech(seconds=4.5, rng=rng)
	else:
	logger.info("Downloaded real speech for %s (%.1fs)", filename, wav.size / SR)
	sf.write(str(path), wav, SR, subtype="PCM_16")
	created += 1

	# --- FAKE slots: always synthesise ---------------------------------- #
	fake_specs = [
	("fake_tts_commercial.wav", lambda r: synth_fake_tts_commercial(4.5, r)),
	("fake_voice_clone.wav", lambda r: synth_fake_voice_clone(4.5, r)),
	("fake_neural_tts.wav", lambda r: synth_fake_neural_tts(4.5, r)),
	]
	for filename, gen in fake_specs:
	path = target_dir / filename
	if path.exists() and path.stat().st_size > 4096:
	continue
	wav = gen(rng)
	sf.write(str(path), wav, SR, subtype="PCM_16")
	logger.info("Generated fake sample: %s", filename)
	created += 1

	if created:
	logger.info("Sample audio bootstrap complete: %d new files in %s", created, target_dir)
	return created