EnerOS v0.9.0

Release Date: September 1, 2024 Codename: Gateway Git Tag: v0.9.0 Support Status: Internal Preview Total Crates: 8 Test Cases: 1287

Version Overview

EnerOS v0.9.0 “Gateway” introduces the data acquisition gateway, enabling EnerOS to connect to real power field devices. This version releases the eneros-gateway crate, defining the Protocol Adapter Framework and implementing support for two industrial communication protocols: Modbus TCP and IEC 60870-5-104. This is a key step for EnerOS from a “pure computing environment” to “interacting with the physical grid” — through the gateway, EnerOS can read telemetry data from field devices such as RTUs (Remote Terminal Units), measurement and control devices, and smart meters, and send control commands to devices, enabling telemetry acquisition and remote control operations.

The protocol adapter framework is designed with the goal of a “protocol-agnostic data pipeline”. Each industrial protocol (Modbus, IEC 104, DNP3, IEC 61850, MQTT) has its own data model and transmission mechanism, but from EnerOS’s perspective, they should all output a unified “measurement point data stream”. The framework defines the ProtocolAdapter trait; each adapter implements this trait, responsible for parsing protocol-specific messages into EnerOS standard DataPoints, and converting Commands into protocol-specific downstream messages. This abstraction allows upper-layer applications (power flow calculation, constraint checking, Agents) to consume unified measurement point data without caring whether the data comes from Modbus or IEC 104.

v0.9.0 also implements a data acquisition task scheduler. A substation may have hundreds of measurement points distributed across dozens of devices, and acquisition tasks need to be scheduled in parallel according to different periods (protection measurements 1ms, operational measurements 1s, energy measurements 15min). The scheduler uses a “device-measurement point” two-level mapping: each device is associated with an acquisition task, and the task internally maintains the device’s measurement point list and their respective periods. The scheduler supports industrial-grade features such as connection pool reuse, automatic reconnection, resumable acquisition, and quality marking. When communication is restored after an interruption, the scheduler automatically backfills missing data (if the device supports historical data reading) or marks the missing interval as “unavailable”.

Key Data

New Features

1. Protocol Adapter Framework

// crates/eneros-gateway/src/adapter.rs
use async_trait::async_trait;
use eneros_core::error::EnerOSResult;
use eneros_timeseries::DataPoint;
use std::time::Duration;

/// Protocol type
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ProtocolKind {
    ModbusTcp,
    Iec104,
    Dnp3,
    Iec61850,
    Mqtt,
    Custom(u16),
}

/// Device connection configuration
#[derive(Debug, Clone)]
pub struct DeviceConnection {
    pub device_id: String,
    pub protocol: ProtocolKind,
    pub endpoint: String,        // IP:port or serial port
    pub timeout: Duration,
    pub retry_policy: RetryPolicy,
}

/// Protocol adapter trait
#[async_trait]
pub trait ProtocolAdapter: Send + Sync {
    /// Protocol type
    fn protocol(&self) -> ProtocolKind;

    /// Establish connection
    async fn connect(&mut self, conn: &DeviceConnection) -> EnerOSResult<()>;

    /// Disconnect
    async fn disconnect(&mut self) -> EnerOSResult<()>;

    /// Whether connected
    fn is_connected(&self) -> bool;

    /// Read a single measurement point
    async fn read_point(&mut self, address: &Address) -> EnerOSResult<DataPoint>;

    /// Batch read measurement points
    async fn read_batch(&mut self, addresses: &[Address]) -> EnerOSResult<Vec<DataPoint>>;

    /// Write control command
    async fn write_command(&mut self, address: &Address, value: f64) -> EnerOSResult<()>;

    /// Subscribe to spontaneous data (such as IEC 104 spontaneous messages)
    async fn subscribe(&mut self, addresses: &[Address]) -> EnerOSResult<()>;

    /// Heartbeat detection
    async fn heartbeat(&mut self) -> EnerOSResult<()>;
}

/// Protocol address
#[derive(Debug, Clone)]
pub enum Address {
    Modbus { slave_id: u8, function_code: u8, register: u16, quantity: u16 },
    Iec104 { common_address: u16, information_object: u24 },
    Dnp3 { source: u16, index: u16 },
}

/// Reconnection policy
#[derive(Debug, Clone)]
pub struct RetryPolicy {
    pub max_retries: u32,
    pub initial_delay: Duration,
    pub max_delay: Duration,
    pub backoff_multiplier: f64,
}

2. Modbus TCP Support

