SCADA 数据采集集成

本教程演示如何将既有 SCADA 系统通过 IEC 60870-5-104 协议接入 EnerOS，实现遥测（TM）、遥信（TS）、遥脉（PI）与远程控制（TC）的双向流转。EnerOS 通过 eneros-scada crate 提供原生 IEC 104 支持，所有数据自动落入内核时序引擎，供 Agent、分析与可视化使用。

IEC 60870-5-104 协议概览

IEC 60870-5-104（简称 IEC 104）是电力系统远动通信的事实标准，基于 TCP/IP 传输（默认端口 2404），其核心概念：

概念	说明
ASDU	应用服务数据单元，协议报文的基本单位
IOA	信息对象地址（Information Object Address），唯一标识一个测点
Common Address	公共地址，标识一个 RTU/IED
TI	类型标识（Type Identification），定义 ASDU 的数据类型
COT	传送原因（Cause of Transmission），如周期、突发、命令响应

常见类型标识（TI）：

TI	类型	说明
1	Single-point	单点遥信（开关状态）
3	Double-point	双点遥信（断路器：分/合/中间）
9	Normalized measured value	归一化遥测（-1 ~ +1）
13	Short measured value	短浮点遥测（IEEE 754）
15	Integrated totals	电度量（积分值）
45	Single command	单点遥控（合/分）
50	Set-point command short	设点命令（浮点）

准备工作

一台支持 IEC 104 的 SCADA 主站或模拟器（推荐 lib60870-C 或 OpenMUC）
网络可达，且完成 mTLS 证书签发（参考零信任 mTLS）
已部署 eneros-gateway 与 eneros-scada 服务

本教程涉及以下 crate：

Crate	作用
`eneros-scada`	IEC 104 客户端、数据源、点位映射
`eneros-device`	底层协议适配（ASDU 解析、APCI 帧）
`eneros-gateway`	协议网关与安全命令下发
`eneros-timeseries`	时序数据持久化
`eneros-constraint`	命令安全校验
`eneros-audit`	WORM 审计链

步骤 1：配置协议网关

在 eneros-gateway 中新增 IEC 104 通道。配置文件位于 /etc/eneros/gateway.yaml：

gateways:
  - id: scada-104-master
    type: iec104
    role: master                    # EnerOS 作为主站（客户端）
    endpoints:
      - 10.0.0.20:2404              # RTU/IED 地址
      - 10.0.0.21:2404              # 备用 RTU
    tls:
      mode: mutual                  # 强制 mTLS
      ca: /etc/eneros/ca.crt
      cert: /etc/eneros/certs/gateway.crt
      key: /etc/eneros/certs/gateway.key
      verify_peer: true
    common_address: 1               # 公共地址（RTU 标识）
    # IEC 104 协议参数（T1/T3 超时、K/W 窗口）
    t1: 15                          # 确认超时（秒）
    t2: 10                          # ACK 超时（秒）
    t3: 25                          # 连接探测间隔（秒）
    k: 12                           # 发送窗口大小
    w: 8                            # ACK 窗口大小
    # 启动时自动总召唤
    auto_interrogation: true
    interrogation_interval_s: 30    # 总召唤周期
    # 数据质量过滤
    quality_filter:
      ignore_invalid: true          # 忽略 quality.invalid = 1 的测点
      ignore_substituted: true      # 忽略 quality.substituted = 1 的测点

协议参数说明

参数	默认值	说明
`t1`	15s	发送报文后等待 ACK 的超时，超时则断连重连
`t2`	10s	接收报文后发送 ACK 的延迟，避免 ACK 频繁发送
`t3`	25s	长时间无数据时发送 TESTFR 探测的间隔
`k`	12	未确认 I 帧的最大数量（发送窗口）
`w`	8	接收多少帧后必须发送 ACK（接收窗口）

:::tip k 与 w 的关系应满足 w < k，否则会导致 ACK 频繁发送，浪费带宽。 :::

步骤 2：定义点位映射

将 SCADA 信息体地址（IOA）映射到 EnerOS 时序点位。EnerOS 提供 IoaMappingTable 管理映射关系：

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

/// 构建 IEEE 14-bus 系统的 SCADA 点位映射表
fn build_ioa_mapping() -> IoaMappingTable {
    let mut table = IoaMappingTable::new();

    // ============ 遥测点（TM）============
    // 母线电压幅值（TI=13，短浮点）
    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));

    // 机组有功出力（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));

    // 机组无功出力（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));

    // 线路有功潮流（MW），仅展示关键支路
    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));

    // ============ 遥信点（TS）============
    // 断路器状态（TI=3，双点遥信：0=中间, 1=分, 2=合, 3=不确定）
    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));

    // 隔离开关状态（TI=1，单点遥信）
    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));

    // ============ 遥脉点（PI）============
    // 电度量（TI=15，积分值）
    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));

    // ============ 控制点（TC）============
    // 断路器控制命令（TI=45，单点命令）
    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));

    // 机组出力设点命令（TI=50，设点命令短浮点）
    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!("点位映射表: {} 个测点", mapping.len());

