Aditya N Bharadwaj
Published © CC BY-NC-ND

Eco-Sentinel: AI-Powered Urban Bird Conservation

A low-cost Arduino UNO Q edge-AI station that listens for urban birds, validates calls, visualizes biodiversity, and guides local habitat co

AdvancedFull instructions providedOver 1 day24
Eco-Sentinel: AI-Powered Urban Bird Conservation

Things used in this project

Hardware components

Arduino UNO Q
Arduino UNO Q
Thank you Hackster for the giveaway. It is the 4GB variant
×1
NodeMCU ESP8266 Breakout Board
NodeMCU ESP8266 Breakout Board
I specifically used LoLin NodeMCU V3 unit which has VU pin for 5V output
×1
I2C 16x2 Arduino LCD Display Module
DFRobot I2C 16x2 Arduino LCD Display Module
×1
Breadboard (generic)
Breadboard (generic)
×1
Male/Female Jumper Wires
Male/Female Jumper Wires
×4
USB-A to Micro-USB Cable
USB-A to Micro-USB Cable
×2
9V 1A Switching Wall Power Supply
9V 1A Switching Wall Power Supply
I used an old power bank which is around 7000mAh and gave around 8hrs of runtime. Good for portability
×1
Microphone Amplifier Breakout
Adafruit Microphone Amplifier Breakout
I used an old Android phone, instead of a mic for future Image processing project implementation.
×1

Software apps and online services

Arduino App Lab
On Arduino.cc
Arduino IDE
Arduino IDE
Streamlit
GitHub
streamlit
google colab
for statistical analysis
IP Webcam

Hand tools and fabrication machines

Blade Set, 0.584-1.55mmΒ² Capacity
Blade Set, 0.584-1.55mmΒ² Capacity

Story

Read more

Schematics

Telemetry device connection

for telemetry, on site glance device to view status of the mobile battery, UNO Q, and latest candidate detected in our case along with time.

System Architecture

Code

main.py

Python
Arduino Applab main.py
"""
Real-Time Avian Bio-Acoustic Edge Monitoring Station
BirdNET v2.4 ONNX inference for Arduino UNO Q

Version: 4.0

Main fixes:
1. Automatically detects and quarantines the supplied HTML/corrupt ONNX file.
2. Downloads the ARM-friendly BirdNET v2.4 INT8 ONNX model when needed.
3. Downloads the exact matching 6,522-label file.
4. Validates ONNX input/output contract before starting audio.
5. Validates that labels are in BirdNET "scientific_Common name" format.
6. Waits for a complete 3-second audio window before the first inference.
7. Uses 50% overlapping 3-second windows.
8. Keeps inference on the audio loop and heavy feature extraction on a worker.
9. Sends species, confidence, battery and timestamp to NodeMCU.
10. Logs detections to SQLite and automatically exports detections.csv.
11. Saves a mel spectrogram and acoustic features for every accepted detection.
12. Includes startup diagnostics and clearer failure messages.

BirdNET v2.4 contract:
- 48,000 Hz
- 3 seconds
- 144,000 float32 samples
- 6,522 output classes
- raw logits -> sigmoid confidence

Keep birdnet_model.onnx and birdnet_labels.txt in the same directory as this file.
"""

import csv
import datetime as dt
import json
import os
import queue
import shutil
import subprocess
import threading
import time

import librosa
import librosa.display
import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt
import numpy as np
import onnxruntime as ort
import requests
import sqlite3
from static_ffmpeg import run as ffmpeg_run

# UNO Q Linux MPU -> STM32 MCU Bridge for onboard LED matrix.
try:
    from arduino.app_utils import Bridge
    MATRIX_BRIDGE_AVAILABLE = True
except Exception as exc:
    Bridge = None
    MATRIX_BRIDGE_AVAILABLE = False
    print(f"[MATRIX] Bridge unavailable; monitoring continues: {exc}")


# ============================================================================
# 0. CONFIGURATION
# ============================================================================

BASE_DIR = os.path.dirname(os.path.abspath(__file__))

