Inferencia en Tiempo Real

Conoce cómo funciona el análisis polifónico en tiempo real ejecutado en el dispositivo del usuario y el cálculo matemático de proximidad.

SoundGuard incorpora un motor de inferencia híbrido diseñado para alertar instantáneamente a personas con pérdida auditiva sobre eventos críticos del entorno (sirenas, alarmas, cláxones). Para cumplir con este objetivo, ofrecemos una solución dual que balancea el procesamiento en el dispositivo y la nube.

Arquitectura Híbrida: WASM vs. Cloud

Ofrecemos el procesamiento de sonido bajo dos esquemas complementarios:

Característica	Procesamiento Local (WASM Edge)	Clasificación Cloud (API REST)
Tecnología	MediaPipe AudioClassifier + YAMNet	AWS Lambda + Bedrock Nova + k-NN
Latencia	Cero (Inferencia síncrona en ~260 ms)	~600 ms (Inferencia REST)
Privacidad	Total (el audio no sale del dispositivo)	El audio se transmite al backend en la nube
Costo	Gratuito e ilimitado (corre en el cliente)	Facturación por invocación de API y modelo
Conectividad	Funciona 100% offline (tras descargar modelo)	Requiere conexión estable a Internet
Uso Ideal	Alertas en vivo y streaming continuo	Análisis detallado de archivos y archivo histórico

Modelo de Proximidad Acústica: Cálculo de TTC

Para sirenas y sonidos críticos urbanos, no basta con identificar la clase de sonido; también es indispensable estimar si la fuente de peligro (como una ambulancia o camión de bomberos) se está acercando rápidamente.

SoundGuard utiliza el cálculo de Tiempo para la Colisión (Time-to-Collision - TTC) basado en la intensidad física del sonido:

Ley del Inverso del Cuadrado: Un acercamiento a velocidad constante produce un incremento exponencial de la presión sonora, lo cual se traduce en un aumento lineal en la escala de decibelios ($dB$).
Ventana de Historial Móvil: El sistema almacena una ventana móvil de decibelios en los últimos $1.5$ segundos en la banda de frecuencia del sonido crítico detectado.
Regresión Lineal por Mínimos Cuadrados: Calculamos la tasa de cambio del volumen en el tiempo ($m = dB/s$) mediante la ecuación de la pendiente:

m = [n * sum(t * dB) - sum(t) * sum(dB)] / [n * sum(t^2) - (sum(t))^2]
Fórmula Física del TTC: Estimamos el tiempo en segundos hasta que la fuente sonora alcance el punto cero del usuario:

TTC = 8.686 / m
Acción Coercitiva: Si la pendiente es positiva ($m > 1.6$ dB/s), el TTC es inferior a $4.5$ segundos y el volumen supera los $48$ dB, se dispara una alerta de Peligro Inminente que activa la API de vibración física del dispositivo (navigator.vibrate).

¿Cómo implementarlo como Endpoint / Streaming en Backend?

Si estás construyendo una aplicación móvil nativa o integrando SoundGuard en un dispositivo IoT de hardware limitado (como placas de microcontroladores sin soporte para JS/WASM pesado):

Se puede ofrecer un servicio homólogo de streaming en el backend utilizando WebSockets y un microservicio en Python:

Ejemplo de Arquitectura de Streaming con WebSockets

sequenceDiagram
    participant IoT as Dispositivo IoT / App
    participant WS as WebSocket Server (FastAPI)
    participant Model as Clasificador Polifónico (YAMNet)
    
    IoT->>WS: Conexión WebSocket (WSS)
    Note over IoT,WS: Envío continuo de buffers de audio Float32 (16kHz)
    loop Cada 500ms
        IoT->>WS: Enviar Binary Frame (8000 muestras)
        WS->>Model: Pasar buffer a YAMNet (Inferencia local Python)
        Model-->>WS: Predicciones concurrentes y decibelios por banda
        WS-->>IoT: JSON con clases detectadas + TTC
    end

Código Base del Servidor (FastAPI + TensorFlow YAMNet)

from fastapi import FastAPI, WebSocket
import numpy as np
import tensorflow as tf
import tensorflow_hub as hub

app = FastAPI()
# Cargar YAMNet en memoria una sola vez
yamnet_model = hub.load('https://tfhub.dev/google/yamnet/1')

@app.websocket("/ws/stream")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    # Buffer circular en memoria
    audio_buffer = np.array([], dtype=np.float32)
    
    try:
        while True:
            # Recibir datos binarios Float32 del micrófono
            data = await websocket.receive_bytes()
            samples = np.frombuffer(data, dtype=np.float32)
            audio_buffer = np.concatenate((audio_buffer, samples))
            
            # YAMNet requiere ventanas mínimas de 0.975s (15600 muestras a 16kHz)
            if len(audio_buffer) >= 15600:
                # Tomar la última ventana
                input_data = audio_buffer[-15600:]
                
                # Ejecutar clasificación
                scores, embeddings, spectrogram = yamnet_model(input_data)
                mean_scores = np.mean(scores, axis=0)
                
                # Obtener clases con score alto
                top_indices = np.argsort(mean_scores)[::-1][:3]
                results = []
                for idx in top_indices:
                    prob = float(mean_scores[idx])
                    if prob > 0.05:
                        results.append({"class_index": int(idx), "prob": prob})
                
                # Devolver las clases encontradas
                await websocket.send_json({"detected": results})
                
                # Reducir buffer circular para mantener rendimiento
                audio_buffer = audio_buffer[-16000:]
    except Exception as e:
        print(f"Desconexión o error en stream: {e}")
    finally:
        await websocket.close()

Esta solución es totalmente compatible con la lógica del cliente y permite descargar la CPU local a un servidor escalable.