Upload folder using huggingface_hub

src/__init__.py

@@ -0,0 +1,8 @@
+"""Wake-word dataset creator package.
+
+Generates Edge Impulse / Hugging Face ready wake-word datasets using
+Google Cloud TTS (when an API key is available) with an automatic
+fall back to free, local Piper TTS.
+"""
+
+__all__ = ["config", "audio", "builder", "hf_export", "backends", "edge_impulse"]

src/audio.py

@@ -0,0 +1,240 @@
+"""Audio helpers: WAV I/O, resampling, augmentation and synthetic noise.
+
+Kept dependency-light: only numpy is required (no ffmpeg / scipy), so it
+runs anywhere including a minimal Hugging Face Space.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import io
+import math
+import random
+import re
+import wave
+from pathlib import Path
+from typing import Tuple
+
+import numpy as np
+
+
+# --------------------------------------------------------------------------- #
+# Naming helpers
+# --------------------------------------------------------------------------- #
+
+def slugify(text: str, max_len: int = 80) -> str:
+    text = str(text).strip().lower()
+    text = re.sub(r"[^a-z0-9]+", "_", text)
+    text = re.sub(r"_+", "_", text).strip("_")
+    return (text or "item")[:max_len]
+
+
+def stable_hash(text: str, length: int = 12) -> str:
+    return hashlib.sha1(text.encode("utf-8")).hexdigest()[:length]
+
+
+# --------------------------------------------------------------------------- #
+# WAV read / write / resample
+# --------------------------------------------------------------------------- #
+
+def read_wav_bytes(wav_bytes: bytes) -> Tuple[np.ndarray, int]:
+    """Decode 16-bit PCM WAV bytes into mono float32 samples and sample rate."""
+    with wave.open(io.BytesIO(wav_bytes), "rb") as wf:
+        channels = wf.getnchannels()
+        sample_width = wf.getsampwidth()
+        sample_rate = wf.getframerate()
+        frames = wf.readframes(wf.getnframes())
+
+    if sample_width != 2:
+        raise ValueError(f"Expected 16-bit PCM WAV, got sample width {sample_width}")
+
+    audio = np.frombuffer(frames, dtype=np.int16).astype(np.float32)
+    if channels > 1:
+        audio = audio.reshape(-1, channels).mean(axis=1)
+    return audio, sample_rate
+
+
+def read_wav_file(path: Path) -> Tuple[np.ndarray, int]:
+    return read_wav_bytes(Path(path).read_bytes())
+
+
+def write_wav_file(path: Path, audio: np.ndarray, sample_rate_hz: int) -> None:
+    path = Path(path)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    audio_i16 = np.clip(audio, -32768, 32767).astype(np.int16)
+    with wave.open(str(path), "wb") as wf:
+        wf.setnchannels(1)
+        wf.setsampwidth(2)
+        wf.setframerate(sample_rate_hz)
+        wf.writeframes(audio_i16.tobytes())
+
+
+def resample(audio: np.ndarray, src_rate: int, dst_rate: int) -> np.ndarray:
+    """Linear-interpolation resample. Good enough for a bootstrap dataset."""
+    if src_rate == dst_rate or len(audio) == 0:
+        return audio.astype(np.float32)
+    duration = len(audio) / float(src_rate)
+    dst_len = max(1, int(round(duration * dst_rate)))
+    src_idx = np.linspace(0.0, len(audio) - 1, num=dst_len)
+    return np.interp(src_idx, np.arange(len(audio)), audio).astype(np.float32)
+
+
+# --------------------------------------------------------------------------- #
+# Shaping
+# --------------------------------------------------------------------------- #
+
+def normalize(audio: np.ndarray, peak: float = 28000.0) -> np.ndarray:
+    if len(audio) == 0:
+        return audio
+    m = float(np.max(np.abs(audio)))
+    if m < 1.0:
+        return audio
+    return (audio * (peak / m)).astype(np.float32)
+
+
+def pad_or_trim(audio: np.ndarray, target_samples: int, random_crop: bool = False) -> np.ndarray:
+    current = len(audio)
+    if current == target_samples:
+        return audio
+    if current > target_samples:
+        start = (
+            random.randint(0, current - target_samples)
+            if random_crop
+            else (current - target_samples) // 2
+        )
+        return audio[start:start + target_samples]
+    pad_total = target_samples - current
+    pad_left = pad_total // 2
+    pad_right = pad_total - pad_left
+    return np.pad(audio, (pad_left, pad_right), mode="constant")
+
+
+# --------------------------------------------------------------------------- #
+# Augmentation
+# --------------------------------------------------------------------------- #
+
+def _gain(audio: np.ndarray, gain_db: float) -> np.ndarray:
+    return audio * (10.0 ** (gain_db / 20.0))
+
+
+def _time_shift(audio: np.ndarray, max_shift: int) -> np.ndarray:
+    return np.roll(audio, random.randint(-max_shift, max_shift))
+
+
+def _add_noise(audio: np.ndarray, snr_db: float) -> np.ndarray:
+    noise = np.random.normal(0.0, 1.0, len(audio)).astype(np.float32)
+    clean_power = float(np.mean(audio ** 2))
+    noise_power = float(np.mean(noise ** 2))
+    if clean_power < 1.0 or noise_power < 1e-9:
+        return audio
+    target_noise_power = clean_power / (10.0 ** (snr_db / 10.0))
+    noise *= math.sqrt(target_noise_power / noise_power)
+    return audio + noise
+
+
+def _echo(audio: np.ndarray, sr: int) -> np.ndarray:
+    delay = random.randint(int(0.03 * sr), int(0.12 * sr))
+    decay = random.uniform(0.08, 0.25)
+    out = audio.copy()
+    if 0 < delay < len(audio):
+        out[delay:] += audio[:-delay] * decay
+    return out
+
+
+def augment(audio: np.ndarray, sr: int) -> np.ndarray:
+    out = audio.copy()
+    out = _gain(out, random.uniform(-6.0, 3.0))
+    out = _time_shift(out, max_shift=int(0.12 * sr))
+    if random.random() < 0.75:
+        out = _add_noise(out, random.choice([30, 25, 20, 15, 10]))
+    if random.random() < 0.35:
+        out = _echo(out, sr)
+    return normalize(out, 28000.0)
+
+
+# --------------------------------------------------------------------------- #
+# Synthetic background noise
+# --------------------------------------------------------------------------- #
+
+def _white(n: int) -> np.ndarray:
+    return np.random.normal(0.0, 1.0, n).astype(np.float32)
+
+
+def _pink(n: int) -> np.ndarray:
+    white = _white(n)
+    out = np.zeros_like(white)
+    alpha = 0.985
+    for i in range(1, n):
+        out[i] = alpha * out[i - 1] + (1.0 - alpha) * white[i]
+    return out.astype(np.float32)
+
+
+def _brown(n: int) -> np.ndarray:
+    brown = np.cumsum(_white(n))
+    brown = brown - np.mean(brown)
+    return normalize(brown.astype(np.float32), 1.0)
+
+
+def _hum(n: int, sr: int) -> np.ndarray:
+    t = np.arange(n, dtype=np.float32) / float(sr)
+    hum = (
+        np.sin(2.0 * math.pi * 50.0 * t)
+        + 0.5 * np.sin(2.0 * math.pi * 100.0 * t)
+        + 0.25 * np.sin(2.0 * math.pi * 150.0 * t)
+    )
+    hum += 0.04 * _white(n)
+    return hum.astype(np.float32)
+
+
+def _fan(n: int, sr: int) -> np.ndarray:
+    base = _pink(n)
+    t = np.arange(n, dtype=np.float32) / float(sr)
+    blade_rate = random.uniform(18.0, 45.0)
+    modulation = 0.65 + 0.35 * np.sin(2.0 * math.pi * blade_rate * t)
+    return (base * modulation).astype(np.float32)
+
+
+def _cafe(n: int, sr: int) -> np.ndarray:
+    base = 0.55 * _pink(n) + 0.45 * _white(n)
+    transient_count = max(1, int((n / sr) * random.uniform(2.0, 6.0)))
+    for _ in range(transient_count):
+        pos = random.randint(0, max(0, n - 1))
+        length = random.randint(max(1, int(0.008 * sr)), max(2, int(0.05 * sr)))
+        end = min(n, pos + length)
+        if end <= pos:
+            continue
+        click = np.hanning(end - pos).astype(np.float32)
+        base[pos:end] += click * random.uniform(0.5, 2.0)
+    return base.astype(np.float32)
+
+
+def _street(n: int, sr: int) -> np.ndarray:
+    base = 0.7 * _brown(n) + 0.3 * _white(n)
+    t = np.arange(n, dtype=np.float32) / float(sr)
+    for _ in range(random.randint(1, 3)):
+        center = random.uniform(0.2, max(0.21, t[-1] - 0.2))
+        width = random.uniform(0.2, 0.8)
+        envelope = np.exp(-0.5 * ((t - center) / width) ** 2)
+        freq = random.uniform(70.0, 180.0)
+        base += 0.35 * envelope * np.sin(2.0 * math.pi * freq * t)
+    return base.astype(np.float32)
+
+
+def make_background_noise(noise_type: str, num_samples: int, sr: int) -> np.ndarray:
+    if noise_type == "white":
+        noise = _white(num_samples)
+    elif noise_type == "pink":
+        noise = _pink(num_samples)
+    elif noise_type == "brown":
+        noise = _brown(num_samples)
+    elif noise_type == "hum":
+        noise = _hum(num_samples, sr)
+    elif noise_type == "fan":
+        noise = _fan(num_samples, sr)
+    elif noise_type == "cafe":
+        noise = _cafe(num_samples, sr)
+    elif noise_type == "street":
+        noise = _street(num_samples, sr)
+    else:
+        raise ValueError(f"Unknown noise type: {noise_type}")
+    return normalize(noise, random.uniform(6000, 22000))

src/backends/__init__.py

@@ -0,0 +1,61 @@
+"""TTS backend abstraction and automatic backend selection.
+
+The dataset builder is agnostic to which engine produces audio. Two
+backends are provided:
+
+* :class:`~src.backends.gcp.GCPTTSBackend` — Google Cloud TTS via REST
+  API key. Used when a key is available.
+* :class:`~src.backends.piper.PiperTTSBackend` — free, local Piper TTS.
+  Used as the fall back.
+"""
+
+from __future__ import annotations
+
+from typing import List, Optional
+
+from .base import SynthesisResult, TTSBackend, Voice
+from .gcp import GCPTTSBackend
+from .piper import PiperTTSBackend
+
+__all__ = [
+    "SynthesisResult",
+    "TTSBackend",
+    "Voice",
+    "GCPTTSBackend",
+    "PiperTTSBackend",
+    "select_backend",
+]
+
+
+def select_backend(
+    gcp_api_key: Optional[str],
+    language_prefixes: List[str],
+    max_gcp_voices_per_locale: int,
+    max_piper_voices: int,
+    sample_rate_hz: int,
+) -> TTSBackend:
+    """Return a ready TTS backend.
+
+    Prefers Google Cloud TTS when ``gcp_api_key`` is provided and the key
+    validates. Otherwise (no key, or the key fails to initialise) falls
+    back to the free Piper backend.
+    """
+    if gcp_api_key and gcp_api_key.strip():
+        backend = GCPTTSBackend(
+            api_key=gcp_api_key.strip(),
+            language_prefixes=language_prefixes,
+            max_voices_per_locale=max_gcp_voices_per_locale,
+            sample_rate_hz=sample_rate_hz,
+        )
+        try:
+            backend.prepare()
+            return backend
+        except Exception as exc:  # noqa: BLE001 - any failure => fall back
+            print(f"[backend] Google Cloud TTS unavailable ({exc}); falling back to Piper.")
+
+    backend = PiperTTSBackend(
+        max_voices=max_piper_voices,
+        sample_rate_hz=sample_rate_hz,
+    )
+    backend.prepare()
+    return backend

src/backends/base.py

@@ -0,0 +1,48 @@
+"""Abstract TTS backend interface."""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from dataclasses import dataclass
+from typing import List
+
+import numpy as np
+
+
+@dataclass(frozen=True)
+class Voice:
+    """A speaker/voice exposed by a backend."""
+
+    name: str
+    language_code: str
+    description: str = ""
+
+
+@dataclass
+class SynthesisResult:
+    """Output of a single synthesis call: mono float32 audio at a known rate."""
+
+    audio: np.ndarray
+    sample_rate_hz: int
+
+
+class TTSBackend(ABC):
+    """Common interface for every TTS engine."""
+
+    #: Short identifier stored in dataset metadata (e.g. "google_cloud_tts").
+    source: str = "tts"
+
+    @abstractmethod
+    def prepare(self) -> None:
+        """Validate credentials, download models and list voices.
+
+        Must raise if the backend cannot be used.
+        """
+
+    @abstractmethod
+    def voices(self) -> List[Voice]:
+        """Return the selected voices to synthesize with."""
+
+    @abstractmethod
+    def synthesize(self, text: str, voice: Voice) -> SynthesisResult:
+        """Synthesize ``text`` with ``voice`` into mono float32 audio."""

src/backends/gcp.py

@@ -0,0 +1,126 @@
+"""Google Cloud Text-to-Speech backend using an API key (REST API).
+
+Uses the public REST endpoints so the only requirement is an API key,
+which is the simplest credential to provide inside a Hugging Face Space
+(store it as a Space secret named ``GCP_TTS_API_KEY``):
+
+* ``GET  /v1/voices``          — list available voices
+* ``POST /v1/text:synthesize`` — synthesize speech
+
+Requesting ``LINEAR16`` at 16 kHz returns ready-to-use PCM WAV bytes.
+"""
+
+from __future__ import annotations
+
+import base64
+from typing import List, Optional
+
+import requests
+
+from ..audio import read_wav_bytes
+from .base import SynthesisResult, TTSBackend, Voice
+
+_BASE_URL = "https://texttospeech.googleapis.com/v1"
+_TIMEOUT = 30
+
+
+class GCPTTSBackend(TTSBackend):
+    source = "google_cloud_tts"
+
+    def __init__(
+        self,
+        api_key: str,
+        language_prefixes: List[str],
+        max_voices_per_locale: int,
+        sample_rate_hz: int,
+        prefer_voice_types: Optional[List[str]] = None,
+    ) -> None:
+        self.api_key = api_key
+        self.language_prefixes = [p.lower() for p in language_prefixes]
+        self.max_voices_per_locale = max_voices_per_locale
+        self.sample_rate_hz = sample_rate_hz
+        self.prefer_voice_types = prefer_voice_types or [
+            "chirp",
+            "neural",
+            "wavenet",
+            "studio",
+            "journey",
+            "standard",
+        ]
+        self._voices: List[Voice] = []
+
+    # ------------------------------------------------------------------ #
+
+    def prepare(self) -> None:
+        response = requests.get(
+            f"{_BASE_URL}/voices",
+            params={"key": self.api_key},
+            timeout=_TIMEOUT,
+        )
+        if response.status_code != 200:
+            raise RuntimeError(
+                f"Google TTS voices request failed ({response.status_code}): "
+                f"{response.text[:200]}"
+            )
+
+        raw_voices = response.json().get("voices", [])
+        by_locale: dict[str, List[Voice]] = {}
+        for item in raw_voices:
+            name = item.get("name", "")
+            for locale in item.get("languageCodes", []):
+                prefix = locale.split("-")[0].lower()
+                if prefix not in self.language_prefixes:
+                    continue
+                by_locale.setdefault(locale, []).append(
+                    Voice(name=name, language_code=locale, description="google_cloud_tts")
+                )
+
+        def rank(voice: Voice) -> tuple[int, str]:
+            lower = voice.name.lower()
+            for idx, preferred in enumerate(self.prefer_voice_types):
+                if preferred in lower:
+                    return idx, voice.name
+            return 999, voice.name
+
+        selected: List[Voice] = []
+        for locale in sorted(by_locale):
+            ordered = sorted(by_locale[locale], key=rank)
+            selected.extend(ordered[: self.max_voices_per_locale])
+
+        if not selected:
+            raise RuntimeError("Google TTS returned no voices matching the requested locales.")
+        self._voices = selected
+
+    def voices(self) -> List[Voice]:
+        return list(self._voices)
+
+    def synthesize(self, text: str, voice: Voice) -> SynthesisResult:
+        payload = {
+            "input": {"text": text},
+            "voice": {
+                "languageCode": voice.language_code,
+                "name": voice.name,
+            },
+            "audioConfig": {
+                "audioEncoding": "LINEAR16",
+                "sampleRateHertz": self.sample_rate_hz,
+            },
+        }
+        response = requests.post(
+            f"{_BASE_URL}/text:synthesize",
+            params={"key": self.api_key},
+            json=payload,
+            timeout=_TIMEOUT,
+        )
+        if response.status_code != 200:
+            raise RuntimeError(
+                f"Google TTS synthesize failed ({response.status_code}): "
+                f"{response.text[:200]}"
+            )
+
+        audio_b64 = response.json().get("audioContent")
+        if not audio_b64:
+            raise RuntimeError("Google TTS response contained no audioContent.")
+
+        audio, sample_rate = read_wav_bytes(base64.b64decode(audio_b64))
+        return SynthesisResult(audio=audio, sample_rate_hz=sample_rate)

src/backends/piper.py

@@ -0,0 +1,89 @@
+"""Free, local Piper TTS backend.
+
+Voice models are pulled from the ``rhasspy/piper-voices`` repo on the
+Hugging Face Hub and synthesized with the ``piper`` Python package. No
+network access is needed at synthesis time and no paid service is used.
+"""
+
+from __future__ import annotations
+
+import tempfile
+import wave
+from pathlib import Path
+from typing import List, Optional
+
+from ..audio import read_wav_file, resample
+from ..config import PIPER_VOICES
+from .base import SynthesisResult, TTSBackend, Voice
+
+
+class PiperTTSBackend(TTSBackend):
+    source = "piper_tts"
+
+    def __init__(
+        self,
+        max_voices: int,
+        sample_rate_hz: int,
+        cache_dir: Optional[str] = None,
+    ) -> None:
+        self.max_voices = max_voices
+        self.sample_rate_hz = sample_rate_hz
+        self.cache_dir = Path(cache_dir) if cache_dir else Path(tempfile.gettempdir()) / "piper_voices"
+        self._voices: List[Voice] = []
+        self._models: dict[str, "object"] = {}  # voice name -> PiperVoice instance
+
+    # ------------------------------------------------------------------ #
+
+    def prepare(self) -> None:
+        from huggingface_hub import hf_hub_download  # local import: heavy dep
+        from piper import PiperVoice as PiperModel  # local import: heavy dep
+
+        self.cache_dir.mkdir(parents=True, exist_ok=True)
+        selected = PIPER_VOICES[: self.max_voices]
+
+        for spec in selected:
+            try:
+                onnx_path = hf_hub_download(
+                    repo_id="rhasspy/piper-voices",
+                    filename=f"{spec.repo_path}.onnx",
+                    cache_dir=str(self.cache_dir),
+                )
+                config_path = hf_hub_download(
+                    repo_id="rhasspy/piper-voices",
+                    filename=f"{spec.repo_path}.onnx.json",
+                    cache_dir=str(self.cache_dir),
+                )
+                model = PiperModel.load(onnx_path, config_path=config_path)
+                self._models[spec.voice_id] = model
+                self._voices.append(
+                    Voice(
+                        name=spec.voice_id,
+                        language_code=spec.locale,
+                        description=spec.description,
+                    )
+                )
+            except Exception as exc:  # noqa: BLE001 - skip individual bad voices
+                print(f"[piper] Skipping voice {spec.voice_id}: {exc}")
+
+        if not self._voices:
+            raise RuntimeError("No Piper voices could be downloaded or loaded.")
+
+    def voices(self) -> List[Voice]:
+        return list(self._voices)
+
+    def synthesize(self, text: str, voice: Voice) -> SynthesisResult:
+        model = self._models.get(voice.name)
+        if model is None:
+            raise RuntimeError(f"Piper voice not loaded: {voice.name}")
+
+        with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tmp:
+            tmp_path = Path(tmp.name)
+        try:
+            with wave.open(str(tmp_path), "wb") as wav_file:
+                model.synthesize(text, wav_file)
+            audio, src_rate = read_wav_file(tmp_path)
+        finally:
+            tmp_path.unlink(missing_ok=True)
+
+        audio = resample(audio, src_rate, self.sample_rate_hz)
+        return SynthesisResult(audio=audio, sample_rate_hz=self.sample_rate_hz)

src/builder.py

@@ -0,0 +1,244 @@
+"""Dataset builder: orchestrates TTS synthesis, augmentation and layout.
+
+Produces both an Edge Impulse-ready folder layout and the metadata needed
+for a Hugging Face dataset:
+
+    <out_dir>/
+      edge_impulse_upload/
+        training/  <label>.<id>.wav
+        testing/   <label>.<id>.wav
+      by_label/
+        <label>/   <human-readable>.wav
+      metadata.csv
+      selected_voices.csv
+      dataset_summary.json
+"""
+
+from __future__ import annotations
+
+import csv
+import json
+import random
+import shutil
+from dataclasses import asdict, dataclass, field
+from pathlib import Path
+from typing import Callable, Dict, List, Optional
+
+import numpy as np
+
+from . import audio as A
+from .backends import TTSBackend
+from .config import NOISE_TYPES, DatasetConfig
+
+
+ProgressFn = Callable[[str], None]
+
+
+@dataclass
+class BuildResult:
+    out_dir: str
+    backend_source: str
+    total_samples: int
+    label_counts: Dict[str, int]
+    split_counts: Dict[str, int]
+    voices: List[Dict[str, str]]
+    metadata_csv: str
+    summary_json: str
+    generated_base: int = 0
+    generated_augmented: int = 0
+    failed: int = 0
+    warnings: List[str] = field(default_factory=list)
+
+
+def _choose_split(test_ratio: float) -> str:
+    return "testing" if random.random() < test_ratio else "training"
+
+
+def _reset_dirs(out_dir: Path, labels: List[str]) -> None:
+    if out_dir.exists():
+        shutil.rmtree(out_dir)
+    for split in ("training", "testing"):
+        (out_dir / "edge_impulse_upload" / split).mkdir(parents=True, exist_ok=True)
+    for label in labels:
+        (out_dir / "by_label" / label).mkdir(parents=True, exist_ok=True)
+
+
+def build_dataset(
+    config: DatasetConfig,
+    backend: TTSBackend,
+    progress: Optional[ProgressFn] = None,
+) -> BuildResult:
+    def log(message: str) -> None:
+        if progress:
+            progress(message)
+        else:
+            print(message)
+
+    random.seed(config.seed)
+    np.random.seed(config.seed)
+
+    out_dir = Path(config.out_dir).resolve()
+    labels = [config.wake_label, config.unknown_label, config.noise_label]
+    _reset_dirs(out_dir, labels)
+
+    voices = backend.voices()
+    log(f"Backend: {backend.source} with {len(voices)} voice(s).")
+
+    # Persist selected voices.
+    voice_rows = [
+        {"name": v.name, "language_code": v.language_code, "description": v.description}
+        for v in voices
+    ]
+    with (out_dir / "selected_voices.csv").open("w", newline="", encoding="utf-8") as f:
+        writer = csv.DictWriter(f, fieldnames=["name", "language_code", "description"])
+        writer.writeheader()
+        writer.writerows(voice_rows)
+
+    rows: List[Dict[str, object]] = []
+    warnings: List[str] = []
+    generated_base = 0
+    generated_aug = 0
+    failed = 0
+
+    def save_item(
+        audio: np.ndarray,
+        label: str,
+        phrase: str,
+        voice_name: str,
+        locale: str,
+        source: str,
+        augmentation: str,
+    ) -> None:
+        split = _choose_split(config.test_ratio)
+        uid = A.stable_hash(
+            json.dumps(
+                {
+                    "label": label,
+                    "phrase": phrase,
+                    "voice": voice_name,
+                    "locale": locale,
+                    "source": source,
+                    "augmentation": augmentation,
+                    "rand": random.random(),
+                },
+                sort_keys=True,
+            )
+        )
+        filename = f"{label}.{uid}.wav"
+        ei_path = out_dir / "edge_impulse_upload" / split / filename
+        A.write_wav_file(ei_path, audio, config.sample_rate_hz)
+
+        human = f"{label}__{A.slugify(phrase)}__{A.slugify(locale)}__{A.slugify(voice_name)}__{uid}.wav"
+        by_label_path = out_dir / "by_label" / label / human
+        shutil.copy2(ei_path, by_label_path)
+
+        rows.append(
+            {
+                "filepath": str(by_label_path.relative_to(out_dir)),
+                "edge_impulse_filepath": str(ei_path.relative_to(out_dir)),
+                "label": label,
+                "phrase": phrase,
+                "voice_name": voice_name,
+                "language_code": locale,
+                "sample_rate_hz": config.sample_rate_hz,
+                "duration_seconds": config.duration_seconds,
+                "split": split,
+                "source": source,
+                "augmentation": augmentation,
+            }
+        )
+
+    # -- Speech clips ---------------------------------------------------- #
+    phrase_groups = [
+        (config.wake_label, config.wake_phrases),
+        (config.unknown_label, config.unknown_phrases),
+    ]
+    target_samples = config.target_samples
+
+    for voice in voices:
+        for label, phrases in phrase_groups:
+            for phrase in phrases:
+                for _ in range(config.base_repeats_per_phrase_per_voice):
+                    try:
+                        result = backend.synthesize(phrase, voice)
+                        clip = A.resample(result.audio, result.sample_rate_hz, config.sample_rate_hz)
+                        clip = A.pad_or_trim(clip, target_samples)
+                        clip = A.normalize(clip, 24000.0)
+
+                        save_item(clip, label, phrase, voice.name, voice.language_code, backend.source, "original")
+                        generated_base += 1
+
+                        for aug_idx in range(config.augmentations_per_speech_clip):
+                            aug = A.augment(clip, config.sample_rate_hz)
+                            save_item(
+                                aug, label, phrase, voice.name, voice.language_code,
+                                backend.source, f"aug_{aug_idx:02d}",
+                            )
+                            generated_aug += 1
+
+                        if generated_base % 10 == 0:
+                            log(f"Synthesized {generated_base} base clips...")
+                    except Exception as exc:  # noqa: BLE001
+                        failed += 1
+                        msg = f"Failed voice={voice.name} phrase={phrase!r}: {exc}"
+                        warnings.append(msg)
+                        log(f"WARNING: {msg}")
+
+    # -- Background noise ------------------------------------------------ #
+    for _ in range(config.background_noise_samples):
+        noise_type = random.choice(NOISE_TYPES)
+        clip = A.make_background_noise(noise_type, target_samples, config.sample_rate_hz)
+        save_item(clip, config.noise_label, "", "synthetic_noise", "", "synthetic_noise", noise_type)
+
+    # -- Metadata & summary --------------------------------------------- #
+    label_counts: Dict[str, int] = {}
+    split_counts: Dict[str, int] = {}
+    for row in rows:
+        label_counts[row["label"]] = label_counts.get(row["label"], 0) + 1
+        split_counts[row["split"]] = split_counts.get(row["split"], 0) + 1
+
+    metadata_csv = out_dir / "metadata.csv"
+    fieldnames = list(rows[0].keys()) if rows else [
+        "filepath", "edge_impulse_filepath", "label", "phrase", "voice_name",
+        "language_code", "sample_rate_hz", "duration_seconds", "split", "source", "augmentation",
+    ]
+    with metadata_csv.open("w", newline="", encoding="utf-8") as f:
+        writer = csv.DictWriter(f, fieldnames=fieldnames)
+        writer.writeheader()
+        writer.writerows(rows)
+
+    summary = {
+        "dataset": config.dataset_name,
+        "backend": backend.source,
+        "sample_rate_hz": config.sample_rate_hz,
+        "duration_seconds": config.duration_seconds,
+        "total_samples": len(rows),
+        "generated_base_speech_clips": generated_base,
+        "generated_augmented_speech_clips": generated_aug,
+        "background_noise_samples": config.background_noise_samples,
+        "failed_speech_samples": failed,
+        "labels": label_counts,
+        "splits": split_counts,
+        "voices": voice_rows,
+        "wake_phrases": config.wake_phrases,
+        "unknown_phrases": config.unknown_phrases,
+    }
+    summary_json = out_dir / "dataset_summary.json"
+    summary_json.write_text(json.dumps(summary, indent=2), encoding="utf-8")
+
+    log(f"Done. Total samples: {len(rows)} (base={generated_base}, augmented={generated_aug}, failed={failed}).")
+
+    return BuildResult(
+        out_dir=str(out_dir),
+        backend_source=backend.source,
+        total_samples=len(rows),
+        label_counts=label_counts,
+        split_counts=split_counts,
+        voices=voice_rows,
+        metadata_csv=str(metadata_csv),
+        summary_json=str(summary_json),
+        generated_base=generated_base,
+        generated_augmented=generated_aug,
+        failed=failed,
+        warnings=warnings,
+    )

src/config.py

@@ -0,0 +1,153 @@
+"""Configuration, phrase banks and default catalogs for the dataset creator."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import List
+
+
+# --------------------------------------------------------------------------- #
+# Default phrase banks (target phrase: "Hey Android")
+# --------------------------------------------------------------------------- #
+
+DEFAULT_WAKE_PHRASES: List[str] = [
+    "Hey Android",
+    "Hey, Android",
+    "Hello Android",
+    "Okay Android",
+    "OK Android",
+    "Android",
+]
+
+DEFAULT_UNKNOWN_PHRASES: List[str] = [
+    "Hey Andrew",
+    "Hey Adrian",
+    "Hey Andrea",
+    "Hello phone",
+    "Open settings",
+    "Start recording",
+    "Stop recording",
+    "Good morning",
+    "Where is my phone",
+    "Androids are useful",
+    "This is a test",
+    "Turn on the light",
+    "Turn off the light",
+    "Increase volume",
+    "Decrease volume",
+    "Play music",
+    "Pause music",
+]
+
+# The three primary keyword-spotting classes.
+WAKE_LABEL = "hey_android"
+UNKNOWN_LABEL = "unknown"
+NOISE_LABEL = "background_noise"
+
+NOISE_TYPES: List[str] = ["white", "pink", "brown", "hum", "fan", "cafe", "street"]
+
+
+# --------------------------------------------------------------------------- #
+# Piper voice catalog (free, downloaded from rhasspy/piper-voices on the Hub)
+# --------------------------------------------------------------------------- #
+
+@dataclass(frozen=True)
+class PiperVoice:
+    """A downloadable Piper voice model on the Hugging Face Hub."""
+
+    voice_id: str
+    repo_path: str  # path inside rhasspy/piper-voices (without extension)
+    locale: str
+    description: str
+
+
+PIPER_VOICES: List[PiperVoice] = [
+    PiperVoice(
+        "en_US-lessac-medium",
+        "en/en_US/lessac/medium/en_US-lessac-medium",
+        "en-US",
+        "English US",
+    ),
+    PiperVoice(
+        "en_GB-alba-medium",
+        "en/en_GB/alba/medium/en_GB-alba-medium",
+        "en-GB",
+        "English GB",
+    ),
+    PiperVoice(
+        "en_GB-northern_english_male-medium",
+        "en/en_GB/northern_english_male/medium/en_GB-northern_english_male-medium",
+        "en-GB",
+        "English GB Northern male",
+    ),
+    PiperVoice(
+        "nl_NL-mls-medium",
+        "nl/nl_NL/mls/medium/nl_NL-mls-medium",
+        "nl-NL",
+        "Dutch NL",
+    ),
+    PiperVoice(
+        "de_DE-thorsten-medium",
+        "de/de_DE/thorsten/medium/de_DE-thorsten-medium",
+        "de-DE",
+        "German",
+    ),
+    PiperVoice(
+        "fr_FR-siwis-medium",
+        "fr/fr_FR/siwis/medium/fr_FR-siwis-medium",
+        "fr-FR",
+        "French",
+    ),
+    PiperVoice(
+        "es_ES-sharvard-medium",
+        "es/es_ES/sharvard/medium/es_ES-sharvard-medium",
+        "es-ES",
+        "Spanish",
+    ),
+]
+
+
+# --------------------------------------------------------------------------- #
+# Dataset generation config
+# --------------------------------------------------------------------------- #
+
+@dataclass
+class DatasetConfig:
+    """All knobs controlling a single dataset build."""
+
+    # Output
+    out_dir: str = "output"
+    dataset_name: str = "hey_android"
+
+    # Audio
+    sample_rate_hz: int = 16000
+    duration_seconds: float = 2.0
+
+    # Classes / phrases
+    wake_label: str = WAKE_LABEL
+    unknown_label: str = UNKNOWN_LABEL
+    noise_label: str = NOISE_LABEL
+    wake_phrases: List[str] = field(default_factory=lambda: list(DEFAULT_WAKE_PHRASES))
+    unknown_phrases: List[str] = field(default_factory=lambda: list(DEFAULT_UNKNOWN_PHRASES))
+
+    # Size controls
+    base_repeats_per_phrase_per_voice: int = 1
+    augmentations_per_speech_clip: int = 8
+    background_noise_samples: int = 200
+    max_piper_voices: int = 7
+    max_gcp_voices_per_locale: int = 3
+
+    # GCP voice selection
+    gcp_language_prefixes: List[str] = field(
+        default_factory=lambda: ["en", "nl", "de", "fr", "es"]
+    )
+
+    # Split
+    test_ratio: float = 0.2
+
+    # Reproducibility
+    seed: int = 1337
+
+    @property
+    def target_samples(self) -> int:
+        return int(self.sample_rate_hz * self.duration_seconds)

