---
pipeline_tag: audio-classification
---
This model is an ONNX version of a YAMNet-based classifier trained to recognize environmental and human-made sounds (e.g., speech, music, silence, barking, etc.).

Format: ONNX (.onnx)

Input: float32 mono audio, sampled at 16,000 Hz

Output: Score matrix per class [num_frames, num_classes]

Classes: See mapping in yamnet_class_map.csv

Example for uses
```python

import sounddevice as sd
import numpy as np
import onnxruntime
import scipy.signal
import csv
import threading
import time
from collections import deque, Counter

# Path to the ONNX model file
MODEL_PATH = "./yamnet.onnx"
# Create an inference session with ONNX runtime using CPU provider
session = onnxruntime.InferenceSession(MODEL_PATH, providers=['CPUExecutionProvider'])
# Get the name of the model's input node
input_name = session.get_inputs()[0].name

def load_class_map(csv_path="yamnet_class_map.csv"):
    """Load class names from a CSV file into a dictionary mapping class IDs to names."""
    class_map = {}
    with open(csv_path, newline='') as csvfile:
        reader = csv.reader(csvfile)
        next(reader)  # Skip header row
        for row in reader:
            class_id = int(row[0])
            class_name = row[2]  # Class name is in the third column
            class_map[class_id] = class_name
    return class_map

# Load the class mapping from CSV
class_map = load_class_map()

# Audio buffer settings (1.5 seconds at 16kHz sample rate)
BUFFER_SIZE = int(16000 * 1.5) 
# Circular buffer to store audio samples
audio_buffer = deque(maxlen=BUFFER_SIZE)

# Queues for storing and consolidating detection results
detections_queue = deque(maxlen=3)  # Stores recent detections
consolidated_results = deque(maxlen=3)  # Stores consolidated results
last_printed_result = None  # Last result printed to console
last_inference_time = 0  # Timestamp of last inference

# Configuration for sleep mode (temporarily pausing inference)
sleep_triggers = {"music", "speech", "silence"}  # Classes that trigger sleep mode
sleep_duration = 3.0  # How long to sleep after detecting trigger classes
same_class_count = 0  # Counter for consecutive same-class detections
last_detected_class = None  # Last class detected
is_sleeping_until = 0  # Timestamp until which we're in sleep mode

def resample_if_needed(audio, original_sr, target_sr=16000):
    """Resample audio to target sample rate if needed."""
    if original_sr != target_sr:
        audio = scipy.signal.resample_poly(audio, target_sr, int(original_sr))
    return audio

def run_inference(audio_chunk):
    """Run the ONNX model inference on an audio chunk and process results."""
    global detections_queue, consolidated_results, last_printed_result
    global is_sleeping_until, same_class_count, last_detected_class

    try:
        # Run the ONNX model
        outputs = session.run(None, {input_name: audio_chunk.astype(np.float32)})
        scores = outputs[0]
        mean_scores = np.mean(scores, axis=0)  # Average scores across frames

        # Get the class with highest score
        class_id = np.argmax(mean_scores)
        confidence = mean_scores[class_id]
        class_name = class_map.get(class_id, f"Class {class_id}")

        # Track consecutive detections of the same class
        if class_name == last_detected_class:
            same_class_count += 1
        else:
            same_class_count = 1
            last_detected_class = class_name

        # Enter sleep mode if we detect trigger classes consecutively
        if same_class_count >= 3 and class_name in sleep_triggers:
            is_sleeping_until = time.time() + sleep_duration
            print(f" Pausing inferences for {sleep_duration} seconds (detected: {class_name})")
            same_class_count = 0

        # Add detection to queue
        detections_queue.append((class_name, confidence))

        # When queue is full, consolidate results
        if len(detections_queue) == 1:
            most_common_name, _ = Counter([d[0] for d in detections_queue]).most_common(1)[0]
            relevant_conf = [conf for n, conf in detections_queue if n == most_common_name]
            avg_conf = np.mean(relevant_conf) * 100
            if avg_conf > 20:  # Only consider results with >20% confidence
                consolidated_results.append((most_common_name, avg_conf))
            detections_queue.clear()

        # When we have enough consolidated results, print the final result
        if len(consolidated_results) == 3:
            names = [r[0] for r in consolidated_results]
            confidences = [r[1] for r in consolidated_results]
            most_common_name, _ = Counter(names).most_common(1)[0]
            avg_conf = np.mean([c for n, c in consolidated_results if n == most_common_name])
            msg = f" Consolidated result: {most_common_name} with average confidence {avg_conf:.2f}%"
            if msg != last_printed_result:
                print(msg)
                last_printed_result = msg

    except Exception as e:
        print(f" ONNX Error: {e}")

def audio_callback(indata, frames, time_info, status):
    """Callback function for audio input stream."""
    global last_inference_time, is_sleeping_until

    if status:
        print(f" Status: {status}")

    # Get mono audio and resample if needed
    audio = indata[:, 0]
    original_sr = sd.query_devices(sd.default.device[0], 'input')['default_samplerate']
    audio = resample_if_needed(audio, original_sr)

    # Normalize audio
    max_val = np.max(np.abs(audio))
    if max_val > 0:
        audio = audio / max_val

    # Add audio to buffer
    audio_buffer.extend(audio)

    # Run inference if buffer is full and not in sleep mode
    if len(audio_buffer) >= BUFFER_SIZE:
        now = time.time()

        if now < is_sleeping_until:
            return

        # Throttle inference to once per second
        if now - last_inference_time > 1.0:
            last_inference_time = now
            audio_chunk = np.array(audio_buffer)
            # Run inference in a separate thread
            threading.Thread(target=run_inference, args=(audio_chunk,), daemon=True).start()

def main():
    """Main function to start audio streaming and processing."""
    print(" Starting real-time listening (Ctrl+C to stop)...")

    # List available input devices
    devices = sd.query_devices()
    input_devices = [i for i, d in enumerate(devices) if d['max_input_channels'] > 0]

    print("\nAvailable input devices:")
    for i in input_devices:
        print(f"{i}: {devices[i]['name']}")

    # Let user select input device
    device_id = int(input("Select input device ID: "))
    sd.default.device = (device_id, None)

    # Configure audio settings
    samplerate = int(sd.query_devices(device_id, 'input')['default_samplerate'])
    sd.default.samplerate = samplerate
    sd.default.channels = 1  # Mono audio

    print("🎧 Listening...")

    # Start audio stream with callback
    with sd.InputStream(callback=audio_callback, channels=1, samplerate=samplerate):
        try:
            while True:
                time.sleep(0.1)  # Keep main thread alive
        except KeyboardInterrupt:
            print("\n Stopping listener...")

if __name__ == "__main__":
    main()

```