1 of 7

EI Python SDK

While the Edge Impulse Studio is a great interface for guiding you through the process of collecting data and training a model, the edgeimpulse Python SDK allows you to programmatically Bring Your Own Model (BYOM), developed and trained on any platform. See here for the Python SDK API reference documentation.

With the following tutorials, you will learn how to use the Edge Impulse Python SDK with a number of other machine-learning frameworks and platforms:

Using the Edge Impulse Python SDK with TensorFlow and Keras
Using the Edge Impulse Python SDK with Hugging Face
Using the Edge Impulse Python SDK with Weights & Biases
Using the Edge Impulse Python SDK with SageMaker Studio
Using the Edge Impulse Python SDK to upload and download data

Using the Edge Impulse Python SDK with TensorFlow and Keras

TensorFlow is an open source library for training machine learning models. Keras is an open source Python library that makes creating neural networks in TensorFlow much easier. We use these two libraries together to very quickly train a model to identify handwritten digits. From there, we use the Edge Impulse Python SDK library to profile the model to see how inference will perform on a target edge device. Then, we use the SDK again to convert our trained model to a C++ library that can be deployed to an edge hardware platform, such as a microcontroller.

Follow the code below to see how to train a simple machine learning model and deploy it to a C++ library using Edge Impulse.

To learn more about using the Python SDK, please see: Edge Impulse Python SDK Overview.

# If you have not done so already, install the following dependencies
!python -m pip install tensorflow==2.12.0 edgeimpulse

from tensorflow import keras
import edgeimpulse as ei

You will need to obtain an API key from an Edge Impulse project. Log into edgeimpulse.com and create a new project. Open the project, navigate to Dashboard and click on the Keys tab to view your API keys. Double-click on the API key to highlight it, right-click, and select Copy.

Note that you do not actually need to use the project in the Edge Impulse Studio. We just need the API Key.

Paste that API key string in the ei.API_KEY value in the following cell:

# Settings
ei.API_KEY = "ei_dae2..." # Change this to your Edge Impulse API key
labels = ["0", "1", "2", "3", "4", "5", "6", "7", "8", "9"]
num_classes = len(labels)
deploy_filename = "my_model_cpp.zip"

Train a machine learning model

We want to create a classifier that can uniquely identify handwritten digits. To start, we will use TensorFlow and Keras to train a very simple convolutional neural network (CNN) on the classic MNIST dataset, which consists of handwritten digits from 0 to 9.

# Load MNIST data
(x_train, y_train), (x_test, y_test) = keras.datasets.mnist.load_data()
x_train = keras.utils.normalize(x_train, axis=1)
x_test = keras.utils.normalize(x_test, axis=1)
y_train = keras.utils.to_categorical(y_train, num_classes)
y_test = keras.utils.to_categorical(y_test, num_classes)
input_shape = x_train[0].shape

# Build the model 
model = keras.Sequential([
    keras.layers.Flatten(),
    keras.layers.Dense(32, activation='relu', input_shape=input_shape),
    keras.layers.Dense(num_classes, activation='softmax')
])

# Compile the model
model.compile(optimizer='adam',
              loss='categorical_crossentropy',
              metrics=['accuracy'])

# Train the model
model.fit(x_train, 
          y_train, 
          epochs=5)

# Evaluate model on test set
score = model.evaluate(x_test, y_test, verbose=0)
print(f"Test loss: {score[0]}")
print(f"Test accuracy: {score[1]}")

Profile your model

To start, we need to list the possible target devices we can use for profiling. We need to pick from this list.

# List the available profile target devices
ei.model.list_profile_devices()

You should see a list printed such as:

['alif-he',
 'alif-hp',
 'arduino-nano-33-ble',
 'arduino-nicla-vision',
 'portenta-h7',
 'brainchip-akd1000',
 'cortex-m4f-80mhz',
 'cortex-m7-216mhz',
 ...
 'ti-tda4vm']

A common option is the cortex-m4f-80mhz, as this is a relatively low-power microcontroller family. From there, we can use the Edge Impulse Python SDK to generate a profile for your model to ensure it fits on your target hardware and meets your timing requirements.

# Estimate the RAM, ROM, and inference time for our model on the target hardware family
try:
    profile = ei.model.profile(model=model,
                               device='cortex-m4f-80mhz')
    print(profile.summary())
except Exception as e:
    print(f"Could not profile: {e}")

Deploy your model

Once you are happy with the performance of the model, you can deploy it to a number of possible hardware targets. To see the available hardware targets, run the following:

# List the available profile target devices
ei.model.list_deployment_targets()

You should see a list printed such as:

['zip',
 'arduino',
 'tinkergen',
 'cubemx',
 'wasm',
 ...
 'runner-linux-aarch64-tda4vm']

The most generic target is to download a .zip file that holds a C++ library containing the inference runtime and your trained model, so we choose 'zip' from the above list. To do that, we first need to create a Classification object which contains our label strings (and other optional information about the model). These strings will be added to the C++ library metadata so you can access them in your edge application.

Note that instead of writing the raw bytes to a file, you can also specify an output_directory argument in the .deploy() function. Your deployment file(s) will be downloaded to that directory.

Important! The deployment targets list will change depending on the values provided for model, model_output_type, and model_input_type in the next part. For example, you will not see openmv listed once you upload a model (e.g. using .profile() or .deploy()) if model_input_type is not set to ei.model.input_type.ImageInput(). If you attempt to deploy to an unavailable target, you will receive the error Could not deploy: deploy_target: .... If model_input_type is not provided, it will default to OtherInput. See this page for more information about input types.

# Set model information, such as your list of labels
model_output_type = ei.model.output_type.Classification(labels=labels)

# Set model input type
model_input_type = ei.model.input_type.OtherInput()

# Create C++ library with trained model
deploy_bytes = None
try:
    
    deploy_bytes = ei.model.deploy(model=model,
                                   model_output_type=model_output_type,
                                   model_input_type=model_input_type,
                                   deploy_target='zip')
except Exception as e:
    print(f"Could not deploy: {e}")
    
# Write the downloaded raw bytes to a file
if deploy_bytes:
    with open(deploy_filename, 'wb') as f:
        f.write(deploy_bytes.getvalue())

Your model C++ library should be downloaded as the file my_model_cpp.zip in the same directory as this notebook. You are now ready to use your C++ model in your embedded and edge device application! To use the C++ model for local inference, see our documentation here.

Using the Edge Impulse Python SDK to run EON Tuner

The EON Tuner is Edge Impulse's automated machine learning (AutoML) tool to help you find the best combination of blocks and hyperparameters for your model and within your hardware constraints. This example will walk you through uploading data, running the EON Tuner, and interpreting the results.

WARNING: This notebook will add and delete data in your Edge Impulse project, so be careful! We recommend creating a throwaway project when testing this notebook.

To start, create a new project in Edge Impulse. Do not add any data to it.

# If you have not done so already, install the following dependencies
# !python -m pip install matplotlib pandas edgeimpulse

import edgeimpulse as ei
from edgeimpulse.experimental.data import (
    upload_directory
)
from edgeimpulse.experimental.tuner import (
    check_tuner,
    set_impulse_from_trial,
    start_tuner,
    start_custom_tuner,
    tuner_report_as_df,
)
from edgeimpulse.experimental.impulse import (
    build,
)

Note that you do not actually need to use the project in the Edge Impulse Studio. We just need the API Key.

Paste that API key string in the ei.API_KEY value in the following cell:

# Settings
ei.API_KEY = "ei_dae2..." # Change this to your Edge Impulse API key
deploy_filename = "my_model_cpp.zip"

# Get the project ID
api = ei.experimental.api.EdgeImpulseApi()
project_id = api.default_project_id()

Upload dataset

We start by downloading the continuous motion dataset and uploading it to our project.

# Download and unzip gesture dataset
!mkdir -p dataset/
!wget -P dataset -q https://cdn.edgeimpulse.com/datasets/gestures.zip
!unzip -q dataset/gestures.zip -d dataset/gestures/

# Upload training dataset
resp = upload_directory(
    directory="dataset/gestures/training",
    category="training",
)
print(f"Uploaded {len(resp.successes)} training samples")

# Upload test dataset
resp = upload_directory(
    directory="dataset/gestures/testing",
    category="testing",
)
print(f"Uploaded {len(resp.successes)} testing samples")

# Uncomment the following if you want to delete the temporary dataset folder
#!rm -rf dataset/

Run the Tuner

To start, we need to list the possible target devices we can use for profiling. We need to pick from this list.

# List the available profile targets
ei.model.list_profile_devices()

You should see a list printed such as:

['alif-he',
 'alif-hp',
 'arduino-nano-33-ble',
 'arduino-nicla-vision',
 'portenta-h7',
 'brainchip-akd1000',
 'cortex-m4f-80mhz',
 'cortex-m7-216mhz',
 ...
 'ti-tda4vm']

From there, we start the tuner with start_tuner() and wait for completion via check_tuner(). In this example, we configure the tuner to target for the cortex-m4f-80mhz device. Since we want to classify the motion, we choose classification for our classifcation_type and our dataset as motion continuous. We constrain our model to a latency of 100ms for running the impulse.

NOTE: We set the max trials to 3 here. In a real life situation, you will omit this so the tuner decides the best number of trials.

Once the tuner is done, you can print out the results to determine the best combination of blocks and hyperparameters.

# Choose a device from the list
target_device = "cortex-m4f-80mhz"

# Start tuner. This will take 15+ minutes.
start_tuner(
    target_device=target_device,
    classification_type="classification",
    dataset_category="motion_continuous",
    target_latency=100,
    tuning_max_trials=3,
)

# Wait while checking the tuner's progress.
state = check_tuner(
    wait_for_completion=True
)

Print EON Tuner results

To visualize the results of the tuner trials, you can head to the project page on Edge Impulse Studio.

Alternatively, you can access the results programmatically: the configuration settings and output of the EON Tuner is stored in the variable state. You can access the results of the various trials with state.trials. Note that some trials can fail, so it's a good idea to test the status of each trial.

From there, you will want to sort the results based on some metric. In this example, we will sort based on int8 test set accuracy from highest to lowest.

