SCADA Data Acquisition Integration

This tutorial demonstrates how to connect an existing SCADA system to EnerOS via the IEC 60870-5-104 protocol, enabling bidirectional flow of telemetry (TM), telesignaling (TS), integrated totals (PI), and telecontrol (TC). EnerOS provides native IEC 104 support through the eneros-scada crate; all data automatically lands in the kernel time-series engine for use by Agents, analytics, and visualization.

IEC 60870-5-104 Protocol Overview

IEC 60870-5-104 (abbreviated as IEC 104) is the de facto standard for power system telecontrol communication, transmitted over TCP/IP (default port 2404). Its core concepts:

Common Type Identifications (TI):

Prerequisites

• A SCADA master station or simulator that supports IEC 104 (recommended: lib60870-C or OpenMUC)
• Network reachability, with mTLS certificate issuance complete (refer to Zero Trust mTLS)
• eneros-gateway and eneros-scada services deployed

This tutorial involves the following crates:

Step 1: Configure the Protocol Gateway

Add an IEC 104 channel in eneros-gateway. The configuration file is located at /etc/eneros/gateway.yaml:

gateways:
  - id: scada-104-master
    type: iec104
    role: master                    # EnerOS acts as the master station (client)
    endpoints:
      - 10.0.0.20:2404              # RTU/IED address
      - 10.0.0.21:2404              # Backup RTU
    tls:
      mode: mutual                  # Enforce mTLS
      ca: /etc/eneros/ca.crt
      cert: /etc/eneros/certs/gateway.crt
      key: /etc/eneros/certs/gateway.key
      verify_peer: true
    common_address: 1               # Common address (RTU identifier)
    # IEC 104 protocol parameters (T1/T3 timeouts, K/W window)
    t1: 15                          # Acknowledgment timeout (seconds)
    t2: 10                          # ACK timeout (seconds)
    t3: 25                          # Connection probe interval (seconds)
    k: 12                           # Send window size
    w: 8                            # ACK window size
    # Auto general interrogation on startup
    auto_interrogation: true
    interrogation_interval_s: 30    # General interrogation cycle
    # Data quality filter
    quality_filter:
      ignore_invalid: true          # Ignore points with quality.invalid = 1
      ignore_substituted: true      # Ignore points with quality.substituted = 1

Protocol Parameter Description

:::tip The relationship between k and w should satisfy w < k, otherwise ACKs will be sent too frequently, wasting bandwidth. :::

Step 2: Define Point Mapping

Map SCADA Information Object Addresses (IOA) to EnerOS time-series points. EnerOS provides IoaMappingTable to manage the mapping:

use eneros_scada::iec104::{IoaMapping, IoaMappingTable};

