Energy Quotient™ Developer Documentation

Synapse Power Monitor (PMON) Architecture

Sensor → Zenoh → PMON-DAQ → Storage

The Synapse Power Monitor (PMON) pipeline provides a distributed power quality monitoring system using Zenoh middleware for real-time data streaming and processing.

Data Flow Overview:

1. Hardware/Simulation Layer: Physical sensors or simulation generates power quality measurements
2. Zenoh Publisher: Data published to Zenoh middleware with topic-based routing
3. PMON Subscriber: Receives power quality data via Zenoh pub/sub mechanisms
4. Data Processing: Deserializes and validates power quality measurements
5. Storage Backend: Writes data to Avro files with configurable rotation policies
6. Management Interface: Web API for monitoring and control operations

Key Architecture Features:

• Zenoh Middleware: Decentralized pub/sub with automatic discovery and routing
• Flexible Data Sources: Supports both real sensors and high-fidelity simulation
• Pluggable Storage: Configurable Avro backend with compression and rotation
• Builder Pattern: Type-safe configuration with compile-time sensor mode selection
• Async Runtime: Full Tokio integration for concurrent operations
• Graceful Shutdown: Signal handling with proper resource cleanup

Reliability Features:

• Connection Resilience: Automatic Zenoh session recovery and retry logic
• Data Validation: Schema-enforced deserialization with error handling
• File Rotation: Time/size-based file management with overflow protection
• Configuration Management: Centralized config with environment-specific defaults
• Comprehensive Logging: Structured tracing with configurable log levels

synapse-pmon: Data Flow and Type Architecture

The synapse-pmon system processes power quality measurements through a Zenoh-based distributed architecture, supporting both real-time sensor data and high-fidelity simulation modes.

Core Data Flow

PowerQualityData (simulation/sensor)
    ↓ [Zenoh Publisher]
Zenoh Topic: "sim_sensor/power_quality"
    ↓ [Network/IPC Transport]
Zenoh Subscriber (FIFO Channel)
    ↓ [Bincode Deserializer]
PowerQualityData Record
    ↓ [Avro Writer Backend]
Compressed Avro Files (ZSTD)

Key Data Types

#![allow(unused)]
fn main() {
// Core power quality measurement structure
#[derive(Debug, Serialize, Deserialize, AvroSchema, Encode, Decode)]
pub struct PowerQualityData {
    pub time_us: i64,                           // Microsecond timestamp
    pub freq: f32,                              // System frequency (Hz)
    pub phase_a: Measurements,                  // Phase A measurements
    pub phase_b: Option<Measurements>,          // Phase B (optional)
    pub phase_c: Option<Measurements>,          // Phase C (optional)
    pub nirms: f32,                            // Neutral current RMS
}

// Individual phase measurements
#[derive(Debug, Serialize, Deserialize, AvroSchema, Encode, Decode)]
pub struct Measurements {
    pub vrms: f32,   // RMS voltage
    pub irms: f32,   // RMS current  
    pub watt: f32,   // Active power
    pub fvrms: f32,  // Fundamental RMS voltage
    pub firms: f32,  // Fundamental RMS current
    pub fwatt: f32,  // Fundamental active power
    pub fvar: f32,   // Fundamental reactive power
}

// System configuration variants
#[derive(Debug, Serialize, Deserialize, Default, AvroSchema)]
pub enum PhaseVariants {
    Single,    // Single-phase system
    Two,       // Two-phase system
    #[default]
    Three,     // Three-phase system
}
}

Processing Architecture

1. Data Ingestion

• Zenoh Publisher: Publishes PowerQualityData to "sim_sensor/power_quality" topic
• FIFO Channel: Buffered subscriber with configurable queue depth (default: 1024)
• Session Management: Automatic connection handling with retry policies

2. Data Transformation

• Bincode Deserialization: High-performance binary deserialization from Zenoh payload
• Schema Validation: Compile-time type safety with Avro schema enforcement
• Error Recovery: Robust error handling with detailed logging and continue-on-error semantics

3. Storage Backend

• Avro Writer: Schema-based columnar storage with ZSTD compression
• File Rotation: Configurable policies (time, size, count) for data management
• Directory Structure: Organized output with configurable base directories

Sensor Modes and Builder Pattern

#![allow(unused)]
fn main() {
// Type-safe builder pattern with sensor mode markers
impl<W> PowerMonitorBuilder<W, RealSensor> {
    pub fn new() -> Self { /* Real sensor configuration */ }
    pub fn with_sim(self) -> PowerMonitorBuilder<W, SimulationSensor> { /* Switch to simulation */ }
}

impl<W> PowerMonitorBuilder<W, SimulationSensor> {
    pub async fn build(self) -> Result<PowerMonitor<W>, PowerMonitorError> {
        // Includes sensor simulation setup
    }
}

// Runtime sensor selection
pub struct PowerMonitor<W: WriteBackend> {
    zenoh_session: Arc<Session>,
    sub_channel: Option<Subscriber<FifoChannelHandler<Sample>>>,
    writer: W,
    sensor: Option<Sensor<PmonSimSensor<'static>>>,    // Only for simulation
    sensor_handle: Option<tokio::task::JoinHandle<()>>, // Async sensor task
    is_simulation: bool,
    is_connection_terminated: bool,
}
}

Zenoh Configuration

#![allow(unused)]
fn main() {
// Default Zenoh session configuration
let mut config = zenoh::config::Config::default();

config.insert_json5("connect/endpoints", r#"["tcp/127.0.0.1:7447"]"#)?;
config.insert_json5("connect/timeout_ms", r#"{"peer": -1}"#)?;

// Topic-based subscription
let subscriber = session
    .declare_subscriber("sim_sensor/power_quality")
    .with(FifoChannel::new(1024))
    .wait()?;
}

Error Handling & Performance

• Connection Management: Automatic Zenoh session recovery with exponential backoff
• Data Integrity: Schema validation prevents malformed data from corrupting storage
• Async Processing: Non-blocking I/O with proper task isolation
• Resource Cleanup: Graceful shutdown with proper resource deallocation
• Memory Management: Bounded channels prevent memory exhaustion under load

Configuration Examples

#![allow(unused)]
fn main() {
// Writer configuration with rotation
let writer_config = WriterConfig::<AvroConfig>::builder()
    .with_writer_config(AvroConfig::default())
    .with_compression(CompressionConfig::default())
    .with_file_rotation(RotationPolicy::Size(1500))      // Rotate at 1500 bytes
    // .with_file_rotation(RotationPolicy::Time(Duration::from_secs(60)))  // Time-based
    .with_base_dir(PathBuf::from("./data"))
    .build()?;

// Simulation sensor setup
let sensor = SensorBuilder::<PmonSimSensor, PowerQualityGenerator>::new()
    .with_config(SynapseConfig::new_with_defaults())
    .with_data_generator(PowerQualityGenerator)
    .build()?;
}

Binary Targets

pmon (Power Monitor):

• Production binary for real sensor data processing
• Configured for continuous operation with proper signal handling
• Supports both simulation and real sensor modes

sim (Simulation):

• Development/testing binary with built-in pub/sub demonstration
• Shows both publisher and subscriber behavior in single process
• Useful for system validation and integration testing

Integration Points

• System Service: Systemd integration for production deployment
• Config Management: Centralized configuration with environment overrides
• Monitoring: Structured logging with configurable verbosity levels