Edge Impulse Integration with Python — Deep Dive

Automate Edge Impulse workflows in Python — data ingestion, impulse configuration, model training, on-device deployment, and CI/CD integration.

Python SDK Setup

pip install edgeimpulse edgeimpulse-api

# Set your API key (from Edge Impulse project dashboard → Keys)
export EI_API_KEY="ei_abc123..."

The SDK has two layers:

edgeimpulse — high-level functions for common workflows
edgeimpulse_api — low-level API client mirroring every REST endpoint

Uploading Data Programmatically

Uploading Sensor Data (Time-Series)

import edgeimpulse as ei
import numpy as np

ei.API_KEY = "ei_abc123..."

# Upload numpy arrays directly
# Shape: (num_samples, num_features) with sample rate info
sensor_data = np.random.randn(1000, 3)  # 1000 samples, 3-axis accelerometer

response = ei.experimental.data.upload_numpy(
    data=sensor_data,
    label="normal_vibration",
    filename="machine_01_normal.csv",
    category="training",           # or "testing"
    sample_rate_ms=10,             # 100 Hz
    sensors=[
        {"name": "accX", "units": "m/s2"},
        {"name": "accY", "units": "m/s2"},
        {"name": "accZ", "units": "m/s2"},
    ]
)
print(f"Uploaded sample ID: {response.sample_id}")

Uploading Images

import edgeimpulse as ei
from pathlib import Path

ei.API_KEY = "ei_abc123..."

# Upload directory of labeled images
image_dir = Path("dataset/training/")

for label_dir in image_dir.iterdir():
    if not label_dir.is_dir():
        continue
    label = label_dir.name

    for img_path in label_dir.glob("*.jpg"):
        ei.experimental.data.upload_file(
            filepath=str(img_path),
            label=label,
            category="training"
        )
        print(f"Uploaded {img_path.name} as '{label}'")

Bulk Upload with the Data Acquisition API

For large datasets, the REST API is more efficient:

import requests
import json

API_KEY = "ei_abc123..."
PROJECT_ID = 12345

headers = {
    "x-api-key": API_KEY,
    "Content-Type": "application/json"
}

# Upload CBOR-encoded sensor data (most efficient format)
def upload_sample(values, label, interval_ms, sensors):
    payload = {
        "protected": {
            "ver": "v1",
            "alg": "none"
        },
        "signature": "0" * 64,  # Empty signature for API key auth
        "payload": {
            "device_name": "python-pipeline",
            "device_type": "PYTHON_SDK",
            "interval_ms": interval_ms,
            "sensors": sensors,
            "values": values
        }
    }

    response = requests.post(
        f"https://ingestion.edgeimpulse.com/api/training/data",
        headers={
            "x-api-key": API_KEY,
            "x-label": label,
            "Content-Type": "application/json"
        },
        json=payload
    )
    return response.json()

Configuring and Training an Impulse

Using the API to Set Up Processing and Learning Blocks

from edgeimpulse_api import (
    Configuration, ApiClient,
    ImpulseApi, LearnApi, DSPApi
)

config = Configuration(host="https://studio.edgeimpulse.com/v1")
config.api_key["ApiKeyAuthentication"] = API_KEY

client = ApiClient(config)
impulse_api = ImpulseApi(client)
learn_api = LearnApi(client)
dsp_api = DSPApi(client)

# Get current impulse design
impulse = impulse_api.get_impulse(PROJECT_ID)
print(f"Processing blocks: {len(impulse.impulse.input_blocks)}")
print(f"Learning blocks: {len(impulse.impulse.learn_blocks)}")

# Start a training job
job = learn_api.start_training_job(PROJECT_ID, block_id=learn_block_id)
print(f"Training job started: {job.id}")

# Poll for completion
import time
while True:
    status = learn_api.get_training_job_status(PROJECT_ID, job.id)
    if status.job.finished:
        print(f"Training complete. Accuracy: {status.job.result.accuracy}%")
        break
    time.sleep(10)

Running Inference in Python

Using the Python SDK Runner

import edgeimpulse as ei

ei.API_KEY = "ei_abc123..."

# Download and run model locally
model = ei.Model(project_id=PROJECT_ID)

# Classify a single sample
import numpy as np
sample = np.random.randn(1000, 3).flatten().tolist()

result = model.classify(sample)

for label, score in result.classification.items():
    print(f"{label}: {score:.4f}")

if result.anomaly:
    print(f"Anomaly score: {result.anomaly:.4f}")

Linux Deployment with the Edge Impulse Runner

For continuous inference on Raspberry Pi or similar:

import subprocess
import json

# The edge-impulse-linux-runner handles camera/mic input automatically
proc = subprocess.Popen(
    ["edge-impulse-linux-runner", "--api-key", API_KEY],
    stdout=subprocess.PIPE,
    text=True
)

for line in proc.stdout:
    if line.startswith("{"):
        result = json.loads(line)
        if "classification" in result:
            for label, score in result["classification"].items():
                if score > 0.8:
                    print(f"Detected: {label} ({score:.2f})")

Custom Inference with the Exported Model

# Export a TFLite model from Edge Impulse
import edgeimpulse_api as ei_api

deployment_api = ei_api.DeploymentApi(client)

# Request a TFLite (float32) build
build = deployment_api.build_deployment(
    PROJECT_ID,
    type="runner-linux-aarch64"  # or "wasm", "arduino", etc.
)

