TinyML & Edge ML — Machine Learning on IoT Devices Guide

import tensorflow as tf
import numpy as np

# Train a simple model
model = tf.keras.Sequential([
    tf.keras.layers.Dense(16, activation='relu', input_shape=(3,)),
    tf.keras.layers.Dense(8, activation='relu'),
    tf.keras.layers.Dense(1, activation='sigmoid')
])
model.compile(optimizer='adam', loss='binary_crossentropy', metrics=['accuracy'])

# Generate synthetic sensor data
X_train = np.random.randn(1000, 3)
y_train = (X_train[:, 0] + X_train[:, 1] > 0).astype(float)

model.fit(X_train, y_train, epochs=10, verbose=0)
print(f"Test accuracy: {model.evaluate(X_train, y_train, verbose=0)[1]:.3f}")

# Convert to TensorFlow Lite
converter = tf.lite.TFLiteConverter.from_keras_model(model)
tflite_model = converter.convert()
print(f"Float model size: {len(tflite_model)} bytes")

# Quantize to 8-bit integer
converter.optimizations = [tf.lite.Optimize.DEFAULT]
converter.target_spec.supported_types = [tf.int8]
converter.representative_dataset = lambda: [
    {'lstm_input': X_train[i:i+1].astype(np.float32)}
    for i in range(100)
]
quantized_model = converter.convert()
print(f"Quantized model size: {len(quantized_model)} bytes")

#include <TensorFlowLite_ESP32.h>
#include "sensor_model.h"  // Converted .h file from TFLite

tflite::MicroMutableOpResolver<10> resolver;
tflite::ErrorReporter* error_reporter;
const tflite::Model* model;
tflite::MicroInterpreter* interpreter;

// Tensor arena (model workspace) — must be at least model_size * 10
constexpr int kTensorArenaSize = 8 * 1024;
uint8_t tensor_arena[kTensorArenaSize];

void setup() {
  Serial.begin(115200);

  model = tflite::GetModel(sensor_model_tflite);
  if (model->version() != TFLITE_SCHEMA_VERSION) {
    Serial.println("Model schema mismatch");
    return;
  }

  resolver.AddFullyConnected();
  resolver.AddSoftmax();

  interpreter = new tflite::MicroInterpreter(
    model, resolver, tensor_arena, kTensorArenaSize, error_reporter
  );

  interpreter->AllocateTensors();
  Serial.println("TinyML model loaded");
}

void loop() {
  // Read sensor values (temperature, humidity, light)
  float sensor_data[3] = {24.5, 55.2, 0.8};

  // Copy into model input tensor
  TfLiteTensor* input = interpreter->input(0);
  for (int i = 0; i < 3; i++) {
    input->data.f[i] = sensor_data[i];
  }

  // Run inference
  interpreter->Invoke();

  // Read output
  TfLiteTensor* output = interpreter->output(0);
  float anomaly_score = output->data.f[0];

  if (anomaly_score > 0.8) {
    Serial.println("ALERT: Anomaly detected!");
  }

  delay(1000);
}

import numpy as np
import pickle

class OnDeviceAnomalyDetector:
    def __init__(self, threshold: float = 3.0):
        self.mean = None
        self.std = None
        self.threshold = threshold
        self.trained = False

    def train(self, data: np.ndarray):
        self.mean = np.mean(data, axis=0)
        self.std = np.std(data, axis=0)
        self.trained = True

    def predict(self, sample: np.ndarray) -> tuple[bool, float]:
        if not self.trained:
            raise ValueError("Model not trained")
        z_scores = np.abs((sample - self.mean) / (self.std + 1e-8))
        score = np.max(z_scores)
        return bool(score > self.threshold), score

    def export_c_header(self, filename: str = "anomaly_model.h"):
        with open(filename, "w") as f:
            f.write(f"#ifndef ANOMALY_MODEL_H\n#define ANOMALY_MODEL_H\n\n")
            f.write(f"const float MODEL_MEAN[] = {{")
            f.write(", ".join(f"{v:.4f}f" for v in self.mean))
            f.write("};\n\n")
            f.write(f"const float MODEL_STD[] = {{")
            f.write(", ".join(f"{v:.4f}f" for v in self.std))
            f.write("};\n\n")
            f.write(f"const float MODEL_THRESHOLD = {self.threshold}f;\n")
            f.write(f"#endif\n")

# Training data from a healthy motor
normal_vibrations = np.random.randn(1000, 3) * 0.5 + np.array([1.0, 0.5, 0.2])

detector = OnDeviceAnomalyDetector(threshold=3.0)
detector.train(normal_vibrations)

# Test on new readings
normal_reading = np.array([0.8, 0.6, 0.3])
anomaly_reading = np.array([4.2, 3.8, 1.5])

for label, sample in [("Normal", normal_reading), ("Anomaly", anomaly_reading)]:
    is_anomaly, score = detector.predict(sample)
    print(f"{label}: anomaly={is_anomaly}, score={score:.2f}")

detector.export_c_header()

import tensorflow as tf
import numpy as np

# Simplified keyword spotting model
def build_keyword_model(input_size: int = 49) -> tf.keras.Model:
    inputs = tf.keras.Input(shape=(input_size, 4))
    x = tf.keras.layers.DepthwiseConv2D((3, 3), activation='relu')(inputs)
    x = tf.keras.layers.GlobalAveragePooling2D()(x)
    x = tf.keras.layers.Dense(16, activation='relu')(x)
    outputs = tf.keras.layers.Dense(2, activation='softmax')(x)
    model = tf.keras.Model(inputs, outputs)
    return model

model = build_keyword_model()
model.summary()

converter = tf.lite.TFLiteConverter.from_keras_model(model)
converter.optimizations = [tf.lite.Optimize.DEFAULT]
quantized_model = converter.convert()
print(f"Keyword spotter model: {len(quantized_model)} bytes")

import random, math, time

class TinyAnomalyDetector:
    def __init__(self):
        self.baseline = {"mean": [0.5, 0.3, 0.1], "std": [0.2, 0.15, 0.08]}
        self.threshold = 3.0

    def read_vibration(self) -> list[float]:
        x = random.gauss(0.5, 0.2)
        y = random.gauss(0.3, 0.15)
        z = random.gauss(0.1, 0.08)
        return [x, y, z]

    def detect(self, sample: list[float]) -> tuple[bool, float]:
        z_scores = []
        for i in range(3):
            z = abs(sample[i] - self.baseline["mean"][i]) / self.baseline["std"][i]
            z_scores.append(z)
        score = max(z_scores)
        return (score > self.threshold), round(score, 2)

detector = TinyAnomalyDetector()

for i in range(10):
    sample = detector.read_vibration()
    is_anomaly, score = detector.detect(sample)
    status = "ANOMALY" if is_anomaly else "OK"
    print(f"[{status}] Score: {score:.2f} | X:{sample[0]:.3f} Y:{sample[1]:.3f} Z:{sample[2]:.3f}")
    time.sleep(0.2)