Log raw data for BME AI-Studio

What this example shows

Bosch's BME AI-Studio trains a gas classification algorithm from recorded sensor data. This example records that data: it runs the sensor in parallel mode with a ten-step heater profile, logs every valid gas measurement for 10 minutes into a .bmerawdata file on an SD card, in the BME AI-Studio Raw Data Format, and then closes the file. You import the file into BME AI-Studio to label the data.

The button on pin 0 switches the label tag while the recording runs, which marks the start of a new Specimen in AI-Studio. Set LABEL_BUTTON_PIN to None to record the whole session as a single Specimen.

What you need, as the example lists it:

• an ESP32 board, because WiFi and NTP set the real time clock;
• the BME690 breakout on the I2C pins, or connected via Qwiic (BME690() uses SDA on pin 21 and SCL on pin 22 on the esp32 port);
• an SD card module on the SPI pins: SPI bus 2 with SCK 18, MISO 19, MOSI 23 and chip select 5;
• the sdcard driver from micropython-lib, installed with mpremote mip install sdcard. The library's README says mip installs it along with this library.

Before you run it, put your WiFi name and password in SSID and PASSWORD. The example only uses WiFi to get the current time over NTP. Give every board its own BOARD_ID: AI-Studio uses it to tell recordings of different boards apart.

This example also uses the sdcard driver from micropython-lib, which is not part of the Soldered BME690 library.

The example code

# FILE: bme690-aiStudioLogger.py
# AUTHOR: Josip Simun Kuci @ Soldered
# BRIEF: Record BME690 raw data into a .bmerawdata file on an SD card, in the
#        Bosch BME AI-Studio Raw Data Format, so that it can be imported into
#        BME AI-Studio and used to train a gas classification algorithm
# WORKS WITH: BME690 breakout board: solde.red/333411
# LAST UPDATED: 2026-09-16
#
# The example runs the sensor in parallel mode with a ten step heater profile,
# logs every valid gas measurement for LOG_DURATION_MS and then closes the
# file.
#
# AI-Studio splits a recording into Specimens wherever the label tag changes,
# and it measures the duration of a Specimen from its first to its last point.
# A recording whose label tag never changes therefore imports as a Specimen of
# zero length. The recording starts already tagged, and the button on
# LABEL_BUTTON_PIN switches the tag while it runs, which marks the moment as
# the start of a new Specimen. That is the same thing buttons S1 and S2 do on
# the BME688 Development Kit.
#
# Needed hardware:
# - an ESP32 board (WiFi and NTP are used for the real time clock)
# - a BME690 breakout on the I2C pins, or connected via Qwiic
# - an SD card module on the SPI pins, chip select on SD_CS
#
# Needed libraries:
# - sdcard, the SD card driver of micropython-lib, installed with
#   mpremote mip install sdcard
#
# The AI-Studio workflow is described here:
# https://www.bosch-sensortec.com/software/bme/docs/overview/getting-started.html

from bme690 import (
    BME690,
    BME69X_ERROR,
    BME69X_GASM_VALID_MSK,
    BME69X_HEAT_STAB_MSK,
    BME69X_NEW_DATA_MSK,
    BME69X_PARALLEL_MODE,
    BME69X_WARNING,
)
from machine import Pin, SPI
import json
import network
import ntptime
import os
import random
import sdcard
import time

# WiFi credentials, only used to get the current time over NTP.
SSID = "YOUR_SSID_HERE"
PASSWORD = "YOUR_PASSWORD_HERE"

# SPI bus and pins of the SD card module. The pins below are the native VSPI
# pins of the ESP32.
SD_SPI_ID = 2
SD_SCK = 18
SD_MISO = 19
SD_MOSI = 23
SD_CS = 5

# Bus speed used while the card is initialized, and afterwards. The card only
# answers slowly until it is out of its initialization sequence.
SD_INIT_BAUDRATE = 400000
SD_BAUDRATE = 20000000

# Button which marks the start of a new Specimen while recording. It is
# active low, so the internal pull up is enough. Set it to None to record the
# whole session as a single Specimen.
LABEL_BUTTON_PIN = 0

# Debounce time of that button.
LABEL_BUTTON_DEBOUNCE_MS = 250

# Total duration of one measurement session.
LOG_DURATION_MS = 10 * 60000

# Duration of a single heater profile step in milliseconds. The Raw Data
# Format expects the heater profile time base to be 140 ms.
MEAS_DUR = 140

