How to Stream Data

Stream data to Zelos at any rate: steady, in bursts, from many sources, or with your own timestamps.

Basic Streaming

The simplest way to stream data is to log events in a loop:

PythonRustGo

import time
import random  # For demo purposes
import zelos_sdk

zelos_sdk.init()
source = zelos_sdk.TraceSource("sensors")

# Define schema (optional but recommended)
temperature_event = source.add_event("temperature", [
    zelos_sdk.TraceEventFieldMetadata("value", zelos_sdk.DataType.Float64, "°C")
])

while True:
    # Read your sensor (simulated here)
    temperature = 20.0 + random.uniform(-2, 2)

    # Log to Zelos
    temperature_event.log(value=temperature)

    # Control rate
    time.sleep(1.0)  # 1 Hz

use std::time::Duration;
use tokio::time;
use rand::Rng;  // For demo purposes

// Define schema (required in Rust)
let temp_event = source
    .build_event("temperature")
    .add_f64_field("value", Some("°C".to_string()))
    .build()?;

let mut interval = time::interval(Duration::from_secs(1));
let mut rng = rand::thread_rng();

loop {
    interval.tick().await;

    // Read your sensor (simulated here)
    let temperature = 20.0 + rng.gen_range(-2.0..2.0);

    temp_event.build()
        .try_insert_f64("value", temperature)?
        .emit()?;
}

import (
    "math/rand"
    "time"
)

// Define schema (required in Go)
unitC := "°C"
tempEvent, _ := source.BuildEvent("temperature").
    AddFloat64Field("value", &unitC).
    Build()

ticker := time.NewTicker(time.Second)
defer ticker.Stop()

for range ticker.C {
    // Read your sensor (simulated here)
    temperature := 20.0 + rand.Float64()*4 - 2

    builder, _ := tempEvent.Build().TryInsertFloat64("value", temperature)
    builder.Emit()
}

High-Frequency Streaming

Compensate for loop overhead so the rate does not drift:

PythonRustGo

import time
import math
import zelos_sdk

zelos_sdk.init()
source = zelos_sdk.TraceSource("high_freq")

# Define schema for better performance
data_event = source.add_event("data", [
    zelos_sdk.TraceEventFieldMetadata("value", zelos_sdk.DataType.Float64, "V")
])

# 1 kHz = 1ms period
period = 0.001
start_time = time.time()
next_time = start_time

while True:
    current_time = time.time() - start_time

    # Generate signal (example: 100 Hz sine wave)
    value = math.sin(2 * math.pi * 100 * current_time)

    # Log data
    data_event.log(value=value)

    # Precise sleep to maintain rate
    next_time += period
    sleep_duration = next_time - time.time()
    if sleep_duration > 0:
        time.sleep(sleep_duration)

use tokio::time::{interval, MissedTickBehavior};
use std::time::{Duration, Instant};

let data_event = source
    .build_event("data")
    .add_f64_field("value", Some("V".to_string()))
    .build()?;

// 1 kHz = 1ms period
let mut interval = interval(Duration::from_millis(1));
// Skip missed ticks instead of bursting
interval.set_missed_tick_behavior(MissedTickBehavior::Skip);

let start = Instant::now();

loop {
    interval.tick().await;

    let elapsed = start.elapsed().as_secs_f64();
    // Generate signal (example: 100 Hz sine wave)
    let value = (2.0 * std::f64::consts::PI * 100.0 * elapsed).sin();

    data_event.build()
        .try_insert_f64("value", value)?
        .emit()?;
}

import (
    "math"
    "time"
)

unitVolt := "V"
dataEvent, _ := source.BuildEvent("data").
    AddFloat64Field("value", &unitVolt).
    Build()

// 1 kHz = 1ms period
ticker := time.NewTicker(time.Millisecond)
defer ticker.Stop()

start := time.Now()

for range ticker.C {
    elapsed := time.Since(start).Seconds()
    // Generate signal (example: 100 Hz sine wave)
    value := math.Sin(2 * math.Pi * 100 * elapsed)

    builder, _ := dataEvent.Build().TryInsertFloat64("value", value)
    builder.Emit()
}

Burst Streaming

Stream data in bursts when events occur:

PythonRustGo

import time
import zelos_sdk

zelos_sdk.init()
source = zelos_sdk.TraceSource("events")

# Define event schemas
source.add_event("event_start", [
    zelos_sdk.TraceEventFieldMetadata("trigger", zelos_sdk.DataType.String),
    zelos_sdk.TraceEventFieldMetadata("timestamp_ms", zelos_sdk.DataType.Int64)
])

