Générer des modules WASM pour les flux de données

Environnement de développement :

• Visual Studio Code
• (Facultatif) Extension REDHat YAML pour VS Code
• extension Opérations Azure IoT Data Flow pour VS Code.
• Extension CodeLLDB pour VS Code pour activer le débogage des modules WASM
• Docker

Images Docker :

docker pull mcr.microsoft.com/azureiotoperations/processor-app:1.1.5
docker tag mcr.microsoft.com/azureiotoperations/processor-app:1.1.5 host-app

docker pull mcr.microsoft.com/azureiotoperations/devx-runtime:0.1.8
docker tag mcr.microsoft.com/azureiotoperations/devx-runtime:0.1.8 devx

docker pull mcr.microsoft.com/azureiotoperations/statestore-cli:0.0.2
docker tag mcr.microsoft.com/azureiotoperations/statestore-cli:0.0.2 statestore-cli

docker pull eclipse-mosquitto

• CLI dataflow-dev. Installer avec npx @azure-tools/dataflow-dev ou globalement avec npm install -g @azure-tools/dataflow-dev.
• Azure CLI
• Docker

Images Docker :

docker pull mcr.microsoft.com/azureiotoperations/processor-app:1.1.5
docker tag mcr.microsoft.com/azureiotoperations/processor-app:1.1.5 host-app

docker pull mcr.microsoft.com/azureiotoperations/devx-runtime:0.1.8
docker tag mcr.microsoft.com/azureiotoperations/devx-runtime:0.1.8 devx

docker pull mcr.microsoft.com/azureiotoperations/statestore-cli:0.0.2
docker tag mcr.microsoft.com/azureiotoperations/statestore-cli:0.0.2 statestore-cli

docker pull eclipse-mosquitto

Si vous développez à l’aide de Rust, installez la chaîne d’outils Rust et ajoutez la cible WASI Preview 2 :

# Install Rust toolchain
curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y

# Add WASM target (required for Azure IoT Operations WASM components)
rustup target add wasm32-wasip2

# Install build tools for validating and packaging WASM artifacts
cargo install wasm-tools --version '=1.201.0' --locked

Si vous développez à l'aide de Python, vérifiez que vous avez installé Python 3.8 ou version ultérieure, ainsi que l'outil componentize-py pour empaqueter du code Python en tant que modules WASM :

# Install componentize-py for building Python WASM modules
pip install "componentize-py==0.14"

# Clone the samples repo (provides WIT schemas and project structure)
git clone https://github.com/Azure-Samples/explore-iot-operations.git

Cet exemple utilise un exemple d’espace de travail qui contient toutes les ressources nécessaires pour générer et exécuter une application graphe localement à l’aide de l’extension VS Code.

Ouvrir l’exemple d’espace de travail dans VS Code

Clonez le référentiel Explore IoT Operations si ce n'est déjà fait.

Ouvrez le dossier samples/wasm dans Visual Studio Code en sélectionnant File > Ouvrir le dossier et en accédant au dossier samples/wasm.

Générer les opérateurs

Appuyez sur Ctrl+Shift+P pour ouvrir la palette de commandes et rechercher Opérations Azure IoT : Générer tous les opérateurs. Sélectionnez la mise en production en tant que mode de génération.

Cette commande génère tous les opérateurs de l’espace de travail et crée des .wasm fichiers dans le operators dossier. Vous utilisez les fichiers .wasm pour exécuter l’application graphe localement.

Exécuter l’application graphe localement

Pour démarrer l’environnement d’exécution local, appuyez sur Ctrl+Shift+P pour ouvrir la palette de commandes et rechercher Opérations Azure IoT : Démarrer l’environnement de développement. Sélectionnez la mise en production comme mode d’exécution.

Lorsque l’environnement d’exécution local est en cours d’exécution, appuyez sur Ctrl+Shift+P pour ouvrir la palette de commandes et rechercher Opérations Azure IoT : Exécuter Application Graph. Sélectionnez la mise en production comme mode d’exécution. Cette commande exécute localement l’application graphe à l’aide de l’environnement d’exécution local avec le fichier graph.dataflow.yaml de l’espace de travail.

Il lit également à partir de hostapp.env.list pour définir la variable TK_CONFIGURATION_PARAMETERS d’environnement pour les paramètres de configuration de l’opérateur de flux de données.

Lorsque vous êtes invité à entrer des données d’entrée, sélectionnez le data-and-images dossier dans l’espace de travail. Ce dossier contient les fichiers de données d’entrée de l’application graphique, y compris les données de température et d’humidité, ainsi que certaines images du module d’instantané.