Note: Edge Impulse supports only one learning block per project at this time (excluding anomaly detection blocks). As a result, we will use the first learning block (e.g. learning_blocks[0]) in the list to extract metrics.

import json

# The easiest way to view the results is to look at the EON Tuner page on your project
print(f"Navigate to https://studio.edgeimpulse.com/studio/{project_id}/tuner to see the results")

# Set quantization ("float32" or "int8")
qtzn = "int8"

# Filter out all failed trials
results = [r for r in state.trials if r.status == "completed"]

# Extract int8 accuracies from the trial results
accuracies = []
for result in results:
    accuracy = result.impulse.learn_blocks[0]["metrics"]["test"][qtzn]["accuracy"]
    accuracies.append(accuracy)

# Sort the results based on int8 accuracies
acc_results = zip(accuracies, results)
sorted_results = sorted(acc_results, reverse=True, key=lambda x: list(x)[0])
sorted_results = [result for _, result in sorted_results]

Now that we have the sorted results, we can extract the values we care about. We will print out the following metrics along with the impulse configuration (processing/learning block configuration and hyperparameters) of the top-performing trial.

This will help you determine if the impulse can fit on your target hardware and run fast enough for your needs. The impulse configuration can be used to recreate the processing and learning blocks on Edge Impulse. Later, we will set the project impulse based on the trial ID to simply deploy (rather than re-train).

Note: we assume the first learning block has the metrics we care about.

def get_metrics(results, qtzn, idx):
    """Calculate metrics for a given trial index"""

    metrics = {}

    # Get model accuracy results
    result_metrics = results[idx].impulse.learn_blocks[0]["metrics"]
    metrics["val-acc"] = result_metrics['validation'][qtzn]['accuracy']
    metrics["test-acc"] = result_metrics['test'][qtzn]['accuracy']
    
    # Calculate processing block RAM
    metrics["processing-block-ram"] = 0
    for i, dsp_block in enumerate(results[idx].impulse.dsp_blocks):
        metrics["processing-block-ram"] += dsp_block["performance"]["ram"]

    # Get latency, RAM, and ROM usage
    device_performance = results[idx].device_performance[qtzn]
    metrics["learning-block-latency-ms"] = device_performance['latency']
    metrics["learning-block-tflite-ram"] = device_performance['tflite']['ramRequired']
    metrics["learning-block-tflite-rom"] = device_performance['tflite']['romRequired']
    metrics["learning-block-eon-ram"] = device_performance['eon']['ramRequired']
    metrics["learning-block-eon-rom"] = device_performance['eon']['romRequired']

    return metrics

# The top performing impulse is the first element (sorted by highest int8 accuracy on test set)
trial_idx = 0

# Print info about the processing (DSP) blocks and store RAM usage
print("Processing blocks")
print("===")
for i, dsp_block in enumerate(sorted_results[trial_idx].impulse.dsp_blocks):
    print(f"Processing block {i}")
    print("---")
    print("Block:")
    print(json.dumps(dsp_block["block"], indent=2))
    print("Config:")
    print(json.dumps(dsp_block["config"], indent=2))
print()

# Print info about the learning blocks
print("Learning blocks")
print("===")
for i, learn_block in enumerate(sorted_results[trial_idx].impulse.learn_blocks):
    print(f"Learn block {i}")
    print("---")
    print("Block:")
    print(json.dumps(learn_block["block"], indent=2))
    print("Config:")
    print(json.dumps(learn_block["config"], indent=2))
    metadata = learn_block["metadata"]
    qtzn_metadata = [m for m in metadata["modelValidationMetrics"] if m.get("type") == qtzn]
print()

# Print metrics
metrics = get_metrics(sorted_results, qtzn, trial_idx)
print(f"Metrics ({qtzn}) for best trial")
print("===")
print(f"Validation accuracy: {metrics['val-acc']}")
print(f"Test accuracy: {metrics['test-acc']}")
print(f"Estimated processing blocks RAM (bytes): {metrics['processing-block-ram']}")
print(f"Estimated learning blocks latency (ms): {metrics['learning-block-latency-ms']}")
print(f"Estimated learning blocks RAM (bytes): {metrics['learning-block-tflite-ram']}")
print(f"Estimated learning blocks ROM (bytes): {metrics['learning-block-tflite-rom']}")
print(f"Estimated learning blocks RAM with EON Compiler (bytes): {metrics['learning-block-eon-ram']}")
print(f"Estimated learning blocks ROM with EON Compiler (bytes): {metrics['learning-block-eon-rom']}")

Graph results

You can optionally use a plotting package like matplotlib to graph the results from the top results to compare the metrics.

import matplotlib.pyplot as plt

# Get metrics for the top 3 trials (sorted by int8 test set accuracy)
num_trials = 3
top_metrics = [get_metrics(sorted_results, qtzn, idx) for idx in range(num_trials)]

# Construct metrics for plotting
test_accs = [top_metrics[x]['test-acc'] for x in range(num_trials)]
proc_rams = [top_metrics[x]['processing-block-ram'] for x in range(num_trials)]
learn_latencies = [top_metrics[x]['learning-block-latency-ms'] for x in range(num_trials)]
learn_tflite_rams = [top_metrics[x]['learning-block-tflite-ram'] for x in range(num_trials)]
learn_tflite_roms = [top_metrics[x]['learning-block-tflite-rom'] for x in range(num_trials)]
learn_eon_rams = [top_metrics[x]['learning-block-eon-ram'] for x in range(num_trials)]
learn_eon_roms = [top_metrics[x]['learning-block-eon-rom'] for x in range(num_trials)]

# Create plots
fig, axs = plt.subplots(7, 1, figsize=(8, 15))
indices = range(num_trials)

# Plot test accuracies
axs[0].barh(indices, test_accs)
axs[0].set_title("Test set accuracy")
axs[0].set_xlabel("Accuracy")
axs[0].set_ylabel("Trial")

# Plot processing block RAM
axs[1].barh(indices, proc_rams)
axs[1].set_title("Processing block RAM")
axs[1].set_xlabel("RAM (bytes)")
axs[1].set_ylabel("Trial")

# Plot learning block latency
axs[2].barh(indices, learn_latencies)
axs[2].set_title("Learning block latency")
axs[2].set_xlabel("Latency (ms)")
axs[2].set_ylabel("Trial")

# Plot learning block RAM (TFLite)
axs[3].barh(indices, learn_tflite_rams)
axs[3].set_title("Learning block RAM (TFLite)")
axs[3].set_xlabel("RAM (bytes)")
axs[3].set_ylabel("Trial")

# Plot learning block ROM (TFLite)
axs[4].barh(indices, learn_tflite_roms)
axs[4].set_title("Learning block ROM (TFLite)")
axs[4].set_xlabel("ROM (bytes)")
axs[4].set_ylabel("Trial")

# Plot learning block RAM (EON)
axs[5].barh(indices, learn_eon_rams)
axs[5].set_title("Learning block RAM (EON)")
axs[5].set_xlabel("RAM (bytes)")
axs[5].set_ylabel("Trial")

# Plot learning block ROM (EON)
axs[6].barh(indices, learn_eon_roms)
axs[6].set_title("Learning block ROM (EON)")
axs[6].set_xlabel("ROM (bytes)")
axs[6].set_ylabel("Trial")

# Prevent overlap
plt.tight_layout()

Results as a DataFrame

If you have pandas installed, you can make the previous section much easier by reporting metrics as a DataFrame.

import pandas as pd

# Convert the state metrics into a DataFrame
df = tuner_report_as_df(state)
df.head()

# Print column names
for col in df.columns:
    print(col)

# Sort the DataFrame by validation (int8) accuracy
df = df.sort_values(by="test_int8_accuracy", ascending=False)

# Print out best trial metrics
print(f"Trial ID: {df.iloc[0]['id']}")
print(f"Test accuracy (int8): {df.iloc[0]['test_int8_accuracy']}")
print(f"Estimated learning blocks latency (ms): {df.iloc[0]['device_performance_int8_latency']}")
print(f"Estimated learning blocks RAM (bytes): {df.iloc[0]['device_performance_int8_tflite_ram_required']}")
print(f"Estimated learning blocks ROM (bytes): {df.iloc[0]['device_performance_int8_tflite_rom_required']}")
print(f"Estimated learning blocks RAM with EON Compiler (bytes): {df.iloc[0]['device_performance_int8_eon_ram_required']}")
print(f"Estimated learning blocks ROM with EON Compiler (bytes): {df.iloc[0]['device_performance_int8_eon_rom_required']}")

Set trial as impulse and deploy

We can replace the current impulse with the top performing trial from the EON Tuner. From there, we can deploy it, just like we would any impulse.

# Get the ID for the top-performing trial and set that to our impulse. This will take about a minute.
trial_id = df.iloc[0].trial_id
response = set_impulse_from_trial(trial_id)
job_id = response.id

# Make sure the impulse update was successful
if not hasattr(response, "success") or getattr(response, "success") == False:
    raise RuntimeError("Could not set project impulse to trial impulse")

# List the available profile target devices
ei.model.list_deployment_targets()

You should see a list printed such as:

['zip',
 'arduino',
 'cubemx',
 'wasm',
 ...
 'runner-linux-aarch64-jetson-orin-6-0']

Note that instead of writing the raw bytes to a file, you can also specify an output_directory argument in the .deploy() function. Your deployment file(s) will be downloaded to that directory.

# Build and download C++ library with the trained model
deploy_bytes = None
try:
    deploy_bytes = build(
        deploy_model_type=qtzn,
        engine="tflite",
        deploy_target="zip"
    )
except Exception as e:
    print(f"Could not deploy: {e}")
    
# Write the downloaded raw bytes to a file
if deploy_bytes:
    with open(deploy_filename, 'wb') as f:
        f.write(deploy_bytes.getvalue())

Configure custom search space

By default, the EON Tuner will make a guess at a search space based on the type of data you uploaded (e.g. using spectral-analysis blocks for feature extraction). As a result, you can run the tuner without needing to construct a search space. However, you may want to define your own search space.

The best way to define a search space is to open your project (after uploading data), head to the EON Tuner page, click Run EON Tuner, and select the Space tab.

