eneros-twin

eneros-twin 提供电网数字孪生引擎，维护与物理电网同步的实时镜像，支持 What-If 假设分析、历史回放与虚拟传感器。孪生体基于 eneros-topology 共享同一份拓扑，叠加时序状态与仿真结果，是 EnerOS 实现「先模拟后执行」安全范式的核心组件。

职责描述

eneros-twin 承担以下核心职责：

• 实时镜像：订阅 SCADA / PMU 量测，秒级同步至孪生体，保持与物理电网一致
• What-If 分析：在孪生体上模拟指令效果（开关变位、出力调节），不污染真实电网
• 历史回放：按时间点重建任意时刻电网状态，支持事件复盘与根因分析
• 虚拟传感器：在无物理量测处估算状态（基于状态估计），扩充可观测范围
• 快照管理：定时与事件触发的快照，支持差异对比
• 多场景并行：支持多个独立孪生体并行运行，互不干扰
• 不确定性传播：量测误差与状态估计不确定性在孪生体中传播

关键类型与接口

TwinModel

use eneros_topology::NetworkGraph;
use eneros_powerflow::PowerFlowResult;
use eneros_timeseries::TimeSeriesEngine;
use chrono::{DateTime, Utc};

pub struct TwinModel {
    network: NetworkGraph,
    state: NetworkState,
    history: HistoryStore,
    config: TwinConfig,
}

#[derive(Debug, Clone)]
pub struct NetworkState {
    pub bus_voltages: std::collections::HashMap<eneros_core::BusId, eneros_core::Voltage>,
    pub branch_flows: Vec<eneros_powerflow::BranchFlow>,
    pub updated_at: DateTime<Utc>,
    pub confidence: f64,
}

#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct TwinConfig {
    pub sync_interval_ms: u64,
    pub snapshot_interval_s: u64,
    pub history_retention_days: u32,
    pub enable_virtual_sensor: bool,
}

impl Default for TwinConfig {
    fn default() -> Self {
        Self {
            sync_interval_ms: 1000,
            snapshot_interval_s: 60,
            history_retention_days: 30,
            enable_virtual_sensor: true,
        }
    }
}

impl TwinModel {
    pub fn new(network: NetworkGraph) -> Self {
        Self {
            network,
            state: NetworkState {
                bus_voltages: std::collections::HashMap::new(),
                branch_flows: Vec::new(),
                updated_at: Utc::now(),
                confidence: 0.0,
            },
            history: HistoryStore::new(),
            config: TwinConfig::default(),
        }
    }

    pub fn load(network_id: &str) -> Result<Self> {
        let network = NetworkGraph::load(network_id)?;
        Ok(Self::new(network))
    }

    pub async fn sync(&mut self, measurements: &[Measurement]) {
        for m in measurements {
            self.apply_measurement(m);
        }
        self.state.updated_at = Utc::now();
        self.state.confidence = self.compute_confidence();
        self.history.push(self.state.clone());
    }

    fn apply_measurement(&mut self, m: &Measurement) {
        match m {
            Measurement::Voltage { bus, magnitude, angle } => {
                self.state.bus_voltages.insert(
                    bus.clone(),
                    eneros_core::Voltage::new(*magnitude, *angle),
                );
            }
            Measurement::Power { branch, p, q } => {
                // 更新支路潮流
            }
        }
    }

    fn compute_confidence(&self) -> f64 {
        let coverage = self.state.bus_voltages.len() as f64
            / self.network.bus_count().max(1) as f64;
        coverage.min(1.0)
    }

    pub fn snapshot(&self) -> Snapshot {
        Snapshot {
            timestamp: self.state.updated_at,
            state: self.state.clone(),
        }
    }

    pub fn what_if(&self) -> WhatIfEngine<'_> {
        WhatIfEngine { twin: self }
    }

    pub fn replay(&self, at: DateTime<Utc>) -> Option<Snapshot> {
        self.history.get(at)
    }

    pub fn network(&self) -> &NetworkGraph {
        &self.network
    }

    pub fn state(&self) -> &NetworkState {
        &self.state
    }
}