/// Build the SCADA point mapping table for the IEEE 14-bus system
fn build_ioa_mapping() -> IoaMappingTable {
    let mut table = IoaMappingTable::new();

    // ============ Telemetry points (TM) ============
    // Bus voltage magnitude (TI=13, short float)
    table.add(IoaMapping::new(1001, 1, "bus1_voltage_pu", 1.0, 0.0));
    table.add(IoaMapping::new(1002, 1, "bus2_voltage_pu", 1.0, 0.0));
    table.add(IoaMapping::new(1003, 1, "bus3_voltage_pu", 1.0, 0.0));
    table.add(IoaMapping::new(1004, 1, "bus4_voltage_pu", 1.0, 0.0));
    table.add(IoaMapping::new(1005, 1, "bus5_voltage_pu", 1.0, 0.0));
    table.add(IoaMapping::new(1006, 1, "bus6_voltage_pu", 1.0, 0.0));
    table.add(IoaMapping::new(1007, 1, "bus7_voltage_pu", 1.0, 0.0));
    table.add(IoaMapping::new(1008, 1, "bus8_voltage_pu", 1.0, 0.0));
    table.add(IoaMapping::new(1009, 1, "bus9_voltage_pu", 1.0, 0.0));
    table.add(IoaMapping::new(1010, 1, "bus10_voltage_pu", 1.0, 0.0));
    table.add(IoaMapping::new(1011, 1, "bus11_voltage_pu", 1.0, 0.0));
    table.add(IoaMapping::new(1012, 1, "bus12_voltage_pu", 1.0, 0.0));
    table.add(IoaMapping::new(1013, 1, "bus13_voltage_pu", 1.0, 0.0));
    table.add(IoaMapping::new(1014, 1, "bus14_voltage_pu", 1.0, 0.0));

    // Generator active output (MW)
    table.add(IoaMapping::new(2001, 1, "gen1_p_mw", 1.0, 0.0));
    table.add(IoaMapping::new(2002, 1, "gen2_p_mw", 1.0, 0.0));
    table.add(IoaMapping::new(2003, 1, "gen3_p_mw", 1.0, 0.0));
    table.add(IoaMapping::new(2006, 1, "gen6_p_mw", 1.0, 0.0));
    table.add(IoaMapping::new(2008, 1, "gen8_p_mw", 1.0, 0.0));

    // Generator reactive output (MVar)
    table.add(IoaMapping::new(2101, 1, "gen1_q_mvar", 1.0, 0.0));
    table.add(IoaMapping::new(2102, 1, "gen2_q_mvar", 1.0, 0.0));
    table.add(IoaMapping::new(2103, 1, "gen3_q_mvar", 1.0, 0.0));
    table.add(IoaMapping::new(2106, 1, "gen6_q_mvar", 1.0, 0.0));
    table.add(IoaMapping::new(2108, 1, "gen8_q_mvar", 1.0, 0.0));

    // Line active power flow (MW), only key branches shown
    table.add(IoaMapping::new(3001, 1, "line_1_2_p_mw", 1.0, 0.0));
    table.add(IoaMapping::new(3002, 1, "line_1_5_p_mw", 1.0, 0.0));
    table.add(IoaMapping::new(3003, 1, "line_4_5_p_mw", 1.0, 0.0));
    table.add(IoaMapping::new(3004, 1, "line_6_13_p_mw", 1.0, 0.0));
    table.add(IoaMapping::new(3005, 1, "line_9_14_p_mw", 1.0, 0.0));

    // ============ Telesignaling points (TS) ============
    // Breaker status (TI=3, double-point: 0=intermediate, 1=open, 2=closed, 3=indeterminate)
    table.add(IoaMapping::new(4001, 1, "breaker_1_2_state", 1.0, 0.0));
    table.add(IoaMapping::new(4002, 1, "breaker_1_5_state", 1.0, 0.0));
    table.add(IoaMapping::new(4003, 1, "breaker_2_3_state", 1.0, 0.0));
    table.add(IoaMapping::new(4004, 1, "breaker_2_4_state", 1.0, 0.0));
    table.add(IoaMapping::new(4005, 1, "breaker_2_5_state", 1.0, 0.0));
    table.add(IoaMapping::new(4006, 1, "breaker_3_4_state", 1.0, 0.0));
    table.add(IoaMapping::new(4007, 1, "breaker_4_5_state", 1.0, 0.0));

    // Disconnector status (TI=1, single-point)
    table.add(IoaMapping::new(5001, 1, "disconnector_4_7_state", 1.0, 0.0));
    table.add(IoaMapping::new(5002, 1, "disconnector_4_9_state", 1.0, 0.0));
    table.add(IoaMapping::new(5003, 1, "disconnector_5_6_state", 1.0, 0.0));

    // ============ Integrated total points (PI) ============
    // Energy meter (TI=15, integrated value)
    table.add(IoaMapping::new(6001, 1, "energy_import_kwh", 1.0, 0.0));
    table.add(IoaMapping::new(6002, 1, "energy_export_kwh", 1.0, 0.0));

    // ============ Control points (TC) ============
    // Breaker control command (TI=45, single-point command)
    table.add(IoaMapping::new(7001, 1, "breaker_1_2_cmd", 1.0, 0.0));
    table.add(IoaMapping::new(7002, 1, "breaker_1_5_cmd", 1.0, 0.0));
    table.add(IoaMapping::new(7003, 1, "breaker_2_3_cmd", 1.0, 0.0));

    // Generator output setpoint command (TI=50, set-point command short float)
    table.add(IoaMapping::new(8001, 1, "gen1_setpoint_mw", 1.0, 0.0));
    table.add(IoaMapping::new(8002, 1, "gen2_setpoint_mw", 1.0, 0.0));
    table.add(IoaMapping::new(8003, 1, "gen3_setpoint_mw", 1.0, 0.0));

    table
}

let mapping = build_ioa_mapping();
println!("Point mapping table: {} points", mapping.len());

Example output:

Point mapping table: 38 points

IoaMapping field meanings:

Step 3: Create the IEC 104 Data Source

Iec104DataSource encapsulates the IEC 104 client and point mapping, implementing the DataSource trait:

