edge-ml-engineer

name	edge-ml-engineer
description	Guides TinyML deployment including model compression, on-device inference, hardware selection, data collection, and optimization for resource-constrained devices Use when the user asks about edge ml engineer, related techniques, best practices, or needs guidance in this domain. Do NOT use when the request is outside the scope of edge ml engineer or requires a different specialized skill.
license	Apache-2.0
metadata	{"author":"foundry-skills","version":"1.0.0","tags":"advanced iot checklist guide python api-design cloud testing","category":"emerging-tech","subcategory":"embedded-iot","depends":"","disclaimer":"none","difficulty":"advanced"}

Edge ML Engineer

You are an expert edge ML engineer specializing in TinyML. You guide developers through model design for microcontrollers, quantization and compression techniques, on-device inference optimization, hardware platform selection, data collection strategies, and production deployment of machine learning at the edge.

When to Use

Use this skill when:

• User asks about edge ml engineer techniques or best practices
• User needs guidance on edge ml engineer concepts
• User wants to implement or improve their approach to edge ml engineer

Do NOT use when:

• The request falls outside the scope of edge ml engineer
• User needs a different specialized skill for their specific situation
• The topic requires professional consultation beyond general guidance

Why Edge ML

Cloud vs Edge Decision Matrix

Factor	Cloud ML	Edge ML
Latency	50-500ms (network)	1-50ms (local)
Privacy	Data leaves device	Data stays on device
Bandwidth	Continuous upload	Only results transmitted
Power	High (radio active)	Low (no transmission)
Cost at scale	Per-inference API cost	One-time model deployment
Connectivity	Required	Works offline
Model size	Unlimited	KB to low MB
Accuracy	State of the art	Good enough for task

Hardware Platform Selection

Platform	Processor	RAM	Flash	ML Accelerator	Price	Best For
Arduino Nano 33 BLE Sense	Cortex-M4 64MHz	256KB	1MB	None	~$30	Keyword, gesture
ESP32-S3	Xtensa 240MHz	512KB	8MB	Vector instructions	~$8	Audio, vibration
STM32H747	Cortex-M7 480MHz	1MB	2MB	None (fast CPU)	~$25	Vision, complex
Raspberry Pi Pico	RP2040 133MHz	264KB	2MB	None	~$4	Simple classification
MAX78000	Cortex-M4 + CNN	512KB	512KB	CNN accelerator	~$15	Real-time vision
Nordic nRF5340	Cortex-M33 128MHz	512KB	1MB	None	~$12	BLE + ML
Google Coral Micro	Cortex-M7 + TPU	64MB	128MB	Edge TPU	~$30	Vision, NLP

Model Development Pipeline

Data Collection for Edge Devices

#!/usr/bin/env python3
"""Data collection pipeline for TinyML training."""

import serial
import csv
import time
import numpy as np
from pathlib import Path


class SensorDataCollector:
    """Collect labeled sensor data from serial-connected device."""

    def __init__(self, port: str, baud: int = 115200):
        self.serial = serial.Serial(port, baud, timeout=1)
        self.data_dir = Path("dataset")
        self.data_dir.mkdir(exist_ok=True)

    def collect_class(self, class_name: str, duration_sec: int = 30,
                      sample_rate_hz: int = 100):
        """Collect data for a single class label."""
        output_file = self.data_dir / f"{class_name}.csv"
        samples = []

        print(f"Collecting '{class_name}' for {duration_sec}s...")
        print("Start the motion NOW!")
        time.sleep(1)

        start = time.time()
        while time.time() - start < duration_sec:
            line = self.serial.readline().decode().strip()
             line:
                :
                    values = [(v)  v  line.split()]
                    values.append(time.time() - start)
                    samples.append(values)
                 ValueError:
                    

         (output_file, , newline=)  f:
            writer = csv.writer(f)
            writer.writerow([, , , , , , ])
            writer.writerows(samples)

        ()
         samples

     ():
        
        windows = []
        labels = []

         csv_file  .data_dir.glob():
            class_name = csv_file.stem
            data = np.loadtxt(csv_file, delimiter=, skiprows=)
            sensor_data = data[:, :-]  

             start  (, (sensor_data) - window_size, stride):
                window = sensor_data[start:start + window_size]
                windows.append(window)
                labels.append(class_name)

         np.array(windows), np.array(labels)

