eneros-twin

eneros-twin provides a grid Digital Twin engine, maintaining a real-time mirror synchronized with the physical grid, supporting What-If hypothesis analysis, historical replay, and virtual sensors. The Twin shares the same topology based on eneros-topology, overlaying time series state and simulation results. It is the core component of EnerOS’s “simulate before execute” safety paradigm.

Responsibilities

eneros-twin handles the following core responsibilities:

• Real-time Mirror: Subscribes to SCADA / PMU measurements, synchronizes to the Twin at second-level, maintaining consistency with the physical grid
• What-If Analysis: Simulates command effects (switch operations, output adjustment) on the Twin without polluting the real grid
• Historical Replay: Reconstructs grid state at any point in time, supporting event review and root cause analysis
• Virtual Sensors: Estimates state at locations without physical measurements (based on state estimation), expanding observable range
• Snapshot Management: Scheduled and event-triggered snapshots, supporting differential comparison
• Multi-Scenario Parallelism: Supports multiple independent Twins running in parallel without interference
• Uncertainty Propagation: Measurement errors and state estimation uncertainty propagate within the Twin

Key Types and Interfaces

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 } => {
                // Update branch flow
            }
        }
    }

    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 } => {
                // Adjust bus generation
            }
            _ => {}
        }
    }

    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);
        }
        // Estimate virtual measurement based on state estimation
        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,
}

Core API

Usage Examples

Load Twin and Synchronize

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!(
    "Sync complete: confidence={:.1}%, bus_count={}",
    twin.state().confidence * 100.0,
    twin.state().bus_voltages.len()
);

What-If Simulation

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!("Simulation result:");
println!("  Load Flow converged: {}", result.powerflow.converged);
println!("  Violation count: {}", result.violations.len());
println!("  Loss change: {:.4} pu", result.delta.loss_change);

for (bus, old, new) in &result.delta.voltage_changes {
    println!("  Bus {} voltage: {:.4} → {:.4}", bus, old, new);
}

Historical Replay

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

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

// Replay state from 1 hour ago
let one_hour_ago = Utc::now() - Duration::hours(1);
if let Some(snapshot) = twin.replay(one_hour_ago) {
    println!(
        "Replay time: {}, bus_count: {}",
        snapshot.timestamp,
        snapshot.state.bus_voltages.len()
    );
}

// Time range query
let from = Utc::now() - Duration::hours(24);
let to = Utc::now();
for snapshot in twin.history.range(from, to) {
    println!(
        "[{}] confidence: {:.1}%",
        snapshot.timestamp,
        snapshot.state.confidence * 100.0
    );
}

Virtual Sensor

use eneros_twin::{TwinModel, VirtualSensor};

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

let report = vs.coverage_report(&twin);
println!(
    "Total buses: {}, measured: {}, virtual: {}, coverage: {:.1}%",
    report.total_buses,
    report.measured_buses,
    report.virtual_buses,
    report.coverage_percent
);

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

State Comparison

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!("Compared with same time yesterday:");
    println!("  Bus count with voltage changes: {}", delta.voltage_changes.len());
    println!("  Loss change: {:.4} pu", delta.loss_change);
}

Performance Metrics

Test environment: 4 cores / 8GB / Ubuntu 22.04 / Rust 1.78 release build.

Configuration Parameter Table

TwinConfig Fields

Dependencies

Own Dependencies

[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"] }

Depended On By

• eneros-agent — Agents perform What-If on the Twin before decision-making
• eneros-dashboard — Dashboard displays Twin state
• eneros-simulator — Simulator builds scenarios based on the Twin
• eneros-aiops — AIOps performs root cause analysis on the Twin

Version and Compatibility

Related Documentation

• Digital Twin Engine — Capability details
• Real-time Dual Execution Domain — Real-time/analysis dual domains
• eneros-powerflow — Load Flow solving
• eneros-analysis — State estimation
• eneros-gateway — What-If simulation command construction
• Crate Index — Complete list