source.add_event("event_sample", [
    zelos_sdk.TraceEventFieldMetadata("value", zelos_sdk.DataType.Float64),
    zelos_sdk.TraceEventFieldMetadata("index", zelos_sdk.DataType.UInt32)
])

source.add_event("event_end", [
    zelos_sdk.TraceEventFieldMetadata("duration_ms", zelos_sdk.DataType.Float64),
    zelos_sdk.TraceEventFieldMetadata("sample_count", zelos_sdk.DataType.UInt32)
])

def handle_burst_event(trigger_name: str, samples: list):
    """Log a burst of data when an event occurs"""
    start_time = time.time()

    # Log event start
    source.log("event_start", {
        "trigger": trigger_name,
        "timestamp_ms": int(start_time * 1000)
    })

    # Log burst of samples
    for i, value in enumerate(samples):
        source.log("event_sample", {
            "value": value,
            "index": i
        })

    # Log event end
    duration_ms = (time.time() - start_time) * 1000
    source.log("event_end", {
        "duration_ms": duration_ms,
        "sample_count": len(samples)
    })

# Example usage
while True:
    # Wait for trigger condition
    time.sleep(5)

    # Generate burst data (simulated)
    burst_data = [i * 0.1 for i in range(100)]
    handle_burst_event("threshold_exceeded", burst_data)

use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use tokio::time;

// Define event schemas
let start_event = source
    .build_event("event_start")
    .add_string_field("trigger", None)
    .add_i64_field("timestamp_ms", None)
    .build()?;

let sample_event = source
    .build_event("event_sample")
    .add_f64_field("value", None)
    .add_u32_field("index", None)
    .build()?;

let end_event = source
    .build_event("event_end")
    .add_f64_field("duration_ms", None)
    .add_u32_field("sample_count", None)
    .build()?;

async fn handle_burst_event(
    trigger_name: &str,
    samples: Vec<f64>,
    start_event: &TraceEvent,
    sample_event: &TraceEvent,
    end_event: &TraceEvent,
) -> Result<()> {
    let start_time = Instant::now();

    // Log event start
    start_event.build()
        .try_insert_string("trigger", trigger_name.to_string())?
        .try_insert_i64(
            "timestamp_ms",
            SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_millis() as i64,
        )?
        .emit()?;

    // Log burst of samples
    for (i, value) in samples.iter().enumerate() {
        sample_event.build()
            .try_insert_f64("value", *value)?
            .try_insert_u32("index", i as u32)?
            .emit()?;
    }

    // Log event end
    let duration_ms = start_time.elapsed().as_secs_f64() * 1000.0;
    end_event.build()
        .try_insert_f64("duration_ms", duration_ms)?
        .try_insert_u32("sample_count", samples.len() as u32)?
        .emit()?;

    Ok(())
}

// Example usage
loop {
    // Wait for trigger condition
    time::sleep(Duration::from_secs(5)).await;

    // Generate burst data (simulated)
    let burst_data: Vec<f64> = (0..100).map(|i| i as f64 * 0.1).collect();
    handle_burst_event("threshold_exceeded", burst_data, &start_event, &sample_event, &end_event).await?;
}

import (
    "time"
)

// Define event schemas
startEvent, _ := source.BuildEvent("event_start").
    AddStringField("trigger", nil).
    AddInt64Field("timestamp_ms", nil).
    Build()

sampleEvent, _ := source.BuildEvent("event_sample").
    AddFloat64Field("value", nil).
    AddUint32Field("index", nil).
    Build()

endEvent, _ := source.BuildEvent("event_end").
    AddFloat64Field("duration_ms", nil).
    AddUint32Field("sample_count", nil).
    Build()

func handleBurstEvent(triggerName string, samples []float64) error {
    startTime := time.Now()

    // Log event start
    builder, _ := startEvent.Build().TryInsertString("trigger", triggerName)
    builder, _ = builder.TryInsertInt64("timestamp_ms", startTime.UnixMilli())
    if err := builder.Emit(); err != nil {
        return err
    }

    // Log burst of samples
    for i, value := range samples {
        b, _ := sampleEvent.Build().TryInsertFloat64("value", value)
        b, _ = b.TryInsertUint32("index", uint32(i))
        if err := b.Emit(); err != nil {
            return err
        }
    }

    // Log event end
    durationMs := float64(time.Since(startTime).Milliseconds())
    builder, _ = endEvent.Build().TryInsertFloat64("duration_ms", durationMs)
    builder, _ = builder.TryInsertUint32("sample_count", uint32(len(samples)))
    return builder.Emit()
}