Measurement / Snapshot

use eneros_core::{BusId, BranchId};

#[derive(Debug, Clone)]
pub enum Measurement {
    Voltage {
        bus: BusId,
        magnitude: f64,
        angle: f64,
    },
    Power {
        branch: BranchId,
        p: f64,
        q: f64,
    },
    Frequency {
        value: f64,
    },
}

#[derive(Debug, Clone)]
pub struct Snapshot {
    pub timestamp: DateTime<Utc>,
    pub state: NetworkState,
}

pub struct HistoryStore {
    snapshots: Vec<Snapshot>,
    max_size: usize,
}

impl HistoryStore {
    pub fn new() -> Self {
        Self {
            snapshots: Vec::new(),
            max_size: 100_000,
        }
    }

    pub fn push(&mut self, state: NetworkState) {
        let snapshot = Snapshot {
            timestamp: state.updated_at,
            state,
        };
        if self.snapshots.len() >= self.max_size {
            self.snapshots.remove(0);
        }
        self.snapshots.push(snapshot);
    }

    pub fn get(&self, at: DateTime<Utc>) -> Option<Snapshot> {
        self.snapshots
            .iter()
            .min_by_key(|s| (s.timestamp - at).num_seconds().abs())
            .cloned()
    }

    pub fn range(&self, from: DateTime<Utc>, to: DateTime<Utc>) -> Vec<&Snapshot> {
        self.snapshots
            .iter()
            .filter(|s| s.timestamp >= from && s.timestamp <= to)
            .collect()
    }

    pub fn count(&self) -> usize {
        self.snapshots.len()
    }
}

WhatIfEngine

use eneros_gateway::Command;

pub struct WhatIfEngine<'a> {
    twin: &'a TwinModel,
}

#[derive(Debug, Clone)]
pub struct TwinSimulationResult {
    pub command: Command,
    pub powerflow: eneros_powerflow::PowerFlowResult,
    pub violations: Vec<eneros_constraint::Violation>,
    pub delta: StateDelta,
}

#[derive(Debug, Clone, Default)]
pub struct StateDelta {
    pub voltage_changes: Vec<(BusId, f64, f64)>,
    pub loading_changes: Vec<(BranchId, f64, f64)>,
    pub loss_change: f64,
}

impl<'a> WhatIfEngine<'a> {
    pub fn apply(&self, cmd: &Command) -> TwinSimulationResult {
        let mut simulated_network = self.twin.network.clone();
        self.apply_command(&mut simulated_network, cmd);

        let solver = eneros_powerflow::PowerFlowSolver::new();
        let powerflow = solver.solve(&simulated_network).unwrap_or_else(|_| {
            eneros_powerflow::PowerFlowResult {
                converged: false,
                iterations: 0,
                residual: f64::INFINITY,
                duration: std::time::Duration::ZERO,
                buses: Vec::new(),
                branch_flows: Vec::new(),
            }
        });

        let engine = eneros_constraint::ConstraintEngine::default();
        let violations = engine.check(&powerflow);

        let delta = self.compute_delta(&powerflow);

        TwinSimulationResult {
            command: cmd.clone(),
            powerflow,
            violations,
            delta,
        }
    }

    fn apply_command(&self, network: &mut NetworkGraph, cmd: &Command) {
        use eneros_gateway::Action;
        match &cmd.action {
            Action::SwitchBranch { branch, status } => {
                if let Some(b) = find_branch_mut(network, branch) {
                    b.status = *status;
                }
            }
            Action::SetGeneration { bus, mw } => {
                // 调整母线发电
            }
            _ => {}
        }
    }