输出示例：

点位映射表: 38 个测点

IoaMapping 各字段含义：

字段	类型	说明
`ioa`	u32	信息对象地址（RTU 侧的测点地址）
`common_address`	u16	公共地址（RTU 标识）
`point_name`	String	EnerOS 时序引擎中的点位名称
`scale`	f64	缩放系数（原始值 × scale + offset = 物理值）
`offset`	f64	零点偏移

步骤 3：创建 IEC 104 数据源

Iec104DataSource 封装了 IEC 104 客户端与点位映射，实现 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. 配置 IEC 104 客户端
    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. 创建数据源（含点位映射）
    let mapping = build_ioa_mapping();
    let mut data_source = Iec104DataSource::new(config, mapping);

    // 3. 连接 RTU
    data_source.connect().await?;
    println!("✓ 已连接 RTU 10.0.0.20:2404");

    // 4. 启动总召唤
    data_source.start_interrogation().await?;
    println!("✓ 总召唤已启动");

    // 5. 创建采集器，开始数据采集
    let mut collector = ScadaCollector::new(data_source);
    collector.start().await?;

    // 6. 持续接收实时数据
    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(())
}

输出示例：

✓ 已连接 RTU 10.0.0.20:2404
✓ 总召唤已启动
[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

步骤 4：数据落入时序引擎

将采集到的实时数据写入 eneros-timeseries，供 Agent 与分析模块查询：

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)               // 批量写入大小
        .flush_interval_ms(100);        // 自动刷盘间隔

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

        // 同时写入质量标记
        ts.write_tag(
            &reading.point_name,
            "quality",
            &reading.quality.to_string(),
        ).await?;
    }

    Ok(())
}

时序引擎支持以下查询模式：

// 1. 单点最新值查询
let latest = ts.query_latest("bus1_voltage_pu").await?;

// 2. 时间范围查询
let range = ts.query_range(
    "bus1_voltage_pu",
    "2026-07-06T00:00:00Z".parse()?,
    "2026-07-06T23:59:59Z".parse()?,
).await?;

// 3. 聚合查询（5 分钟均值）
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?;

步骤 5：下发遥控命令

命令下发前必须通过约束引擎校验，避免误操作。EnerOS 强制执行 SBO（Select-Before-Operate）流程：

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;

/// 安全下发断路器控制命令（SBO 流程）
async fn send_breaker_command(
    data_source: &mut Iec104DataSource,
    constraints: &ConstraintEngine,
    ioa: u32,
    close: bool,
) -> Result<(), Box<dyn std::error::Error>> {
    // 1. 构造 EnerOS 内部命令
    let cmd = Command::new(
        CommandType::SwitchToggle,
        ioa as u64,
        CommandPriority::High,
        "scada-operator",
    )
    .with_parameter("closed", if close { 1.0 } else { 0.0 });

    // 2. 约束引擎校验
    let verdict = constraints.validate(&cmd).await?;
    if !verdict.passed {
        eprintln!("命令被约束引擎拒绝:");
        for v in &verdict.violations {
            eprintln!("  - {}", v);
        }
        return Err("命令校验失败".into());
    }
    println!("✓ 约束校验通过");

    // 3. SBO: Select 阶段（预选）
    let select_asdu = build_single_command(
        ioa,                              // IOA
        if close { 1 } else { 0 },        // SCS: 1=合, 0=分
        CauseOfTransmission::Activation,  // 激活
        /* qu = */ 0,                     // 默认
        /* select = */ true,              // SBO 模式
    )?;
    data_source.send_command(select_asdu).await?;
    println!("✓ Select 已发送 (ioa={}, close={})", ioa, close);

    // 4. 等待 Select 确认（应返回 COT=ActivationConfirmation）
    let ack = data_source.wait_command_ack(Duration::from_secs(2)).await?;
    if !ack.is_positive() {
        return Err(format!("Select 被拒绝: {:?}", ack).into());
    }
    println!("✓ Select 确认成功");

    // 5. SBO: Operate 阶段（执行）
    let operate_asdu = build_single_command(
        ioa,
        if close { 1 } else { 0 },
        CauseOfTransmission::Activation,
        0,
        /* select = */ false,             // 执行模式
    )?;
    data_source.send_command(operate_asdu).await?;
    println!("✓ Operate 已发送");

    // 6. 等待执行确认
    let exec_ack = data_source.wait_command_ack(Duration::from_secs(5)).await?;
    if exec_ack.is_positive() {
        println!("✓ 命令执行成功");
    } else {
        return Err(format!("命令执行失败: {:?}", exec_ack).into());
    }

    // 7. 写入审计链
    audit_log::record(
        "scada-operator",
        &format!("breaker_{}_{}", ioa, if close { "close" } else { "open" }),
        "success",
    ).await?;

    Ok(())
}

