Rohdaten für BME AI-Studio aufzeichnen

Was dieses Beispiel zeigt

Bosch BME AI-Studio trainiert einen Algorithmus zur Gasklassifizierung anhand aufgezeichneter Sensordaten. Dieses Beispiel zeichnet diese Daten auf: Es betreibt den Sensor im Parallel-Modus mit einem Heizprofil aus zehn Schritten, protokolliert 10 Minuten lang jede gültige Gasmessung in eine .bmerawdata-Datei auf einer SD-Karte, im BME AI-Studio Raw Data Format, und schließt die Datei anschließend. Sie importieren die Datei in BME AI-Studio, um die Daten zu labeln.

Der Taster an Pin 0 wechselt während der Aufzeichnung das Label-Tag, was den Beginn eines neuen Specimen in AI-Studio markiert. Setzen Sie LABEL_BUTTON_PIN auf None, um die gesamte Sitzung als ein einziges Specimen aufzuzeichnen.

Was Sie benötigen, wie es das Beispiel auflistet:

• ein ESP32-Board, weil WiFi und NTP die Echtzeituhr stellen;
• das BME690 Breakout an den I2C-Pins oder über Qwiic angeschlossen (BME690() verwendet SDA an Pin 21 und SCL an Pin 22 auf dem esp32-Port);
• ein SD-Karten-Modul an den SPI-Pins: SPI-Bus 2 mit SCK 18, MISO 19, MOSI 23 und Chip Select 5;
• den sdcard-Driver aus micropython-lib, installiert mit mpremote mip install sdcard. Laut README der Bibliothek installiert mip ihn zusammen mit dieser Bibliothek.

Tragen Sie vor dem Ausführen Ihren WiFi-Namen und Ihr Passwort in SSID und PASSWORD ein. Das Beispiel nutzt WiFi nur, um die aktuelle Uhrzeit über NTP zu beziehen. Geben Sie jedem Board eine eigene BOARD_ID: AI-Studio nutzt sie, um Aufzeichnungen verschiedener Boards voneinander zu unterscheiden.

Dieses Beispiel verwendet außerdem den sdcard-Driver aus micropython-lib, der nicht Teil der Soldered BME690-Bibliothek ist.

Der Beispielcode

# 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.")

🔗Vollständiges Beispiel öffnen

Examples/bme690-aiStudioLogger.py bei Commit 91a07b5

→

Erwartetes Ergebnis: Die Shell zeigt den Beginn der Sitzung:

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.

Die Anzahl der Punkte nach Connecting to WiFi und Waiting for NTP hängt davon ab, wie lange der jeweilige Schritt dauert. Während der Aufzeichnung gibt jede gültige Messung eine Zeile der Form T: ... C | P: ... hPa | H: ... % | R: ... Ohm | step: ... aus, und jeder Tastendruck gibt New specimen, label tag is now mit dem neuen Tag aus. Am Ende gibt das Beispiel aus, wie viele Zeilen es geschrieben hat und wie viele Zyklen es als verloren markiert hat, den Namen der Datei auf der Karte und Import it into BME AI-Studio to label the data and train an algorithm.

Verwendete Funktionen

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

Erstellt das Sensorobjekt. Es setzt den Sensor per Soft-Reset zurück, prüft dessen Chip-ID (0x61) und liest die Kalibrierdaten. Es löst eine Exception aus, wenn der Sensor nicht antwortet oder sich ein anderer Chip an der Adresse befindet; jede andere Funktion hält ihr Ergebnis im Status fest, statt eine Exception auszulösen.

ReturnsDas Sensorobjekt.

Parameter

Typ	Name	Beschreibung
I2C	i2c	Ein fertiges machine.I2C-Objekt. Wird es weggelassen, öffnet das Modul I2C(0) mit SCL an Pin 22 und SDA an Pin 21 auf den Ports esp32 und esp8266 sowie einer Soldered-Board-Firmware und löst auf jedem anderen Port eine Exception aus.
int	address	Die I2C-Adresse: BME69X_I2C_ADDR_LOW (0x76, der Standardwert) oder BME69X_I2C_ADDR_HIGH (0x77).
int	amb_temp	Umgebungstemperatur in °C, zur Berechnung der Heizereinstellung verwendet. Standardwert 25.

sensor.check_status()returns int

Fasst den Status zusammen, den die letzte Funktion hinterlassen hat. Die Funktion check_sensor_status() des Beispiels stoppt bei einem Fehler und gibt eine Warnung aus.

ReturnsBME69X_ERROR (-1) nach einem Fehler, BME69X_WARNING (1) nach einer Warnung, sonst BME69X_OK (0).

sensor.status_string()returns str

