Skip to content

Zelos SDK

Build data pipelines from your code to the Zelos App with minimal overhead and maximum flexibility.

What You Can Build

  • Real-time Monitoring

    Stream telemetry, metrics, and events from any application

  • Data Recording

    Capture traces for debugging and post-mortem analysis

  • System Control

    Create actions to control your systems from the Zelos App

  • Automated Testing

    Validate behavior with data assertions and trace analysis

Architecture

graph TB
    subgraph "Your Application"
        A[Your Code]
        B[Zelos SDK]
    end

    subgraph "Zelos"
        D[Zelos Agent]
        F[Visualization]
        G[Trace Files]
    end

    A --> B
    B --> D
    D --> F
    D --> G

Language Support

import zelos_sdk

# Initialize once
zelos_sdk.init()

# Create logical sources
motor = zelos_sdk.TraceSource("motor_controller")
sensors = zelos_sdk.TraceSource("sensor_array")

# Log structured events
motor.log("status", {
    "state": "running",
    "rpm": 1500,
    "torque_nm": 25.5,
    "efficiency_percent": 92.3
})

use tokio_util::sync::CancellationToken;
use zelos::{TraceRouter, TraceSource};
use zelos::trace_grpc::publish::{TracePublishClient, TracePublishClientConfig};

// Set up the trace infrastructure
let cancel = CancellationToken::new();
let (router, router_task) = TraceRouter::new(cancel.clone());
tokio::spawn(router_task);

// Connect to agent
let config = TracePublishClientConfig::default();
let (client, client_task) = TracePublishClient::new(router.clone(), config);
tokio::spawn(client_task);

// Create sources and events
let motor = TraceSource::new("motor_controller", router.sender());

let status_event = motor
    .build_event("status")
    .add_string_field("state", None)
    .add_f64_field("rpm", None)
    .add_f64_field("torque_nm", Some("Nm".to_string()))
    .build()?;

// Emit events
status_event.build()
    .try_insert_string("state", "running".to_string())?
    .try_insert_f64("rpm", 1500.0)?
    .try_insert_f64("torque_nm", 25.5)?
    .emit()?;

import (
    zelos "github.com/zeloscloud/zelos/go"
)

// Initialize the router
router, sender, receiver := zelos.NewTraceRouter(ctx)

// Connect to agent
config := zelos.DefaultTracePublishClientConfig()
client := zelos.NewTracePublishClient(ctx, receiver, config)
go client.Run()

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

// Define event schemas
statusEvent, _ := motor.BuildEvent("status").
    AddStringField("state", nil).
    AddFloat64Field("rpm", nil).
    AddFloat64Field("torque_nm", nil).
    Build()

// Log events
statusEvent.Build().
    TryInsertString("state", "running").
    TryInsertFloat64("rpm", 1500.0).
    TryInsertFloat64("torque_nm", 25.5).
    Emit()

Testing Integration

The SDK includes first-class support for hardware-in-the-loop testing:

  • pytest integration for Python users
  • Time-aware assertions that understand hardware timing
  • Automatic trace recording during test execution
# Write expressive hardware tests
def test_motor_acceleration(motor, check):
    """Verify motor reaches target speed within spec."""
    check.that(
        motor.status.rpm,
        "is greater than", 2000,
        within_duration_s=0.5
    )

Next Steps

Resources