IoT Ubiquitous Access

EnerOS v0.45.0 enhances IoT ubiquitous access capabilities, with built-in MQTT 5.0 broker, CoAP server, LwM2M device management, sensor network aggregation, and edge device batch OTA. Massive IoT devices can directly access EnerOS without external IoT platforms. The access capability is provided by the eneros-iot crate.

IoT Access Architecture

┌──────────────────────────────────────────────────────────────┐
│                     Edge Device Layer                        │
│  PMU / RTU / IED / Smart Meter / Sensor / Edge Gateway       │
└────────┬───────────────┬───────────────┬─────────────────────┘
         │               │               │
┌────────▼──────┐ ┌──────▼──────┐ ┌──────▼──────────────────┐
│ MQTT 5.0      │ │ CoAP        │ │ LwM2M                   │
│ Broker        │ │ Server      │ │ Device Mgmt             │
│ (TCP/TLS)     │ │ (UDP/DTLS)  │ │ (CoAP/DTLS)             │
└────────┬──────┘ └──────┬──────┘ └────────┬────────────────┘
         │               │                 │
┌────────▼───────────────▼─────────────────▼──────────────────┐
│                     Protocol Adaptation Layer                │
│  IEC 61850 / IEC 60870 / DNP3 / Modbus / DL/T 698.45        │
│  C37.118 (PMU) / OPC UA                                     │
└────────┬─────────────────────────────────────────────────────┘
         │
┌────────▼─────────────────────────────────────────────────────┐
│                     Data Processing Layer                    │
│  Edge Aggregation / Filtering / Time-Series Write / OTA Mgmt │
└──────────────────────────────────────────────────────────────┘

MQTT 5.0 Broker

Built-in high-performance MQTT broker, supporting million-level connections, shared subscriptions, and session persistence:

use eneros_iot::mqtt::{MqttBroker, BrokerConfig, Persistence, QoS};

let broker = MqttBroker::new(BrokerConfig {
    bind: "0.0.0.0:1883".parse()?,
    tls_bind: Some("0.0.0.0:8883".parse()?),
    websocket_bind: Some("0.0.0.0:9001".parse()?),
    max_connections: 1_000_000,
    max_packet_size: 1 << 20,                  // 1MB
    persistence: Persistence::Sled,
    session_expiry: Duration::from_secs(86400),
    message_queue_size: 1024,
    keep_alive: Duration::from_secs(60),
});

broker.start().await?;

// Subscribe to topic
broker.subscribe("grid/+/voltage", QoS::AtLeastOnce, |topic, payload| {
    let bus_id = topic.split('/').nth(1).unwrap();
    let v: f64 = serde_json::from_slice(payload)?;
    timeseries.write(bus_id, "voltage", ts, v).await?;
    Ok(())
}).await?;

// Publish message
broker.publish("device/rtu_001/command", QoS::ExactlyOnce, &cmd_payload).await?;

Topic Conventions

grid/{bus_id}/voltage       — Voltage
grid/{bus_id}/current       — Current
grid/{bus_id}/power         — Power
grid/{bus_id}/frequency     — Frequency
device/{device_id}/status   — Device status
device/{device_id}/event    — Device event
device/{device_id}/command  — Device command
grid/{feeder_id}/load       — Feeder load

Shared Subscription

Multiple consumers share a subscription group to achieve load balancing:

// Subscription group $share/consumer_group/topic
broker.subscribe("$share/processors/grid/+/voltage", QoS::AtLeastOnce, |topic, payload| {
    process_voltage(topic, payload).await
}).await?;

MQTT Configuration Parameters

CoAP Server

Supports CoAP protocol for constrained devices, suitable for low-power, low-bandwidth scenarios:

use eneros_iot::coap::{CoapServer, Resource, Response, Method};

let server = CoapServer::new("0.0.0.0:5683")?
    .dtls("0.0.0.0:5684", dtls_config);

server.resource(Resource::get("voltage", |req| async move {
    let bus_id = req.uri_path().last().unwrap_or("default");
    let v = read_voltage(bus_id).await?;
    Response::content(format!("{:.4}", v))
}));

server.resource(Resource::post("command", |req| async move {
    let cmd: Command = serde_json::from_slice(req.payload())?;
    handle_command(cmd).await?;
    Response::changed()
}));

server.resource(Resource::observe("load", |sub| async move {
    // Push real-time load
    while let Some(load) = load_stream.recv().await {
        sub.notify(format!("{:.2}", load)).await?;
    }
}));

server.start().await?;

CoAP Method Support

LwM2M Device Management

Supports OMA LwM2M device management protocol, covering registration, object access, firmware update, and firmware upgrade throughout the lifecycle:

use eneros_iot::lwm2m::{Lwm2mServer, ObjectId, FirmwareUpdate};

let server = Lwm2mServer::new()
    .endpoint("coaps://eneros:5684")
    .bootstrap(true)
    .registration_lifetime(Duration::from_secs(86400));

// Device registration callback
server.on_register(|dev| async move {
    println!("Device registered: {} ({})", dev.endpoint, dev.objects);
    register_in_topology(&dev).await?;
    Ok(())
});

server.on_deregister(|dev| async move {
    println!("Device deregistered: {}", dev.endpoint);
    deregister_from_topology(&dev).await?;
    Ok(())
});

// Read device object
let battery = server.read(device_id, ObjectId::Battery).await?;
println!("Battery level: {}%", battery.level);
println!("Voltage: {}mV", battery.voltage);

let location = server.read(device_id, ObjectId::Location).await?;
println!("Location: ({}, {})", location.latitude, location.longitude);