    fn compute_delta(&self, result: &eneros_powerflow::PowerFlowResult) -> StateDelta {
        let mut delta = StateDelta::default();
        for bus in &result.buses {
            if let Some(old) = self.twin.state.bus_voltages.get(&bus.id) {
                let diff = bus.voltage.magnitude.0 - old.magnitude.0;
                if diff.abs() > 1e-6 {
                    delta.voltage_changes.push((
                        bus.id.clone(),
                        old.magnitude.0,
                        bus.voltage.magnitude.0,
                    ));
                }
            }
        }
        delta.loss_change = result.total_loss_pu()
            - self.twin.state.branch_flows.iter().map(|f| f.loss_pu).sum::<f64>();
        delta
    }

    pub fn compare(&self, other_state: &NetworkState) -> StateDelta {
        let mut delta = StateDelta::default();
        for (bus, voltage) in &self.twin.state.bus_voltages {
            if let Some(other) = other_state.bus_voltages.get(bus) {
                let diff = voltage.magnitude.0 - other.magnitude.0;
                if diff.abs() > 1e-6 {
                    delta.voltage_changes.push((
                        bus.clone(),
                        other.magnitude.0,
                        voltage.magnitude.0,
                    ));
                }
            }
        }
        delta
    }
}

fn find_branch_mut<'a>(
    network: &'a mut NetworkGraph,
    id: &eneros_core::BranchId,
) -> Option<&'a mut eneros_topology::Branch> {
    None
}

VirtualSensor

use eneros_analysis::StateEstimator;

pub struct VirtualSensor {
    estimator: StateEstimator,
}

impl VirtualSensor {
    pub fn new() -> Self {
        Self {
            estimator: StateEstimator::new(),
        }
    }

    pub fn estimate(&self, twin: &TwinModel, bus: &eneros_core::BusId) -> Option<eneros_core::Voltage> {
        if let Some(v) = twin.state.bus_voltages.get(bus) {
            return Some(*v);
        }
        // 基于状态估计估算虚拟量测
        self.estimator.estimate_voltage(&twin.network, bus)
    }

    pub fn coverage_report(&self, twin: &TwinModel) -> CoverageReport {
        let total = twin.network.bus_count();
        let measured = twin.state.bus_voltages.len();
        let virtual_count = twin
            .network
            .bus_ids()
            .filter(|id| !twin.state.bus_voltages.contains_key(id) && self.estimate(twin, id).is_some())
            .count();
        CoverageReport {
            total_buses: total,
            measured_buses: measured,
            virtual_buses: virtual_count,
            coverage_percent: (measured + virtual_count) as f64 / total as f64 * 100.0,
        }
    }
}

#[derive(Debug, Clone)]
pub struct CoverageReport {
    pub total_buses: usize,
    pub measured_buses: usize,
    pub virtual_buses: usize,
    pub coverage_percent: f64,
}

核心 API

使用示例

加载孪生体并同步

use eneros_twin::{TwinModel, Measurement};

let mut twin = TwinModel::load("net_8f3a2b")?;

let measurements = vec![
    Measurement::Voltage {
        bus: "bus_1".into(),
        magnitude: 1.05,
        angle: 0.0,
    },
    Measurement::Voltage {
        bus: "bus_2".into(),
        magnitude: 0.98,
        angle: -0.12,
    },
    Measurement::Power {
        branch: "branch_1_2".into(),
        p: 0.5,
        q: 0.1,
    },
    Measurement::Frequency { value: 50.0 },
];

twin.sync(&measurements).await;
println!(
    "同步完成: 置信度={:.1}%, 母线数={}",
    twin.state().confidence * 100.0,
    twin.state().bus_voltages.len()
);

What-If 模拟

use eneros_twin::TwinModel;
use eneros_gateway::{Command, Action, BranchStatus};

let twin = TwinModel::load("net_8f3a2b")?;

let cmd = Command::new("planner_01").action(Action::SwitchBranch {
    branch: "branch_1_2".into(),
    status: BranchStatus::Open,
});