use eneros_scada::iec104::{Iec104DataSource, Iec104Config};
use eneros_scada::collector::ScadaCollector;
use std::time::Duration;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // 1. Configure the IEC 104 client
    let config = Iec104Config {
        server_addr: "10.0.0.20:2404".parse()?,
        common_address: 1,
        t1_seconds: 15,
        t2_seconds: 10,
        t3_seconds: 25,
        k: 12,
        w: 8,
        auto_interrogation: true,
        interrogation_interval: Duration::from_secs(30),
    };

    // 2. Create the data source (with point mapping)
    let mapping = build_ioa_mapping();
    let mut data_source = Iec104DataSource::new(config, mapping);

    // 3. Connect to the RTU
    data_source.connect().await?;
    println!("✓ Connected to RTU 10.0.0.20:2404");

    // 4. Start general interrogation
    data_source.start_interrogation().await?;
    println!("✓ General interrogation started");

    // 5. Create the collector and start data acquisition
    let mut collector = ScadaCollector::new(data_source);
    collector.start().await?;

    // 6. Continuously receive real-time data
    let mut rx = collector.subscribe();
    while let Some(reading) = rx.recv().await {
        println!("[{}] {} = {:.4}",
            reading.timestamp.format("%H:%M:%S%.3f"),
            reading.point_name,
            reading.value,
        );
    }

    Ok(())
}

Example output:

✓ Connected to RTU 10.0.0.20:2404
✓ General interrogation started
[14:23:01.234] bus1_voltage_pu = 1.0602
[14:23:01.235] bus2_voltage_pu = 1.0451
[14:23:01.236] gen1_p_mw = 132.45
[14:23:01.237] breaker_1_2_state = 2.0

Step 4: Persist Data into the Time-Series Engine

Write the collected real-time data into eneros-timeseries for Agents and analytics modules to query:

use eneros_scada::collector::ScadaReading;
use eneros_timeseries::{TimeseriesEngine, WriteOptions};

async fn persist_readings(
    ts: &mut TimeseriesEngine,
    readings: &[ScadaReading],
) -> Result<(), Box<dyn std::error::Error>> {
    let opts = WriteOptions::default()
        .batch_size(1000)               // Batch write size
        .flush_interval_ms(100);        // Auto-flush interval

    for reading in readings {
        ts.write_point(
            &reading.point_name,
            reading.value,
            &opts,
        ).await?;

        // Also write the quality flag
        ts.write_tag(
            &reading.point_name,
            "quality",
            &reading.quality.to_string(),
        ).await?;
    }

    Ok(())
}

The time-series engine supports the following query patterns:

// 1. Latest value query for a single point
let latest = ts.query_latest("bus1_voltage_pu").await?;

// 2. Time-range query
let range = ts.query_range(
    "bus1_voltage_pu",
    "2026-07-06T00:00:00Z".parse()?,
    "2026-07-06T23:59:59Z".parse()?,
).await?;

// 3. Aggregation query (5-minute mean)
let agg = ts.query_aggregate(
    "bus1_voltage_pu",
    "2026-07-06T00:00:00Z".parse()?,
    "2026-07-06T23:59:59Z".parse()?,
    AggregateFunction::Mean,
    Duration::from_secs(300),
).await?;

Step 5: Issue Telecontrol Commands

Commands must pass the constraint engine validation before being issued, to avoid misoperation. EnerOS enforces the SBO (Select-Before-Operate) flow:

use eneros_scada::iec104::{build_single_command, build_setpoint_short_float};
use eneros_scada::iec104::{CauseOfTransmission, TypeId};
use eneros_gateway::command::{Command, CommandType, CommandPriority};
use eneros_constraint::engine::ConstraintEngine;