// Example usage
for {
    // Wait for trigger condition
    time.Sleep(5 * time.Second)

    // Generate burst data (simulated)
    burstData := make([]float64, 100)
    for i := range burstData {
        burstData[i] = float64(i) * 0.1
    }

    if err := handleBurstEvent("threshold_exceeded", burstData); err != nil {
        log.Printf("Error handling burst: %v", err)
    }
}

Multiple Sources in Parallel

Stream from different components at different rates:

PythonRustGo

import threading
import time
import random
import zelos_sdk

zelos_sdk.init()

def stream_motor():
    """100 Hz motor telemetry"""
    source = zelos_sdk.TraceSource("motor")

    # Define schema
    telemetry = source.add_event("telemetry", [
        zelos_sdk.TraceEventFieldMetadata("rpm", zelos_sdk.DataType.Float64, "rpm"),
        zelos_sdk.TraceEventFieldMetadata("torque", zelos_sdk.DataType.Float64, "Nm")
    ])

    while True:
        telemetry.log(
            rpm=2000 + random.uniform(-100, 100),
            torque=50 + random.uniform(-5, 5)
        )
        time.sleep(0.01)  # 100 Hz

def stream_battery():
    """1 Hz battery status"""
    source = zelos_sdk.TraceSource("battery")

    # Define schema
    status = source.add_event("status", [
        zelos_sdk.TraceEventFieldMetadata("voltage", zelos_sdk.DataType.Float64, "V"),
        zelos_sdk.TraceEventFieldMetadata("current", zelos_sdk.DataType.Float64, "A"),
        zelos_sdk.TraceEventFieldMetadata("soc", zelos_sdk.DataType.Float64, "%")
    ])

    soc = 85.0  # State of charge

    while True:
        # Simulate battery discharge
        soc = max(20.0, soc - 0.1)

        status.log(
            voltage=48.0 + random.uniform(-0.5, 0.5),
            current=random.uniform(-10, 50),
            soc=soc
        )
        time.sleep(1.0)  # 1 Hz

def stream_gps():
    """10 Hz GPS updates"""
    source = zelos_sdk.TraceSource("gps")

    # Define schema
    position = source.add_event("position", [
        zelos_sdk.TraceEventFieldMetadata("lat", zelos_sdk.DataType.Float64, "deg"),
        zelos_sdk.TraceEventFieldMetadata("lon", zelos_sdk.DataType.Float64, "deg"),
        zelos_sdk.TraceEventFieldMetadata("alt", zelos_sdk.DataType.Float64, "m")
    ])

    # Base coordinates (Palo Alto, CA)
    base_lat, base_lon = 37.4419, -122.1430

    while True:
        position.log(
            lat=base_lat + random.uniform(-0.001, 0.001),
            lon=base_lon + random.uniform(-0.001, 0.001),
            alt=30 + random.uniform(-1, 1)
        )
        time.sleep(0.1)  # 10 Hz

# Run all streams in parallel
threads = [
    threading.Thread(target=stream_motor, daemon=True),
    threading.Thread(target=stream_battery, daemon=True),
    threading.Thread(target=stream_gps, daemon=True)
]

for t in threads:
    t.start()

# Keep main thread alive
try:
    for t in threads:
        t.join()
except KeyboardInterrupt:
    print("Stopping streams...")

use tokio;
use rand::Rng;
use std::time::Duration;

async fn stream_motor(source: TraceSource) -> Result<()> {
    let telemetry = source
        .build_event("telemetry")
        .add_f64_field("rpm", Some("rpm".to_string()))
        .add_f64_field("torque", Some("Nm".to_string()))
        .build()?;

    let mut interval = tokio::time::interval(Duration::from_millis(10)); // 100 Hz
    let mut rng = rand::thread_rng();

    loop {
        interval.tick().await;

        telemetry.build()
            .try_insert_f64("rpm", 2000.0 + rng.gen_range(-100.0..100.0))?
            .try_insert_f64("torque", 50.0 + rng.gen_range(-5.0..5.0))?
            .emit()?;
    }
}

