数字孪生引擎

EnerOS 数字孪生引擎在内存中维护电网的实时镜像，支持 What-If 推演、历史回放、虚拟传感器等能力。端到端延迟 < 100ms，让 Agent 能够「先仿真后执行」，将决策风险降到最低。

数字孪生是 Power-Native 内核的「沙箱」：在孪生镜像上模拟决策效果，验证约束与稳定性后再投放到真实电网。这让 Agent 拥有了「试错空间」，避免了在真实系统上直接决策带来的风险。

架构总览

┌─────────────────────────────────────────────────────┐
│  Agent：先仿真后执行                                │
└──────────────────┬──────────────────────────────────┘
                   │ what_if / replay / fork
┌──────────────────┴──────────────────────────────────┐
│  eneros-twin (数字孪生引擎)                         │
│  ┌──────────┐ ┌──────────┐ ┌──────────┐            │
│  │ Mirror   │ │ WhatIf   │ │ Replay   │            │
│  │ 镜像维护 │ │ 推演引擎 │ │ 历史回放 │            │
│  └──────────┘ └──────────┘ └──────────┘            │
│  ┌──────────┐ ┌──────────┐ ┌──────────┐            │
│  │ VSensor  │ │ Calibrate│ │ Compare  │            │
│  │ 虚拟传感 │ │ 模型校准 │ │ 多孪生对比│            │
│  └──────────┘ └──────────┘ └──────────┘            │
└──────────────────┬──────────────────────────────────┘
                   │ 订阅内核事件
┌──────────────────┴──────────────────────────────────┐
│  Kernel：拓扑 / 设备 / 时序 / 测量                  │
└─────────────────────────────────────────────────────┘

镜像维护

孪生引擎订阅所有拓扑变更、设备状态、测量数据事件，实时更新镜像：

use eneros_twin::{DigitalTwin, TwinConfig, Fidelity};
use std::time::Duration;

let mut twin = DigitalTwin::new(TwinConfig {
    sync_interval: Duration::from_millis(50),
    history_window: Duration::from_secs(3600),
    fidelity: Fidelity::Detailed,
    max_forks: 8,
    auto_calibrate: true,
});

// 自动订阅内核事件并同步
twin.start(&kernel).await?;

// 查询镜像状态
let status = twin.status();
println!("镜像延迟: {:.1}ms", status.lag_ms);
println!("历史窗口: {:.0}s", status.history_window.as_secs());
println!("活跃 fork 数: {}", status.active_forks);

TwinConfig 字段

Fidelity 变体

What-If 推演

在不影响真实电网的前提下，模拟决策的执行效果：

use eneros_twin::{WhatIf, Scenario, Event, Mode};
use std::time::Duration;

let scenario = Scenario::new()
    .event(Event::OpenBranch(branch_id))
    .event(Event::IncreaseLoad(bus_id, 50.0))
    .event(Event::SetGenerator(gen_id, 80.0))
    .duration(Duration::from_secs(60));

let result = twin.what_if(scenario).await?;

// 分析推演结果
println!("违反约束数: {}", result.violations.len());
for violation in &result.violations {
    println!("  - {:?}: {}", violation.kind, violation.message);
}

println!("最大电压偏差: {:.2}%", result.max_voltage_deviation * 100.0);
println!("最大频率偏差: {:.2} Hz", result.max_frequency_deviation);
println!("收敛: {}", result.converged);

WhatIfResult 字段

推演策略

支持稳态推演与暂态推演两种模式：

// 稳态：快速，适用于调度决策
let steady = twin.what_if(scenario.clone())
    .mode(Mode::SteadyState)
    .await?;

// 暂态：精确，适用于保护验证
let transient = twin.what_if(scenario)
    .mode(Mode::Transient(Duration::from_secs(10)))
    .step(Duration::from_millis(1))  // 仿真步长
    .await?;

// 概率推演：考虑不确定性
let monte_carlo = twin.what_if(scenario)
    .mode(Mode::MonteCarlo { samples: 100 })
    .await?;

推演模式对比

历史回放

回放任意时间点的电网状态，支持事件级精确复现：

use eneros_twin::{Replay, Range};

let mut replay = twin.replay()
    .range(Range::between(t_start, t_end))
    .speed(10.0)       // 10 倍速
    .events(true)      // 包含事件
    .start().await?;

while let Some(snapshot) = replay.next().await? {
    // 在快照上运行分析
    let result = analyze(&snapshot)?;
    println!("{}: V_max = {:.4}", snapshot.timestamp, result.max_voltage);
}

// 跳转到指定时刻
replay.seek_to(specific_time).await?;

// 暂停/恢复
replay.pause().await?;
replay.resume().await?;

ReplayConfig 字段

虚拟传感器

在孪生镜像中部署虚拟测量点，覆盖真实传感器未安装的位置：

use eneros_twin::VirtualSensor;
use std::time::Duration;