# New data, gas measurement valid and heater stable, all at once.
BME690_VALID_DATA = (
    BME69X_NEW_DATA_MSK | BME69X_GASM_VALID_MSK | BME69X_HEAT_STAB_MSK
)

# Identifier of the board which recorded the data. AI-Studio uses it to tell
# recordings of different boards apart, so give every board its own value.
BOARD_ID = "E0E2E69BA804"

# Unique id of the sensor element, used to trace a row back to one sensor.
SENSOR_ID = 1903381786

# Index of the sensor on the board. A single BME690 breakout only has one.
SENSOR_INDEX = 0

# Heater temperature profile in degrees Celsius.
HEATER_TEMP = [320, 100, 100, 100, 200, 200, 200, 320, 320, 320]

# Multipliers of the shared heater duration, one per profile step.
HEATER_MUL = [5, 2, 10, 30, 5, 5, 5, 5, 5, 5]

HEATER_LEN = 10

# MicroPython counts the seconds from the year 2000, the format wants them
# counted from 1970.
UNIX_EPOCH_OFFSET = 946684800

# The column description of the data block. The order of the columns here is
# the order of the values in every recorded row.
COLUMNS = [
    ("Sensor Index", "", "integer", "sensor_index"),
    ("Sensor ID", "", "integer", "sensor_id"),
    (
        "Time Since PowerOn",
        "Milliseconds",
        "integer",
        "timestamp_since_poweron",
    ),
    (
        "Real time clock",
        "Unix Timestamp: seconds since Jan 01 1970. (UTC); 0 = missing",
        "integer",
        "real_time_clock",
    ),
    ("Temperature", "DegreesCelcius", "float", "temperature"),
    ("Pressure", "Hectopascals", "float", "pressure"),
    ("Relative Humidity", "Percent", "float", "relative_humidity"),
    ("Resistance Gassensor", "Ohms", "float", "resistance_gassensor"),
    (
        "Heater Profile Step Index",
        "",
        "integer",
        "heater_profile_step_index",
    ),
    ("Scanning Mode Enabled", "", "boolean", "scanning_enabled"),
    ("Scanning Cycle Index", "", "integer", "scanning_cycle_index"),
    ("Label Tag", "", "integer", "label_tag"),
    ("Error Code", "", "integer", "error_code"),
]


def check_sensor_status(sensor):
    """Stop the example if the sensor reported an error, print warnings."""
    status = sensor.check_status()

    if status == BME69X_ERROR:
        raise Exception("BME690 error: " + sensor.status_string())

    if status == BME69X_WARNING:
        print("BME690 warning: " + sensor.status_string())


def iso8601(unix_time):
    """Format a unix timestamp as an ISO 8601 UTC string."""
    parts = time.gmtime(unix_time - UNIX_EPOCH_OFFSET)

    return "{:04d}-{:02d}-{:02d}T{:02d}:{:02d}:{:02d}+00:00".format(
        parts[0], parts[1], parts[2], parts[3], parts[4], parts[5]
    )


def file_name_date(unix_time):
    """Format a unix timestamp as the yyyy_mm_dd_hh_mm used in a file name."""
    parts = time.gmtime(unix_time - UNIX_EPOCH_OFFSET)

    return "{:04d}_{:02d}_{:02d}_{:02d}_{:02d}".format(
        parts[0], parts[1], parts[2], parts[3], parts[4]
    )


def make_seed():
    """
    Build the random seed which labels all files of one session.

    The format expects sixteen lowercase alphanumeric characters.
    """
    alphabet = "abcdefghijklmnopqrstuvwxyz0123456789"

    return "".join(random.choice(alphabet) for _ in range(16))


def connect_wifi():
    """Connect to WiFi, the format stores an absolute creation date."""
    station = network.WLAN(network.STA_IF)
    station.active(True)

    if not station.isconnected():
        station.connect(SSID, PASSWORD)
        print("Connecting to WiFi", end="")
        while not station.isconnected():
            print(".", end="")
            time.sleep_ms(500)
        print("\nWiFi connected")

    return station


def sync_time():
    """Set the real time clock over NTP and return the unix time."""
    print("Waiting for NTP", end="")
    while True:
        try:
            ntptime.settime()
            break
        except Exception:
            print(".", end="")
            time.sleep_ms(200)
    print("\nTime synced")

    return time.time() + UNIX_EPOCH_OFFSET