MODEL_PATH = os.path.join(BASE_DIR, "birdnet_model.onnx")
LABELS_PATH = os.path.join(BASE_DIR, "birdnet_labels.txt")
DB_PATH = os.path.join(BASE_DIR, "birds.db")
CSV_PATH = os.path.join(BASE_DIR, "detections.csv")
SPECTROGRAM_DIR = os.path.join(BASE_DIR, "spectrograms")
os.makedirs(SPECTROGRAM_DIR, exist_ok=True)

# Officially documented BirdNET v2.4 ONNX source used for this project.
# INT8 ARM is the preferred variant for a low-RAM ARM edge computer.
MODEL_URL = (
    "https://huggingface.co/tphakala/BirdNET-v2.4/"
    "resolve/main/BirdNET_v2.4_int8_arm.onnx"
)
LABELS_URL = (
    "https://huggingface.co/tphakala/BirdNET-v2.4/"
    "resolve/main/labels.txt"
)

# Field network from the supplied project configuration.
COOLPAD_IP = "My_phone_IP" #paste your IP  address on IP webcam
NODEMCU_IP = "NodeMCU_IP" #connect nodemcu to phone to view IP

# Indian Standard Time (IST = UTC+05:30).
IST_OFFSET_SECONDS = 19800
INDIA_NTP_SERVERS = (
    "time.google.com",
    "pool.ntp.org",
    "time.cloudflare.com",
)
TIME_SYNC_INTERVAL_SECONDS = 6 * 60 * 60
AUDIO_STREAM = f"http://{COOLPAD_IP}/audio.wav"
TELEMETRY_URL = f"{NODEMCU_IP}/telemetry"

# BirdNET v2.4 audio contract.
SAMPLE_RATE = 48000
CHUNK_LEN = 3.0
CHUNK_SAMPLES = int(SAMPLE_RATE * CHUNK_LEN)
HOP_SAMPLES = CHUNK_SAMPLES // 2  # 50% overlap

# Detection settings.
# Low-SNR mode: allow weak candidates, then require repeated evidence.
CANDIDATE_CONFIDENCE_THRESHOLD = 0.05
CONFIRM_CONFIDENCE_THRESHOLD = 0.05
RMS_SILENCE_THRESHOLD = 0.01
BATTERY_POLL_SECONDS = 30

EVIDENCE_WINDOW_SECONDS = 15.0
EVIDENCE_MIN_HITS = 2
EVIDENCE_MIN_TOTAL_SCORE = 0.18
CANDIDATE_DECAY_SECONDS = 4.5
TOP_K_CANDIDATES = 5

CONFIDENCE_THRESHOLD = CANDIDATE_CONFIDENCE_THRESHOLD

# Prevent the same continuous call from filling the database every 1.5 s.
# Set to 0 to log every accepted inference.
DUPLICATE_SUPPRESSION_SECONDS = 10

# How often a "Scanning..." heartbeat is sent to the NodeMCU.
HEARTBEAT_SECONDS = 12

# Matrix control is asynchronous so Bridge/RPC cannot block BirdNET inference.
_matrix_q = queue.Queue(maxsize=4)
_matrix_last_state = None


def request_matrix_state(state):
    global _matrix_last_state
    state = str(state)

    if state == _matrix_last_state:
        return

    _matrix_last_state = state
    try:
        _matrix_q.put_nowait(state)
    except queue.Full:
        try:
            _matrix_q.get_nowait()
        except queue.Empty:
            pass
        try:
            _matrix_q.put_nowait(state)
        except queue.Full:
            pass


def _matrix_worker():
    while True:
        state = _matrix_q.get()
        try:
            if MATRIX_BRIDGE_AVAILABLE and Bridge is not None:
                Bridge.call("set_bird_matrix_state", state)
                print(f"[MATRIX] State -> {state}")
        except Exception as exc:
            print(f"[MATRIX] Bridge call failed ({state}): {exc}")
        finally:
            _matrix_q.task_done()


threading.Thread(
    target=_matrix_worker,
    name="matrix-worker",
    daemon=True,
).start()