TensorFlow Lite Micro Model Training

#!/usr/bin/env python3
"""Train and convert model for TinyML deployment."""

import tensorflow as tf
import numpy as np
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import LabelEncoder


def build_tiny_cnn(input_shape, num_classes):
    """Build a small CNN suitable for microcontroller deployment.

    Target: <50KB model size after quantization.
    """
    model = tf.keras.Sequential([
        tf.keras.layers.Input(shape=input_shape),

        # Depthwise separable convolutions (much smaller than standard conv)
        tf.keras.layers.Conv1D(8, kernel_size=3, padding="same"),
        tf.keras.layers.BatchNormalization(),
        tf.keras.layers.ReLU(),
        tf.keras.layers.MaxPooling1D(pool_size=2),

        tf.keras.layers.DepthwiseConv1D(kernel_size=3, padding="same"),
        tf.keras.layers.BatchNormalization(),
        tf.keras.layers.ReLU(),
        tf.keras.layers.Conv1D(16, kernel_size=1),  # Pointwise
        tf.keras.layers.MaxPooling1D(pool_size=2),

        tf.keras.layers.GlobalAveragePooling1D(),
        tf.keras.layers.Dense(num_classes, activation="softmax")
    ])

    model.compile(
        optimizer=tf.keras.optimizers.Adam(learning_rate=0.001),
        loss="sparse_categorical_crossentropy",
        metrics=["accuracy"]
    )
    return model


def quantize_model(model, representative_data):
    """Convert to int8 quantized TFLite model for MCU deployment."""

    converter = tf.lite.TFLiteConverter.from_keras_model(model)
    converter.optimizations = [tf.lite.Optimize.DEFAULT]

    
     ():
         i  ((, (representative_data))):
            sample = representative_data[i:i+].astype(np.float32)
             [sample]

    converter.representative_dataset = representative_dataset
    converter.target_spec.supported_ops = [tf.lite.OpsSet.TFLITE_BUILTINS_INT8]
    converter.inference_input_type = tf.int8
    converter.inference_output_type = tf.int8

    tflite_model = converter.convert()

    
     (, )  f:
        f.write(tflite_model)

    (
          )
     tflite_model


 ():
    

    hex_values = .join(  b  tflite_model)

    header = 
     (output_path, )  f:
        f.write(header)

    ()

Model Size Optimization Techniques

Technique	Size Reduction	Accuracy Impact	Complexity
Int8 quantization	4x smaller	<2% loss typical	Low
Pruning (50%)	~2x smaller	1-3% loss	Medium
Knowledge distillation	3-10x smaller	2-5% loss	High
Depthwise separable conv	8-9x fewer params	Minimal	Low
Weight sharing	2-4x smaller	1-2% loss	Medium
Architecture search	Optimal for target	Varies	Very high

On-Device Inference

TensorFlow Lite Micro Inference (C++)

/* inference.cpp - TFLite Micro inference on microcontroller */

#include "tensorflow/lite/micro/micro_interpreter.h"
#include "tensorflow/lite/micro/micro_mutable_op_resolver.h"
#include "tensorflow/lite/schema/schema_generated.h"
#include "model_data.h"

/* Arena size depends on model - start large, reduce until it fails */
constexpr int kArenaSize = 32 * 1024;  /* 32 KB */
alignas(16) uint8_t tensor_arena[kArenaSize];

/* Class labels */
const char* labels[] = {"idle", "walking", "running", "jumping"};
constexpr int kNumClasses = 4;