Beschreibt den aktuellen Statuscode in wenigen Worten, etwa Communication failure oder No new data.

ReturnsDie Beschreibung, ein leerer String, wenn der Status OK ist.

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

Legt das Oversampling für Temperatur, Druck und Luftfeuchtigkeit fest. Ohne Argumente aufgerufen, wie hier, setzt es die Standardwerte der Arduino-Bibliothek.

ReturnsNichts. Das Ergebnis steht im Status.

Parameter

Typ	Name	Beschreibung
int	os_temp	Oversampling der Temperatur, BME69X_OS_NONE bis BME69X_OS_16X.
int	os_pres	Oversampling des Drucks, BME69X_OS_NONE bis BME69X_OS_16X.
int	os_hum	Oversampling der Luftfeuchtigkeit, BME69X_OS_NONE bis BME69X_OS_16X.

sensor.get_meas_dur(op_mode=BME69X_SLEEP_MODE)returns int

Berechnet, wie lange eine Messung von Temperatur, Druck und Luftfeuchtigkeit dauert, anhand des mit set_tph() gesetzten Oversamplings, rufen Sie also zuerst set_tph() auf. Im Parallel-Modus wird keine Aufwachzeit hinzugerechnet.

ReturnsDie Messdauer in Mikrosekunden.

Parameter

Typ	Name	Beschreibung
int	op_mode	Der Modus, für den berechnet wird. Das Beispiel übergibt BME69X_PARALLEL_MODE, weil der Modus noch nicht gesetzt ist.

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

Legt das Heizprofil des Gassensors fest und schaltet den Heizer ein. Mit einer Liste von Temperaturen, einer Liste von Multiplikatoren und einer gemeinsamen Heizdauer, wie hier, schreibt es ein Profil für den Parallel-Modus: Jeder Schritt dauert sein Vielfaches der gemeinsamen Dauer. Temperaturen über 400 °C werden auf 400 °C begrenzt.

ReturnsNichts. Das Ergebnis steht im Status.

Parameter

Typ	Name	Beschreibung
list	temp	Heizertemperatur jedes Schritts in °C. Das Beispiel verwendet HEATER_TEMP.
list	dur	Im Parallel-Modus der Multiplikator der gemeinsamen Dauer für jeden Schritt. Das Beispiel verwendet HEATER_MUL.
int	shared_heatr_dur	Die gemeinsame Heizdauer in ms, hier MEAS_DUR (140 ms) abzüglich der Messdauer.

sensor.set_op_mode(op_mode)returns None

Versetzt den Sensor in den Schlafmodus und setzt dann den gewünschten Modus. Mit BME69X_PARALLEL_MODE misst der Sensor weiter und durchfährt den Heizer selbstständig. Der Modus wird für fetch_data() und get_data() gespeichert.

ReturnsNichts. Das Ergebnis steht im Status.

Parameter

Typ	Name	Beschreibung
int	op_mode	BME69X_SLEEP_MODE, BME69X_FORCED_MODE, BME69X_PARALLEL_MODE oder BME69X_SEQUENTIAL_MODE.

sensor.fetch_data()returns int

Liest die drei Ergebnisfelder des Sensors in den Puffer des Moduls und sortiert sie so, dass der älteste neue Messwert zuerst kommt. Sind keine neuen Daten da, lautet der Status No new data.

ReturnsDie Anzahl neuer Datenfelder, 0, wenn keine vorhanden sind.

sensor.get_data()returns tuple

Gibt das nächste neue Feld aus dem Puffer als BME690Data-Objekt zurück, mit temperature (°C), pressure (Pa), humidity (% r.H.), gas_resistance (Ω), status, gas_index (der Profilschritt) und meas_index.

Returns(data, fields_left): das BME690Data-Objekt oder None, wenn es keine neuen Daten gibt, und wie viele neue Felder nach diesem noch übrig sind.

Alles zusammengesetzt

Das Beispiel verbindet sich mit WiFi, stellt die Uhr über NTP und bindet die SD-Karte unter /sd ein. Danach richtet es den Sensor für den Parallel-Modus genauso ein wie in Heizer im Parallel-Modus durchfahren, mit einer Zeitbasis von 140 ms. Es schreibt den Dateikopf und fragt den Sensor 10 Minuten lang alle 70 ms ab. Jedes gültige Feld wird zu einer Zeile in der Datei, wobei der Druck von Pa in hPa umgerechnet wird. Fehlt zwischen zwei Zeilen ein Heizschritt, erhält die Zeile den Fehlercode 2, und AI-Studio verwirft diesen Zyklus beim Import. Nach der Aufzeichnung schließt das Beispiel die Datei, hängt die Karte aus und trennt die WiFi-Verbindung.