# ============================================================================
# Bengaluru project whitelist
# ============================================================================
# These are the common-name strings used after parsing birdnet_labels.txt.
# The whitelist is intentionally kept from the supplied project rather than
# silently replacing it with an external species list.
NATIVE_WHITELIST = {
    "House Sparrow",
    "Common Myna",
    "Rose-ringed Parakeet",
    "Black Kite",
    "Asian Koel",
    "Jungle Myna",
    "Brahminy Kite",
    "Rock Pigeon",
    "Greater Coucal",
    "White-throated Kingfisher",
    "Spotted Dove",
    "Indian Pond Heron",
    "Red-whiskered Bulbul",
    "Red-vented Bulbul",
    "Purple Sunbird",
    "Coppersmith Barbet",
    "Indian Robin",
    "House Crow",
}


# ============================================================================
# 1. UTILITY FUNCTIONS
# ============================================================================

_time_lock = threading.Lock()
_time_offset = None
_last_time_sync = 0.0


def sync_india_time(force=False):
    """Get network time using NTP and maintain the offset to the UNO Q clock."""
    global _time_offset, _last_time_sync

    now = time.time()
    with _time_lock:
        if (
            not force
            and _time_offset is not None
            and now - _last_time_sync < TIME_SYNC_INTERVAL_SECONDS
        ):
            return True

    import socket
    import struct

    packet = b"\x1b" + 47 * b"\0"

    for host in INDIA_NTP_SERVERS:
        try:
            t0 = time.time()
            with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
                sock.settimeout(2.0)
                sock.sendto(packet, (host, 123))
                data, _ = sock.recvfrom(512)
            t1 = time.time()

            if len(data) < 48:
                continue

            words = struct.unpack("!12I", data[:48])
            ntp_seconds = words[10]
            ntp_fraction = words[11]
            ntp_time = (
                ntp_seconds - 2208988800
                + ntp_fraction / 4294967296.0
            )

            offset = ntp_time - ((t0 + t1) / 2.0)

            with _time_lock:
                _time_offset = offset
                _last_time_sync = t1

            print(
                f"[TIME] Network time synchronized via {host}; "
                f"offset={offset:+.3f}s; timezone=IST (UTC+05:30)"
            )
            return True

        except Exception as exc:
            print(f"[TIME] NTP server {host} unavailable: {exc}")

    print("[TIME] Network time sync failed; using existing UNO Q clock.")
    return False


def now_local_iso():
    with _time_lock:
        offset = _time_offset

    current = time.time()
    if offset is not None:
        current += offset

    # UTC -> IST.
    current += IST_OFFSET_SECONDS

    current_dt = dt.datetime.fromtimestamp(
        current,
        dt.timezone.utc,
    )
    return current_dt.isoformat(timespec="seconds")


def _time_sync_worker():
    while True:
        time.sleep(TIME_SYNC_INTERVAL_SECONDS)
        sync_india_time(force=True)



def safe_species_filename(species):
    """Make a filesystem-safe species name."""
    allowed = set("abcdefghijklmnopqrstuvwxyz"
                  "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
                  "0123456789-_")
    return "".join(c if c in allowed else "_" for c in species)


def is_probably_html(path):
    """Detect the exact failure seen in the supplied ONNX file."""
    try:
        with open(path, "rb") as f:
            head = f.read(512).lstrip().lower()
        return (
            head.startswith(b"<!doctype html")
            or head.startswith(b"<html")
            or b"<html" in head[:200]
            or b"<head" in head[:200]
        )
    except OSError:
        return False


def quarantine_bad_file(path):
    """Rename a corrupt download instead of silently deleting it."""
    if not os.path.exists(path):
        return

    stamp = time.strftime("%Y%m%d_%H%M%S")
    backup = f"{path}.invalid_{stamp}"

    try:
        os.replace(path, backup)
        print(f"[SETUP] Invalid file moved to: {os.path.basename(backup)}")
    except OSError as exc:
        raise RuntimeError(
            f"Could not quarantine invalid file {path}: {exc}"
        ) from exc