def mount_sd():
    """
    Mount the SD card on /sd.

    A soft reset leaves an earlier mount in place, so mounting again would
    fail with EBUSY. The old mount is therefore dropped first, which does
    nothing when there is none.
    """
    try:
        os.umount("/sd")
    except OSError:
        pass

    spi = SPI(
        SD_SPI_ID,
        baudrate=SD_INIT_BAUDRATE,
        sck=Pin(SD_SCK),
        mosi=Pin(SD_MOSI),
        miso=Pin(SD_MISO),
    )
    card = sdcard.SDCard(spi, Pin(SD_CS))
    spi.init(baudrate=SD_BAUDRATE)

    os.mount(card, "/sd")
    print("SD card ready")

    return spi


def build_header(now, seed_power_on_off):
    """
    Build everything which comes before the measurements themselves.

    A ten minute recording holds a few thousand rows, which is far more than
    fits in RAM as one document, so the header is written out first with its
    data block left open, and the rows are appended as they are measured.
    """
    iso_now = iso8601(now)

    # The board configuration, the same two objects are stored in a
    # .bmeconfig file by AI-Studio.
    config_header = {
        "dateCreated_ISO": iso_now,
        "appVersion": "2.2.0",
        "boardType": "soldered_bme690",
        "boardMode": "burn_in",
        "boardLayout": "grouped",
    }

    config_body = {
        "heaterProfiles": [
            {
                "id": "heater_354",
                "timeBase": MEAS_DUR,
                "temperatureTimeVectors": [
                    [HEATER_TEMP[i], HEATER_MUL[i]]
                    for i in range(HEATER_LEN)
                ],
            }
        ],
        # The sensor scans continuously, so there are no sleeping cycles.
        "dutyCycleProfiles": [
            {
                "id": "duty_1",
                "numberScanningCycles": 1,
                "numberSleepingCycles": 0,
            }
        ],
        "sensorConfigurations": [
            {
                "sensorIndex": SENSOR_INDEX,
                "heaterProfile": "heater_354",
                "dutyCycleProfile": "duty_1",
            }
        ],
    }

    # The header of the recording itself, the counters and the seed repeat
    # the matching parts of the file name.
    raw_data_header = {
        "counterPowerOnOff": 1,
        "seedPowerOnOff": seed_power_on_off,
        "counterFileLimit": 0,
        "dateCreated": str(now),
        "dateCreated_ISO": iso_now,
        "firmwareVersion": "1.0.0",
        "boardId": BOARD_ID,
    }

    data_columns = [
        {
            "name": column[0],
            "unit": column[1],
            "format": column[2],
            "key": column[3],
            "colId": index + 1,
        }
        for index, column in enumerate(COLUMNS)
    ]

    # The objects are joined by hand instead of dumping one document, so that
    # the data block stays the last member of the last object.
    return (
        '{"configHeader":'
        + json.dumps(config_header)
        + ',"configBody":'
        + json.dumps(config_body)
        + ',"rawDataHeader":'
        + json.dumps(raw_data_header)
        + ',"rawDataBody":{"dataColumns":'
        + json.dumps(data_columns)
        + ',"dataBlock":['
    )


