A distributed, fault-tolerant speech data acquisition and annotation infrastructure system designed to transform raw audio/video sources into validated, training-ready ASR/TTS datasets with diarization, emotion tags, style tags, language labels, consensus quality scores, and structured manifests.
The platform is designed around a multi-stage, distributed architecture managed by Celery queues and persisted in a durable PostgreSQL/SQLite state store.
┌──────────────────────┐
│ FastAPI API / UI │
└──────────┬───────────┘
│ (JWT / API Key Protected)
▼
┌──────────────────────┐
│ Redis Broker │
└──────────┬───────────┘
│
┌────────────────────────┼────────────────────────┐
▼ (cpu_queue) ▼ (gpu_queue) ▼ (cpu_queue)
┌───────────┐ ┌───────────┐ ┌───────────┐
│ Celery │ │ Celery │ │ Celery │
│ Worker │ │ Worker │ │ Worker │
│ (Ingest) │ │ (VAD) │ │ (Export) │
└─────┬─────┘ └─────┬─────┘ └─────┬─────┘
│ │ │
▼ ▼ ▼
┌───────────────┐ ┌───────────────┐ ┌───────────────┐
│ S3/MinIO Obj │ │ Postgres DB │ │DatasetRegistry│
│ (Audio Files) │ │ (Metadata/QA) │ │(Versioned Rel)│
└───────────────┘ └───────────────┘ └───────────────┘
EdgepulseAI/
├── pipeline/
│ ├── db.py ← SQLite database connection lifecycle & context managers
│ ├── logger.py ← Structured JSON logging for ELK stack
│ ├── ingest.py ← Ingestion stage via yt-dlp with exponential retries
│ ├── process.py ← FFmpeg normalization stage converting to 16kHz mono WAV
│ ├── metadata.py ← Metadata extraction stage (LID via Whisper + duration)
│ ├── segment.py ← Voice Activity Detection (Silero VAD + RMS fallback)
│ ├── diarize.py ← Speaker clustering (Pyannote + acoustic fallback)
│ ├── style_emotion.py ← Expression & style labeling (wav2vec2 + heuristics)
│ ├── validate.py ← Quality score checking (overlap, SNR, confidence boundaries)
│ ├── export.py ← Export stage (JSON, CSV, RTTM, manifest.jsonl)
│ └── dataset_stats.py ← Stats compiler tracking stage elapsed-time benchmarks
│
├── edgepulse/
│ ├── storage/
│ │ ├── base.py ← Abstract interface for Object Storage
│ │ ├── local.py ← Local file implementation of ObjectStorageBackend
│ │ ├── s3.py ← AWS S3 / MinIO backend driver
│ │ └── job_store/
│ │ ├── base.py ← Abstract base interface for state & registry storage
│ │ ├── sqlite_store.py ← SQLite JobStore engine with explicit lock managers
│ │ └── postgres_store.py ← Production Postgres state & registry engine
│ │
│ ├── workers/
│ │ ├── celery_app.py ← Celery queue configuration (CPU vs GPU split, Redis broker)
│ │ └── tasks.py ← Decoupled tasks defining sequential pipeline execution
│ │
│ ├── models/
│ │ ├── lid.py ← Code-mixed Whisper LID and Romanized script classifier
│ │ └── speaker_graph.py ← ECAPA-TDNN speaker verification and global matching
│ │
│ ├── services/
│ │ ├── validate.py ← Three-factor Consensus Quality Engine
│ │ ├── optimizer.py ← Rarity, diversity, and quality optimization algorithm
│ │ ├── registry.py ← Release registry tracking version lifecycle states
│ │ ├── feedback.py ← Active learning calibrator with L1 data drift metrics
│ │ └── reproducibility.py ← SHA-256 config, model, and manifest verification
│ │
│ ├── api/
│ │ ├── routes.py ← API routes (RBAC protected, SlowAPI rate-limited, /health endpoint)
│ │ └── security.py ← JWT Token (32-byte secret) & API Key authentication layer
│ │
│ ├── observability/
│ │ └── prometheus_exporter.py ← Prometheus metrics exposing gauges and counters
│ │
│ └── tests/
│ ├── load_testing/
│ │ └── locustfile.py ← Locust load testing script tracking job completion latency
│ │
│ ├── test_platform.py ← Base platform tests (lineage, consensus quality)
│ ├── test_celery_workers.py ← Workers execution and stage updates verification
│ ├── test_active_learning.py ← TVD drift calculations and threshold calibrations
│ └── test_security_hardened.py ← JWT token generation, RBAC, /health, and rate limits
│
├── grafana/
│ └── provisioning/
│ ├── datasources/
│ │ └── datasource.yaml ← Prometheus datasource configuration
│ └── dashboards/
│ ├── dashboards.yaml ← Dashboards provider configuration
│ └── pipeline_dashboard.json ← Provisioned monitoring charts
│
├── infra/
│ └── terraform/
│ ├── main.tf ← VPC, ALB, Route53, S3, RDS PostgreSQL, ElastiCache Redis, and ECS Fargate
│ ├── variables.tf ← Deployment parameter variables
│ └── outputs.tf ← Connection outputs for RDS, Redis, and S3
│
├── app.py ← FastAPI server instance attaching routers & metrics mounts
├── run_pipeline.py ← Main orchestrator CLI (supporting --workers and --resume)
├── config.yaml ← Centralized model IDs, parameters, and thresholds
├── Dockerfile ← Container configuration installing FFmpeg and packages
├── docker-compose.yml ← Services definition mapping persistent database paths
└── requirements.txt ← Pinned pipeline dependencies
- Distributed Queue Execution: Workloads are completely offloaded from FastAPI background threads into an isolated Celery task cluster, separated into
cpu_queue(ingest, transcode, metadata, validate, export) andgpu_queue(VAD, diarization, style/emotion tagging). - Durable Result Backends: Tasks use a transient Redis broker for routing, but persist final result records directly in a PostgreSQL/JobStore database.
- Drift-Aware Active Learning: Evaluates label Total Variation Distance (TVD) to raise quality validation score thresholds under data drift alerts.
- Security Hardening: Protected by a 32-byte key JWT signature, SlowAPI rate limiters (10/min), and Role-Based Access Control (
admin,reviewer,user). - Declarative Grafana Provisioning: Auto-loaded charts visualizing queue depths, completed/failed counters, consensus quality scores, and active learning drift metrics.
Run the full platform verification test suite:
python -m unittest discover -s edgepulse/tests -p "test_*.py"Run Locust load testing:
locust -f tests/load_testing/locustfile.py --headless -u 50 -r 5 --run-time 2mRajveer Singh Saggu
Distributed under the MIT License. See LICENSE for more information.