def download_atomic(url, path, what, min_bytes):
    """
    Download to a temporary file and replace the target only after the
    download passes basic sanity checks.
    """
    tmp_path = path + ".download"

    print(f"[SETUP] Downloading {what}...")
    print(f"[SETUP] URL: {url}")

    try:
        with requests.get(
            url,
            allow_redirects=True,
            timeout=(10, 180),
            stream=True,
            headers={"User-Agent": "Avian-BioAcoustic-Edge/3.0"},
        ) as response:
            response.raise_for_status()

            content_type = response.headers.get("Content-Type", "")
            total = 0

            with open(tmp_path, "wb") as out:
                for chunk in response.iter_content(chunk_size=1024 * 1024):
                    if not chunk:
                        continue
                    out.write(chunk)
                    total += len(chunk)

        if total < min_bytes:
            raise RuntimeError(
                f"{what} download is too small: {total} bytes."
            )

        if is_probably_html(tmp_path):
            raise RuntimeError(
                f"{what} download is HTML instead of the real file. "
                f"Content-Type was {content_type!r}."
            )

        os.replace(tmp_path, path)
        print(f"[SETUP] {what} downloaded: {total / 1e6:.2f} MB")

    except Exception:
        try:
            if os.path.exists(tmp_path):
                os.remove(tmp_path)
        except OSError:
            pass
        raise


def ensure_labels():
    """Download labels only when absent or obviously invalid."""
    if os.path.exists(LABELS_PATH):
        try:
            with open(LABELS_PATH, encoding="utf-8") as f:
                lines = [line.strip() for line in f if line.strip()]

            if len(lines) >= 6000:
                print(f"[SETUP] Existing labels found: {len(lines)} lines")
                return

            print("[SETUP] Existing labels file is unexpectedly small.")
        except (OSError, UnicodeError) as exc:
            print(f"[SETUP] Existing labels file cannot be read: {exc}")

        quarantine_bad_file(LABELS_PATH)

    download_atomic(
        LABELS_URL,
        LABELS_PATH,
        "BirdNET labels",
        min_bytes=100_000,
    )


def load_labels():
    """
    Parse the exact BirdNET labels format:
        Scientific name_Common name

    Return both full labels and common names.
    """
    with open(LABELS_PATH, encoding="utf-8") as f:
        raw_labels = [line.strip() for line in f if line.strip()]

    common_names = []
    for line in raw_labels:
        if "_" not in line:
            raise RuntimeError(
                f"Invalid label format: {line!r}. "
                "Expected 'Scientific name_Common name'."
            )
        common_names.append(line.rsplit("_", 1)[1].strip())

    if len(raw_labels) != 6522:
        raise RuntimeError(
            f"Expected 6,522 BirdNET v2.4 labels, found {len(raw_labels)}. "
            "Do not mix labels from another BirdNET model."
        )

    return raw_labels, common_names


# ============================================================================
# 2. MODEL VALIDATION AND LOADING
# ============================================================================

EXPECTED_INPUT_SAMPLES = 144000
EXPECTED_CLASSES = 6522


def shape_is_compatible(shape, expected_last_dim):
    """
    Accept [144000], [None,144000], [1,144000], etc., but require the
    final fixed dimension to match BirdNET v2.4.
    """
    if not shape:
        return False
    last = shape[-1]
    return isinstance(last, int) and last == expected_last_dim


def create_ort_session(path):
    """Create an optimized CPU ONNX Runtime session."""
    options = ort.SessionOptions()
    options.graph_optimization_level = ort.GraphOptimizationLevel.ORT_ENABLE_ALL
    options.log_severity_level = 3

    return ort.InferenceSession(
        path,
        sess_options=options,
        providers=["CPUExecutionProvider"],
    )


def validate_model_contract(session):
    """Validate BirdNET v2.4 input/output dimensions."""
    inputs = session.get_inputs()
    outputs = session.get_outputs()

    if not inputs:
        raise RuntimeError("ONNX model has no input tensors.")
    if not outputs:
        raise RuntimeError("ONNX model has no output tensors.")

    model_input = inputs[0]
    model_output = outputs[0]

    print(
        f"[SETUP] ONNX input:  name={model_input.name!r}, "
        f"type={model_input.type}, shape={model_input.shape}"
    )
    print(
        f"[SETUP] ONNX output: name={model_output.name!r}, "
        f"type={model_output.type}, shape={model_output.shape}"
    )

    if model_input.type != "tensor(float)":
        raise RuntimeError(
            f"Unexpected model input type: {model_input.type}. "
            "BirdNET v2.4 should use float32."
        )

    if not shape_is_compatible(model_input.shape, EXPECTED_INPUT_SAMPLES):
        raise RuntimeError(
            f"Unexpected BirdNET input shape: {model_input.shape}. "
            "Expected a final dimension of 144000 samples."
        )

    if not shape_is_compatible(model_output.shape, EXPECTED_CLASSES):
        raise RuntimeError(
            f"Unexpected BirdNET output shape: {model_output.shape}. "
            "Expected a final dimension of 6522 classes."
        )

    return model_input.name, model_output.name