Attendez de voir une notification VS Code indiquant que les journaux sont prêts : Log files for the run can be found at ...\wasm\data-and-images\output\logs.

La sortie se trouve dans le output dossier sous le data-and-images dossier. Vous pouvez ouvrir le output dossier dans l’espace de travail pour afficher les fichiers de sortie. Le .txt fichier avec la date et l’heure du nom de fichier contient les données traitées, et il ressemble à l’exemple suivant :

{"tst":"2025-09-19T04:19:13.530381+0000","topic":"sensors","qos":0,"retain":0,"payloadlen":312,"properties":{"payload-format-indicator":1,"message-expiry-interval":10,"correlation-data":"...","user-properties":{"__ts":"001758255553528:00000:...","__protVer":"1.0","__srcId":"mqtt-source"},"content-type":"application/json"},"payload":{"temperature":[{"count":2,"max":653.888888888889,"min":204.44444444444449,"average":429.16666666666669,"last":204.44444444444449,"unit":"C","overtemp":true}],"humidity":[{"count":3,"max":85,"min":45,"average":69.666666666666671,"last":79}],"object":[{"result":"notebook, notebook computer; sliding door"}]}}

La sortie indique que l’application de graphique a traité les données d’entrée et généré la sortie. La sortie inclut les données de température et d’humidité, ainsi que les objets détectés dans les images.

Clonez le référentiel Explore IoT Operations si vous ne l'avez pas déjà fait :

git clone https://github.com/Azure-Samples/explore-iot-operations.git

Générer les opérateurs

Accédez à votre répertoire d’application graphe et générez tous les opérateurs :

cd explore-iot-operations/samples/wasm
dataflow-dev build --app .

L’indicateur --app spécifie le répertoire qui contient le graph.dataflow.yaml fichier et le operators dossier. L’interface CLI génère tous les opérateurs dans l’espace de travail et crée des .wasm fichiers.

Exécuter l’application graphe localement

Démarrez l’environnement d’exécution Docker local, puis exécutez le graphique :

dataflow-dev run start
dataflow-dev test --app . test-runner/tests/t03-complex-full-pipeline
dataflow-dev run stop

La test commande exécute le graphique défini dans le fichier du cas de .test.yaml test, alimente les données d’entrée et compare la sortie par rapport aux résultats attendus. L’exemple de référentiel inclut plusieurs scénarios de test tels que :

Pour exécuter tous les tests :

dataflow-dev run start
dataflow-dev test --app . test-runner/tests
dataflow-dev run stop

Si vous utilisez les outils Rust ou Python installés localement, aucun exemple n'est prêt à s'exécuter. Vous devez créer votre propre application et opérateurs de graphe, puis les générer et les exécuter localement. Suivez les instructions de la section suivante pour créer une application de graphe avec des modules WASM personnalisés.

Au lieu d’utiliser un exemple d’espace de travail existant, vous créez un espace de travail à partir de zéro. Ce processus vous permet de découvrir comment créer une application de graphe et programmer les opérateurs dans Python et Rust.

Créer un projet d’application graphe dans Python

Appuyez sur Ctrl+Shift+P pour ouvrir la palette de commandes VS Code et rechercher Opérations Azure IoT : Créer une application :

1. Pour le dossier, sélectionnez un dossier dans lequel vous souhaitez créer le projet. Vous pouvez créer un dossier pour ce projet.
2. Entrez my-graph comme nom.
3. Sélectionnez Python comme langue.
4. Sélectionnez Mapper comme type.
5. Entrez map comme nom.

Vous disposez maintenant d’un nouvel espace de travail VS Code avec la structure de projet de base et les fichiers de démarrage. Les fichiers de démarrage incluent le fichier graph.dataflow.yaml et le code source du modèle d’opérateur de mappage.

Ajouter Python code pour le module d’opérateur de carte

Ouvrez le fichier et remplacez le operators/map/map.py contenu par le code suivant pour convertir une valeur de température entrante de Fahrenheit en Celsius :

import json
from map_impl import exports
from map_impl import imports
from map_impl.imports import types