// Trigger firmware update
server.write(device_id, ObjectId::FirmwareUpdate,
    FirmwareUpdate::download("https://ota.eneros/fw/v2.bin")
        .signature(Signature::ecdsa(pubkey))
        .apply_mode(ApplyMode::AfterDownload)).await?;

// Listen for firmware update progress
server.on_firmware_update_progress(|dev, progress| async move {
    println!("Device {} firmware update: {:.0}%", dev.endpoint, progress * 100.0);
});

LwM2M Objects

Sensor Network Aggregation

Supports edge aggregation of massive sensor data, reducing bandwidth and central processing pressure:

use eneros_iot::sensor::{SensorNetwork, Aggregation, Filter};
use std::time::Duration;

let network = SensorNetwork::new()
    .sensor("pmu_1", SensorType::PMU, Duration::from_micros(100))
    .sensor("pmu_2", SensorType::PMU, Duration::from_micros(100))
    .sensor("feeder_1", SensorType::Feeder, Duration::from_millis(100))
    .sensor("transformer_1", SensorType::Transformer, Duration::from_secs(1));

// Edge aggregation
network.aggregate(Aggregation::windowed(
    Duration::from_millis(100),
    |samples| {
        // Median filter (remove outliers)
        let mut values: Vec<_> = samples.iter().map(|s| s.value).collect();
        values.sort_by(|a, b| a.partial_cmp(b).unwrap());
        values.get(values.len() / 2).copied().unwrap_or(0.0)
    },
)).await?;

// Multi-stage filtering
network.filter(Filter::chain(vec![
    Filter::median(5),           // Median filter
    Filter::lowpass(0.1),        // Low-pass filter
    Filter::outlier(3.0),        // Outlier removal (3σ)
]));

// Automatically write to time-series database
network.forward_to_timeseries(&timeseries).await?;

// Anomaly detection
network.on_anomaly(|sensor, value, baseline| async move {
    notify_ops(format!("{} anomaly: {} vs baseline {}", sensor, value, baseline)).await
});

Aggregation Strategies

Edge Device OTA

Batch firmware updates, supporting differential upgrades, canary releases, and automatic rollback:

use eneros_iot::ota::{OtaManager, OtaCampaign, Strategy, Firmware, Signature, OnFailure};

let ota = OtaManager::new()
    .concurrency(50)                  // Concurrent upgrade count
    .timeout(Duration::from_secs(300));

let campaign = OtaCampaign::new("firmware-v2.3")
    .target("rtu.*")                  // Match all RTUs
    .strategy(Strategy::canary(10)    // 10% canary
        .observe(Duration::from_secs(300))
        .promote_on(HealthCheck::all_healthy()))
    .firmware(Firmware::from_file("firmware.bin")
        .size(2 * 1024 * 1024))       // 2MB
    .signature(Signature::ecdsa(private_key))
    .delta(Delta::from_previous("v2.2"))  // Differential upgrade
    .on_failure(OnFailure::rollback()
        .threshold(0.05))             // Failure rate > 5% rollback all
    .on_progress(|device, progress| async move {
        println!("{}: {:.0}%", device, progress * 100.0);
    });

let result = ota.run(campaign).await?;
println!("Success: {}/{}", result.success, result.total);
println!("Failures: {}", result.failures);
println!("Total time: {:.1}s", result.elapsed.as_secs_f64());

if result.success < result.total * 90 / 100 {
    println!("Success rate too low, triggering full rollback");
    ota.rollback_all().await?;
}

OTA Strategy Comparison

Device Model

use eneros_iot::device::{Device, DeviceType, Protocol, GeoPoint};

let device = Device::new("rtu_001")
    .ty(DeviceType::RTU)
    .protocol(Protocol::IEC61850)
    .location(GeoPoint::new(31.23, 121.47))
    .tags(vec!["feeder_5", "substation_a"])
    .metadata(json!({
        "vendor": "Siemens",
        "model": "SICAM",
        "install_date": "2024-03-15",
    }));

device.register(&iot).await?;

// Device status changes
device.on_status_change(|status| async move {
    match status {
        Status::Online => println!("Device {} online", device.id),
        Status::Offline => notify_ops("Device offline").await,
        Status::Faulty => create_work_order(&device).await,
    }
});

Supported Device Types

Security

All IoT communications support encryption, with differentiated configuration by device type:

use eneros_iot::mqtt::{MqttBroker, TlsConfig, Auth};

let broker = MqttBroker::new(config)
    .tls(TlsConfig::mutual()
        .client_cert_required(true)
        .ca("/etc/eneros/iot-ca.crt")
        .cipher_suites(CipherSuite::tls_1_3()))
    .auth(Auth::token(jwt_validator)
        .claim_check(|claims| async {
            // Verify device certificate and token binding
            claims.device_id == claims.certificate_cn
        }));

// Device-level rate limiting
broker.rate_limit(RateLimit::per_device(
    100,                                  // 100 msg/s
    Duration::from_secs(1),
    Action::Throttle(Duration::from_secs(60)),
));

Performance Metrics

Relationship with Other Capabilities

• Time-Series Native Operations: IoT data written to time-series database, see Time-Series Native Operations
• Zero Trust and Security Enhancement: IoT communication encryption, see Zero Trust and Security Enhancement
• Internationalization and Compliance: DL/T 698.45 meter protocol, see Internationalization and Compliance
• Grid Topology First-Class Citizen: Devices registered to topology, see Grid Topology First-Class Citizen
• Agent Intelligence Advanced: Anomaly detection based on IoT data

Related Documentation

• Time-Series Native Operations — IoT data written to time-series database
• Zero Trust and Security Enhancement — IoT communication encryption
• Internationalization and Compliance — DL/T 698.45 meter protocol
• Grid Topology First-Class Citizen — Devices registered to topology
• Agent Intelligence Advanced — Anomaly detection