def load_validated_model():
    """
    Load the local model.

    If the supplied model is HTML/corrupt or cannot be parsed, quarantine it,
    download the known BirdNET v2.4 ARM model, then validate again.
    """
    ensure_labels()
    raw_labels, labels = load_labels()

    if not os.path.exists(MODEL_PATH):
        download_atomic(
            MODEL_URL,
            MODEL_PATH,
            "BirdNET v2.4 INT8 ARM ONNX model",
            min_bytes=20_000_000,
        )

    if is_probably_html(MODEL_PATH) or os.path.getsize(MODEL_PATH) < 20_000_000:
        print("[SETUP] Local birdnet_model.onnx is not a real BirdNET model.")
        quarantine_bad_file(MODEL_PATH)
        download_atomic(
            MODEL_URL,
            MODEL_PATH,
            "BirdNET v2.4 INT8 ARM ONNX model",
            min_bytes=20_000_000,
        )

    try:
        print(
            f"[SETUP] Loading ONNX model "
            f"({os.path.getsize(MODEL_PATH) / 1e6:.2f} MB)..."
        )
        session = create_ort_session(MODEL_PATH)
        input_name, output_name = validate_model_contract(session)

    except Exception as first_error:
        print(f"[SETUP] Local ONNX load failed: {first_error}")
        print("[SETUP] The local model will be quarantined and re-downloaded.")

        quarantine_bad_file(MODEL_PATH)

        download_atomic(
            MODEL_URL,
            MODEL_PATH,
            "BirdNET v2.4 INT8 ARM ONNX model",
            min_bytes=20_000_000,
        )

        try:
            session = create_ort_session(MODEL_PATH)
            input_name, output_name = validate_model_contract(session)
        except Exception as second_error:
            raise RuntimeError(
                "BirdNET ONNX model could not be loaded even after a fresh "
                f"download.\nOriginal error: {first_error}\n"
                f"Fresh-download error: {second_error}"
            ) from second_error

    if len(raw_labels) != EXPECTED_CLASSES:
        raise RuntimeError(
            f"Labels contain {len(raw_labels)} entries, but BirdNET v2.4 "
            f"requires {EXPECTED_CLASSES}."
        )

    print(
        f"[SETUP] BirdNET model validated successfully: "
        f"{EXPECTED_CLASSES} classes, 48 kHz, 3 seconds."
    )
    print(
        f"[SETUP] Label examples: {raw_labels[:3]}"
    )

    return session, input_name, output_name, raw_labels, labels


session, input_name, output_name, raw_labels, labels = load_validated_model()

request_matrix_state("listening")


# ============================================================================
# 3. BACKGROUND TELEMETRY
# ============================================================================

_telemetry_q = queue.Queue(maxsize=20)


def _telemetry_worker():
    while True:
        payload = _telemetry_q.get()
        try:
            response = requests.post(
                TELEMETRY_URL,
                json=payload,
                timeout=2,
            )
            if not response.ok:
                print(
                    f"[TELEMETRY] NodeMCU HTTP {response.status_code}"
                )
        except requests.RequestException as exc:
            # Do not stop bird detection because the display is offline.
            print(f"[TELEMETRY] NodeMCU unavailable: {exc}")
        finally:
            _telemetry_q.task_done()


threading.Thread(
    target=_telemetry_worker,
    name="telemetry-worker",
    daemon=True,
).start()


def send_telemetry(species, confidence, battery):
    payload = {
        "species": species,
        "confidence": round(float(confidence) * 100.0, 1),
        "battery": battery,
        "timestamp": now_local_iso(),
        "timezone": "Asia/Kolkata",
    }

    try:
        _telemetry_q.put_nowait(payload)
    except queue.Full:
        pass


