Practical case: Raspberry Pi Noise Traffic Light

#!/usr/bin/env python3
"""
noise_traffic_light.py
Monitors ambient noise via an I2S microphone and controls a NeoPixel ring.
Includes a dry-run mode for educational validation on standard PCs.
"""

import sys
import math
import struct
import time
import argparse
import logging

# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

# Hardware libraries (wrapped in try-except for dry-run compatibility)
try:
    import pyaudio
    HAS_PYAUDIO = True
except ImportError:
    HAS_PYAUDIO = False

try:
    import board
    import neopixel
    HAS_NEOPIXEL = True
except ImportError:
    HAS_NEOPIXEL = False


class MockAudioStream:
    """Generates synthetic PCM audio data to simulate a fluctuating noise environment."""
    def __init__(self, chunk_size, rate):
        self.chunk_size = chunk_size
        self.rate = rate
        self.start_time = time.time()

    def read(self, num_frames, exception_on_overflow=False):
        # Simulate elapsed time
        time.sleep(num_frames / self.rate)
        t = time.time() - self.start_time

        # Create a pulsing envelope: goes from 10% to 100% of max amplitude over 10 seconds
        envelope = 0.1 + 0.9 * ((math.sin(t * 2 * math.pi / 10) + 1) / 2)
        max_amp = 32767 * envelope

        data = bytearray()
        for i in range(num_frames):
            # Generate a 440Hz sine wave sample
            sample = int(max_amp * math.sin(2 * math.pi * 440 * (i / self.rate)))
            # Pack as 16-bit signed integer (little-endian)
            data.extend(struct.pack('<h', sample))
        return bytes(data)

    def close(self):
        pass


class MockStatusRing:
    """Simulates a NeoPixel ring by printing colors to the console."""
    def __init__(self, num_pixels):
        self.num_pixels = num_pixels
        self.current_color = None

    def fill(self, color):
        if color != self.current_color:
            self.current_color = color
            logging.info(f"[MOCK LED] Ring color changed to RGB: {color}")


class RealAudioStream:
    """Wrapper for the PyAudio I2S stream."""
    def __init__(self, chunk_size, rate):
        self.pa = pyaudio.PyAudio()
        self.stream = self.pa.open(
            format=pyaudio.paInt16,
            channels=1,
            rate=rate,
            input=True,
            frames_per_buffer=chunk_size
        )

    def read(self, num_frames, exception_on_overflow=False):
        return self.stream.read(num_frames, exception_on_overflow=exception_on_overflow)

    def close(self):
        self.stream.stop_stream()
        self.stream.close()
        self.pa.terminate()


class RealStatusRing:
    """Wrapper for the hardware NeoPixel ring."""
    def __init__(self, pin, num_pixels):
        # board.D12 corresponds to GPIO 12
        self.pixels = neopixel.NeoPixel(pin, num_pixels, brightness=0.5, auto_write=True)

    def fill(self, color):
        self.pixels.fill(color)


def calculate_dbfs(pcm_data):
    """
    Calculates the Decibels Relative to Full Scale (dBFS) of a PCM audio chunk.
    0 dBFS is the maximum possible digital value. Normal audio is negative.
    """
    # 16-bit audio = 2 bytes per frame
    count = len(pcm_data) // 2
    if count == 0:
        return -96.0

    # Unpack raw bytes into signed 16-bit integers
    shorts = struct.unpack(f'<{count}h', pcm_data)

    # Calculate Root Mean Square (RMS)
    sum_squares = sum(s * s for s in shorts)
    rms = math.sqrt(sum_squares / count)

    if rms == 0:
        return -96.0

    # Calculate dBFS (Max RMS for 16-bit signed is approx 32768)
    dbfs = 20 * math.log10(rms / 32768.0)
    return dbfs


def main():
    parser = argparse.ArgumentParser(description="Workshop Noise Traffic Light")
    parser.add_argument("--dry-run", action="store_true", help="Run with mock hardware interfaces")
    parser.add_argument("--threshold-yellow", type=float, default=-25.0, help="dBFS threshold for Yellow")
    parser.add_argument("--threshold-red", type=float, default=-15.0, help="dBFS threshold for Red")
    args = parser.parse_args()

    # Configuration constants
    CHUNK = 4096
    RATE = 44100
    NUM_PIXELS = 12

    # Initialize Hardware / Mocks
    if args.dry_run:
        logging.info("Starting in DRY-RUN mode. Using mock hardware.")
        audio = MockAudioStream(CHUNK, RATE)
        ring = MockStatusRing(NUM_PIXELS)
    else:
        if not HAS_PYAUDIO or not HAS_NEOPIXEL:
            logging.error("Hardware libraries missing. Run environment setup or use --dry-run.")
            sys.exit(1)
        logging.info("Starting in HARDWARE mode.")
        audio = RealAudioStream(CHUNK, RATE)
        ring = RealStatusRing(board.D12, NUM_PIXELS)

    # Traffic Light Colors (R, G, B)
    COLOR_GREEN = (0, 255, 0)
    COLOR_YELLOW = (255, 255, 0)
    COLOR_RED = (255, 0, 0)

    logging.info("Listening for ambient noise... Press Ctrl+C to stop.")

    try:
        while True:
            # 1. Capture audio chunk
            pcm_data = audio.read(CHUNK, exception_on_overflow=False)

            # 2. Process audio to dBFS
            dbfs = calculate_dbfs(pcm_data)

            # 3. Determine state and set visual indicator
            if dbfs >= args.threshold_red:
                state = "LOUD"
                ring.fill(COLOR_RED)
            elif dbfs >= args.threshold_yellow:
                state = "WARNING"
                ring.fill(COLOR_YELLOW)
            else:
                state = "QUIET"
                ring.fill(COLOR_GREEN)

            logging.info(f"Level: {dbfs:.2f} dBFS | State: {state}")

    except KeyboardInterrupt:
        logging.info("Stopping...")
    finally:
        audio.close()


if __name__ == "__main__":
    main()