async fn stream_battery(source: TraceSource) -> Result<()> {
    let status = source
        .build_event("status")
        .add_f64_field("voltage", Some("V".to_string()))
        .add_f64_field("current", Some("A".to_string()))
        .add_f64_field("soc", Some("%".to_string()))
        .build()?;

    let mut interval = tokio::time::interval(Duration::from_secs(1)); // 1 Hz
    let mut rng = rand::thread_rng();
    let mut soc = 85.0;

    loop {
        interval.tick().await;

        // Simulate battery discharge
        soc = (soc - 0.1).max(20.0);

        status.build()
            .try_insert_f64("voltage", 48.0 + rng.gen_range(-0.5..0.5))?
            .try_insert_f64("current", rng.gen_range(-10.0..50.0))?
            .try_insert_f64("soc", soc)?
            .emit()?;
    }
}

async fn stream_gps(source: TraceSource) -> Result<()> {
    let position = source
        .build_event("position")
        .add_f64_field("lat", Some("deg".to_string()))
        .add_f64_field("lon", Some("deg".to_string()))
        .add_f64_field("alt", Some("m".to_string()))
        .build()?;

    let mut interval = tokio::time::interval(Duration::from_millis(100)); // 10 Hz
    let mut rng = rand::thread_rng();

    // Base coordinates (Palo Alto, CA)
    let base_lat = 37.4419;
    let base_lon = -122.1430;

    loop {
        interval.tick().await;

        position.build()
            .try_insert_f64("lat", base_lat + rng.gen_range(-0.001..0.001))?
            .try_insert_f64("lon", base_lon + rng.gen_range(-0.001..0.001))?
            .try_insert_f64("alt", 30.0 + rng.gen_range(-1.0..1.0))?
            .emit()?;
    }
}

#[tokio::main]
async fn main() -> Result<()> {
    // Setup router and client (see quickstart for full setup)
    // ...

    // Create sources
    let motor = TraceSource::new("motor", router.sender());
    let battery = TraceSource::new("battery", router.sender());
    let gps = TraceSource::new("gps", router.sender());

    // Run all streams concurrently; the first error aborts the rest
    tokio::try_join!(
        stream_motor(motor),
        stream_battery(battery),
        stream_gps(gps),
    )?;
    Ok(())
}

import (
    "math"
    "math/rand"
    "sync"
    "time"
)

func streamMotor(source *zelos.TraceSource, wg *sync.WaitGroup) {
    defer wg.Done()

    unitRpm := "rpm"
    unitNm := "Nm"
    telemetry, _ := source.BuildEvent("telemetry").
        AddFloat64Field("rpm", &unitRpm).
        AddFloat64Field("torque", &unitNm).
        Build()

    ticker := time.NewTicker(10 * time.Millisecond) // 100 Hz
    defer ticker.Stop()

    for range ticker.C {
        builder, _ := telemetry.Build().
            TryInsertFloat64("rpm", 2000+rand.Float64()*200-100)
        builder, _ = builder.
            TryInsertFloat64("torque", 50+rand.Float64()*10-5)
        builder.Emit()
    }
}

func streamBattery(source *zelos.TraceSource, wg *sync.WaitGroup) {
    defer wg.Done()

    unitV := "V"
    unitA := "A"
    unitPct := "%"
    status, _ := source.BuildEvent("status").
        AddFloat64Field("voltage", &unitV).
        AddFloat64Field("current", &unitA).
        AddFloat64Field("soc", &unitPct).
        Build()

    ticker := time.NewTicker(time.Second) // 1 Hz
    defer ticker.Stop()

    soc := 85.0

    for range ticker.C {
        // Simulate battery discharge
        soc = math.Max(20.0, soc-0.1)

        builder, _ := status.Build().
            TryInsertFloat64("voltage", 48.0+rand.Float64()-0.5)
        builder, _ = builder.
            TryInsertFloat64("current", rand.Float64()*60-10)
        builder, _ = builder.
            TryInsertFloat64("soc", soc)
        builder.Emit()
    }
}

func streamGPS(source *zelos.TraceSource, wg *sync.WaitGroup) {
    defer wg.Done()

    unitDeg := "deg"
    unitM := "m"
    position, _ := source.BuildEvent("position").
        AddFloat64Field("lat", &unitDeg).
        AddFloat64Field("lon", &unitDeg).
        AddFloat64Field("alt", &unitM).
        Build()

    ticker := time.NewTicker(100 * time.Millisecond) // 10 Hz
    defer ticker.Stop()

    // Base coordinates (Palo Alto, CA)
    baseLat := 37.4419
    baseLon := -122.1430

    for range ticker.C {
        builder, _ := position.Build().
            TryInsertFloat64("lat", baseLat+rand.Float64()*0.002-0.001)
        builder, _ = builder.
            TryInsertFloat64("lon", baseLon+rand.Float64()*0.002-0.001)
        builder, _ = builder.
            TryInsertFloat64("alt", 30+rand.Float64()*2-1)
        builder.Emit()
    }
}