def _telemetry_heartbeat_worker():
    """
    Keep the NodeMCU informed that the UNO Q Python process is alive.

    This is deliberately independent of FFmpeg/BirdNET. Otherwise, if the
    audio stream stalls, the NodeMCU can incorrectly conclude that the UNO Q
    itself is offline even though Python is still running normally.
    """
    # Send one immediately after the telemetry worker has started.
    send_telemetry("UNO Q Online", 0.0, _battery_level)

    while True:
        time.sleep(10)
        send_telemetry("UNO Q Online", 0.0, _battery_level)


# ============================================================================
# 4. COOLPAD BATTERY POLLING
# ============================================================================

_battery_level = "N/A"

threading.Thread(
    target=_telemetry_heartbeat_worker,
    name="telemetry-heartbeat",
    daemon=True,
).start()


def deep_scan_json(data, target_keys):
    """Find battery-like fields anywhere in a nested JSON response."""
    if isinstance(data, dict):
        for key, value in data.items():
            key_lower = str(key).lower()

            if any(target in key_lower for target in target_keys):
                if isinstance(value, (int, float, str)):
                    return value

                if isinstance(value, dict) and "data" in value:
                    return value["data"]

            found = deep_scan_json(value, target_keys)
            if found is not None:
                return found

    elif isinstance(data, list):
        for item in data:
            found = deep_scan_json(item, target_keys)
            if found is not None:
                return found

    return None


def _battery_worker():
    global _battery_level

    keys = ["battery", "percent", "capacity", "charge"]

    while True:
        value = None

        for endpoint in ("status.json", "sensors.json"):
            try:
                response = requests.get(
                    f"http://{COOLPAD_IP}/{endpoint}",
                    timeout=2,
                )

                if response.ok:
                    value = deep_scan_json(response.json(), keys)
                    if value is not None:
                        break

            except (
                requests.RequestException,
                ValueError,
            ):
                continue

        if value is not None:
            try:
                battery = int(round(float(value)))
                _battery_level = max(0, min(100, battery))
            except (TypeError, ValueError):
                pass

        time.sleep(BATTERY_POLL_SECONDS)


threading.Thread(
    target=_battery_worker,
    name="battery-worker",
    daemon=True,
).start()


# ============================================================================
# 5. SQLITE + CSV DATA STORAGE
# ============================================================================

DB_LOCK = threading.Lock()


def init_db():
    with DB_LOCK:
        conn = sqlite3.connect(DB_PATH)

        conn.execute("""
            CREATE TABLE IF NOT EXISTS inventory (
                id INTEGER PRIMARY KEY,
                ts TEXT,
                species TEXT,
                confidence REAL,
                battery INTEGER,
                rms REAL,
                detection_status TEXT,
                whitelisted INTEGER,
                spectrogram_path TEXT,
                features_json TEXT
            )
        """)

        # Migration for older project databases.  The original database may
        # already contain an inventory table but be missing one or more of
        # the columns used by the current application.
        migrations = [
            ("ts", "TEXT"),
            ("species", "TEXT"),
            ("confidence", "REAL"),
            ("battery", "INTEGER"),
            ("rms", "REAL"),
            ("detection_status", "TEXT"),
            ("whitelisted", "INTEGER"),
            ("spectrogram_path", "TEXT"),
            ("features_json", "TEXT"),
        ]

        for column, column_type in migrations:
            try:
                conn.execute(
                    f"ALTER TABLE inventory ADD COLUMN {column} {column_type}"
                )
                print(f"[DB] Added missing column: {column}")
            except sqlite3.OperationalError:
                # Column already exists; this is expected on normal starts.
                pass

        conn.commit()
        conn.close()


def log_detection(
    species,
    confidence,
    battery,
    rms,
    detection_status="CONFIRMED",
    whitelisted=False,
):
    """
    Store EVERY meaningful BirdNET event, including rejected/false-positive
    candidates.

    detection_status examples:
      CONFIRMED               -> temporal evidence accepted it
      LOW_CONFIDENCE          -> below the 5% candidate threshold
      NON_WHITELISTED         -> BirdNET predicted a species outside our list
      CANDIDATE               -> weak whitelisted candidate still accumulating
                                  evidence
    """
    with DB_LOCK:
        conn = sqlite3.connect(DB_PATH)

        cursor = conn.execute(
            """
            INSERT INTO inventory
                (
                    ts,
                    species,
                    confidence,
                    battery,
                    rms,
                    detection_status,
                    whitelisted
                )
            VALUES (?, ?, ?, ?, ?, ?, ?)
            """,
            (
                now_local_iso(),
                species,
                float(confidence),
                battery if isinstance(battery, int) else None,
                float(rms),
                detection_status,
                1 if whitelisted else 0,
            ),
        )

        conn.commit()
        row_id = cursor.lastrowid
        conn.close()

    return row_id