src/edge_impulse.py

@@ -0,0 +1,126 @@
+"""Upload a generated dataset directly to an Edge Impulse project.
+
+Uses the Edge Impulse ingestion REST API with a project API key
+(Project → Dashboard → Keys). WAV filenames use the ``label.<id>.wav``
+convention, so Edge Impulse auto-assigns labels from the filename prefix.
+
+Ingestion endpoints:
+    POST https://ingestion.edgeimpulse.com/api/training/files
+    POST https://ingestion.edgeimpulse.com/api/testing/files
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Callable, List, Optional
+
+import requests
+
+_INGESTION_URL = "https://ingestion.edgeimpulse.com/api/{category}/files"
+_TIMEOUT = 60
+_BATCH_SIZE = 25
+
+ProgressFn = Callable[[str], None]
+
+
+@dataclass
+class UploadResult:
+    uploaded: int = 0
+    failed: int = 0
+    errors: List[str] = field(default_factory=list)
+
+
+def verify_api_key(api_key: str) -> bool:
+    """Return True if the ingestion API accepts the key.
+
+    A key with no files still returns a non-auth error, so we treat any
+    non-401/403 response as "key looks usable".
+    """
+    if not api_key or not api_key.strip():
+        return False
+    try:
+        response = requests.post(
+            _INGESTION_URL.format(category="training"),
+            headers={"x-api-key": api_key.strip()},
+            timeout=_TIMEOUT,
+        )
+    except requests.RequestException:
+        return False
+    return response.status_code not in (401, 403)
+
+
+def _upload_batch(
+    category: str,
+    api_key: str,
+    wav_paths: List[Path],
+    allow_duplicates: bool,
+) -> tuple[int, Optional[str]]:
+    headers = {"x-api-key": api_key}
+    if not allow_duplicates:
+        headers["x-disallow-duplicates"] = "1"
+
+    files = []
+    handles = []
+    try:
+        for path in wav_paths:
+            handle = path.open("rb")
+            handles.append(handle)
+            files.append(("data", (path.name, handle, "audio/wav")))
+        response = requests.post(
+            _INGESTION_URL.format(category=category),
+            headers=headers,
+            files=files,
+            timeout=_TIMEOUT,
+        )
+    finally:
+        for handle in handles:
+            handle.close()
+
+    if response.status_code == 200:
+        return len(wav_paths), None
+    return 0, f"HTTP {response.status_code}: {response.text[:200]}"
+
+
+def upload_dataset(
+    dataset_dir: str,
+    api_key: str,
+    allow_duplicates: bool = False,
+    progress: Optional[ProgressFn] = None,
+) -> UploadResult:
+    """Upload the ``edge_impulse_upload/{training,testing}`` WAVs to a project."""
+    def log(message: str) -> None:
+        if progress:
+            progress(message)
+        else:
+            print(message)
+
+    api_key = (api_key or "").strip()
+    if not api_key:
+        raise ValueError("An Edge Impulse API key is required to upload.")
+
+    base = Path(dataset_dir) / "edge_impulse_upload"
+    result = UploadResult()
+
+    for src_split, category in (("training", "training"), ("testing", "testing")):
+        split_dir = base / src_split
+        if not split_dir.exists():
+            continue
+        wavs = sorted(split_dir.glob("*.wav"))
+        if not wavs:
+            continue
+
+        log(f"Uploading {len(wavs)} {category} file(s) to Edge Impulse...")
+        for start in range(0, len(wavs), _BATCH_SIZE):
+            batch = wavs[start:start + _BATCH_SIZE]
+            uploaded, error = _upload_batch(category, api_key, batch, allow_duplicates)
+            if error is None:
+                result.uploaded += uploaded
+                log(f"  {category}: {result.uploaded} uploaded")
+            else:
+                result.failed += len(batch)
+                result.errors.append(f"{category} batch @ {start}: {error}")
+                log(f"  WARNING: {category} batch failed: {error}")
+
+    log(f"Edge Impulse upload complete: {result.uploaded} uploaded, {result.failed} failed.")
+    return result

src/hf_export.py

@@ -0,0 +1,168 @@
+"""Hugging Face dataset export: audio layout, metadata, card, optional push."""
+
+from __future__ import annotations
+
+import csv
+import json
+import shutil
+from pathlib import Path
+from typing import Dict, List, Optional
+
+from .builder import BuildResult
+from .config import DatasetConfig
+
+
+def _dataset_card(config: DatasetConfig, result: BuildResult, repo_id: str) -> str:
+    labels_table = "\n".join(
+        f"| `{label}` | {count} |" for label, count in sorted(result.label_counts.items())
+    )
+    return f"""---
+license: cc-by-4.0
+pretty_name: {config.dataset_name} wake word synthetic speech dataset
+task_categories:
+- audio-classification
+tags:
+- audio
+- speech
+- wake-word
+- keyword-spotting
+- text-to-speech
+- synthetic-data
+- edge-impulse
+- tinyml
+size_categories:
+- n<1K
+---
+
+# {config.dataset_name} — Wake Word Synthetic Speech Dataset
+
+Synthetic, augmented audio for training a small wake-word / keyword-spotting
+model. Generated with **{result.backend_source}** and local audio augmentation.
+
+## Classes
+
+| Label | Samples |
+|---|---|
+{labels_table}
+
+- `{config.wake_label}` — the target wake phrase and close variants.
+- `{config.unknown_label}` — near-miss and unrelated short phrases.
+- `{config.noise_label}` — synthetic background noise.
+
+## Audio Specification
+
+| Property | Value |
+|---|---|
+| Format | WAV |
+| Channels | Mono |
+| Sample rate | {config.sample_rate_hz} Hz |
+| Clip length | {config.duration_seconds} seconds |
+
+## Layout
+
+```text
+audio/
+  train/  {config.wake_label}.<id>.wav ...
+  test/   {config.wake_label}.<id>.wav ...
+metadata.csv
+hf_metadata.csv
+selected_voices.csv
+dataset_summary.json
+```
+
+## Loading
+
+```python
+from datasets import load_dataset, Audio
+
+ds = load_dataset("{repo_id}")
+ds = ds.cast_column("audio", Audio(sampling_rate={config.sample_rate_hz}))
+print(ds)
+```
+
+## Edge Impulse
+
+Filenames follow the Edge Impulse label-prefix convention
+(`{config.wake_label}.<id>.wav`) so they upload directly:
+
+```bash
+edge-impulse-uploader --category training audio/train/*.wav
+edge-impulse-uploader --category testing audio/test/*.wav
+```
+
+## Limitations
+
+Synthetic TTS is a bootstrap, not a production benchmark. Add real device
+and environment recordings before deploying a wake-word product.
+
+## License
+
+CC BY 4.0. Verify that your use of the generated synthetic speech complies
+with the terms of the voice models and tools used to create it.
+"""
+
+
+def export_hf_dataset(
+    config: DatasetConfig,
+    result: BuildResult,
+    hf_dir: str,
+    repo_id: str = "your-username/your-dataset",
+) -> str:
+    """Build a Hugging Face-ready folder from a completed build. Returns its path."""
+    source_dir = Path(result.out_dir)
+    hf_path = Path(hf_dir).resolve()
+    if hf_path.exists():
+        shutil.rmtree(hf_path)
+    (hf_path / "audio" / "train").mkdir(parents=True, exist_ok=True)
+    (hf_path / "audio" / "test").mkdir(parents=True, exist_ok=True)
+
+    for src_split, hf_split in (("training", "train"), ("testing", "test")):
+        src = source_dir / "edge_impulse_upload" / src_split
+        dst = hf_path / "audio" / hf_split
+        for wav in sorted(src.glob("*.wav")):
+            shutil.copy2(wav, dst / wav.name)
+
+    for name in ("metadata.csv", "selected_voices.csv", "dataset_summary.json"):
+        src = source_dir / name
+        if src.exists():
+            shutil.copy2(src, hf_path / name)
+
+    # HF metadata.csv-style index (audio path + label).
+    hf_rows: List[Dict[str, str]] = []
+    for split in ("train", "test"):
+        for wav in sorted((hf_path / "audio" / split).glob("*.wav")):
+            hf_rows.append(
+                {
+                    "audio": str(wav.relative_to(hf_path)),
+                    "label": wav.name.split(".")[0],
+                    "split": split,
+                    "filename": wav.name,
+                }
+            )
+    with (hf_path / "hf_metadata.csv").open("w", newline="", encoding="utf-8") as f:
+        writer = csv.DictWriter(f, fieldnames=["audio", "label", "split", "filename"])
+        writer.writeheader()
+        writer.writerows(hf_rows)
+
+    (hf_path / "README.md").write_text(
+        _dataset_card(config, result, repo_id), encoding="utf-8"
+    )
+    (hf_path / "hf_dataset_summary.json").write_text(
+        json.dumps({"total_wavs": len(hf_rows), "repo_id": repo_id}, indent=2),
+        encoding="utf-8",
+    )
+    return str(hf_path)
+
+
+def push_to_hub(hf_dir: str, repo_id: str, token: str, private: bool = False) -> str:
+    """Upload the HF dataset folder to the Hub. Returns the dataset URL."""
+    from huggingface_hub import HfApi
+
+    api = HfApi(token=token)
+    api.create_repo(repo_id=repo_id, repo_type="dataset", exist_ok=True, private=private)
+    api.upload_folder(
+        folder_path=hf_dir,
+        repo_id=repo_id,
+        repo_type="dataset",
+    )
+    return f"https://huggingface.co/datasets/{repo_id}"