class Map(exports.Map):
    def init(self, configuration) -> bool:
        imports.logger.log(imports.logger.Level.INFO, "module4/map", "Init invoked")
        return True

    def process(self, message: types.DataModel) -> types.DataModel:
        # TODO: implement custom logic for map operator
        imports.logger.log(imports.logger.Level.INFO, "module4/map", "processing from python")

        # Ensure the input is of the expected type
        if not isinstance(message, types.DataModel_Message):
            raise ValueError("Unexpected input type: Expected DataModel_Message")

        # Extract and decode the payload
        payload_variant = message.value.payload
        if isinstance(payload_variant, types.BufferOrBytes_Buffer):
            # It's a Buffer handle - read from host
            imports.logger.log(imports.logger.Level.INFO, "module4/map", "Reading payload from Buffer")
            payload = payload_variant.value.read()
        elif isinstance(payload_variant, types.BufferOrBytes_Bytes):
            # It's already bytes
            imports.logger.log(imports.logger.Level.INFO, "module4/map", "Reading payload from Bytes")
            payload = payload_variant.value
        else:
            raise ValueError("Unexpected payload type")

        decoded = payload.decode("utf-8")

        # Parse the JSON data
        json_data = json.loads(decoded)

        # Check and update the temperature value
        if "temperature" in json_data and "value" in json_data["temperature"]:
            temp_f = json_data["temperature"]["value"]
            if isinstance(temp_f, int):
                # Convert Fahrenheit to Celsius
                temp_c = round((temp_f - 32) * 5.0 / 9.0)

                # Update the JSON data
                json_data["temperature"]["value"] = temp_c
                json_data["temperature"]["unit"] = "C"

                # Serialize the updated JSON back to bytes
                updated_payload = json.dumps(json_data).encode("utf-8")

                # Update the message payload
                message.value.payload = types.BufferOrBytes_Bytes(value=updated_payload)

        return message

Vérifiez que Docker est en cours d’exécution. Ensuite, appuyez sur Ctrl+Shift+P pour ouvrir la palette de commandes et rechercher Opérations Azure IoT : Générer tous les opérateurs. Créez un module de mise en production .

Le processus de génération place le map.wasm fichier de l’opérateur map dans le operators/map/bin/release dossier.

Ajouter du code Rust pour le module d’opérateur de filtre

Créez un opérateur en appuyant sur Ctrl+Shift+P pour ouvrir la palette de commandes et rechercher Opérations Azure IoT : Créer un opérateur :

1. Sélectionnez Rust comme langue.
2. Sélectionnez Filtrer comme type d’opérateur.
3. Entrez filter comme nom.

Ouvrez le operators/filter/src/lib.rs fichier et remplacez le contenu par le code suivant pour filtrer les valeurs où la température est supérieure à 500 °C :

mod filter_operator {
    use wasm_graph_sdk::macros::filter_operator;
    use serde_json::Value;

    fn filter_init(_configuration: ModuleConfiguration) -> bool {
        // Add code here to process the module init properties and module schemas from the configuration

        true
    }

    #[filter_operator(init = "filter_init")]
    fn filter(input: DataModel) -> Result<bool, Error> {

        // Extract payload from input to process
        let payload = match input {
            DataModel::Message(Message {
                payload: BufferOrBytes::Buffer(buffer),
                ..
            }) => buffer.read(),
            DataModel::Message(Message {
                payload: BufferOrBytes::Bytes(bytes),
                ..
            }) => bytes,
            _ => return Err(Error { message: "Unexpected input type".to_string() }),
        };

        // ... perform filtering logic here and return boolean
        if let Ok(payload_str) = std::str::from_utf8(&payload) {
            if let Ok(json) = serde_json::from_str::<Value>(payload_str) {
                if let Some(temp_c) = json["temperature"]["value"].as_i64() {
                    // Return true if temperature is above 500°C
                    return Ok(temp_c > 500);
                }
            }
        }

        Ok(false)
   }
}

Ouvrez le operators/filter/Cargo.toml fichier et ajoutez les dépendances suivantes :

[dependencies]
# ...
serde = { version = "1.0", features = ["derive"] }
serde_json = "1.0"

Vérifiez que Docker est en cours d’exécution. Ensuite, appuyez sur Ctrl+Shift+P pour ouvrir la palette de commandes et rechercher Opérations Azure IoT : Générer tous les opérateurs. Créez un module de mise en production .

Le processus de génération place le filter.wasm fichier de l’opérateur filter dans le operators/filter/bin/release dossier.

Pour créer et générer de nouveaux opérateurs avec l’interface CLI de développement de flux de données, utilisez l’un des exemples de projets en tant que modèle et créez de nouveaux fichiers sources d’opérateur dans le operators dossier. Ensuite, utilisez la dataflow-dev build commande pour générer les opérateurs et générer les .wasm fichiers :

