Compilare moduli WASM per i flussi di dati

Ambiente di sviluppo:

• Visual Studio Code
• (Opzionale) RedHat YAML estensione per VS Code
• Estensione Azure IoT Operations Data Flow per VS Code.
• Estensione CodeLLDB per VS Code per abilitare il debug dei moduli WASM
• Docker

Immagini 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

• aio-dataflow CLI. Eseguire l'installazione con npx @azure-tools/dataflow-dev o a livello globale con npm install -g @azure-tools/dataflow-dev.
• Azure CLI
• Docker

Immagini 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

Se si sta sviluppando usando Rust, installare la toolchain Rust e aggiungere la destinazione 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

Se si sviluppa usando Python, assicurarsi di aver installato Python 3.8 o versione successiva, insieme allo strumento componentize-py per la creazione di pacchetti di codice Python come moduli 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

Questo esempio usa un'area di lavoro di esempio che contiene tutte le risorse necessarie per compilare ed eseguire un'applicazione graph in locale usando l'estensione VS Code.

Aprire l'area di lavoro di esempio in VS Code

Clonare il repository Explore IoT Operations se non è già stato fatto.

Aprire la cartella samples/wasm in Visual Studio Code selezionando File > Apri cartella e passando alla cartella samples/wasm.

Costruire gli operatori

Premere Ctrl+Shift+P per aprire il riquadro comandi e cercare Azure IoT Operations: Compila tutti gli operatori. Selezionare Release (Versione ) come modalità di compilazione.

Questo comando costruisce tutti gli operatori nell'area di lavoro e crea .wasm file nella operators cartella. Usi i file .wasm per eseguire l'applicazione grafica in locale.

Eseguire l'applicazione graph in locale

Per avviare l'ambiente di esecuzione locale, premere Ctrl+Shift+P per aprire il riquadro comandi e cercare Azure IoT Operations: Avvia ambiente di sviluppo. Selezionare release come modalità di esecuzione.

Quando l'ambiente di esecuzione locale è in esecuzione, premere Ctrl+Shift+P per aprire il riquadro comandi e cercare Azure IoT Operations: Esegui Application Graph. Selezionare release come modalità di esecuzione. Questo comando esegue l'applicazione graph in locale usando l'ambiente di esecuzione locale con il graph.dataflow.yaml file nell'area di lavoro.

Legge anche da hostapp.env.list per impostare la variabile di ambiente TK_CONFIGURATION_PARAMETERS per i parametri di configurazione dell'operatore di flusso di dati.

Quando viene richiesto di immettere i dati, selezionare la data-and-images cartella nell'area di lavoro. Questa cartella contiene i file di dati di input per l'applicazione graph, inclusi i dati relativi a temperatura e umidità e alcune immagini per il modulo snapshot.

Attendere che venga visualizzata una notifica di VS Code che indica che i log sono pronti: Log files for the run can be found at ...\wasm\data-and-images\output\logs.

L'output si trova nella cartella output sotto la cartella data-and-images. È possibile aprire la output cartella nell'area di lavoro per visualizzare i file di output. Il .txt file con la data e l'ora per il nome file contiene i dati elaborati ed è simile all'esempio seguente:

{"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"}]}}

L'output mostra che l'applicazione grafico ha elaborato i dati di input e generato l'output. L'output include i dati relativi a temperatura e umidità e gli oggetti rilevati nelle immagini.

Clonare il repository Explore IoT Operations se non è già stato fatto:

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

Costruire gli operatori

Passare alla directory dell'applicazione graph e compilare tutti gli operatori:

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

Il --app flag specifica la directory che contiene il graph.dataflow.yaml file e la operators cartella. La CLI compila tutti gli operatori nell'area di lavoro e crea file .wasm.

Eseguire l'applicazione graph in locale

Avviare l'ambiente di esecuzione Docker locale, quindi eseguire il grafico:

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

Il test comando esegue il grafico definito nel file del .test.yaml test case, inserisce i dati di input e confronta l'output con i risultati previsti. Il repository di esempio include più scenari di test, ad esempio:

Per eseguire tutti i test:

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

Se si usano gli strumenti Rust o Python installati in locale, non sono disponibili esempi pronti per l'esecuzione. È necessario creare un'applicazione graph e operatori personalizzati, quindi compilarli ed eseguirli in locale. Segui le istruzioni nella sezione successiva per creare una nuova applicazione grafica con moduli WASM personalizzati.

Anziché usare un'area di lavoro di esempio esistente, si crea una nuova area di lavoro da zero. Questo processo consente di apprendere come creare una nuova applicazione graph e programmare gli operatori in Python e Rust.

Creare un nuovo progetto di applicazione graph in Python

Premere Ctrl+Shift+P per aprire il riquadro comandi di VS Code e cercare Azure IoT Operations: Crea applicazione:

1. Per la cartella, selezionare una cartella in cui si vuole creare il progetto. È possibile creare una nuova cartella per questo progetto.
2. Immettere my-graph come nome.
3. Selezionare Python come lingua.
4. Selezionare Mappa come tipo.
5. Immettere map come nome.