func main() {
    // Setup router and client (see quickstart for full setup)
    // ...

    // Create sources
    motor, _ := zelos.NewTraceSource("motor", sender)
    battery, _ := zelos.NewTraceSource("battery", sender)
    gps, _ := zelos.NewTraceSource("gps", sender)
    defer motor.Close()
    defer battery.Close()
    defer gps.Close()

    // Run all streams in parallel
    var wg sync.WaitGroup
    wg.Add(3)

    go streamMotor(motor, &wg)
    go streamBattery(battery, &wg)
    go streamGPS(gps, &wg)

    wg.Wait()
}

Stream with Custom Timestamps

Use specific timestamps for replay or synchronization:

PythonRustGo

import time
import zelos_sdk

zelos_sdk.init()
source = zelos_sdk.TraceSource("replay")

# Define schema
measurement = source.add_event("measurement", [
    zelos_sdk.TraceEventFieldMetadata("value", zelos_sdk.DataType.Float64)
])

# Method 1: Current timestamp (automatic)
measurement.log(value=1.0)

# Method 2: Specific timestamp using log_at
specific_time_ns = 1699564234567890123
measurement.log_at(specific_time_ns, value=2.0)

# Method 3: Calculated timestamp
past_time_ns = time.time_ns() - (60 * 1_000_000_000)  # 1 minute ago
measurement.log_at(past_time_ns, value=3.0)

# Method 4: Synchronized timestamps
def stream_synchronized(sync_offset_ns: int):
    """Stream with synchronized timestamps"""
    while True:
        # Calculate synchronized timestamp
        timestamp_ns = time.time_ns() + sync_offset_ns

        # Read sensor value
        value = read_sensor()  # Your sensor reading function

        # Log with specific timestamp
        measurement.log_at(timestamp_ns, value=value)

        time.sleep(0.1)

# Method 5: Replay historical data
def replay_historical_data(records: list):
    """Replay data with original timestamps"""
    for record in records:
        # Use the original timestamp from the record
        measurement.log_at(
            record['timestamp_ns'],
            value=record['value']
        )

        # Optional: add delay to simulate real-time replay
        time.sleep(0.001)

use zelos_trace::time::now_time_ns;
use std::time::Duration;
use tokio::time;

let measurement = source
    .build_event("measurement")
    .add_f64_field("value", None)
    .build()?;

// Method 1: Current timestamp (automatic)
measurement.build()
    .try_insert_f64("value", 1.0)?
    .emit()?;

// Method 2: Specific timestamp using emit_at
let specific_time_ns = 1699564234567890123i64;
measurement.build()
    .try_insert_f64("value", 2.0)?
    .emit_at(specific_time_ns)?;

// Method 3: Calculated timestamp
let past_time_ns = now_time_ns() - (60 * 1_000_000_000);
measurement.build()
    .try_insert_f64("value", 3.0)?
    .emit_at(past_time_ns)?;

// Method 4: Synchronized timestamps
async fn stream_synchronized(
    sync_offset_ns: i64,
    measurement: &TraceEvent
) -> Result<()> {
    let mut interval = time::interval(Duration::from_millis(100));

    loop {
        interval.tick().await;

        // Calculate synchronized timestamp
        let timestamp_ns = now_time_ns() + sync_offset_ns;

        // Read sensor value
        let value = read_sensor(); // Your sensor reading function

        // Log with specific timestamp
        measurement.build()
            .try_insert_f64("value", value)?
            .emit_at(timestamp_ns)?;
    }
}

// Method 5: Replay historical data
struct Record {
    timestamp_ns: i64,
    value: f64,
}

async fn replay_historical_data(
    records: Vec<Record>,
    measurement: &TraceEvent
) -> Result<()> {
    for record in records {
        // Use the original timestamp from the record
        measurement.build()
            .try_insert_f64("value", record.value)?
            .emit_at(record.timestamp_ns)?;

        // Optional: add delay to simulate real-time replay
        time::sleep(Duration::from_millis(1)).await;
    }
    Ok(())
}