• Pour les projets Rust, consultez les exemples du explore-iot-operations/samples/wasm dossier.
• Pour Python projets, consultez les exemples du dossier explore-iot-operations/samples/wasm-python.

Créez un convertisseur de température qui transforme Fahrenheit en Celsius.

Si vous utilisez Rust :

cargo new --lib temperature-converter
cd temperature-converter

Créez .cargo/config.toml pour configurer le Registre du Kit de développement logiciel (SDK) :

[registries]
aio-wg = { index = "sparse+https://pkgs.dev.azure.com/azure-iot-sdks/iot-operations/_packaging/preview/Cargo/index/" }

[build]
target = "wasm32-wasip2"

Modifiez Cargo.toml:

[package]
name = "temperature-converter"
version = "0.1.0"
edition = "2021"

[dependencies]
wit-bindgen = "0.22"
wasm_graph_sdk = { version = "=1.1.3", registry = "aio-wg" }
serde = { version = "1", default-features = false, features = ["derive"] }
serde_json = { version = "1", default-features = false, features = ["alloc"] }

[lib]
crate-type = ["cdylib"]

Écrire src/lib.rs:

use serde_json::{json, Value};

use wasm_graph_sdk::logger::{self, Level};
use wasm_graph_sdk::macros::map_operator;

fn fahrenheit_to_celsius_init(_configuration: ModuleConfiguration) -> bool {
    logger::log(Level::Info, "temperature-converter", "Init invoked");
    true
}

#[map_operator(init = "fahrenheit_to_celsius_init")]
fn fahrenheit_to_celsius(input: DataModel) -> Result<DataModel, Error> {
    let DataModel::Message(mut result) = input else {
        return Err(Error {
            message: "Unexpected input type".to_string(),
        });
    };

    let payload = &result.payload.read();
    if let Ok(data_str) = std::str::from_utf8(payload) {
        if let Ok(mut data) = serde_json::from_str::<Value>(data_str) {
            if let Some(temp) = data["temperature"]["value"].as_f64() {
                let celsius = (temp - 32.0) * 5.0 / 9.0;
                data["temperature"] = json!({
                    "value_celsius": celsius,
                    "original_fahrenheit": temp
                });

                if let Ok(output_str) = serde_json::to_string(&data) {
                    result.payload = BufferOrBytes::Bytes(output_str.into_bytes());
                }
            }
        }
    }

    Ok(DataModel::Message(result))
}

Pour générer le module, exécutez :

cargo build --release --target wasm32-wasip2
cp target/wasm32-wasip2/release/temperature_converter.wasm .

Vous pouvez utiliser une build conteneurisée pour CI/CD ou quand vous ne souhaitez pas configurer une chaîne d’outils Rust locale :

# Release build
docker run --rm -v "$(pwd):/workspace" \
  ghcr.io/azure-samples/explore-iot-operations/rust-wasm-builder \
  --app-name temperature-converter

# Debug build (includes symbols)
docker run --rm -v "$(pwd):/workspace" \
  ghcr.io/azure-samples/explore-iot-operations/rust-wasm-builder \
  --app-name temperature-converter --build-mode debug

--app-name doit correspondre à votre nom de crate de Cargo.toml. Pour plus d’options, consultez la documentation du générateur Docker Rust.

Si vous utilisez Python :

cd explore-iot-operations/samples/wasm-python/operators/map

Créez temperature_converter.py :

import json
from map_impl import exports
from map_impl import imports
from map_impl.imports import types

class Map(exports.Map):
    def init(self, configuration) -> bool:
        imports.logger.log(imports.logger.Level.INFO, "temperature-converter", "Init invoked")
        return True

    def process(self, message: types.DataModel) -> types.DataModel:
        if not isinstance(message, types.DataModel_Message):
            raise ValueError("Unexpected input type: Expected DataModel_Message")

        payload_variant = message.value.payload
        if isinstance(payload_variant, types.BufferOrBytes_Buffer):
            payload = payload_variant.value.read()
        elif isinstance(payload_variant, types.BufferOrBytes_Bytes):
            payload = payload_variant.value
        else:
            raise ValueError("Unexpected payload type")

        decoded = payload.decode("utf-8")
        data = json.loads(decoded)

        if "temperature" in data and "value" in data["temperature"]:
            temp_f = data["temperature"]["value"]
            if isinstance(temp_f, (int, float)):
                temp_c = (temp_f - 32) * 5.0 / 9.0
                data["temperature"]["value"] = temp_c
                data["temperature"]["unit"] = "C"
                updated_payload = json.dumps(data).encode("utf-8")
                message.value.payload = types.BufferOrBytes_Bytes(value=updated_payload)

        return message