The search space is defined in JSON format, so we can just copy that to create a dictionary. This is a good place to start for tuning blocks and hyperparameters.

Note: Functions to get available blocks and search space parameters coming soon

from edgeimpulse_api import (
    OptimizeConfig,
    TunerSpaceImpulse,
)

# Configure the search space
space = {
    "inputBlocks": [
      {
        "type": "time-series",
        "window": [
          {"windowSizeMs": 9000, "windowIncreaseMs": 9000},
          {"windowSizeMs": 10000, "windowIncreaseMs": 10000}
        ],
        "frequencyHz": [62.5],
        "padZeros": [True]
      }
    ],
    "dspBlocks": [
      {
        "type": "spectral-analysis",
        "analysis-type": ["FFT"],
        "fft-length": [16, 64],
        "scale-axes": [1],
        "filter-type": ["none"],
        "filter-cutoff": [3],
        "filter-order": [6],
        "do-log": [True],
        "do-fft-overlap": [True]
      },
      {
        "type": "spectral-analysis",
        "analysis-type": ["Wavelet"],
        "wavelet": ["haar", "bior1.3"],
        "wavelet-level": [1, 2]
      },
      {"type": "raw", "scale-axes": [1]}
    ],
    "learnBlocks": [
      {
        "id": 4,
        "type": "keras",
        "dimension": ["dense"],
        "denseBaseNeurons": [40, 20],
        "denseLayers": [2, 3],
        "dropout": [0.25, 0.5],
        "learningRate": [0.0005],
        "trainingCycles": [30]
      }
    ]
  }

# Wrap the search space
ts = TunerSpaceImpulse.from_dict(space)

# Create a custom configuration
config = OptimizeConfig(
    name=None,
    target_device={"name": "cortex-m4f-80mhz"},
    classification_type="classification",
    dataset_category="motion_continuous",
    target_latency=100,
    tuning_max_trials=2,
    space=[ts]
)

# Start tuner and wait for it to complete
start_custom_tuner(
    config=config
)
state = check_tuner(
    wait_for_completion=True
)

# The easiest way to view the results is to look at the EON Tuner page on your project
print(f"Navigate to https://studio.edgeimpulse.com/studio/{project_id}/tuner to see the results")

# Set quantization ("float32" or "int8")
qtzn = "int8"

# Filter out all failed trials
results = [r for r in state.trials if r.status == "completed"]

# Extract float32 accuracies from the trial results
accuracies = []
for result in results:
    accuracy = result.impulse.learn_blocks[0]["metrics"]["test"][qtzn]["accuracy"]
    accuracies.append(accuracy)

# Sort the results based on int8 accuracies
acc_results = zip(accuracies, results)
sorted_results = sorted(acc_results, reverse=True, key=lambda x: list(x)[0])
sorted_results = [result for _, result in sorted_results]

# The top performing impulse is the first element (sorted by highest int8 accuracy on test set)
trial_idx = 0

# Print info about the processing (DSP) blocks and store RAM usage
print("Processing blocks")
print("===")
for i, dsp_block in enumerate(sorted_results[trial_idx].impulse.dsp_blocks):
    print(f"Processing block {i}")
    print("---")
    print("Block:")
    print(json.dumps(dsp_block["block"], indent=2))
    print("Config:")
    print(json.dumps(dsp_block["config"], indent=2))
print()

# Print info about the learning blocks
print("Learning blocks")
print("===")
for i, learn_block in enumerate(sorted_results[trial_idx].impulse.learn_blocks):
    print(f"Learn block {i}")
    print("---")
    print("Block:")
    print(json.dumps(learn_block["block"], indent=2))
    print("Config:")
    print(json.dumps(learn_block["config"], indent=2))
    metadata = learn_block["metadata"]
    qtzn_metadata = [m for m in metadata["modelValidationMetrics"] if m.get("type") == qtzn]
print()

# Print metrics
metrics = get_metrics(sorted_results, qtzn, trial_idx)
print(f"Metrics ({qtzn}) for best trial")
print("===")
print(f"Validation accuracy: {metrics['val-acc']}")
print(f"Test accuracy: {metrics['test-acc']}")
print(f"Estimated processing blocks RAM (bytes): {metrics['processing-block-ram']}")
print(f"Estimated learning blocks latency (ms): {metrics['learning-block-latency-ms']}")
print(f"Estimated learning blocks RAM (bytes): {metrics['learning-block-tflite-ram']}")
print(f"Estimated learning blocks ROM (bytes): {metrics['learning-block-tflite-rom']}")
print(f"Estimated learning blocks RAM with EON Compiler (bytes): {metrics['learning-block-eon-ram']}")
print(f"Estimated learning blocks ROM with EON Compiler (bytes): {metrics['learning-block-eon-rom']}")

Using the Edge Impulse Python SDK with Hugging Face

🤗 Hugging Face offers a suite of tools that assist with various AI applications. Most notably, they provide a hub for people to share their pre-trained models. In this tutorial, we will demonstrate how to download a simple ResNet model from the Hugging Face hub, profile it, and convert it to a C++ library for use in your edge application. This particular model was trained to identify species of bean plants using the bean dataset.

To learn more about using the Python SDK, please see: Edge Impulse Python SDK Overview

# If you have not done so already, install the following dependencies
!python -m pip install huggingface_hub edgeimpulse

import json
from huggingface_hub import hf_hub_download
import edgeimpulse as ei

Note that you do not actually need to use the project in the Edge Impulse Studio. We just need the API Key.

Paste that API key string in the ei.API_KEY value in the following cell:

# Edge Impulse Settings
ei.API_KEY = "ei_dae2..."
target_device = 'cortex-m4f-80mhz'
deploy_filename = "my_model_cpp.zip"

To download a model from the Hugging Face hub, we need to first find a model. Head to huggingface.co/models. On the left side, click Image Classification to filter under the Tasks tab and under the Libraries tab, filter by ONNX (as the Edge Impulse Python SDK easily accepts ONNX models). You should see the resnet-tiny-beans model trained by user fxmarty.

Click on the resnet-tiny-beans entry (or follow this link) to read about the model and view the files. If you click on the Files* tab, you can see all of the files available in this particular model.

Set the name of the repo (username/repo-name) and the file we want to download.

# Define file location for our model
repo_name = "fxmarty/resnet-tiny-beans"
download_dir = "./"
model_filename = "model.onnx"

# Download pre-trained model
hf_hub_download(repo_id=repo_name,
                filename=model_filename,
                local_dir=download_dir)

Profile your model

To start, we need to list the possible target devices we can use for profiling. We need to pick from this list.

# List the available profile target devices
ei.model.list_profile_devices()

You should see a list printed such as:

['alif-he',
 'alif-hp',
 'arduino-nano-33-ble',
 'arduino-nicla-vision',
 'portenta-h7',
 'brainchip-akd1000',
 'cortex-m4f-80mhz',
 'cortex-m7-216mhz',
 ...
 'ti-tda4vm']

# Estimate the RAM, ROM, and inference time for our model on the target hardware family
try:
    profile = ei.model.profile(model=model_filename,
                               device='cortex-m4f-80mhz')
    print(profile.summary())
except Exception as e:
    print(f"Could not profile: {e}")

Deploy your model

Once you are happy with the performance of the model, you can deploy it to a number of possible hardware targets. To see the available hardware targets, run the following:

# List the available profile target devices
ei.model.list_deployment_targets()

You should see a list printed such as:

['zip',
 'arduino',
 'tinkergen',
 'cubemx',
 'wasm',
 ...
 'runner-linux-aarch64-tda4vm']

The most generic target is to download a .zip file containing a C++ library containing the inference runtime and your trained model, so we choose 'zip' from the above list. We also need to tell Edge Impulse how we are planning to use the model. In this case, we want to perform classification, so we set the output type to Classification.

Note that instead of writing the raw bytes to a file, you can also specify an output_directory argument in the .deploy() function. Your deployment file(s) will be downloaded to that directory.

# Create C++ library with trained model
deploy_bytes = None
try:
    deploy_bytes = ei.model.deploy(model=model_filename,
                                   model_output_type=ei.model.output_type.Classification(),
                                   deploy_target='zip')
except Exception as e:
    print(f"Could not deploy: {e}")
    
# Write the downloaded raw bytes to a file
if deploy_bytes:
    with open(deploy_filename, 'wb') as f:
        f.write(deploy_bytes.getvalue())

Using the Edge Impulse Python SDK with Weights & Biases

Weights & Biases is an online framework for helping manage machine learning training, data versioning, and experiments. When running experiments for edge-focused ML projects, it can be helpful to see the required memory (RAM and ROM) along with estimated inference times of your model for your target hardware. By viewing these metrics, you can quickly gauge if your model will fit onto your target device!

Follow the code below to see how to train a simple machine learning model with different hyperparameters and log those values to the Weights & Biases dashboard.

To learn more about using the Python SDK, please see: Edge Impulse Python SDK Overview

# If you have not done so already, install the following dependencies
!python -m pip install tensorflow==2.12.0 wandb edgeimpulse

from tensorflow import keras
import wandb
import edgeimpulse as ei

Note that you do not actually need to use the project in the Edge Impulse Studio. We just need the API Key.

Paste that API key string in the ei.API_KEY value in the following cell:

# Settings
ei.API_KEY = "ei_dae2..." # Change this to your Edge Impulse API key
labels = ["0", "1", "2", "3", "4", "5", "6", "7", "8", "9"]
num_classes = len(labels)
num_epochs = 5
profile_device = 'cortex-m4f-80mhz' # Run ei.model.list_profile_devices() to see available devices
deploy_filename = "my_model_cpp.zip"

# Define experiment hyperparameters - sweep across number of nodes
project_name = "nodes-sweep"
num_nodes_sweep = [8, 16, 32, 64, 128]

To use Weights and Biases, you will need to create an account on wandb.ai and call the wandb.login() function. This will prompt you to log in to your account. Your credentials should be stored, which allows you to use the wandb package in your Python library.

# Log in to Weights and Biases (will open a prompt)
wandb.login()

Gather a dataset