import (
    "time"
)

measurement, _ := source.BuildEvent("measurement").
    AddFloat64Field("value", nil).
    Build()

// Method 1: Current timestamp (automatic)
builder, _ := measurement.Build().TryInsertFloat64("value", 1.0)
builder.Emit()

// Method 2: Specific timestamp using EmitAt
specificTimeNs := int64(1699564234567890123)
builder, _ = measurement.Build().TryInsertFloat64("value", 2.0)
builder.EmitAt(specificTimeNs)

// Method 3: Calculated timestamp
pastTimeNs := time.Now().Add(-1 * time.Minute).UnixNano()
builder, _ = measurement.Build().TryInsertFloat64("value", 3.0)
builder.EmitAt(pastTimeNs)

// Method 4: Synchronized timestamps
func streamSynchronized(syncOffsetNs int64, measurement *zelos.TraceSourceEvent) {
    ticker := time.NewTicker(100 * time.Millisecond)
    defer ticker.Stop()

    for range ticker.C {
        // Calculate synchronized timestamp
        timestampNs := time.Now().UnixNano() + syncOffsetNs

        // Read sensor value
        value := readSensor() // Your sensor reading function

        // Log with specific timestamp
        builder, _ := measurement.Build().TryInsertFloat64("value", value)
        builder.EmitAt(timestampNs)
    }
}

// Method 5: Replay historical data
type Record struct {
    TimestampNs int64
    Value       float64
}

func replayHistoricalData(records []Record, measurement *zelos.TraceSourceEvent) error {
    for _, record := range records {
        // Use the original timestamp from the record
        builder, _ := measurement.Build().
            TryInsertFloat64("value", record.Value)
        if err := builder.EmitAt(record.TimestampNs); err != nil {
            return err
        }

        // Optional: add delay to simulate real-time replay
        time.Sleep(time.Millisecond)
    }
    return nil
}

Performance Tips

1. Define Schemas Upfront

A declared schema fixes field names, types, and units up front, and skips the per-call event lookup:

# Good: Schema defined once, reused many times
event = source.add_event("data", [
    zelos_sdk.TraceEventFieldMetadata("value", zelos_sdk.DataType.Float64)
])
for i in range(1000000):
    event.log(value=i * 0.1)

# Less efficient: looks the event up by name on every call
for i in range(1000000):
    source.log("data", {"value": i * 0.1})

2. Use Appropriate Data Types

Choose the smallest data type that fits your needs:

# Use Int8 for small values (-128 to 127)
source.add_event("status", [
    zelos_sdk.TraceEventFieldMetadata("state", zelos_sdk.DataType.Int8)
])

# Use Float32 when Float64 precision isn't needed
source.add_event("sensor", [
    zelos_sdk.TraceEventFieldMetadata("temperature", zelos_sdk.DataType.Float32, "°C")
])

3. Batch Related Fields

Log related fields together to reduce overhead:

# Good: one event carries every field on one timestamp
imu.log(accel_x=ax, accel_y=ay, accel_z=az,
        gyro_x=gx, gyro_y=gy, gyro_z=gz)

# Less efficient: two events, and the readings no longer share a timestamp
accel.log(x=ax, y=ay, z=az)
gyro.log(x=gx, y=gy, z=gz)

4. Control Timing Precisely

For high-frequency streaming, use precise timing methods shown in the examples above rather than simple sleep() calls.

Common Patterns

State Machine Monitoring

state_event = source.add_event("state", [
    zelos_sdk.TraceEventFieldMetadata("current", zelos_sdk.DataType.UInt8),
    zelos_sdk.TraceEventFieldMetadata("previous", zelos_sdk.DataType.UInt8),
    zelos_sdk.TraceEventFieldMetadata("transition_time_ms", zelos_sdk.DataType.Float64)
])

# Add value table for readable state names
source.add_value_table("state", "current", {
    0: "IDLE", 1: "INIT", 2: "RUNNING", 3: "ERROR"
})
source.add_value_table("state", "previous", {
    0: "IDLE", 1: "INIT", 2: "RUNNING", 3: "ERROR"
})

Sensor Array Streaming

# Define schema for array of sensors
sensor_array = source.add_event("array", [
    zelos_sdk.TraceEventFieldMetadata(f"sensor_{i}", zelos_sdk.DataType.Float32)
    for i in range(16)
])

# Log all sensors with single timestamp
sensor_array.log(**{f"sensor_{i}": values[i] for i in range(16)})