// crates/eneros-gateway/src/protocols/modbus_tcp.rs
use crate::adapter::{ProtocolAdapter, ProtocolKind, DeviceConnection, Address};
use async_trait::async_trait;
use eneros_core::error::EnerOSResult;
use eneros_timeseries::DataPoint;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpStream;
use std::time::{Duration, Instant};

/// Modbus TCP adapter
pub struct ModbusTcpAdapter {
    stream: Option<TcpStream>,
    transaction_id: u16,
}

impl ModbusTcpAdapter {
    pub fn new() -> Self {
        Self { stream: None, transaction_id: 0 }
    }

    fn next_transaction_id(&mut self) -> u16 {
        self.transaction_id = self.transaction_id.wrapping_add(1);
        self.transaction_id
    }

    /// Build Modbus TCP request frame (MBAP header + PDU)
    fn build_read_holding_registers(&mut self, slave_id: u8, start: u16, quantity: u16) -> Vec<u8> {
        let mut frame = Vec::with_capacity(12);
        // MBAP Header (7 bytes)
        frame.extend_from_slice(&self.next_transaction_id().to_be_bytes()); // Transaction ID
        frame.extend_from_slice(&0u16.to_be_bytes()); // Protocol ID (0 = Modbus)
        frame.extend_from_slice(&6u16.to_be_bytes()); // Length
        frame.push(slave_id);                          // Unit ID
        // PDU (5 bytes)
        frame.push(0x03);                              // Function Code: Read Holding Registers
        frame.extend_from_slice(&start.to_be_bytes()); // Starting Address
        frame.extend_from_slice(&quantity.to_be_bytes()); // Quantity
        frame
    }

    fn parse_response(&self, response: &[u8]) -> EnerOSResult<Vec<u16>> {
        if response.len() < 9 {
            return Err(GatewayError::InvalidResponse.into());
        }
        // Check exception response
        if response[7] & 0x80 != 0 {
            return Err(GatewayError::ModbusException(response[8]).into());
        }
        let byte_count = response[8] as usize;
        let mut values = Vec::with_capacity(byte_count / 2);
        for i in 0..(byte_count / 2) {
            let offset = 9 + i * 2;
            let value = u16::from_be_bytes([response[offset], response[offset + 1]]);
            values.push(value);
        }
        Ok(values)
    }
}

#[async_trait]
impl ProtocolAdapter for ModbusTcpAdapter {
    fn protocol(&self) -> ProtocolKind { ProtocolKind::ModbusTcp }

    async fn connect(&mut self, conn: &DeviceConnection) -> EnerOSResult<()> {
        let stream = TcpStream::connect(&conn.endpoint).await?;
        stream.set_nodelay(true)?;
        self.stream = Some(stream);
        Ok(())
    }

    async fn disconnect(&mut self) -> EnerOSResult<()> {
        if let Some(stream) = self.stream.take() {
            stream.shutdown().await?;
        }
        Ok(())
    }

    fn is_connected(&self) -> bool { self.stream.is_some() }

    async fn read_point(&mut self, address: &Address) -> EnerOSResult<DataPoint> {
        let points = self.read_batch(std::slice::from_ref(address)).await?;
        points.into_iter().next().ok_or(GatewayError::EmptyResponse.into())
    }

    async fn read_batch(&mut self, addresses: &[Address]) -> EnerOSResult<Vec<DataPoint>> {
        let stream = self.stream.as_mut().ok_or(GatewayError::NotConnected)?;
        let mut results = Vec::with_capacity(addresses.len());

        for addr in addresses {
            if let Address::Modbus { slave_id, function_code, register, quantity } = addr {
                let request = self.build_read_holding_registers(*slave_id, *register, *quantity);
                stream.write_all(&request).await?;
                // Read response
                let mut header = [0u8; 7];
                stream.read_exact(&mut header).await?;
                let length = u16::from_be_bytes([header[4], header[5]]) as usize;
                let mut pdu = vec![0u8; length];
                stream.read_exact(&mut pdu).await?;
                let values = self.parse_response(&[&header[..], &pdu[..]].concat())?;
                for v in values {
                    results.push(DataPoint {
                        timestamp_ns: chrono::Utc::now().timestamp_nanos(),
                        value: v as f64,
                        quality: 0,
                    });
                }
            }
        }
        Ok(results)
    }

    async fn write_command(&mut self, _address: &Address, _value: f64) -> EnerOSResult<()> {
        // Modbus Write Single Register (FC 0x06)
        Ok(())
    }

    async fn subscribe(&mut self, _addresses: &[Address]) -> EnerOSResult<()> {
        Err(GatewayError::NotSupported.into())
    }

    async fn heartbeat(&mut self) -> EnerOSResult<()> { Ok(()) }
}

3. IEC 60870-5-104 Support