È ora disponibile una nuova area di lavoro di VS Code con la struttura di progetto di base e i file di avvio. I file di avvio includono il graph.dataflow.yaml file e il codice sorgente del modello di operatore map.

Aggiungere codice Python per il modulo operatore map

Aprire il operators/map/map.py file e sostituire il contenuto con il codice seguente per convertire un valore di temperatura in ingresso da Fahrenheit a 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

Assicurarsi che Docker sia in esecuzione. Premere quindi Ctrl+Shift+P per aprire il riquadro comandi e cercare Azure IoT Operations: Compila tutti gli operatori. Creare un modulo di versione .

Il processo di compilazione inserisce il map.wasm file per l'operatore map nella operators/map/bin/release cartella .

Aggiungere il codice Rust per il modulo dell'operatore di filtro

Creare un nuovo operatore premendo Ctrl+Shift+P per aprire il riquadro comandi e cercare Azure IoT Operations: Crea operatore:

1. Selezionare Rust come lingua.
2. Selezionare Filter come tipo di operatore.
3. Immettere filter come nome.

Aprire il operators/filter/src/lib.rs file e sostituire il contenuto con il codice seguente per filtrare i valori in cui la temperatura è superiore a 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)
   }
}

Aprire il operators/filter/Cargo.toml file e aggiungere le dipendenze seguenti:

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

Assicurarsi che Docker sia in esecuzione. Premere quindi Ctrl+Shift+P per aprire il riquadro comandi e cercare Azure IoT Operations: Compila tutti gli operatori. Creare un modulo di versione .

Il processo di compilazione inserisce il filter.wasm file per l'operatore filter nella operators/filter/bin/release cartella .

Per creare e compilare nuovi operatori con l'interfaccia della riga di comando aio-dataflow, usare uno dei progetti di esempio come modello e creare nuovi file di origine dell'operatore nella operators cartella . Usare quindi il aio-dataflow build comando per compilare gli operatori e generare i .wasm file:

• Per i progetti Rust, vedere gli esempi nella explore-iot-operations/samples/wasm cartella .
• Per i progetti Python, vedere gli esempi nella cartella explore-iot-operations/samples/wasm-python.

Creare un convertitore di temperatura che trasforma Fahrenheit in Celsius.

Se si usa Rust:

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

Creare .cargo/config.toml per configurare il Registro di sistema SDK:

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

[build]
target = "wasm32-wasip2"

Modificare 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"]

Scrivere 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))
}

Per compilare il modulo, eseguire:

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

È possibile usare una compilazione in contenitori per CI/CD o quando non si vuole configurare una toolchain 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 deve corrispondere al nome del crate presente in Cargo.toml. Per altre opzioni, vedere la documentazione Rust Docker Builder.

Se si usa Python:

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

Creare 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

Per compilare il modulo, eseguire:

# 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

È possibile usare una compilazione in contenitori per CI/CD o quando non si vuole configurare una toolchain locale Python:

# 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 deve corrispondere al nome file Python (senza .py). --app-type deve corrispondere al tipo di operatore (map, filter, branche così via). Per altre opzioni, vedere la documentazione del Docker builder di Python.

Aprire il graph.dataflow.yaml file e sostituire il contenuto con il codice seguente:

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

Copiare la data cartella contenente i dati di esempio dal repository explore-iot-operations\samples\wasm\data degli esempi clonati nell'area di lavoro corrente. La data cartella contiene tre file JSON con dati di temperatura di input di esempio.

Se in precedenza è stato arrestato l'ambiente di esecuzione locale, premere Ctrl+Shift+P per aprire il riquadro comandi e cercare Azure IoT Operations: Avvia ambiente di sviluppo. Selezionare release come modalità di esecuzione.

Premere Ctrl+Shift+P per aprire il riquadro comandi e cercare Azure IoT Operations: Esegui Application Graph:

1. Selezionare il file del graph.dataflow.yaml grafico.
2. Selezionare release come modalità di esecuzione.
3. Selezionare la data cartella copiata nell'area di lavoro.

Il contenitore DevX viene avviato per eseguire il grafico. Il risultato elaborato viene salvato nella data/output cartella . Il file di testo nella output cartella contiene i dati elaborati con la temperatura convertita in Celsius e filtrati in base alla soglia.

Per informazioni su come distribuire moduli WASM personalizzati e grafici nell'istanza di Azure IoT Operations, vedere Deploy WASM modules and data flow graphs.

Per eseguire l'applicazione grafica in locale con il CLI aio-dataflow, usare il comando aio-dataflow test con un test case che punta alla configurazione del grafo e ai dati di input. È possibile creare un nuovo file YAML per il caso di test nella cartella test-runner/tests che fa riferimento ai file di dati di input graph.dataflow.yaml. Eseguire quindi il test con il comando seguente:

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

Esaminare i file del test case di esempio nella test-runner/tests cartella per informazioni di riferimento su come strutturare il file YAML del test case. È possibile testare sia i moduli Python che quelli Rust con lo strumento CLI puntando alla configurazione del grafo e ai dati di input appropriati nel file del caso di test.

Se si usano gli strumenti standard installati in locale, non è disponibile alcuno strumento incluso per l'esecuzione dell'applicazione graph con i moduli WASM personalizzati nell'ambiente locale.