def log_features(row_id, spectrogram_path, features_json):
    with DB_LOCK:
        conn = sqlite3.connect(DB_PATH)

        conn.execute(
            """
            UPDATE inventory
            SET spectrogram_path = ?, features_json = ?
            WHERE id = ?
            """,
            (
                spectrogram_path,
                features_json,
                row_id,
            ),
        )

        conn.commit()
        conn.close()


def export_csv():
    """
    Export the complete SQLite inventory into a normal CSV file so the data
    can be opened in Excel, R Studio, Minitab, etc.
    """
    with DB_LOCK:
        conn = sqlite3.connect(DB_PATH)

        cursor = conn.execute(
            """
            SELECT
                id,
                ts,
                species,
                confidence,
                battery,
                rms,
                detection_status,
                whitelisted,
                spectrogram_path,
                features_json
            FROM inventory
            ORDER BY id
            """
        )

        rows = cursor.fetchall()
        headers = [
            "id",
            "timestamp",
            "species",
            "confidence",
            "battery",
            "rms",
            "detection_status",
            "whitelisted",
            "spectrogram_path",
            "features_json",
        ]

        tmp_csv = CSV_PATH + ".tmp"

        with open(
            tmp_csv,
            "w",
            newline="",
            encoding="utf-8",
        ) as f:
            writer = csv.writer(f)
            writer.writerow(headers)
            writer.writerows(rows)

        os.replace(tmp_csv, CSV_PATH)
        conn.close()


init_db()
export_csv()


# ============================================================================
# 6. ACOUSTIC FEATURES + SPECTROGRAMS
# ============================================================================

def extract_features(audio, sr):
    """
    Scalar acoustic features for the future R Studio / Minitab phase.

    The complete feature vector is stored as JSON so additional fields can be
    added later without changing the database schema.
    """
    mfccs = librosa.feature.mfcc(
        y=audio,
        sr=sr,
        n_mfcc=13,
    )

    centroid = librosa.feature.spectral_centroid(
        y=audio,
        sr=sr,
    )

    bandwidth = librosa.feature.spectral_bandwidth(
        y=audio,
        sr=sr,
    )

    rolloff = librosa.feature.spectral_rolloff(
        y=audio,
        sr=sr,
    )

    zcr = librosa.feature.zero_crossing_rate(audio)

    rms = librosa.feature.rms(
        y=audio,
    )

    flatness = librosa.feature.spectral_flatness(
        y=audio,
    )

    return {
        "mfcc_mean": [float(x) for x in mfccs.mean(axis=1)],
        "mfcc_std": [float(x) for x in mfccs.std(axis=1)],
        "spectral_centroid_mean": float(centroid.mean()),
        "spectral_bandwidth_mean": float(bandwidth.mean()),
        "spectral_rolloff_mean": float(rolloff.mean()),
        "zero_crossing_rate_mean": float(zcr.mean()),
        "rms_mean": float(rms.mean()),
        "spectral_flatness_mean": float(flatness.mean()),
    }


def save_spectrogram(audio, sr, out_path):
    mel = librosa.feature.melspectrogram(
        y=audio,
        sr=sr,
        n_mels=128,
        fmax=sr // 2,
    )

    mel_db = librosa.power_to_db(
        mel,
        ref=np.max,
    )

    fig, ax = plt.subplots(figsize=(5, 3))

    librosa.display.specshow(
...

This file has been truncated, please download it to see its full contents.

Eco Sentinel project repo

All details are provided in the Github repository. You just have to copy paste in your app lab

Credits

Aditya N Bharadwaj
4 projects β€’ 3 followers
Average SBCs & Microcontroller Enthusiast!

Comments