# Load MNIST data
(x_train, y_train), (x_test, y_test) = keras.datasets.mnist.load_data()
x_train = keras.utils.normalize(x_train, axis=1)
x_test = keras.utils.normalize(x_test, axis=1)
y_train = keras.utils.to_categorical(y_train, num_classes)
y_test = keras.utils.to_categorical(y_test, num_classes)
input_shape = x_train[0].shape

Create an experiment

We want to vary the hyperparameters in our model and see how it affects the accuracy and predicted RAM, ROM, and inference time on our target platform. To do that, we construct a function that builds a simple model using Keras, trains the model, and computes the accuracy and loss from our holdout test set. We then use the Edge Impulse Python SDK to generate a profile of our model for our target hardware. We log the hyperparameter (number of nodes in the hidden layer), test loss, test accuracy, estimated RAM, estimated ROM, and estimated inference time (ms) to our Weights and Biases console.

# Define experiment - Train and test model, log metrics
def do_experiment(num_nodes):

    # Create W&B project
    run = wandb.init(project=project_name,
                     name=f"{num_nodes}-nodes")

    # Build the model (vary number of nodes in the hidden layer)
    model = keras.Sequential([
        keras.layers.Flatten(),
        keras.layers.Dense(num_nodes, activation='relu', input_shape=input_shape),
        keras.layers.Dense(num_classes, activation='softmax')
    ])

    # Compile the model
    model.compile(optimizer='adam',
                  loss='categorical_crossentropy',
                  metrics=['accuracy'])

    # Train the model
    model.fit(x_train, 
              y_train, 
              epochs=num_epochs)
  
    # Evaluate model
    test_loss, test_accuracy = model.evaluate(x_test, y_test)
    
    # Profile model on target device
    try:
        profile = ei.model.profile(model=model,
                                   device=profile_device)
    except Exception as e:
        print(f"Could not profile: {e}")

    # Log metrics
    if profile.success:
        print("Profiling successful. Logging.")
        wandb.log({
            'num_nodes': num_nodes,
            'test_loss': test_loss,
            'test_accuracy': test_accuracy,
            'profile_ram': profile.model.profile_info.float32.memory.tflite.ram,
            'profile_rom': profile.model.profile_info.float32.memory.tflite.rom,
            'inference_time_ms': profile.model.profile_info.float32.time_per_inference_ms
        })
    else:
        print(f"Profiling unsuccessful. Error: {job_resp.error}")

    # Close run
    wandb.finish()

Run the experiment

Now, it's time to run the experiment and log the results in Weights and Biases. Simply call our function and provide a new hyperparameter value for the number of nodes.

# Perform the experiments - check your dashboard in WandB!
for num_nodes in num_nodes_sweep:
    do_experiment(num_nodes)

Head to wandb.ai and log in (if you have not already done so). Under My projects on the left, click on the nodes-sweep project. You can visualize the results of your experiments with the various charts that Weights & Biases offers. For example, here is a parallel coordinates plot that allows you to quickly visualize the different hyperparameters and metrics (including our new edge profile metrics).

If you would like to deploy your model to your target hardware, the Python SDK can help you with that, too. See our documentation here.

Deploy Your Model

Once you are happy with the performance of your model, you can then deploy it to your target hardware. We will assume that 32 nodes in our hidden layer provided the best trade-off of RAM, flash, inference time, and accuracy for our needs. To start, we will retrain the model:

# Build the model 
model = keras.Sequential([
    keras.layers.Flatten(),
    keras.layers.Dense(32, activation='relu', input_shape=input_shape),
    keras.layers.Dense(num_classes, activation='softmax')
])


# Compile the model
model.compile(optimizer='adam',
              loss='categorical_crossentropy',
              metrics=['accuracy'])


# Train the model
model.fit(x_train, 
          y_train, 
          epochs=5)

Next, we should evaluate the model on our holdout test set.

# Evaluate model on test set
score = model.evaluate(x_test, y_test, verbose=0)
print(f"Test loss: {score[0]}")
print(f"Test accuracy: {score[1]}")

From there, we can see the available hardware targets for deployment:

# List the available profile target devices
ei.model.list_deployment_targets()

You should see a list printed such as:

['zip',
 'arduino',
 'tinkergen',
 'cubemx',
 'wasm',
 ...
 'runner-linux-aarch64-tda4vm']

The most generic target is the .zip file that holds a C++ library containing our trained model and inference runtime. To pass our labels to the C++ library, we create a Classification object, which contains our label strings.

Note that instead of writing the raw bytes to a file, you can also specify an output_directory argument in the .deploy() function. Your deployment file(s) will be downloaded to that directory.

# Set model information, such as your list of labels
model_output_type = ei.model.output_type.Classification(labels=labels)

# Create C++ library with trained model
deploy_bytes = None
try:
    
    deploy_bytes = ei.model.deploy(model=model,
                                   model_output_type=model_output_type,
                                   deploy_target='zip')
except Exception as e:
    print(f"Could not deploy: {e}")
    
# Write the downloaded raw bytes to a file
if deploy_bytes:
    with open(deploy_filename, 'wb') as f:
        f.write(deploy_bytes.getvalue())

Using the Edge Impulse Python SDK with SageMaker Studio

Amazon SageMaker Studio is an integrated development environment (IDE) that provides a single web-based visual interface where you can access purpose-built tools to perform all machine learning (ML) development steps, from preparing data to building, training, and deploying your ML models, improving data science team productivity by up to 10x. You can quickly upload data, create new notebooks, train and tune models, move back and forth between steps to adjust experiments, collaborate seamlessly within your organization, and deploy models to production without leaving SageMaker Studio.

To learn more about using the Python SDK, please see: Edge Impulse Python SDK Overview.

This guide has been built from AWS reference project Introduction to SageMaker TensorFlow - Image Classification, please have a look at this AWS documentation page.

Below are the changes made to the original training script and configuration:

The Python 3 (Data Science 3.0) kernel was used.
The dataset has been changed to classify images as car vs unknown. You can download the dataset from this Edge Impulse public project and store it in your S3 bucket.
The dataset has been imported in the Edge Impulse S3 bucket configured when creating the SageMaker Studio domain. Make sure to adapt to your path or use the AWS reference project.
The training instance used is ml.m5.large.

</div> Install dependencies

# If you have not done so already, install the following dependencies
!python -m pip install tensorflow==2.12.0 edgeimpulse

Transfer Learning

Dataset

Below is the structure of our dataset in our S3 bucket

car-vs-unknown
    |--training
        |--car
            |--abc.jpg
            |--def.jpg
        |--unknown
            |--ghi.jpg
            |--jkl.jpg
    |--testing
            |--car
                |--mno.jpg
                |--prs.jpg
            |--unknown
                |--tuv.jpg
                |--wxy.jpg

We have used the default bucket created when configuring SageMaker Studio domain:

import sagemaker, boto3, json
from sagemaker.session import Session

sagemaker_session = Session()
aws_role = sagemaker_session.get_caller_identity_arn()
aws_region = boto3.Session().region_name
print(aws_region)
sess = sagemaker.Session()

bucket = sess.default_bucket()
subfolder = 'car-vs-unknown/training/'

s3 = boto3.client('s3')
files = s3.list_objects(Bucket=bucket, Prefix=subfolder)['Contents']
print(f"Number of images: {len(files)}")
# or print the files
# for f in files:
#     print(f['Key'])

Train

You can continue with the default model, or can choose a different model from the list. Note that this tutorial has been tested with MobileNetv2 based models. A complete list of SageMaker pre-trained models can also be accessed at Sagemaker pre-trained Models.

from sagemaker.jumpstart.notebook_utils import list_jumpstart_models

# Retrieves all image classification models available by SageMaker Built-In Algorithms.
filter_value = "task in ['ic']"
ic_models = list_jumpstart_models(filter=filter_value)
# od_models = list_jumpstart_models()

print(f"Number of models available for inference: {len(ic_models)}")

# display the model-ids.
for model in ic_models:
    print(model)

from sagemaker import image_uris, model_uris

model_id, model_version = "tensorflow-ic-imagenet-mobilenet-v3-small-075-224", "*" # You can change the based model with one from the list generated above

# Retrieve the base model uri
base_model_uri = model_uris.retrieve( model_id=model_id, model_version=model_version, model_scope="inference")

print(base_model_uri)

Optional, ship this next cell if you don't want to retrain the model. And uncomment the last line of the cell after

from sagemaker import image_uris, model_uris, script_uris, hyperparameters
from sagemaker.estimator import Estimator

training_instance_type = "ml.m5.large"

# Retrieve the Docker image
train_image_uri = image_uris.retrieve(model_id=model_id,model_version=model_version,image_scope="training",instance_type=training_instance_type,region=None,framework=None)

# Retrieve the training script
train_source_uri = script_uris.retrieve(model_id=model_id, model_version=model_version, script_scope="training")

# Retrieve the pretrained model tarball for transfer learning
train_model_uri = model_uris.retrieve(model_id=model_id, model_version=model_version, model_scope="training")

# Retrieve the default hyper-parameters for fine-tuning the model
hyperparameters = hyperparameters.retrieve_default(model_id=model_id, model_version=model_version)

# [Optional] Override default hyperparameters with custom values
hyperparameters["epochs"] = "5"

# The sample training data is available in the following S3 bucket
training_data_bucket = f"{bucket}"
training_data_prefix = f"{subfolder}"
# training_data_bucket = f"jumpstart-cache-prod-{aws_region}"
# training_data_prefix = "training-datasets/tf_flowers/"

training_dataset_s3_path = f"s3://{training_data_bucket}/{training_data_prefix}"

output_bucket = sess.default_bucket()
output_prefix = "ic-car-vs-unknown"
s3_output_location = f"s3://{output_bucket}/{output_prefix}/output"

# Create SageMaker Estimator instance
tf_ic_estimator = Estimator(
    role=aws_role,
    image_uri=train_image_uri,
    source_dir=train_source_uri,
    model_uri=train_model_uri,
    entry_point="transfer_learning.py",
    instance_count=1,
    instance_type=training_instance_type,
    max_run=360000,
    hyperparameters=hyperparameters,
    output_path=s3_output_location
)