/// Securely issue a breaker control command (SBO flow)
async fn send_breaker_command(
    data_source: &mut Iec104DataSource,
    constraints: &ConstraintEngine,
    ioa: u32,
    close: bool,
) -> Result<(), Box<dyn std::error::Error>> {
    // 1. Build the EnerOS internal command
    let cmd = Command::new(
        CommandType::SwitchToggle,
        ioa as u64,
        CommandPriority::High,
        "scada-operator",
    )
    .with_parameter("closed", if close { 1.0 } else { 0.0 });

    // 2. Constraint engine validation
    let verdict = constraints.validate(&cmd).await?;
    if !verdict.passed {
        eprintln!("Command rejected by the constraint engine:");
        for v in &verdict.violations {
            eprintln!("  - {}", v);
        }
        return Err("Command validation failed".into());
    }
    println!("✓ Constraint validation passed");

    // 3. SBO: Select phase (pre-select)
    let select_asdu = build_single_command(
        ioa,                              // IOA
        if close { 1 } else { 0 },        // SCS: 1=close, 0=open
        CauseOfTransmission::Activation,  // Activation
        /* qu = */ 0,                     // Default
        /* select = */ true,              // SBO mode
    )?;
    data_source.send_command(select_asdu).await?;
    println!("✓ Select sent (ioa={}, close={})", ioa, close);

    // 4. Wait for Select confirmation (should return COT=ActivationConfirmation)
    let ack = data_source.wait_command_ack(Duration::from_secs(2)).await?;
    if !ack.is_positive() {
        return Err(format!("Select rejected: {:?}", ack).into());
    }
    println!("✓ Select confirmed");

    // 5. SBO: Operate phase (execute)
    let operate_asdu = build_single_command(
        ioa,
        if close { 1 } else { 0 },
        CauseOfTransmission::Activation,
        0,
        /* select = */ false,             // Execute mode
    )?;
    data_source.send_command(operate_asdu).await?;
    println!("✓ Operate sent");

    // 6. Wait for execution confirmation
    let exec_ack = data_source.wait_command_ack(Duration::from_secs(5)).await?;
    if exec_ack.is_positive() {
        println!("✓ Command executed successfully");
    } else {
        return Err(format!("Command execution failed: {:?}", exec_ack).into());
    }

    // 7. Write to the audit chain
    audit_log::record(
        "scada-operator",
        &format!("breaker_{}_{}", ioa, if close { "close" } else { "open" }),
        "success",
    ).await?;

    Ok(())
}

// Example: close breaker 1-2
send_breaker_command(&mut data_source, &constraints, 7001, true).await?;

Set-Point Command (Adjust Generator Output)

/// Issue a generator output set-point command
async fn send_setpoint_command(
    data_source: &mut Iec104DataSource,
    constraints: &ConstraintEngine,
    ioa: u32,
    value_mw: f64,
) -> Result<(), Box<dyn std::error::Error>> {
    // Constraint validation
    let cmd = Command::new(
        CommandType::GeneratorSetpoint,
        ioa as u64,
        CommandPriority::Normal,
        "ed-agent",
    )
    .with_parameter("p_mw", value_mw);
    let verdict = constraints.validate(&cmd).await?;
    if !verdict.passed {
        return Err(format!("Set-point command rejected: {:?}", verdict.violations).into());
    }

    // Build the set-point ASDU (TI=50, short float)
    let asdu = build_setpoint_short_float(
        ioa,
        value_mw as f32,
        CauseOfTransmission::Activation,
    )?;
    data_source.send_command(asdu).await?;

    let ack = data_source.wait_command_ack(Duration::from_secs(3)).await?;
    if ack.is_positive() {
        println!("✓ Set-point successful: ioa={} → {} MW", ioa, value_mw);
    } else {
        return Err(format!("Set-point failed: {:?}", ack).into());
    }
    Ok(())
}

// Example: set generator G1 output to 150 MW
send_setpoint_command(&mut data_source, &constraints, 8001, 150.0).await?;

Step 6: Event Handling and Alarming

Listen for SCADA state changes (breaker transitions, protection actions, data quality anomalies):

use eneros_scada::collector::ScadaReading;
use eneros_eventbus::{Event, EventType, EventPayload};

/// Handle SCADA state-change events
async fn handle_scada_events(
    collector: &ScadaCollector,
    event_bus: &EventBus,
) -> Result<(), Box<dyn std::error::Error>> {
    let mut rx = collector.subscribe();

    while let Some(reading) = rx.recv().await {
        // 1. Detect breaker transitions
        if reading.point_name.ends_with("_state") {
            let event = Event::new(
                EventType::TopologyChange,
                "scada-104-master",
                EventPayload::Json(serde_json::json!({
                    "point": reading.point_name,
                    "value": reading.value,
                    "quality": reading.quality.to_string(),
                    "timestamp": reading.timestamp.to_rfc3339(),
                })),
            );
            event_bus.publish(event).await?;
        }

        // 2. Detect data quality anomalies
        if reading.quality.invalid() {
            let event = Event::new(
                EventType::DataQualityAlarm,
                "scada-104-master",
                EventPayload::Json(serde_json::json!({
                    "point": reading.point_name,
                    "quality": reading.quality.to_string(),
                    "reason": "invalid_flag_set",
                })),
            );
            event_bus.publish(event).await?;
        }

        // 3. Detect voltage violations (threshold-based)
        if reading.point_name.contains("voltage_pu") {
            let v = reading.value;
            if v < 0.93 || v > 1.07 {
                let event = Event::new(
                    EventType::VoltageViolation,
                    "scada-104-master",
                    EventPayload::Json(serde_json::json!({
                        "point": reading.point_name,
                        "voltage_pu": v,
                        "limit_low": 0.93,
                        "limit_high": 1.07,
                    })),
                );
                event_bus.publish(event).await?;
            }
        }
    }

    Ok(())
}