let sensor = VirtualSensor::new("virtual.bus_5.voltage")
    .calculated_by(|network| {
        // 基于状态估计推算
        network.estimate_voltage(bus_5)
    })
    .update_interval(Duration::from_millis(100))
    .quality(Quality::Calculated);

twin.deploy_sensor(sensor).await?;

// 像真实测量点一样查询
let v = timeseries.query("virtual.bus_5.voltage").await?;

// 部署多个虚拟传感器
let sensors = vec![
    VirtualSensor::new("virtual.bus_5.voltage")
        .calculated_by(|n| n.estimate_voltage(5)),
    VirtualSensor::new("virtual.branch_3.flow")
        .calculated_by(|n| n.estimate_flow(3)),
    VirtualSensor::new("virtual.gen_2.thermal")
        .calculated_by(|n| n.estimate_gen_temp(2)),
];
twin.deploy_sensors(sensors).await?;

虚拟传感器类型

模型校准

孪生引擎持续对比真实测量与仿真结果，自动校准模型参数：

use eneros_twin::{CalibrationTarget, CalibrationMethod};

let calibration = twin.calibrate()
    .target(CalibrationTarget::LineImpedance(branch_id))
    .method(CalibrationMethod::LeastSquares)
    .window(Duration::from_secs(3600))
    .tolerance(1e-4)
    .run().await?;

println!("校准前: z={:.4}, 校准后: z={:.4}",
    calibration.before, calibration.after);
println!("残差: {:.2e}", calibration.residual);
println!("置信度: {:.1}%", calibration.confidence * 100.0);

// 批量校准
let targets = vec![
    CalibrationTarget::LineImpedance(branch_1),
    CalibrationTarget::TransformerTap(xfmr_2),
    CalibrationTarget::LoadModel(bus_3),
];
let report = twin.calibrate_batch(targets).await?;

CalibrationTarget 类型

多孪生实例

支持并行运行多个孪生实例，用于对比不同策略：

// Fork 多个孪生实例
let twin_a = twin.fork("strategy_a").await?;
let twin_b = twin.fork("strategy_b").await?;
let twin_c = twin.fork("strategy_c").await?;

// 在三个孪生上分别推演
let r_a = twin_a.what_if(scenario_a).await?;
let r_b = twin_b.what_if(scenario_b).await?;
let r_c = twin_c.what_if(scenario_c).await?;

// 对比结果
let comparison = twin.compare(&[&r_a, &r_b, &r_c]).await?;
println!("方案 | 违反数 | 最大电压偏差 | 稳定裕度");
for (i, r) in comparison.results.iter().enumerate() {
    println!("  {}  |   {}    |    {:.2}%     |  {:.2}%",
        i, r.violations.len(),
        r.max_voltage_deviation * 100.0,
        r.stability_margin * 100.0);
}

// 选择最优方案
let best = comparison.best();
println!("最优方案: {}", best.label);

// 释放 fork
twin_a.drop().await?;
twin_b.drop().await?;
twin_c.drop().await?;

与 Agent 集成

Agent 调用孪生引擎进行「先仿真后执行」：

use eneros_twin::Scenario;

impl Agent for CautiousDispatchAgent {
    async fn run(&mut self, ctx: &mut AgentContext) -> AgentResult<()> {
        loop {
            let state = ctx.snapshot().await?;
            let decision = self.compute_dispatch(&state)?;

            // 先在孪生上仿真
            let scenario = Scenario::from_decision(&decision);
            let result = ctx.twin().what_if(scenario).await?;

            if result.violations.is_empty() && result.stability_margin > 0.1 {
                // 安全，执行
                ctx.execute(decision).await?;
            } else {
                // 不安全，重新规划
                let adjusted = self.adjust_decision(decision, &result)?;
                ctx.execute(adjusted).await?;
            }

            ctx.sleep(self.interval).await;
        }
    }
}

性能指标

测试环境：4 核 / 8GB / Ubuntu 22.04，节点数 1000。

应用场景

配置参数

eneros.toml 中孪生相关配置：

[twin]
# 同步间隔（毫秒）
sync_interval_ms = 50
# 历史窗口（秒）
history_window_secs = 3600
# 默认模型精度：detailed / standard / coarse / auto
default_fidelity = "auto"
# 最大 fork 数
max_forks = 8
# 是否启用自动校准
auto_calibrate = true
# 校准间隔（秒）
calibrate_interval_secs = 3600
# 镜像保留期（小时）
retention_hours = 24
# 是否启用虚拟传感器
enable_virtual_sensors = true

与其他能力的关系

相关文档

• 电网拓扑一等公民 — 孪生基于拓扑构建
• 物理约束决策 — What-If 调用约束校验
• 时序原生操作 — 回放依赖时序数据
• 设备模型库 — 设备模型驱动仿真
• EnerOS 简介 — Power-Native First 设计哲学