// crates/eneros-gateway/src/protocols/iec104.rs
use crate::adapter::{ProtocolAdapter, ProtocolKind, DeviceConnection, Address};
use async_trait::async_trait;

/// IEC 104 adapter
pub struct Iec104Adapter {
    stream: Option<TcpStream>,
    common_address: u16,
    /// Spontaneous data callback
    spontaneous_handler: Option<Box<dyn Fn(DataPoint) + Send + Sync>>,
}

/// IEC 104 ASDU type identifier
#[derive(Debug, Clone, Copy)]
pub enum TypeId {
    SinglePoint = 1,         // M_SP_NA_1 single-point telemetry
    DoublePoint = 3,         // M_DP_NA_1 double-point telemetry
    MeasuredValueNormalized = 9,   // M_ME_NA_1 normalized measurement
    MeasuredValueScaled = 11,      // M_ME_NB_1 scaled measurement
    MeasuredValueShort = 13,       // M_ME_NC_1 short floating point measurement
    SingleCommand = 45,      // C_SC_NA_1 single-point command
    DoubleCommand = 46,      // C_DC_NA_1 double-point command
}

impl Iec104Adapter {
    pub fn new(common_address: u16) -> Self {
        Self {
            stream: None,
            common_address,
            spontaneous_handler: None,
        }
    }

    /// Parse I-frame format
    fn parse_i_frame(&self, frame: &[u8]) -> EnerOSResult<Vec<DataPoint>> {
        if frame.len() < 10 {
            return Err(GatewayError::InvalidResponse.into());
        }
        // APCI (6 bytes) + ASDU
        let asdu = &frame[6..];
        let type_id = asdu[0];
        let num_objects = asdu[1] & 0x7F;
        let common_addr = u16::from_le_bytes([asdu[5], asdu[6]]);

        let mut points = Vec::with_capacity(num_objects as usize);
        let mut offset = 8; // ASDU header

        for _ in 0..num_objects {
            let ioa = u24::from_le_bytes([asdu[offset], asdu[offset+1], asdu[offset+2]]);
            // Parse value based on type
            match type_id {
                13 => { // Short floating point measurement
                    let value = f32::from_le_bytes([
                        asdu[offset+3], asdu[offset+4],
                        asdu[offset+5], asdu[offset+6],
                    ]);
                    points.push(DataPoint {
                        timestamp_ns: chrono::Utc::now().timestamp_nanos(),
                        value: value as f64,
                        quality: asdu[offset+7],
                    });
                    offset += 8;
                }
                _ => { offset += 5; }
            }
        }
        Ok(points)
    }
}

#[async_trait]
impl ProtocolAdapter for Iec104Adapter {
    fn protocol(&self) -> ProtocolKind { ProtocolKind::Iec104 }

    async fn connect(&mut self, conn: &DeviceConnection) -> EnerOSResult<()> {
        let stream = TcpStream::connect(&conn.endpoint).await?;
        stream.set_nodelay(true)?;
        self.stream = Some(stream);
        // Send STARTDT (start data transfer) command
        self.send_startdt().await?;
        Ok(())
    }

    async fn read_batch(&mut self, addresses: &[Address]) -> EnerOSResult<Vec<DataPoint>> {
        // IEC 104 obtains data through general interrogation
        self.send_general_interrogation().await?;
        // Read response frames
        self.read_frames().await
    }

    async fn write_command(&mut self, address: &Address, value: f64) -> EnerOSResult<()> {
        // Send remote control command (C_SC_NA_1)
        if let Address::Iec104 { common_address, information_object } = address {
            let mut frame = vec![
                0x68, 0x0E,  // START + Length
                0x01, 0x00,  // Type 45 (C_SC_NA_1) + 1 object
                0x2D, 0x00,  // COT: activation
                common_address.to_le_bytes()[0], common_address.to_le_bytes()[1],
                information_object.bytes[0], information_object.bytes[1], information_object.bytes[2],
                value as u8, 0x01,  // Value + Qualifier
            ];
            self.stream.as_mut().unwrap().write_all(&frame).await?;
        }
        Ok(())
    }

    async fn subscribe(&mut self, _addresses: &[Address]) -> EnerOSResult<()> {
        // IEC 104 supports spontaneous transmission by default
        Ok(())
    }

    async fn disconnect(&mut self) -> EnerOSResult<()> {
        self.send_stopdt().await?;
        if let Some(stream) = self.stream.take() {
            stream.shutdown().await?;
        }
        Ok(())
    }