Pour générer le module, exécutez :

# Generate Python bindings from WIT schemas
componentize-py -d ../../schema -w map-impl bindings ./

# Build the WASM module
componentize-py -d ../../schema -w map-impl componentize temperature_converter -o temperature_converter.wasm

# Verify build
file temperature_converter.wasm  # Should show: WebAssembly (wasm) binary module

Vous pouvez utiliser une build conteneurisée pour CI/CD ou quand vous ne souhaitez pas configurer une chaîne d'outils Python locale :

# Release build
docker run --rm -v "$(pwd):/workspace" \
  ghcr.io/azure-samples/explore-iot-operations/python-wasm-builder \
  --app-name temperature_converter --app-type map

# Debug build (includes symbols)
docker run --rm -v "$(pwd):/workspace" \
  ghcr.io/azure-samples/explore-iot-operations/python-wasm-builder \
  --app-name temperature_converter --app-type map --build-mode debug

--app-name doit correspondre à votre nom de fichier Python (sans .py). --app-type doit correspondre au type d’opérateur (map, filter, branchetc.). Pour plus d’options, consultez Python documentation du générateur Docker.

Ouvrez le fichier graph.dataflow.yaml et remplacez son contenu par le code suivant :

metadata:
    $schema: "https://www.schemastore.org/aio-wasm-graph-config-1.0.0.json"
    name: "Temperature Monitoring"
    description: "A graph that converts temperature from Fahrenheit to Celsius, if temperature is above 500°C, then sends the processed data to the sink."
    version: "1.0.0"
    vendor: "Microsoft"
moduleRequirements:
    apiVersion: "1.1.0"
    runtimeVersion: "1.1.0"
operations:
  - operationType: source
    name: source
  - operationType: map
    name: map
    module: map
  - operationType: filter
    name: filter
    module: filter
  - operationType: sink
    name: sink
connections:
  - from:
      name: source
    to:
      name: map
  - from:
      name: map
    to:
      name: filter
  - from:
      name: filter
    to:
      name: sink

Copiez le data dossier qui contient les exemples de données du référentiel explore-iot-operations\samples\wasm\data d’exemples clonés dans l’espace de travail actuel. Le data dossier contient trois fichiers JSON avec des exemples de données de température d’entrée.

Si vous avez précédemment arrêté l’environnement d’exécution local, appuyez sur Ctrl+Shift+P pour ouvrir la palette de commandes et rechercher Opérations Azure IoT : Démarrer l’environnement de développement. Sélectionnez la mise en production comme mode d’exécution.

Appuyez sur Ctrl+Shift+P pour ouvrir la palette de commandes et rechercher Opérations Azure IoT : Exécuter Application Graph :

1. Sélectionnez le graph.dataflow.yaml fichier de graphique.
2. Sélectionnez la mise en production comme mode d’exécution.
3. Sélectionnez le data dossier que vous avez copié dans l’espace de travail.

Le conteneur DevX démarre pour exécuter le graphique. Le résultat traité est enregistré dans le data/output dossier. Le fichier texte du output dossier contient les données traitées avec la température convertie en Celsius et filtrée en fonction du seuil.

Pour savoir comment déployer vos modules ET graphiques WASM personnalisés sur votre instance de Opérations Azure IoT, consultez Déployer des modules WASM et des graphiques de flux de données.

Pour exécuter votre application graphe localement avec l’interface CLI dataflow-dev, utilisez la dataflow-dev test commande avec un cas de test qui pointe vers votre configuration de graphe et vos données d’entrée. Vous pouvez créer un fichier de cas de test YAML dans le dossier test-runner/tests qui référence vos fichiers graph.dataflow.yaml et de données d’entrée. Exécutez ensuite le test avec la commande suivante :

dataflow-dev run start
dataflow-dev test --app . test-runner/tests/my-test-case
dataflow-dev run stop

Examinez les exemples de fichiers de cas de test dans le dossier pour obtenir des informations de référence sur la test-runner/tests façon de structurer votre fichier YAML de cas de test. Vous pouvez tester les modules Python et Rust avec l’outil CLI en pointant vers la configuration de graphique et les données d’entrée appropriées dans votre fichier de cas de test.

Si vous utilisez des outils standard installés localement, il n’existe aucun outil inclus pour l’exécution de l’application graphe avec vos modules WASM personnalisés dans votre environnement local.