// 示例：合上 1-2 号断路器
send_breaker_command(&mut data_source, &constraints, 7001, true).await?;

设点命令（调节机组出力）

/// 下发机组出力设点命令
async fn send_setpoint_command(
    data_source: &mut Iec104DataSource,
    constraints: &ConstraintEngine,
    ioa: u32,
    value_mw: f64,
) -> Result<(), Box<dyn std::error::Error>> {
    // 约束校验
    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!("设点命令被拒绝: {:?}", verdict.violations).into());
    }

    // 构造设点 ASDU（TI=50，短浮点）
    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!("✓ 设点成功: ioa={} → {} MW", ioa, value_mw);
    } else {
        return Err(format!("设点失败: {:?}", ack).into());
    }
    Ok(())
}

// 示例：将 G1 机组出力设为 150 MW
send_setpoint_command(&mut data_source, &constraints, 8001, 150.0).await?;

步骤 6：事件处理与告警

监听 SCADA 状态变化（断路器变位、保护动作、数据质量异常）：

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

/// 处理 SCADA 状态变化事件
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. 检测断路器变位
        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. 检测数据质量异常
        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. 检测电压越限（基于阈值）
        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(())
}

双机冗余采集

生产环境通常部署双机冗余采集（主备模式），避免单点故障：

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

// 构建双机冗余采集组
let dual_group = DualScanGroupBuilder::new()
    .primary("10.0.0.20:2404", "scada-primary")   // 主机
    .secondary("10.0.0.21:2404", "scada-secondary") // 备机
    .options(DualScanOptions {
        switch_timeout_ms: 5000,           // 主备切换超时
        consistency_check: true,           // 启用数据一致性校验
        max_diff_percent: 1.0,             // 允许的最大偏差（%）
        snapshot_interval_s: 5,            // 快照对齐周期
    })
    .build()
    .await?;

// 启动双机采集
let handles = dual_group.start(mapping).await?;

// 主备状态可通过 handles 查看
println!("主通道状态: {:?}", handles.primary_status());
println!("备通道状态: {:?}", handles.secondary_status());

步骤 7：完整端到端示例

将上述步骤整合为一个完整的可运行程序：

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. 配置与连接
    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. 启动采集器
    let mut collector = ScadaCollector::new(ds);
    collector.start().await?;

    // 3. 打开时序引擎
    let mut ts = TimeseriesEngine::open("/var/lib/eneros/ts")?;

    // 4. 持续采集 60 秒
    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!("采集完成: {} 个测点写入", count);
    println!("采集速率: {:.0} 点/秒", count as f64 / 60.0);

    Ok(())
}

输出示例：

采集完成: 4560 个测点写入
采集速率: 76 点/秒

验证

按以下指标验证 SCADA 集成的正确性：

指标	期望值	说明
连接建立	< 1s	TCP + STARTDT con 流程
实时数据延迟	< 1s	RTU 上送 → 时序引擎可见
总召唤周期	30s	配置项 interrogation_interval_s
命令响应延迟	< 2s	SBO 流程完整执行
数据完整性	100%	无丢帧（k/w 窗口正常）
主备切换	< 5s	主机故障后切换到备机
审计记录	100%	所有命令落入 WORM 审计链

调试技巧

# 查看 IEC 104 连接状态
eneros-cli scada status --gateway scada-104-master

# 抓取原始报文
tcpdump -i eth0 -w iec104.pcap port 2404

# 用 wireshark 分析报文
wireshark iec104.pcap

# 查看时序数据写入情况
eneros-cli ts query bus1_voltage_pu --latest 10

# 强制触发总召唤
eneros-cli scada interrogate --gateway scada-104-master --ca 1

# 模拟断路器遥控（需 --confirm 二次确认）
eneros-cli scada send-command breaker_1_2_cmd --value 1 --sbo

常见问题排查

现象	可能原因	解决方案
连接超时	防火墙阻断 2404 端口	检查网络 ACL 与 mTLS 证书
数据不更新	RTU 处于 STOPDT 状态	发送 STARTDT 激活帧
频繁断连	t1 超时过短或网络抖动	增大 t1 至 30s，检查网络质量
命令被拒	约束引擎校验失败	查看 `/var/log/eneros/constraint.log`
数据跳变	scale/offset 配置错误	核对 IoaMapping 的缩放参数
双机数据不一致	时钟不同步	部署 PTP/NTP 时钟源

下一步

物联网泛在接入 — 多协议网关总览
安全守卫 — 命令安全校验详解
调度 Agent 开发 — 基于 SCADA 数据的自动调度
时序存储引擎 — 时序数据查询与聚合