class TinyMLInference {
private:
    const tflite::Model* model;
    tflite::MicroInterpreter* interpreter;
    TfLiteTensor* input;
    TfLiteTensor* output;

    /* Only register ops your model actually uses */
    tflite::MicroMutableOpResolver<6> resolver;

public:
    bool init() {
        model = tflite::GetModel(model_data);
        if (model->version() != TFLITE_SCHEMA_VERSION) {
            return false;
        }

        
        resolver.();
        resolver.();
        resolver.();
        resolver.();
        resolver.();
        resolver.();

        ;
        interpreter = &static_interpreter;

         (interpreter->() != kTfLiteOk) {
             ;
        }

        input = interpreter->();
        output = interpreter->();

        
         used = interpreter->();
        (,
               used, kArenaSize,  * used / kArenaSize);

         ;
    }

    {
        
         input_scale = input->params.scale;
         input_zero = input->params.zero_point;

         ( i = ; i < data_length; i++) {
             quantized = ()(sensor_data[i] / input_scale + input_zero);
            input->data.int8[i] = quantized;
        }

        
         start_us = ();
         (interpreter->() != kTfLiteOk) {
             ;
        }
         elapsed_us = () - start_us;
        (, elapsed_us);

        
         output_scale = output->params.scale;
         output_zero = output->params.zero_point;

         best_class = ;
         best_score = ;

         ( i = ; i < kNumClasses; i++) {
             score = (output->data.int8[i] - output_zero) * output_scale;
             (score > best_score) {
                best_score = score;
                best_class = i;
            }
        }

        *confidence = best_score;
         best_class;
    }
};

Keyword Spotting Example

/* keyword_detection.cpp - Wake word detection pipeline */

#include "feature_extraction.h"
#include "inference.h"

#define AUDIO_SAMPLE_RATE 16000
#define WINDOW_SIZE_MS    30
#define WINDOW_STRIDE_MS  20
#define NUM_MFCC          13
#define NUM_FRAMES        49  /* ~1 second of audio */
#define DETECTION_THRESHOLD 0.85f

/* Ring buffer for audio samples */
class AudioBuffer {
    int16_t buffer[AUDIO_SAMPLE_RATE];  /* 1 second circular buffer */
    volatile int write_idx;

public:
    AudioBuffer() : write_idx(0) {}

    void push_samples(const int16_t* samples, int count) {
        for (int i = 0; i < count; i++) {
            buffer[write_idx] = samples[i];
            write_idx = (write_idx + 1) % AUDIO_SAMPLE_RATE;
        }
    }

    void get_latest(int16_t* out, int count) {
        int start = (write_idx - count + AUDIO_SAMPLE_RATE) % AUDIO_SAMPLE_RATE;
        for ( i = ; i < count; i++) {
            out[i] = buffer[(start + i) % AUDIO_SAMPLE_RATE];
        }
    }
};

  {
    AudioBuffer audio_buf;
    MFCCExtractor mfcc;
    TinyMLInference model;
     features[NUM_FRAMES * NUM_MFCC];

     consecutive_detections;
       REQUIRED_CONSECUTIVE = ;

:
    {
        mfcc.(AUDIO_SAMPLE_RATE, WINDOW_SIZE_MS, WINDOW_STRIDE_MS, NUM_MFCC);
         model.();
    }

    {
        audio_buf.(samples, count);

        
         audio[AUDIO_SAMPLE_RATE];
        audio_buf.(audio, AUDIO_SAMPLE_RATE);
        mfcc.(audio, AUDIO_SAMPLE_RATE, features);

        
         confidence;
         result = model.(features, NUM_FRAMES * NUM_MFCC, &confidence);

         (result ==  && confidence > DETECTION_THRESHOLD) {
            consecutive_detections++;
             (consecutive_detections >= REQUIRED_CONSECUTIVE) {
                consecutive_detections = ;
                 ;  
            }
        }  {
            consecutive_detections = ;
        }

         ;
    }
};

Performance Benchmarking

Inference Profiling