# Use S3 path of the training data to launch SageMaker TrainingJob
tf_ic_estimator.fit({"training": training_dataset_s3_path}, logs=True)

Retrieve and prepare the newly trained model

def download_from_s3(url):
    # Remove 's3://' prefix from URL
    url = url[5:]
    # Split URL by '/' to extract bucket name and key
    parts = url.split('/')
    bucket_name = parts[0]
    s3_key = '/'.join(parts[1:])
    # Download the file from S3
    s3.download_file(bucket_name, s3_key, 'model.tar.gz')

# Download
trained_model_s3_path = f"{s3_output_location}/{tf_ic_estimator._current_job_name}/output/model.tar.gz"
print(trained_model_s3_path)
download_from_s3(trained_model_s3_path)
# or if you just want to use the based model
#download_from_s3(base_model_uri)

import shutil, os

# Extract the .tar.gz file to a temporary directory
temp_directory = 'tmp'  # Replace with your actual temporary directory
tar_gz_file = 'model.tar.gz'  # Replace with the path to your .tar.gz file

# Create directory if does not exist
if not os.path.exists(temp_directory):
    os.makedirs(temp_directory)

shutil.unpack_archive(tar_gz_file, temp_directory)

import tensorflow as tf

print(tf.__version__)

model = tf.keras.models.load_model('tmp/1/')

converter = tf.lite.TFLiteConverter.from_keras_model(model)
tflite_model = converter.convert()

# Save the model.
with open('model.tflite', 'wb') as f:
  f.write(tflite_model)

Edge Impulse

Note that you do not actually need to use the project in the Edge Impulse Studio. We just need the API Key.

Paste that API key string in the ei.API_KEY value in the following cell:

import edgeimpulse as ei
ei.API_KEY = "ei_0a85c3a5ca92a35ee6f61aab18aadb9d9e167bd152f947f2056a4fb6a60977d8" # Change to your key

ei.model.list_profile_devices()

# Estimate the RAM, ROM, and inference time for our model on the target hardware family
try:
    profile = ei.model.profile(model='model.tflite',
                               device='raspberry-pi-4')
    print(profile.summary())
except Exception as e:
    print(f"Could not profile: {e}")

# List the available profile target devices
ei.model.list_deployment_targets()

# Get the labels from the label_info.json
import json

labels_info = open('tmp/labels_info.json')
labels_obj = json.load(labels_info)
labels = labels_obj['labels']
print(labels)

# Set model information, such as your list of labels
model_output_type = ei.model.output_type.Classification(labels=labels)
deploy_filename = "my_model_cpp.zip"

# Create C++ library with trained model
deploy_bytes = None
try:
    
    deploy_bytes = ei.model.deploy(model=model,
                                   model_output_type=model_output_type,
                                   engine='tflite',
                                   deploy_target='zip"')
except Exception as e:
    print(f"Could not deploy: {e}")
    
# Write the downloaded raw bytes to a file
if deploy_bytes:
    with open(deploy_filename, 'wb') as f:
        f.write(deploy_bytes.getvalue())

Voila! You now have a C++ library ready to be compiled and integrated in your embedded targets. Feel free to have a look at Edge Impulse deployment options on the documentation to understand how you can integrate that to your embedded systems.

You can also have a look at the deployment page of your project to test your model on a web browser or test

Using the Edge Impulse Python SDK to run EON Tuner

WARNING: This notebook will add and delete data in your Edge Impulse project, so be careful! We recommend creating a throwaway project when testing this notebook.

To start, create a new project in Edge Impulse. Do not add any data to it.

# If you have not done so already, install the following dependencies
# !python -m pip install matplotlib pandas edgeimpulse

import edgeimpulse as ei
from edgeimpulse.experimental.data import (
    upload_directory
)
from edgeimpulse.experimental.tuner import (
    check_tuner,
    set_impulse_from_trial,
    start_tuner,
    start_custom_tuner,
    tuner_report_as_df,
)
from edgeimpulse.experimental.impulse import (
    build,
)

Note that you do not actually need to use the project in the Edge Impulse Studio. We just need the API Key.

Paste that API key string in the ei.API_KEY value in the following cell:

# Settings
ei.API_KEY = "ei_dae2..." # Change this to your Edge Impulse API key
deploy_filename = "my_model_cpp.zip"

# Get the project ID
api = ei.experimental.api.EdgeImpulseApi()
project_id = api.default_project_id()

Upload dataset

We start by downloading the continuous motion dataset and uploading it to our project.

# Download and unzip gesture dataset
!mkdir -p dataset/
!wget -P dataset -q https://cdn.edgeimpulse.com/datasets/gestures.zip
!unzip -q dataset/gestures.zip -d dataset/gestures/

# Upload training dataset
resp = upload_directory(
    directory="dataset/gestures/training",
    category="training",
)
print(f"Uploaded {len(resp.successes)} training samples")

# Upload test dataset
resp = upload_directory(
    directory="dataset/gestures/testing",
    category="testing",
)
print(f"Uploaded {len(resp.successes)} testing samples")

# Uncomment the following if you want to delete the temporary dataset folder
#!rm -rf dataset/

Run the Tuner

To start, we need to list the possible target devices we can use for profiling. We need to pick from this list.

# List the available profile targets
ei.model.list_profile_devices()

You should see a list printed such as:

['alif-he',
 'alif-hp',
 'arduino-nano-33-ble',
 'arduino-nicla-vision',
 'portenta-h7',
 'brainchip-akd1000',
 'cortex-m4f-80mhz',
 'cortex-m7-216mhz',
 ...
 'ti-tda4vm']

NOTE: We set the max trials to 3 here. In a real life situation, you will omit this so the tuner decides the best number of trials.

Once the tuner is done, you can print out the results to determine the best combination of blocks and hyperparameters.

# Choose a device from the list
target_device = "cortex-m4f-80mhz"

# Start tuner. This will take 15+ minutes.
start_tuner(
    target_device=target_device,
    classification_type="classification",
    dataset_category="motion_continuous",
    target_latency=100,
    tuning_max_trials=3,
)

# Wait while checking the tuner's progress.
state = check_tuner(
    wait_for_completion=True
)

Print EON Tuner results

To visualize the results of the tuner trials, you can head to the project page on Edge Impulse Studio.

From there, you will want to sort the results based on some metric. In this example, we will sort based on int8 test set accuracy from highest to lowest.

Note: Edge Impulse supports only one learning block per project at this time (excluding anomaly detection blocks). As a result, we will use the first learning block (e.g. learning_blocks[0]) in the list to extract metrics.

import json

# The easiest way to view the results is to look at the EON Tuner page on your project
print(f"Navigate to https://studio.edgeimpulse.com/studio/{project_id}/tuner to see the results")

# Set quantization ("float32" or "int8")
qtzn = "int8"

# Filter out all failed trials
results = [r for r in state.trials if r.status == "completed"]

# Extract int8 accuracies from the trial results
accuracies = []
for result in results:
    accuracy = result.impulse.learn_blocks[0]["metrics"]["test"][qtzn]["accuracy"]
    accuracies.append(accuracy)

# Sort the results based on int8 accuracies
acc_results = zip(accuracies, results)
sorted_results = sorted(acc_results, reverse=True, key=lambda x: list(x)[0])
sorted_results = [result for _, result in sorted_results]

Note: we assume the first learning block has the metrics we care about.

def get_metrics(results, qtzn, idx):
    """Calculate metrics for a given trial index"""

    metrics = {}

    # Get model accuracy results
    result_metrics = results[idx].impulse.learn_blocks[0]["metrics"]
    metrics["val-acc"] = result_metrics['validation'][qtzn]['accuracy']
    metrics["test-acc"] = result_metrics['test'][qtzn]['accuracy']
    
    # Calculate processing block RAM
    metrics["processing-block-ram"] = 0
    for i, dsp_block in enumerate(results[idx].impulse.dsp_blocks):
        metrics["processing-block-ram"] += dsp_block["performance"]["ram"]

    # Get latency, RAM, and ROM usage
    device_performance = results[idx].device_performance[qtzn]
    metrics["learning-block-latency-ms"] = device_performance['latency']
    metrics["learning-block-tflite-ram"] = device_performance['tflite']['ramRequired']
    metrics["learning-block-tflite-rom"] = device_performance['tflite']['romRequired']
    metrics["learning-block-eon-ram"] = device_performance['eon']['ramRequired']
    metrics["learning-block-eon-rom"] = device_performance['eon']['romRequired']

    return metrics

# The top performing impulse is the first element (sorted by highest int8 accuracy on test set)
trial_idx = 0

# Print info about the processing (DSP) blocks and store RAM usage
print("Processing blocks")
print("===")
for i, dsp_block in enumerate(sorted_results[trial_idx].impulse.dsp_blocks):
    print(f"Processing block {i}")
    print("---")
    print("Block:")
    print(json.dumps(dsp_block["block"], indent=2))
    print("Config:")
    print(json.dumps(dsp_block["config"], indent=2))
print()

# Print info about the learning blocks
print("Learning blocks")
print("===")
for i, learn_block in enumerate(sorted_results[trial_idx].impulse.learn_blocks):
    print(f"Learn block {i}")
    print("---")
    print("Block:")
    print(json.dumps(learn_block["block"], indent=2))
    print("Config:")
    print(json.dumps(learn_block["config"], indent=2))
    metadata = learn_block["metadata"]
    qtzn_metadata = [m for m in metadata["modelValidationMetrics"] if m.get("type") == qtzn]
print()

# Print metrics
metrics = get_metrics(sorted_results, qtzn, trial_idx)
print(f"Metrics ({qtzn}) for best trial")
print("===")
print(f"Validation accuracy: {metrics['val-acc']}")
print(f"Test accuracy: {metrics['test-acc']}")
print(f"Estimated processing blocks RAM (bytes): {metrics['processing-block-ram']}")
print(f"Estimated learning blocks latency (ms): {metrics['learning-block-latency-ms']}")
print(f"Estimated learning blocks RAM (bytes): {metrics['learning-block-tflite-ram']}")
print(f"Estimated learning blocks ROM (bytes): {metrics['learning-block-tflite-rom']}")
print(f"Estimated learning blocks RAM with EON Compiler (bytes): {metrics['learning-block-eon-ram']}")
print(f"Estimated learning blocks ROM with EON Compiler (bytes): {metrics['learning-block-eon-rom']}")