def record(sensor, file, button):
    """Log every valid gas measurement until the session is over."""
    start = time.ticks_ms()

    # The Raw Data Format counts the Scanning Cycles from one.
    scanning_cycle_index = 1

    rows = 0
    dropped_cycles = 0

    # Recording starts already tagged, so that the whole session forms a
    # Specimen even when the button is never pressed. Pressing the button
    # switches the tag, which starts the next Specimen.
    label_tag = 1
    last_button_press = time.ticks_ms()

    # No step has been seen yet, the first one is expected to be step zero.
    last_step_index = -1

    while time.ticks_diff(time.ticks_ms(), start) < LOG_DURATION_MS:
        # The shortest heater step of the profile lasts two time bases, so
        # polling twice per time base is fast enough not to miss one.
        time.sleep_ms(MEAS_DUR // 2)

        # Mark the start of a new Specimen when the button is pressed. The
        # tag alternates between one and two, the same way the two buttons of
        # the BME688 Development Kit are used.
        if (
            button is not None
            and button.value() == 0
            and time.ticks_diff(time.ticks_ms(), last_button_press)
            > LABEL_BUTTON_DEBOUNCE_MS
        ):
            last_button_press = time.ticks_ms()
            label_tag = 2 if label_tag == 1 else 1
            print("New specimen, label tag is now " + str(label_tag))

        if not sensor.fetch_data():
            continue

        fields_left = 1

        while fields_left:
            data, fields_left = sensor.get_data()

            # Skip the fields which hold no valid gas measurement.
            if data.status != BME690_VALID_DATA:
                continue

            # One scanning cycle is one full sweep through the heater
            # profile, so the counter moves on whenever the step index wraps
            # around.
            if data.gas_index <= last_step_index:
                scanning_cycle_index += 1

            # A step which is not the one after the previous step means the
            # measurement in between was lost. The format reports that as
            # error code 2, and AI-Studio drops the whole cycle on import.
            error_code = 0
            if data.gas_index != (last_step_index + 1) % HEATER_LEN:
                error_code = 2
                dropped_cycles += 1

            last_step_index = data.gas_index

            # The format wants the pressure in hectopascals, the sensor
            # reports it in pascals.
            file.write(
                "{}[{},{},{},{},{:f},{:f},{:f},{:f},{},{},{},{},{}]".format(
                    "" if rows == 0 else ",",
                    SENSOR_INDEX,
                    SENSOR_ID,
                    time.ticks_diff(time.ticks_ms(), start),
                    time.time() + UNIX_EPOCH_OFFSET,
                    data.temperature,
                    data.pressure / 100.0,
                    data.humidity,
                    data.gas_resistance,
                    data.gas_index,
                    1,
                    scanning_cycle_index,
                    label_tag,
                    error_code,
                )
            )
            rows += 1

            print(
                "T: {:.2f} C | P: {:.2f} hPa | H: {:.2f} % | R: {:.1f} "
                "Ohm | step: {}".format(
                    data.temperature,
                    data.pressure / 100.0,
                    data.humidity,
                    data.gas_resistance,
                    data.gas_index,
                )
            )

    return rows, dropped_cycles


print("\nBME690 AI-Studio raw data logger")

station = connect_wifi()
now = sync_time()

# Seed the generator with the synced time so every session gets its own seed
# instead of repeating the same one after every reset.
random.seed(now)
seed_power_on_off = make_seed()

button = None
if LABEL_BUTTON_PIN is not None:
    button = Pin(LABEL_BUTTON_PIN, Pin.IN, Pin.PULL_UP)

spi = mount_sd()

sensor = BME690()
check_sensor_status(sensor)

# Default temperature, pressure and humidity oversampling.
sensor.set_tph()
check_sensor_status(sensor)

# The shared heating duration is the total measurement duration minus the
# time needed for the temperature, pressure and humidity measurement.
shared_heatr_dur = (
    MEAS_DUR - sensor.get_meas_dur(BME69X_PARALLEL_MODE) // 1000
)

sensor.set_heater_prof(HEATER_TEMP, HEATER_MUL, shared_heatr_dur)
check_sensor_status(sensor)

sensor.set_op_mode(BME69X_PARALLEL_MODE)
check_sensor_status(sensor)

# The file name carries the date, the board, the power on counter, the
# session seed and the file counter, each part separated by an underscore.
file_name = (
    "/sd/"
    + file_name_date(now)
    + "_Board_"
    + BOARD_ID
    + "_PowerOnOff_1_"
    + seed_power_on_off
    + "_File_0.bmerawdata"
)

file = open(file_name, "w")
file.write(build_header(now, seed_power_on_off))

print("Recording, this takes {} minutes.".format(LOG_DURATION_MS // 60000))

rows, dropped_cycles = record(sensor, file, button)

# Close the data block, the rawDataBody object and the root object.
file.write("]}}")
file.close()

os.umount("/sd")
spi.deinit()
station.disconnect()

print(
    "Recording finished, {} rows written, {} cycles marked as lost.".format(
        rows, dropped_cycles
    )
)
print("File saved as: " + file_name)
print("Import it into BME AI-Studio to label the data and train an "
      "algorithm.")

🔗Open the complete example

Examples/bme690-aiStudioLogger.py at commit 91a07b5

→

Expected result: the shell shows the start of the session:

ShellMicroPython

BME690 AI-Studio raw data logger
Connecting to WiFi
WiFi connected
Waiting for NTP
Time synced
SD card ready
Recording, this takes 10 minutes.

The number of dots after Connecting to WiFi and Waiting for NTP depends on how long each step takes. During the recording, every valid measurement prints a line of the form T: ... C | P: ... hPa | H: ... % | R: ... Ohm | step: ..., and each button press prints New specimen, label tag is now with the new tag. At the end the example prints how many rows it wrote and how many cycles it marked as lost, the name of the file on the card, and Import it into BME AI-Studio to label the data and train an algorithm.

Functions used

BME690(i2c=None, address=BME69X_I2C_ADDR_LOW, amb_temp=25)returns BME690

Creates the sensor object. It soft-resets the sensor, checks its chip ID (0x61) and reads the calibration data. It raises an exception when the sensor does not answer or another chip is at the address; every other function records its result in the status instead of raising.

ReturnsThe sensor object.

Parameters

Type	Name	Description
I2C	i2c	A ready machine.I2C object. Left out, the module opens I2C(0) with SCL on pin 22 and SDA on pin 21 on the esp32 and esp8266 ports and one Soldered board firmware, and raises an exception on any other port.
int	address	The I2C address: BME69X_I2C_ADDR_LOW (0x76, the default) or BME69X_I2C_ADDR_HIGH (0x77).
int	amb_temp	Ambient temperature in °C, used to work out the heater setting. 25 by default.

sensor.check_status()returns int

Sums up the status the last function left behind. The example's check_sensor_status() stops on an error and prints a warning.

ReturnsBME69X_ERROR (-1) after an error, BME69X_WARNING (1) after a warning, BME69X_OK (0) otherwise.

sensor.status_string()returns str

Describes the current status code in a few words, such as Communication failure or No new data.

ReturnsThe description, an empty string when the status is OK.

sensor.set_tph(os_temp=BME69X_OS_2X, os_pres=BME69X_OS_16X, os_hum=BME69X_OS_1X)returns None

Sets the oversampling for temperature, pressure and humidity. Called without arguments, as here, it sets the defaults of the Arduino library.

ReturnsNothing. The result is in the status.

Parameters

Type	Name	Description
int	os_temp	Temperature oversampling, BME69X_OS_NONE to BME69X_OS_16X.
int	os_pres	Pressure oversampling, BME69X_OS_NONE to BME69X_OS_16X.
int	os_hum	Humidity oversampling, BME69X_OS_NONE to BME69X_OS_16X.

sensor.get_meas_dur(op_mode=BME69X_SLEEP_MODE)returns int

Works out how long one temperature, pressure and humidity measurement takes, from the oversampling set with set_tph(), so call set_tph() first. In parallel mode no wake-up time is added.

ReturnsThe measurement duration in microseconds.

Parameters

Type	Name	Description
int	op_mode	The mode to calculate for. The example passes BME69X_PARALLEL_MODE, because the mode is not set yet.

sensor.set_heater_prof(temp, dur, shared_heatr_dur=None)returns None

Sets the gas heater profile and switches the heater on. With a list of temperatures, a list of multipliers and a shared heating duration, as here, it writes a parallel-mode profile: each step lasts its multiplier times the shared duration. Temperatures above 400 °C are capped at 400 °C.

ReturnsNothing. The result is in the status.

Parameters

Type	Name	Description
list	temp	Heater temperature of each step in °C. The example uses HEATER_TEMP.
list	dur	In parallel mode, the multiplier of the shared duration for each step. The example uses HEATER_MUL.
int	shared_heatr_dur	The shared heating duration in ms, here MEAS_DUR (140 ms) minus the measurement duration.

sensor.set_op_mode(op_mode)returns None

Puts the sensor to sleep, then sets the requested mode. With BME69X_PARALLEL_MODE the sensor keeps measuring and sweeping the heater on its own. The mode is remembered for fetch_data() and get_data().

ReturnsNothing. The result is in the status.

Parameters

Type	Name	Description
int	op_mode	BME69X_SLEEP_MODE, BME69X_FORCED_MODE, BME69X_PARALLEL_MODE or BME69X_SEQUENTIAL_MODE.

sensor.fetch_data()returns int

Reads the sensor's three result fields into the module's buffer and sorts them so the oldest new sample comes first. When nothing new is there, the status reads No new data.

ReturnsThe number of new data fields, 0 when there are none.

sensor.get_data()returns tuple

Hands out the next new field from the buffer as a BME690Data object with temperature (°C), pressure (Pa), humidity (% r.H.), gas_resistance (Ω), status, gas_index (the profile step) and meas_index.

Returns(data, fields_left): the BME690Data, or None when there is no new data, and how many new fields are left after this one.

Putting it together

The example connects to WiFi, sets the clock over NTP and mounts the SD card on /sd. It then sets the sensor up for parallel mode the same way as Sweep the heater in parallel mode, with a 140 ms time base. It writes the file header and polls the sensor every 70 ms for 10 minutes. Each valid field becomes one row in the file, with the pressure converted from Pa to hPa. When a heater step is missing between two rows, the row gets error code 2, and AI-Studio drops that cycle on import. After the recording the example closes the file, unmounts the card and disconnects from WiFi.