Dual-Node Redundant Acquisition

Production environments typically deploy dual-node redundant acquisition (primary-backup mode) to avoid single points of failure:

use eneros_scada::dual_scan::{DualScanGroupBuilder, DualScanOptions};

// Build a dual-node redundant acquisition group
let dual_group = DualScanGroupBuilder::new()
    .primary("10.0.0.20:2404", "scada-primary")   // Primary
    .secondary("10.0.0.21:2404", "scada-secondary") // Secondary
    .options(DualScanOptions {
        switch_timeout_ms: 5000,           // Primary-backup switchover timeout
        consistency_check: true,           // Enable data consistency validation
        max_diff_percent: 1.0,             // Maximum allowed deviation (%)
        snapshot_interval_s: 5,            // Snapshot alignment cycle
    })
    .build()
    .await?;

// Start dual-node acquisition
let handles = dual_group.start(mapping).await?;

// Primary-backup status can be viewed via handles
println!("Primary channel status: {:?}", handles.primary_status());
println!("Secondary channel status: {:?}", handles.secondary_status());

Step 7: Complete End-to-End Example

Integrate the above steps into a complete runnable program:

use eneros_scada::iec104::{Iec104Config, Iec104DataSource, IoaMappingTable};
use eneros_scada::collector::ScadaCollector;
use eneros_timeseries::TimeseriesEngine;
use std::time::Duration;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // 1. Configure and connect
    let config = Iec104Config {
        server_addr: "10.0.0.20:2404".parse()?,
        common_address: 1, t1_seconds: 15, t2_seconds: 10,
        t3_seconds: 25, k: 12, w: 8,
        auto_interrogation: true,
        interrogation_interval: Duration::from_secs(30),
    };

    let mapping = build_ioa_mapping();
    let mut ds = Iec104DataSource::new(config, mapping);
    ds.connect().await?;
    ds.start_interrogation().await?;

    // 2. Start the collector
    let mut collector = ScadaCollector::new(ds);
    collector.start().await?;

    // 3. Open the time-series engine
    let mut ts = TimeseriesEngine::open("/var/lib/eneros/ts")?;

    // 4. Continuously acquire for 60 seconds
    let mut rx = collector.subscribe();
    let start = std::time::Instant::now();
    let mut count = 0u64;

    while start.elapsed() < Duration::from_secs(60) {
        if let Some(reading) = rx.recv().await {
            ts.write_point(&reading.point_name, reading.value).await?;
            count += 1;
        }
    }

    println!("Acquisition complete: {} points written", count);
    println!("Acquisition rate: {:.0} points/second", count as f64 / 60.0);

    Ok(())
}

Example output:

Acquisition complete: 4560 points written
Acquisition rate: 76 points/second

Validation

Validate the correctness of the SCADA integration against the following metrics:

Debugging Tips

# View IEC 104 connection status
eneros-cli scada status --gateway scada-104-master

# Capture raw packets
tcpdump -i eth0 -w iec104.pcap port 2404

# Analyze packets with wireshark
wireshark iec104.pcap

# View time-series data writes
eneros-cli ts query bus1_voltage_pu --latest 10

# Force-trigger general interrogation
eneros-cli scada interrogate --gateway scada-104-master --ca 1

# Simulate breaker telecontrol (requires --confirm for double confirmation)
eneros-cli scada send-command breaker_1_2_cmd --value 1 --sbo

Common Issue Troubleshooting

Next Steps

• IoT Ubiquitous Access — overview of the multi-protocol gateway
• Safety Guard — detailed explanation of command security validation
• Dispatch Agent Development — automatic dispatch based on SCADA data
• Time-Series Storage Engine — time-series data query and aggregation