Graph results

You can optionally use a plotting package like matplotlib to graph the results from the top results to compare the metrics.

import matplotlib.pyplot as plt

# Get metrics for the top 3 trials (sorted by int8 test set accuracy)
num_trials = 3
top_metrics = [get_metrics(sorted_results, qtzn, idx) for idx in range(num_trials)]

# Construct metrics for plotting
test_accs = [top_metrics[x]['test-acc'] for x in range(num_trials)]
proc_rams = [top_metrics[x]['processing-block-ram'] for x in range(num_trials)]
learn_latencies = [top_metrics[x]['learning-block-latency-ms'] for x in range(num_trials)]
learn_tflite_rams = [top_metrics[x]['learning-block-tflite-ram'] for x in range(num_trials)]
learn_tflite_roms = [top_metrics[x]['learning-block-tflite-rom'] for x in range(num_trials)]
learn_eon_rams = [top_metrics[x]['learning-block-eon-ram'] for x in range(num_trials)]
learn_eon_roms = [top_metrics[x]['learning-block-eon-rom'] for x in range(num_trials)]

# Create plots
fig, axs = plt.subplots(7, 1, figsize=(8, 15))
indices = range(num_trials)

# Plot test accuracies
axs[0].barh(indices, test_accs)
axs[0].set_title("Test set accuracy")
axs[0].set_xlabel("Accuracy")
axs[0].set_ylabel("Trial")

# Plot processing block RAM
axs[1].barh(indices, proc_rams)
axs[1].set_title("Processing block RAM")
axs[1].set_xlabel("RAM (bytes)")
axs[1].set_ylabel("Trial")

# Plot learning block latency
axs[2].barh(indices, learn_latencies)
axs[2].set_title("Learning block latency")
axs[2].set_xlabel("Latency (ms)")
axs[2].set_ylabel("Trial")

# Plot learning block RAM (TFLite)
axs[3].barh(indices, learn_tflite_rams)
axs[3].set_title("Learning block RAM (TFLite)")
axs[3].set_xlabel("RAM (bytes)")
axs[3].set_ylabel("Trial")

# Plot learning block ROM (TFLite)
axs[4].barh(indices, learn_tflite_roms)
axs[4].set_title("Learning block ROM (TFLite)")
axs[4].set_xlabel("ROM (bytes)")
axs[4].set_ylabel("Trial")

# Plot learning block RAM (EON)
axs[5].barh(indices, learn_eon_rams)
axs[5].set_title("Learning block RAM (EON)")
axs[5].set_xlabel("RAM (bytes)")
axs[5].set_ylabel("Trial")

# Plot learning block ROM (EON)
axs[6].barh(indices, learn_eon_roms)
axs[6].set_title("Learning block ROM (EON)")
axs[6].set_xlabel("ROM (bytes)")
axs[6].set_ylabel("Trial")

# Prevent overlap
plt.tight_layout()

Results as a DataFrame

If you have pandas installed, you can make the previous section much easier by reporting metrics as a DataFrame.

import pandas as pd

# Convert the state metrics into a DataFrame
df = tuner_report_as_df(state)
df.head()

# Print column names
for col in df.columns:
    print(col)

# Sort the DataFrame by validation (int8) accuracy
df = df.sort_values(by="test_int8_accuracy", ascending=False)

# Print out best trial metrics
print(f"Trial ID: {df.iloc[0]['id']}")
print(f"Test accuracy (int8): {df.iloc[0]['test_int8_accuracy']}")
print(f"Estimated learning blocks latency (ms): {df.iloc[0]['device_performance_int8_latency']}")
print(f"Estimated learning blocks RAM (bytes): {df.iloc[0]['device_performance_int8_tflite_ram_required']}")
print(f"Estimated learning blocks ROM (bytes): {df.iloc[0]['device_performance_int8_tflite_rom_required']}")
print(f"Estimated learning blocks RAM with EON Compiler (bytes): {df.iloc[0]['device_performance_int8_eon_ram_required']}")
print(f"Estimated learning blocks ROM with EON Compiler (bytes): {df.iloc[0]['device_performance_int8_eon_rom_required']}")

Set trial as impulse and deploy

We can replace the current impulse with the top performing trial from the EON Tuner. From there, we can deploy it, just like we would any impulse.

# Get the ID for the top-performing trial and set that to our impulse. This will take about a minute.
trial_id = df.iloc[0].trial_id
response = set_impulse_from_trial(trial_id)
job_id = response.id

# Make sure the impulse update was successful
if not hasattr(response, "success") or getattr(response, "success") == False:
    raise RuntimeError("Could not set project impulse to trial impulse")

# List the available profile target devices
ei.model.list_deployment_targets()

You should see a list printed such as:

['zip',
 'arduino',
 'cubemx',
 'wasm',
 ...
 'runner-linux-aarch64-jetson-orin-6-0']

Note that instead of writing the raw bytes to a file, you can also specify an output_directory argument in the .deploy() function. Your deployment file(s) will be downloaded to that directory.

# Build and download C++ library with the trained model
deploy_bytes = None
try:
    deploy_bytes = build(
        deploy_model_type=qtzn,
        engine="tflite",
        deploy_target="zip"
    )
except Exception as e:
    print(f"Could not deploy: {e}")
    
# Write the downloaded raw bytes to a file
if deploy_bytes:
    with open(deploy_filename, 'wb') as f:
        f.write(deploy_bytes.getvalue())

Configure custom search space

The best way to define a search space is to open your project (after uploading data), head to the EON Tuner page, click Run EON Tuner, and select the Space tab.

The search space is defined in JSON format, so we can just copy that to create a dictionary. This is a good place to start for tuning blocks and hyperparameters.

Note: Functions to get available blocks and search space parameters coming soon

from edgeimpulse_api import (
    OptimizeConfig,
    TunerSpaceImpulse,
)

# Configure the search space
space = {
    "inputBlocks": [
      {
        "type": "time-series",
        "window": [
          {"windowSizeMs": 9000, "windowIncreaseMs": 9000},
          {"windowSizeMs": 10000, "windowIncreaseMs": 10000}
        ],
        "frequencyHz": [62.5],
        "padZeros": [True]
      }
    ],
    "dspBlocks": [
      {
        "type": "spectral-analysis",
        "analysis-type": ["FFT"],
        "fft-length": [16, 64],
        "scale-axes": [1],
        "filter-type": ["none"],
        "filter-cutoff": [3],
        "filter-order": [6],
        "do-log": [True],
        "do-fft-overlap": [True]
      },
      {
        "type": "spectral-analysis",
        "analysis-type": ["Wavelet"],
        "wavelet": ["haar", "bior1.3"],
        "wavelet-level": [1, 2]
      },
      {"type": "raw", "scale-axes": [1]}
    ],
    "learnBlocks": [
      {
        "id": 4,
        "type": "keras",
        "dimension": ["dense"],
        "denseBaseNeurons": [40, 20],
        "denseLayers": [2, 3],
        "dropout": [0.25, 0.5],
        "learningRate": [0.0005],
        "trainingCycles": [30]
      }
    ]
  }

# Wrap the search space
ts = TunerSpaceImpulse.from_dict(space)

# Create a custom configuration
config = OptimizeConfig(
    name=None,
    target_device={"name": "cortex-m4f-80mhz"},
    classification_type="classification",
    dataset_category="motion_continuous",
    target_latency=100,
    tuning_max_trials=2,
    space=[ts]
)

# Start tuner and wait for it to complete
start_custom_tuner(
    config=config
)
state = check_tuner(
    wait_for_completion=True
)

# The easiest way to view the results is to look at the EON Tuner page on your project
print(f"Navigate to https://studio.edgeimpulse.com/studio/{project_id}/tuner to see the results")

# Set quantization ("float32" or "int8")
qtzn = "int8"

# Filter out all failed trials
results = [r for r in state.trials if r.status == "completed"]

# Extract float32 accuracies from the trial results
accuracies = []
for result in results:
    accuracy = result.impulse.learn_blocks[0]["metrics"]["test"][qtzn]["accuracy"]
    accuracies.append(accuracy)

# Sort the results based on int8 accuracies
acc_results = zip(accuracies, results)
sorted_results = sorted(acc_results, reverse=True, key=lambda x: list(x)[0])
sorted_results = [result for _, result in sorted_results]

# The top performing impulse is the first element (sorted by highest int8 accuracy on test set)
trial_idx = 0

# Print info about the processing (DSP) blocks and store RAM usage
print("Processing blocks")
print("===")
for i, dsp_block in enumerate(sorted_results[trial_idx].impulse.dsp_blocks):
    print(f"Processing block {i}")
    print("---")
    print("Block:")
    print(json.dumps(dsp_block["block"], indent=2))
    print("Config:")
    print(json.dumps(dsp_block["config"], indent=2))
print()

# Print info about the learning blocks
print("Learning blocks")
print("===")
for i, learn_block in enumerate(sorted_results[trial_idx].impulse.learn_blocks):
    print(f"Learn block {i}")
    print("---")
    print("Block:")
    print(json.dumps(learn_block["block"], indent=2))
    print("Config:")
    print(json.dumps(learn_block["config"], indent=2))
    metadata = learn_block["metadata"]
    qtzn_metadata = [m for m in metadata["modelValidationMetrics"] if m.get("type") == qtzn]
print()

# Print metrics
metrics = get_metrics(sorted_results, qtzn, trial_idx)
print(f"Metrics ({qtzn}) for best trial")
print("===")
print(f"Validation accuracy: {metrics['val-acc']}")
print(f"Test accuracy: {metrics['test-acc']}")
print(f"Estimated processing blocks RAM (bytes): {metrics['processing-block-ram']}")
print(f"Estimated learning blocks latency (ms): {metrics['learning-block-latency-ms']}")
print(f"Estimated learning blocks RAM (bytes): {metrics['learning-block-tflite-ram']}")
print(f"Estimated learning blocks ROM (bytes): {metrics['learning-block-tflite-rom']}")
print(f"Estimated learning blocks RAM with EON Compiler (bytes): {metrics['learning-block-eon-ram']}")
print(f"Estimated learning blocks ROM with EON Compiler (bytes): {metrics['learning-block-eon-rom']}")