let result = twin.what_if().apply(&cmd);

println!("模拟结果:");
println!("  潮流收敛: {}", result.powerflow.converged);
println!("  越限数量: {}", result.violations.len());
println!("  损耗变化: {:.4} pu", result.delta.loss_change);

for (bus, old, new) in &result.delta.voltage_changes {
    println!("  母线 {} 电压: {:.4} → {:.4}", bus, old, new);
}

历史回放

use eneros_twin::TwinModel;
use chrono::{Utc, Duration};

let twin = TwinModel::load("net_8f3a2b")?;

// 回放 1 小时前的状态
let one_hour_ago = Utc::now() - Duration::hours(1);
if let Some(snapshot) = twin.replay(one_hour_ago) {
    println!(
        "回放时刻: {}, 母线数: {}",
        snapshot.timestamp,
        snapshot.state.bus_voltages.len()
    );
}

// 时间范围查询
let from = Utc::now() - Duration::hours(24);
let to = Utc::now();
for snapshot in twin.history.range(from, to) {
    println!(
        "[{}] 置信度: {:.1}%",
        snapshot.timestamp,
        snapshot.state.confidence * 100.0
    );
}

虚拟传感器

use eneros_twin::{TwinModel, VirtualSensor};

let twin = TwinModel::load("net_8f3a2b")?;
let vs = VirtualSensor::new();

let report = vs.coverage_report(&twin);
println!(
    "总母线: {}, 实测: {}, 虚拟: {}, 覆盖率: {:.1}%",
    report.total_buses,
    report.measured_buses,
    report.virtual_buses,
    report.coverage_percent
);

if let Some(v) = vs.estimate(&twin, &"bus_5".into()) {
    println!("虚拟估算 bus_5: V={:.4} θ={:.4}", v.magnitude.0, v.angle.as_radians());
}

状态对比

use eneros_twin::TwinModel;
use chrono::{Utc, Duration};

let twin = TwinModel::load("net_8f3a2b")?;

let now = twin.state().clone();
let yesterday = twin.replay(Utc::now() - Duration::days(1));

if let Some(snap) = yesterday {
    let delta = twin.what_if().compare(&snap.state);
    println!("与昨日同时刻对比:");
    println!("  电压变化母线数: {}", delta.voltage_changes.len());
    println!("  损耗变化: {:.4} pu", delta.loss_change);
}

性能指标

测试环境：4 核 / 8GB / Ubuntu 22.04 / Rust 1.78 release build。

配置参数表

TwinConfig 字段

依赖关系

自身依赖

[dependencies]
eneros-core = { path = "../core", version = "0.47.0" }
eneros-topology = { path = "../topology", version = "0.47.0" }
eneros-powerflow = { path = "../powerflow", version = "0.47.0" }
eneros-constraint = { path = "../constraint", version = "0.47.0" }
eneros-timeseries = { path = "../timeseries", version = "0.47.0" }
eneros-scada = { path = "../scada", version = "0.47.0" }
eneros-analysis = { path = "../analysis", version = "0.47.0" }
eneros-gateway = { path = "../gateway", version = "0.47.0" }
tokio = { version = "1", features = ["full"] }
serde = { version = "1.0", features = ["derive"] }
serde_json = "1.0"
tracing = "0.1"
chrono = { version = "0.4", features = ["serde"] }

被依赖

• eneros-agent — Agent 决策前在孪生体上 What-If
• eneros-dashboard — 仪表盘展示孪生体状态
• eneros-simulator — 模拟器基于孪生体构建场景
• eneros-aiops — AIOps 在孪生体上做根因分析

版本与兼容性

相关文档

• 数字孪生引擎 — 能力详解
• 实时双执行域 — 实时/分析双域
• eneros-powerflow — 潮流求解
• eneros-analysis — 状态估计
• eneros-gateway — What-If 模拟指令构造
• Crate 索引 — 完整列表