# Download the built artifact
artifact = deployment_api.download_build(PROJECT_ID, type="runner-linux-aarch64")

with open("model_deployment.zip", "wb") as f:
    f.write(artifact)

Custom Processing Blocks

When built-in DSP blocks don’t fit your sensor data, create custom blocks:

# custom_dsp_block.py — runs as a Docker container on Edge Impulse
import json
import numpy as np
from scipy import signal as scipy_signal

def generate_features(draw_graphs, raw_data, axes, sampling_freq, **kwargs):
    """Custom feature extraction for vibration data."""
    features = []

    for axis in range(axes):
        channel = raw_data[axis::axes]

        # Compute power spectral density
        freqs, psd = scipy_signal.welch(
            channel,
            fs=sampling_freq,
            nperseg=min(256, len(channel))
        )

        # Extract band energies
        bands = [(0, 50), (50, 200), (200, 500), (500, sampling_freq/2)]
        for low, high in bands:
            mask = (freqs >= low) & (freqs < high)
            band_energy = np.trapz(psd[mask], freqs[mask])
            features.append(float(band_energy))

        # Peak frequency
        features.append(float(freqs[np.argmax(psd)]))

        # RMS
        features.append(float(np.sqrt(np.mean(channel ** 2))))

    return {
        "features": features,
        "graphs": [],
        "output_config": {
            "type": "flat",
            "shape": {"width": len(features)}
        }
    }

CI/CD Pipeline Integration

Automated Retraining Pipeline

#!/usr/bin/env python3
"""CI/CD script: retrain Edge Impulse model when new data arrives."""

import edgeimpulse as ei
import edgeimpulse_api as ei_api
import sys
import time

API_KEY = os.environ["EI_API_KEY"]
PROJECT_ID = int(os.environ["EI_PROJECT_ID"])
MIN_ACCURACY = float(os.environ.get("EI_MIN_ACCURACY", "85.0"))

ei.API_KEY = API_KEY

# Step 1: Upload new data
new_data_dir = sys.argv[1]
upload_count = upload_dataset(new_data_dir)  # Your upload function
print(f"Uploaded {upload_count} new samples")

# Step 2: Trigger training
config = ei_api.Configuration(host="https://studio.edgeimpulse.com/v1")
config.api_key["ApiKeyAuthentication"] = API_KEY
client = ei_api.ApiClient(config)
learn_api = ei_api.LearnApi(client)

job = learn_api.start_training_job(PROJECT_ID, block_id=LEARN_BLOCK_ID)

# Step 3: Wait for completion
while True:
    status = learn_api.get_training_job_status(PROJECT_ID, job.id)
    if status.job.finished:
        break
    time.sleep(15)

accuracy = status.job.result.accuracy
print(f"Training accuracy: {accuracy}%")

# Step 4: Gate on accuracy
if accuracy < MIN_ACCURACY:
    print(f"FAIL: Accuracy {accuracy}% below threshold {MIN_ACCURACY}%")
    sys.exit(1)

# Step 5: Build deployment artifact
deployment_api = ei_api.DeploymentApi(client)
deployment_api.build_deployment(PROJECT_ID, type="runner-linux-aarch64")
artifact = deployment_api.download_build(PROJECT_ID, type="runner-linux-aarch64")

with open("model_deployment.zip", "wb") as f:
    f.write(artifact)

print(f"SUCCESS: Model built with {accuracy}% accuracy")

Data Augmentation via the API

# Upload augmented versions of existing data
from PIL import Image, ImageEnhance
import io

def augment_and_upload(image_path, label, augmentations=5):
    img = Image.open(image_path)

    for i in range(augmentations):
        aug = img.copy()

        # Random brightness
        enhancer = ImageEnhance.Brightness(aug)
        aug = enhancer.enhance(np.random.uniform(0.7, 1.3))

        # Random rotation
        angle = np.random.uniform(-15, 15)
        aug = aug.rotate(angle, fillcolor=(128, 128, 128))

        # Save to buffer and upload
        buf = io.BytesIO()
        aug.save(buf, format="JPEG")
        buf.seek(0)

        ei.experimental.data.upload_file(
            filepath=buf,
            label=label,
            category="training",
            filename=f"{Path(image_path).stem}_aug{i}.jpg"
        )

Performance Comparison: Edge Impulse Models

Benchmarks on Raspberry Pi 4 (Cortex-A72, no GPU):

Model Type	Input	Size	Latency	RAM
Audio Classification (MFCC + NN)	1s audio	12 KB	3ms	2 KB
Image Classification (MobileNet V2 96×96)	Image	300 KB	85ms	180 KB
Object Detection (FOMO)	Image 96×96	65 KB	15ms	90 KB
Anomaly Detection (K-means)	Sensor data	1 KB	<1ms	0.5 KB

FOMO (Faster Objects, More Objects) is Edge Impulse’s own object detection architecture — optimized for microcontrollers, trading bounding-box precision for extreme efficiency.

The one thing to remember: Edge Impulse’s Python SDK transforms the platform from a web UI into a scriptable ML pipeline — enabling automated data ingestion, training, accuracy gating, and deployment builds that integrate into CI/CD workflows for managing edge AI across device fleets.

pythonmachine-learningedge-computing