Using the Edge Impulse Python SDK to upload and download data

If you want to upload files directly to an Edge Impulse project, we recommend using the CLI uploader tool. However, sometimes you cannot upload your samples directly, as you might need to convert the files to one of the accepted formats or modify the data prior to model training. Edge Impulse offers data augmentation for some types of projects, but you might want to create your own custom augmentation scheme. Or perhaps you want to generate synthetic data and script the upload process.

The Python SDK offers a set of functions to help you move data into and out of your project. This can be extremely helpful when generating or augmenting your dataset. The following cells demonstrate some of these upload and download functions.

You can find the API documentation for the functions found in this tutorial here.

WARNING: This notebook will add and delete data in your Edge Impulse project, so be careful! We recommend creating a throwaway project when testing this notebook.

Note that you might need to refresh the page with your Edge Impulse project to see the samples appear.

# If you have not done so already, install the following dependencies
!python -m pip install edgeimpulse

import edgeimpulse as ei

Note that you do not actually need to use the project in the Edge Impulse Studio. We just need the API Key.

Paste that API key string in the ei.API_KEY value in the following cell:

# Settings
ei.API_KEY = "ei_dae2..." # Change this to your Edge Impulse API key

Upload directory

You can upload all files in a directory using the Python SDK. Note that you can set the category, label, and metadata for all files with a single call. If you want to use a different label for each file set label=None in the function call and name your files with <label>.<name>.<ext>. For example, wave.01.csv will have the label wave when uploaded. See here for more information.

The following file formats are allowed: .cbor, .json, .csv, .wav, .jpg, .png, .mp4, .avi.

from datetime import datetime

# Download image files to use as an example dataset
!mkdir -p dataset
!wget -P dataset -q \
  https://raw.githubusercontent.com/edgeimpulse/notebooks/main/.assets/images/capacitor.01.png \
  https://raw.githubusercontent.com/edgeimpulse/notebooks/main/.assets/images/capacitor.02.png

# Upload the entire directory
response = ei.experimental.data.upload_directory(
    directory="dataset",
    category="training",
    label=None, # Will use the prefix before the '.' on each filename for the label
    metadata={
        "date": datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
        "source": "camera",
    }
)

# Check to make sure there were no failures
assert len(response.fails) == 0, "Could not upload some files"

# Save the sample IDs, as we will need these to retrieve file information and delete samples
ids = []
for sample in response.successes:
    ids.append(sample.sample_id)

# Review the sample IDs and get the associated server-side filename
# Note the lack of extension! Multiple samples on the server can have the same filename.
for id in ids:
    filename = ei.experimental.data.get_filename_by_id(id)
    print(f"Sample ID: {id}, filename: {filename}")

If you head to the Data acquisition page on your project, you should see images in your dataset.

Download files

You can download samples from your Edge Impulse project if you know the sample IDs. You can get sample IDs by calling the ei.data.get_sample_ids() function, which allows you to filter IDs based on filename, category, and label.

# Get sample IDs for everything in the "training" category
infos = ei.experimental.data.get_sample_ids(category="training")

# The SampleInfo should match what we uploaded earlier
ids = []
for info in infos:
    print(info)
    ids.append(info.sample_id)

# Download samples
samples = ei.experimental.data.download_samples_by_ids(ids)

# Save the downloaded files
for sample in samples:
    with open(sample.filename, "wb") as file:
        file.write(sample.data.read())

# View sample information
for sample in samples:
    print(
        f"filename: {sample.filename}\r\n"
        f"  sample ID: {sample.sample_id}\r\n"
        f"  category: {sample.category}\r\n"
        f"  label: {sample.label}\r\n"
        f"  bounding boxes: {sample.bounding_boxes}\r\n"
        f"  metadata: {sample.metadata}"
    )

Take a look at the files in this directory. You should see the downloaded images. They should match the images in the dataset/ directory, which were the original images that we uploaded.

Delete files

If you know the ID of the sample you would like to delete, you can call the delete_sample_by_id() function. You can also delete all the samples in your project by calling delete_all_samples().

# Delete the samples from the Edge Impulse project that we uploaded earlier
for id in ids:
    ei.experimental.data.delete_sample_by_id(id)

Take a look at the data in your project. The samples that we uploaded should be gone.

Upload folder for object detection

For object detection, you can put bounding box information (following the Edge Impulse JSON bounding box format) in a file named info.labels in that same directory.

Important! The annotations file must be named exactly info.labels

# Download images and bounding box annotations to use as an example dataset
!mkdir -p dataset
!rm dataset/capacitor.01.png dataset/capacitor.02.png
!wget -P dataset -q \
  https://raw.githubusercontent.com/edgeimpulse/notebooks/main/.assets/images/dog-ball-toy.01.png \
  https://raw.githubusercontent.com/edgeimpulse/notebooks/main/.assets/images/dog-ball-toy.02.png \
  https://raw.githubusercontent.com/edgeimpulse/notebooks/main/.assets/annotations/info.labels

# Upload the entire directory (including the info.labels file)
response = ei.experimental.data.upload_exported_dataset(
    directory="dataset",
)

# Check to make sure there were no failures
assert len(response.fails) == 0, "Could not upload some files"

# Save the sample IDs, as we will need these to retrieve file information and delete samples
ids = []
for sample in response.successes:
    ids.append(sample.sample_id)

If you head to the Data acquisition page on your project, you should see images in your dataset along with the bounding box information.

# Delete the samples from the Edge Impulse project that we uploaded
for id in ids:
    ei.experimental.data.delete_sample_by_id(id)

Upload individual CSV files

The Edge Impulse ingestion service accepts CSV files, which we can use to upload raw data. Note that if you configure a CSV template using the CSV Wizard, then the expected format of the CSV file might change. If you do not configure a CSV template, then the ingestion service expects CSV data to be in a particular format. See here for details about the default CSV format.

import csv
import io
import os

# Create example CSV data
sample_data = [
    [
        ["timestamp", "accX", "accY", "accZ"],
        [0, -9.81, 0.03, 0.21],
        [10, -9.83, 0.04, 0.27],
        [20, -9.12, 0.03, 0.23],
        [30, -9.14, 0.01, 0.25],
    ],
    [
        ["timestamp", "accX", "accY", "accZ"],
        [0, -9.56, 5.34, 1.21],
        [10, -9.43, 1.37, 1.27],
        [20, -9.22, -4.03, 1.23],
        [30, -9.50, -0.98, 1.25],
    ],
]

# Write to CSV files
filenames = [
    "001.csv",
    "002.csv"
]
for i, filename in enumerate(filenames):
    file_path = os.path.join("dataset", filename)
    with open(file_path, "w", newline="") as file:
        writer = csv.writer(file)
        writer.writerows(sample_data[i])

# Add metadata to the CSV data
my_samples = [
    {
        "filename": filenames[0],
        "data": open(os.path.join("dataset", filenames[0]), "rb"),
        "category": "training",
        "label": "idle",
        "metadata": {
            "source": "accelerometer",
            "collection site": "desk",
        },
    },
    {
        "filename": filenames[1],
        "data": open(os.path.join("dataset", filenames[1]), "rb"),
        "category": "training",
        "label": "wave",
        "metadata": {
            "source": "accelerometer",
            "collection site": "desk",
        },
    },
]

# Wrap the samples in instances of the Sample class
samples = [ei.experimental.data.Sample(**i) for i in my_samples]

# Upload samples to your project
response = ei.experimental.data.upload_samples(samples)

# Check to make sure there were no failures
assert len(response.fails) == 0, "Could not upload some files"

# Save the sample IDs, as we will need these to retrieve file information and delete samples
ids = []
for sample in response.successes:
    ids.append(sample.sample_id)

If you head to the Data acquisition page on your project, you should see your time series data.

# Delete the samples from the Edge Impulse project
for id in ids:
    ei.experimental.data.delete_sample_by_id(id)

Upload JSON data directly

Another way to upload data is to encode it in JSON format. See the data acquisition format specificaion for more information on acceptable key/value pairs. Note that at this time, the signature value can be set to 0.

The raw data must be encoded in an IO object. We convert the dictionary objects to a BytesIO object, but you can also read in data from .json files.

import io
import json

# Create two different example data samples
sample_data_1 = {
    "protected": {
        "ver": "v1",
        "alg": "none",
    },
    "signature": 0,
    "payload": {
        "device_name": "ac:87:a3:0a:2d:1b",
        "device_type": "DISCO-L475VG-IOT01A",
        "interval_ms": 10,
        "sensors": [
            { "name": "accX", "units": "m/s2" },
            { "name": "accY", "units": "m/s2" },
            { "name": "accZ", "units": "m/s2" }
        ],
        "values": [
            [ -9.81, 0.03, 0.21 ],
            [ -9.83, 0.04, 0.27 ],
            [ -9.12, 0.03, 0.23 ],
            [ -9.14, 0.01, 0.25 ]
        ]
    }
}
sample_data_2 = {
    "protected": {
        "ver": "v1",
        "alg": "none",
    },
    "signature": 0,
    "payload": {
        "device_name": "ac:87:a3:0a:2d:1b",
        "device_type": "DISCO-L475VG-IOT01A",
        "interval_ms": 10,
        "sensors": [
            { "name": "accX", "units": "m/s2" },
            { "name": "accY", "units": "m/s2" },
            { "name": "accZ", "units": "m/s2" }
        ],
        "values": [
            [ -9.56, 5.34, 1.21 ],
            [ -9.43, 1.37, 1.27 ],
            [ -9.22, -4.03, 1.23 ],
            [ -9.50, -0.98, 1.25 ]
        ]
    }
}

# Provide a filename, category, label, and optional metadata for each sample
my_samples = [
    {
        "filename": "001.json",
        "data": io.BytesIO(json.dumps(sample_data_1).encode('utf-8')),
        "category": "training",
        "label": "idle",
        "metadata": {
            "source": "accelerometer",
            "collection site": "desk",
        },
    },
    {
        "filename": "002.json",
        "data": io.BytesIO(json.dumps(sample_data_2).encode('utf-8')),
        "category": "training",
        "label": "wave",
        "metadata": {
            "source": "accelerometer",
            "collection site": "desk",
        },
    },
]