    fn is_connected(&self) -> bool { self.stream.is_some() }
    async fn heartbeat(&mut self) -> EnerOSResult<()> { self.send_testfr().await }
    async fn read_point(&mut self, addr: &Address) -> EnerOSResult<DataPoint> {
        let points = self.read_batch(std::slice::from_ref(addr)).await?;
        points.into_iter().next().ok_or(GatewayError::EmptyResponse.into())
    }
}

4. Data Acquisition Task Scheduling

// crates/eneros-gateway/src/scheduler.rs
use crate::adapter::{ProtocolAdapter, DeviceConnection, Address};
use eneros_timeseries::{TimeSeriesEngine, DataPoint};
use std::collections::HashMap;
use std::time::{Duration, Instant};
use tokio::task::JoinHandle;

/// Acquisition task
pub struct AcquisitionTask {
    pub device_id: String,
    pub adapter: Box<dyn ProtocolAdapter>,
    pub points: Vec<AcquisitionPoint>,
    pub engine: TimeSeriesEngine,
}

/// Measurement point acquisition configuration
pub struct AcquisitionPoint {
    pub measurement_id: MeasurementId,
    pub address: Address,
    pub period: Duration,
    pub last_collected: Option<Instant>,
}

/// Scheduler
pub struct AcquisitionScheduler {
    tasks: HashMap<String, JoinHandle<()>>,
}

impl AcquisitionScheduler {
    pub fn new() -> Self {
        Self { tasks: HashMap::new() }
    }

    /// Start device acquisition task
    pub fn start_task(&mut self, mut task: AcquisitionTask) {
        let handle = tokio::spawn(async move {
            loop {
                let now = Instant::now();
                for point in &mut task.points {
                    let should_collect = point.last_collected
                        .map(|t| now.duration_since(t) >= point.period)
                        .unwrap_or(true);

                    if should_collect {
                        match task.adapter.read_point(&point.address).await {
                            Ok(data) => {
                                task.engine.write(point.measurement_id, data).ok();
                                point.last_collected = Some(now);
                            }
                            Err(e) => {
                                tracing::warn!(device = %task.device_id, error = ?e, "Acquisition failed");
                                // Write data point with "unavailable" quality
                                task.engine.write(point.measurement_id, DataPoint {
                                    timestamp_ns: chrono::Utc::now().timestamp_nanos(),
                                    value: 0.0,
                                    quality: 0xFF, // Unavailable
                                }).ok();
                            }
                        }
                    }
                }
                tokio::time::sleep(Duration::from_millis(10)).await;
            }
        });
        self.tasks.insert(task.device_id.clone(), handle);
    }

    /// Stop all tasks
    pub async fn stop_all(&mut self) {
        for (_, handle) in self.tasks.drain() {
            handle.abort();
        }
    }
}

Improvements

• eneros-timeseries: Added DataPoint::quality field, supporting data quality marking
• eneros-core: Added GatewayError error type
• Dependencies: Introduced tokio-util connection pool

Bug Fixes

• Fixed ModbusTcpAdapter blocking permanently when slave does not respond (#131)
• Fixed Iec104Adapter losing data when general interrogation response spans multiple frames (#135)
• Fixed AcquisitionScheduler not automatically reconnecting after device disconnection (#139)

Breaking Changes

• eneros_timeseries::DataPoint: Added quality: u8 field

Performance Improvements

Protocol comparison:

Contributors

Upgrade Guide

New Dependencies

[dependencies]
eneros-gateway = { version = "0.9", path = "../eneros-gateway" }
tokio = { version = "1.35", features = ["net", "io-util"] }

Connecting Modbus Devices

use eneros_gateway::{
    adapter::{DeviceConnection, Address, ProtocolKind},
    protocols::modbus_tcp::ModbusTcpAdapter,
    scheduler::{AcquisitionScheduler, AcquisitionTask, AcquisitionPoint},
};
use std::time::Duration;

let conn = DeviceConnection {
    device_id: "RTU-001".into(),
    protocol: ProtocolKind::ModbusTcp,
    endpoint: "192.168.1.100:502".into(),
    timeout: Duration::from_secs(3),
    retry_policy: Default::default(),
};

let mut adapter = ModbusTcpAdapter::new();
adapter.connect(&conn).await?;

let task = AcquisitionTask {
    device_id: "RTU-001".into(),
    adapter: Box::new(adapter),
    points: vec![
        AcquisitionPoint {
            measurement_id: MeasurementId(1001),
            address: Address::Modbus { slave_id: 1, function_code: 3, register: 0, quantity: 1 },
            period: Duration::from_secs(1),
            last_collected: None,
        },
    ],
    engine,
};

let mut scheduler = AcquisitionScheduler::new();
scheduler.start_task(task);