IAES for Integrators — Industrial Asset Event Standard

3 Steps to Your First Event

Install

Choose your language and install the SDK.

          # Python

          pip install iaes

          # TypeScript

          npm install @iaes/sdk

Create an Event

Build a minimal IAES event — just 12 lines of JSON.

{

  "spec_version": "2.0",

  "event_type": "asset.measurement",

  "event_id": "550e...",

  "timestamp": "2026-03-11T14:30:00Z",

  "source": "acme.gateway",

  "asset": { "asset_id": "PUMP-001" },

  "data": {

    "measurement_type": "vibration_velocity",

    "value": 4.2,

    "unit": "mm/s"

  }

}

Publish

Send via HTTP POST or MQTT — any transport works.

          # HTTP

          curl -X POST \

            -H "Content-Type: application/json" \

            -d @event.json \

            https://your-api/iaes/events

          # MQTT

          mosquitto_pub -t "iaes/org/pump-001/asset.measurement" \

            -f event.json

Code Examples

Complete, working producer code in three languages.

import uuid, json, hashlib

from datetime import datetime, timezone

def create_iaes_event(event_type, asset_id, source, data):

    """Create a minimal IAES v2.0 event."""

    data_json = json.dumps(data, sort_keys=True)

    content_hash = hashlib.sha256(data_json.encode()).hexdigest()[:16]

    return {

        "spec_version": "2.0",

        "event_type": event_type,

        "event_id": str(uuid.uuid4()),

        "correlation_id": str(uuid.uuid4()),

        "timestamp": datetime.now(timezone.utc).isoformat(),

        "source": source,

        "content_hash": content_hash,

        "asset": {"asset_id": asset_id},

        "data": data,

    }

# Example: Vibration measurement

event = create_iaes_event(

    "asset.measurement",

    "MOTOR-001",

    "acme.sensors.vibration",

    {

        "measurement_type": "vibration_velocity",

        "value": 4.2,

        "unit": "mm/s",

        "location": "bearing_drive_end",

    }

)

import { randomUUID } from 'crypto';

import { createHash } from 'crypto';

interface IAESEvent {

  spec_version: string;

  event_type: string;

  event_id: string;

  correlation_id: string;

  timestamp: string;

  source: string;

  content_hash?: string;

  asset: { asset_id: string };

  data: Record<string, unknown>;

}

function createIAESEvent(

  eventType: string,

  assetId: string,

  source: string,

  data: Record<string, unknown>

): IAESEvent {

  const dataJson = JSON.stringify(data);

  const contentHash = createHash('sha256')

    .update(dataJson).digest('hex').slice(0, 16);

  return {

    spec_version: '2.0',

    event_type: eventType,

    event_id: randomUUID(),

    correlation_id: randomUUID(),

    timestamp: new Date().toISOString(),

    source,

    content_hash: contentHash,

    asset: { asset_id: assetId },

    data,

  };

}

package iaes

import (

    "crypto/sha256"

    "encoding/json"

    "fmt"

    "time"

    "github.com/google/uuid"

)

type IAESEvent struct {

    SpecVersion   string                 `json:"spec_version"`

    EventType     string                 `json:"event_type"`

    EventID       string                 `json:"event_id"`

    CorrelationID string                 `json:"correlation_id"`

    Timestamp     string                 `json:"timestamp"`

    Source        string                 `json:"source"`

    ContentHash   string                 `json:"content_hash,omitempty"`

    Asset         AssetIdentity          `json:"asset"`

    Data          map[string]interface{} `json:"data"`

}

type AssetIdentity struct {

    AssetID string `json:"asset_id"`

}

func NewEvent(eventType, assetID, source string,

    data map[string]interface{}) IAESEvent {

    dataBytes, _ := json.Marshal(data)

    hash := sha256.Sum256(dataBytes)

    return IAESEvent{

        SpecVersion:   "2.0",

        EventType:     eventType,

        EventID:       uuid.New().String(),

        CorrelationID: uuid.New().String(),

        Timestamp:     time.Now().UTC().Format(time.RFC3339),

        Source:        source,

        ContentHash:   fmt.Sprintf("%x", hash)[:16],

        Asset:         AssetIdentity{AssetID: assetID},

        Data:          data,

    }

}

Consumer Patterns

How to receive and dispatch IAES events. Any system that can read JSON can consume IAES.

def handle_iaes_event(event: dict):

    """Process an incoming IAES event."""

    event_type = event["event_type"]

    handlers = {

        "asset.measurement": handle_measurement,

        "asset.health": handle_health,

        "maintenance.work_order_intent": handle_wo_intent,

        "maintenance.completion": handle_completion,

    }

    handler = handlers.get(event_type)

    if handler:

        handler(event["asset"], event["data"], event.get("correlation_id"))

    else:

        # MUST tolerate unknown event types (IAES spec requirement)

        logger.info(f"Skipping unknown event type: {event_type}")

def handle_measurement(asset, data, correlation_id):

    """Store measurement, check against alarm thresholds."""

    # Dedup check using content_hash

    existing = db.find_by_content_hash(

        data.get("content_hash"), asset["asset_id"]

    )

    if existing:

        return  # Duplicate — skip

    store_measurement(asset["asset_id"], data)

    check_alarm_thresholds(asset["asset_id"], data)

Integration Scenarios

IAES maps to five common industrial integration patterns.

📡

Sensor Gateway

Normalize raw telemetry from any sensor into structured events.

asset.measurement

🤖

AI / ML Engine

Consume measurements, produce health assessments with severity.

asset.health

🛠

CMMS Connector

Receive work order intents, report completion back.

work_order_intent completion

🗃

Historian / Data Lake

Store all events for time-series analytics and compliance.

all events

📈

Dashboard

Visualize asset health, trends, and KPIs in real time.

asset.health

Correlation & Traceability

One incident, one thread across all systems. Every event in a chain shares the same correlation_id.

Measurement

asset.measurement

→

Health

asset.health

→

Work Order

work_order_intent

→

Completion

completion

correlation_id: 550e8400-e29b-41d4-a716-446655440001

Frequently Asked Questions

No. IAES lives above the transport layer. Your PLCs keep speaking Modbus, OPC-UA, or whatever protocol they use. IAES standardizes the event format after your gateway or middleware reads the data.

Any. IAES is transport-agnostic. Use MQTT for real-time, HTTP for request-response, webhooks for push, or NDJSON files for batch. The standard defines the payload, not the pipe.

No. IAES is a data contract — a JSON schema that defines the shape of industrial asset events. It's analogous to CloudEvents for cloud-native systems, but designed specifically for industrial equipment.

Yes. IAES is open-source under CC BY 4.0. You can implement it in any system, with any vendor, in any language. Wertek AI maintains the reference implementation, but the standard is vendor-neutral.

No. Implement only the event types your system produces or consumes. A sensor gateway might only emit asset.measurement. A CMMS connector might only handle work_order_intent and completion.

For Integrators

How IAES Fits