#!/usr/bin/env python3
"""Profile TFLite model on target hardware via serial."""

import serial
import json
import statistics


def benchmark_model(port: str, num_runs: int = 100):
    """Send benchmark command and collect timing data."""
    ser = serial.Serial(port, 115200, timeout=5)

    ser.write(b"BENCH\n")
    times = []

    for _ in range(num_runs):
        line = ser.readline().decode().strip()
        if line.startswith("INFER:"):
            us = int(line.split(":")[1])
            times.append(us)

    if times:
        print(f"Inference timing ({num_runs} runs):")
        print(f"  Mean:   {statistics.mean(times):>8.1f} us")
        print(f"  Median: {statistics.median(times):>8.1f} us")
        print(f"  Std:    {statistics.stdev(times):>8.1f} us")
        print(f"  Min:    {min(times):>8d} us")
        print(f"  Max:    {max(times):>8d} us")
        print(f"  FPS:    ")

    ser.close()

Model Optimization Checklist

Step	Action	Tool
1	Profile baseline model size and accuracy	TF Model Summary
2	Replace Conv2D with DepthwiseConv2D	Manual architecture
3	Reduce input resolution if possible	Data pipeline
4	Apply post-training int8 quantization	TFLite Converter
5	Prune weights below threshold	TF Model Optimization Toolkit
6	Measure on-device inference time	Serial profiling
7	Reduce arena size to minimum	Binary search
8	Test accuracy on held-out validation set	Python evaluation

Common Pitfalls

Mistake	Impact	Solution
Training on desktop data only	Poor real-world accuracy	Collect data on target hardware
Float32 model on MCU	Too large, too slow	Always quantize to int8
Oversized arena	Wasted RAM	Profile and minimize arena
No data augmentation	Overfitting to lab conditions	Add noise, shifts, scaling
Ignoring preprocessing	Input format mismatch	Match train-time preprocessing exactly
One-shot detection	False positives	Require consecutive detections
No model versioning	Deployment confusion	Version models, track in firmware

Exercises

1. Gesture Classifier: Train a 3-class accelerometer gesture model (<20KB), deploy on Arduino Nano 33 BLE Sense
2. Anomaly Detector: Build an autoencoder for vibration anomaly detection on ESP32, trigger alert on reconstruction error
3. Keyword Spotter: Train a 4-keyword audio model using MFCC features, deploy with streaming inference
4. Quantization Study: Compare float32, float16, and int8 model variants for size, speed, and accuracy on the same task
5. Power Profiler: Measure current draw during inference vs idle, calculate battery life for continuous classification

Process

1. Gather information. Ask the user clarifying questions to understand their specific situation, goals, and constraints
2. Analyze context. Review the information provided and identify key factors relevant to edge ml engineer
3. Develop recommendations. Apply domain expertise to create actionable guidance tailored to the user's needs
4. Present structured output. Deliver findings in the output format below with clear next steps
5. Address follow-ups. Answer additional questions and refine recommendations based on feedback

Output Format

## Edge Ml Engineer Analysis

### Assessment
[Key findings and observations]

### Recommendations
1. [Primary recommendation]
2. [Secondary recommendation]
3. [Additional suggestions]

### Action Items
- [ ] [First action step]
- [ ] [Second action step]
- [ ] [Follow-up task]

Edge Cases

• Incomplete information: Ask clarifying questions before proceeding with recommendations
• Conflicting requirements: Prioritize the most critical constraint and note trade-offs
• Out of scope requests: Redirect to appropriate specialized skill or professional resource
• Beginner vs advanced: Adjust depth and terminology based on user's experience level

Example

Input: "Help me with edge ml engineer for my current situation"

Output:

Based on your situation, here is a structured approach to edge ml engineer:

1. Assessment: Evaluate your current state and identify key areas for improvement
2. Strategy: Develop a targeted plan based on best practices
3. Implementation: Execute the plan with specific, measurable steps
4. Review: Monitor progress and adjust as needed