# Wrap the samples in instances of the Sample class
samples = [ei.data.sample_type.Sample(**i) for i in my_samples]

# Upload samples to your project
response = ei.experimental.data.upload_samples(samples)

# Check to make sure there were no failures
assert len(response.fails) == 0, "Could not upload some files"

# Save the sample IDs, as we will need these to retrieve file information and delete samples
ids = []
for sample in response.successes:
    ids.append(sample.sample_id)

If you head to the Data acquisition page on your project, you should see your time series data.

# Delete the samples from the Edge Impulse project
for id in ids:
    ei.experimental.data.delete_sample_by_id(id)

Upload NumPy arrays

NumPy is powerful Python library for working with large arrays and matrices. You can upload NumPy arrays directly into your Edge Impulse project. Note that the arrays are required to be in a particular format, and must be uploaded with required metadata (such as a list of labels and the sample rate).

Important! NumPy arrays must be in the shape (Number of samples, number of data points, number of sensors)

If you are working with image data in NumPy, we recommend saving those images as .png or .jpg files and using upload_directory().

import numpy as np

# Create example NumPy array with 2 time series samples
sample_data = np.array(
    [
        [ # Sample 1 ("idle")
            [-9.81, 0.03, 0.21],
            [-9.83, 0.04, 0.27],
            [-9.12, 0.03, 0.23],
            [-9.14, 0.01, 0.25],
        ],
        [ # Sample 2 ("wave")
            [-9.56, 5.34, 1.21],
            [-9.43, 1.37, 1.27],
            [-9.22, -4.03, 1.23],
            [-9.50, -0.98, 1.25],
        ],
    ]
)

# Labels for each sample
labels = ["idle", "wave"]

# Names of the sensors and units for the 3 axes
sensors = [
    { "name": "accX", "units": "m/s2" },
    { "name": "accY", "units": "m/s2" },
    { "name": "accZ", "units": "m/s2" },
]

# Optional metadata for all samples being uploaded
metadata = {
    "source": "accelerometer",
    "collection site": "desk",
}

# Upload samples to your project
response = ei.experimental.data.upload_numpy(
    data=sample_data,
    labels=labels,
    sensors=sensors,
    sample_rate_ms=10,
    metadata=metadata,
    category="training",
)

# Check to make sure there were no failures
assert len(response.fails) == 0, "Could not upload some files"

# Save the sample IDs, as we will need these to retrieve file information and delete samples
ids = []
for sample in response.successes:
    ids.append(sample.sample_id)

If you head to the Data acquisition page on your project, you should see your time series data. Note that the sample names are randomly assigned, so we recommend recording the sample IDs when you upload.

# Delete the samples from the Edge Impulse project
for id in ids:
    ei.experimental.data.delete_sample_by_id(id)

Upload pandas (and pandas-like) dataframes

pandas is popular Python library for performing data manipulation and analysis. The Edge Impulse library supports a number of ways to upload dataframes. We will go over each format.

Note that several other packages exist that work as drop-in replacements for pandas. You can use these replacements so long as you import that with the name pd. For example, one of:

import pandas as pd
import modin.pandas as pd
import dask.dataframe as pd
import polars as pd

import pandas as pd

The first option is to upload one dataframe for each sample (non-time series)

# Construct one dataframe for each sample (multidimensional, non-time series)
df_1 = pd.DataFrame([[-9.81, 0.03, 0.21]], columns=["accX", "accY", "accZ"])
df_2 = pd.DataFrame([[-9.56, 5.34, 1.21]], columns=["accX", "accY", "accZ"])

# Optional metadata for all samples being uploaded
metadata = {
    "source": "accelerometer",
    "collection site": "desk",
}

# Upload the first sample
ids = []
response = ei.experimental.data.upload_pandas_sample(
    df_1,
    label="One",
    filename="001",
    metadata=metadata,
    category="training",
)
assert len(response.fails) == 0, "Could not upload some files"
for sample in response.successes:
    ids.append(sample.sample_id)

# Upload the second sample
response = ei.experimental.data.upload_pandas_sample(
    df_2,
    label="Two",
    filename="002",
    metadata=metadata,
    category="training",
)
assert len(response.fails) == 0, "Could not upload some files"
for sample in response.successes:
    ids.append(sample.sample_id)

# Delete the samples from the Edge Impulse project
for id in ids:
    ei.experimental.data.delete_sample_by_id(id)

You can also upload one dataframe for each sample (time series). As with previous examples, we'll assume that the sample rate is 10 ms.

# Create samples (multidimensional, time series)
sample_data_1 = [ # Sample 1 ("idle")
    [-9.81, 0.03, 0.21],
    [-9.83, 0.04, 0.27],
    [-9.12, 0.03, 0.23],
    [-9.14, 0.01, 0.25],
]
sample_data_2 = [ # Sample 1 ("wave")
    [-9.56, 5.34, 1.21],
    [-9.43, 1.37, 1.27],
    [-9.22, -4.03, 1.23],
    [-9.50, -0.98, 1.25],
]

# Construct one dataframe for each sample
df_1 = pd.DataFrame(sample_data_1, columns=["accX", "accY", "accZ"])
df_2 = pd.DataFrame(sample_data_2, columns=["accX", "accY", "accZ"])

# Optional metadata for all samples being uploaded
metadata = {
    "source": "accelerometer",
    "collection site": "desk",
}

# Upload the first sample
ids = []
response = ei.experimental.data.upload_pandas_sample(
    df_1,
    label="Idle",
    filename="001",
    sample_rate_ms=10,
    metadata=metadata,
    category="training",
)
assert len(response.fails) == 0, "Could not upload some files"
for sample in response.successes:
    ids.append(sample.sample_id)

# Upload the second sample
response = ei.experimental.data.upload_pandas_sample(
    df_2,
    label="Wave",
    filename="002",
    sample_rate_ms=10,
    metadata=metadata,
    category="training",
)
assert len(response.fails) == 0, "Could not upload some files"
for sample in response.successes:
    ids.append(sample.sample_id)

# Delete the samples from the Edge Impulse project
for id in ids:
    ei.experimental.data.delete_sample_by_id(id)

You can upload non-time series data where each sample is a row in the dataframe. Note that you need to provide labels in the rows.

# Construct non-time series data, where each row is a different sample
data = [
    ["desk", "training", "One", -9.81, 0.03, 0.21],
    ["field", "training", "Two", -9.56, 5.34, 1.21],
]
columns = ["loc", "category", "label", "accX", "accY", "accZ"]

# Wrap the data in a DataFrame
df = pd.DataFrame(data, columns=columns)

# Upload non-time series DataFrame (with multiple samples) to the project
ids = []
response = ei.experimental.data.upload_pandas_dataframe(
    df,
    feature_cols=["accX", "accY", "accZ"],
    label_col="label",
    category_col="category",
    metadata_cols=["loc"],
)
assert len(response.fails) == 0, "Could not upload some files"
for sample in response.successes:
    ids.append(sample.sample_id)

# Delete the samples from the Edge Impulse project
for id in ids:
    ei.experimental.data.delete_sample_by_id(id)

A "wide" dataframe is one where each column represents a value in the time series data, and the rows become individual samples. Note that you need to provide labels in the rows.

# Construct time series data, where each row is a different sample
data = [
    ["desk", "training", "idle", 0.8, 0.7, 0.8, 0.9, 0.8, 0.8, 0.7, 0.8],
    ["field", "training", "motion", 0.3, 0.9, 0.4, 0.6, 0.8, 0.9, 0.5, 0.4],
]
columns = ["loc", "category", "label", "0", "1", "2", "3", "4", "5", "6", "7"]

# Wrap the data in a DataFrame
df = pd.DataFrame(data, columns=columns)

# Upload time series DataFrame (with multiple samples) to the project
ids = []
response = ei.experimental.data.upload_pandas_dataframe_wide(
    df,
    label_col="label",
    category_col="category",
    metadata_cols=["loc"],
    data_col_start=3,
    sample_rate_ms=100,
)
assert len(response.fails) == 0, "Could not upload some files"
for sample in response.successes:
    ids.append(sample.sample_id)

# Delete the samples from the Edge Impulse project
for id in ids:
    ei.experimental.data.delete_sample_by_id(id)

A DataFrame can also be divided into "groups" so you can upload multidimensional time series data.

# Create samples
sample_data = [
    ["desk", "sample 1", "training", "idle", 0, -9.81, 0.03, 0.21],
    ["desk", "sample 1", "training", "idle", 0.01, -9.83, 0.04, 0.27],
    ["desk", "sample 1", "training", "idle", 0.02, -9.12, 0.03, 0.23],
    ["desk", "sample 1", "training", "idle", 0.03, -9.14, 0.01, 0.25],
    ["field", "sample 2", "training", "wave", 0, -9.56, 5.34, 1.21],
    ["field", "sample 2", "training", "wave", 0.01, -9.43, 1.37, 1.27],
    ["field", "sample 2", "training", "wave", 0.02, -9.22, -4.03, 1.23],
    ["field", "sample 2", "training", "wave", 0.03, -9.50, -0.98, 1.25],
]
columns = ["loc", "sample_name", "category", "label", "timestamp", "accX", "accY", "accZ"]

# Wrap the data in a DataFrame
df = pd.DataFrame(sample_data, columns=columns)

# Upload time series DataFrame (with multiple samples and multiple dimensions) to the project
ids = []
response = ei.experimental.data.upload_pandas_dataframe_with_group(
    df,
    group_by="sample_name",
    timestamp_col="timestamp",
    feature_cols=["accX", "accY", "accZ"],
    label_col="label",
    category_col="category",
    metadata_cols=["loc"]
)
assert len(response.fails) == 0, "Could not upload some files"
for sample in response.successes:
    ids.append(sample.sample_id)

# Delete the samples from the Edge Impulse project
for id in ids:
    ei.experimental.data.delete_sample_by_id(id)