Trace API

The Zelos Cloud Trace API provides a way to collect and analyze structured trace events from your application.

Quick Start

import zelos_sdk

zelos_sdk.init()

source = zelos_sdk.TraceSource("my_source")
source.log("my_event", {"my_data": 123.456})

Event Streaming Example

import time
import random
import zelos_sdk

zelos_sdk.init()

source = zelos_sdk.TraceSource("motor")

while True:
    source.log("stats", {
        "rpm": random.uniform(1000, 5000),
        "torque": random.uniform(10, 50),
        "battery_voltage": random.uniform(46, 54),
    })
    time.sleep(1)

Core Components

TraceSource

Represents a logical source of events within your application.

You can instantiate any number of TraceSource objects to help contextualize events.

Example: Tracing data from a motor subcomponent

import time
import zelos_sdk

zelos_sdk.init()

# Create a source with a name
source = zelos_sdk.TraceSource("motor")

# Log events on the source
source.log(...)

TraceWriter

The TraceWriter class enables capturing trace events to local files for offline analysis, debugging, or archival purposes. It provides a context manager interface for automatic resource management.

import time
import datetime
import zelos_sdk

# Initialize the SDK
zelos_sdk.init()

source = zelos_sdk.TraceSource("my_source")

# Prefix the trace file with the current time
now = datetime.datetime.now().strftime('%Y%m%d_%H%M%S')

# Method 1: Using context manager
with zelos_sdk.TraceWriter(f"{now}_context_managed.trz") as writer:
    for i in range(10):
        source.log("sensor_data", {
            "temperature": 25.0,
            "humidity": 50.0,
        })
        time.sleep(0.1)

# Method 2: Using open/close
try:
    writer = zelos_sdk.TraceWriter(f"{now}_open_close.trz")
    writer.open()

    # Log events - they will be captured to the file
    for i in range(10):
        source.log("sensor_data", {
            "temperature": 25.0,
            "humidity": 50.0,
        })
        time.sleep(0.1)
finally:
    writer.close()

TraceSourceEvent

Create TraceSourceEvent objects with the TraceSource add_event method

TraceSourceEvent objects are also accessible through the corresponding source object's get_event method.

import time
import random
import zelos_sdk

zelos_sdk.init()

source = zelos_sdk.TraceSource("motor")

# Build your strongly-typed event with unit metadata
motor_stats = source.add_event("stats", [
    zelos_sdk.TraceEventFieldMetadata("rpm", zelos_sdk.DataType.Float64),
    zelos_sdk.TraceEventFieldMetadata("torque", zelos_sdk.DataType.Float64, "Nm"),
    zelos_sdk.TraceEventFieldMetadata("battery_voltage", zelos_sdk.DataType.Float32, "V"),
])

while True:
    motor_stats.log(
        rpm=random.uniform(1000, 5000),
        torque=random.uniform(10, 50),
        battery_voltage=random.uniform(46, 54),
    )
    time.sleep(1)

TracePublishClient

Manages connection to a trace ingestion service and handles batching and delivery of events.

# Create with default settings
client = zelos_sdk.TracePublishClient()

# Create with custom configuration
config = zelos_sdk.TracePublishClientConfig(
    url="grpc://localhost:2300",
)
client = zelos_sdk.TracePublishClient(config)

Advanced Usage

Enumeration, Value Table support

import time
from enum import IntEnum
import zelos_sdk

# Define your enumeration
class State(IntEnum):
    ON = 0
    OFF = 1

# Initialize the sdk
zelos_sdk.init()

# Create a source
source = zelos_sdk.TraceSource("my_device")

# Add an event with a state field
state_machine_event = source.add_event("state_machine", [
    zelos_sdk.TraceEventFieldMetadata("state", zelos_sdk.DataType.Int8)
])

# Add value table for the state field to map integer values to human-readable strings
source.add_value_table("state_machine", "state", {state.value: state.name for state in State})

def run_state_machine(current_state):
    match current_state:
        case State.ON:
            return State.OFF
        case State.OFF:
            return State.ON
        case _:
            raise ValueError("Unknown State!")

state = State.OFF
while 1:
    state = run_state_machine(state)
    state_machine_event.log(state=int(state))
    time.sleep(1)

Attribute style access

After defining an event, use attribute-style access for something a bit more Pythonic.

import time
import random
import zelos_sdk

zelos_sdk.init()

source = zelos_sdk.TraceSource("motor")

# Build your strongly-typed event with unit metadata
source.add_event("stats", [
    zelos_sdk.TraceEventFieldMetadata("rpm", zelos_sdk.DataType.Float64),
    zelos_sdk.TraceEventFieldMetadata("torque", zelos_sdk.DataType.Float64, "Nm"),
    zelos_sdk.TraceEventFieldMetadata("battery_voltage", zelos_sdk.DataType.Float32, "V"),
])

# Add a submessage
source.add_event("stats/averages", [
    zelos_sdk.TraceEventFieldMetadata("rpm_avg", zelos_sdk.DataType.Float64),
    zelos_sdk.TraceEventFieldMetadata("torque_avg", zelos_sdk.DataType.Float64, "Nm"),
    zelos_sdk.TraceEventFieldMetadata("battery_voltage_avg", zelos_sdk.DataType.Float32, "V"),
])

while True:
    source.stats.log(
        rpm=random.uniform(1000, 5000),
        torque=random.uniform(10, 50),
        battery_voltage=random.uniform(46, 54),
    )

    # Note '/' replaced with '.' for submessages
    source.stats.averages.log(
        rpm_avg=random.uniform(1000, 5000),
        torque_avg=random.uniform(10, 50),
        battery_voltage_avg=random.uniform(46, 54),
    )
    time.sleep(1)

Initialization with Configuration

# Initialize with custom configuration
zelos_sdk.init(
    "my_application",
    client_config=zelos_sdk.TracePublishClientConfig(
        url="grpc://localhost:2300",
    ),
    log_level="debug"
)

Custom Timestamps

# log event with a specific timestamp (nanoseconds since epoch)
timestamp_ns = int(time.time_ns())
motor_stats.log_at(timestamp_ns,
    rpm=3250.5,
    torque=42.8,
    battery_voltage=48.2,
    motor_temp=62.3
)

Field Types

import time
import zelos_sdk

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

# Define event with all available field types
comprehensive_event = source.add_event("status", [
    # Boolean type
    zelos_sdk.TraceEventFieldMetadata("fault_detected", zelos_sdk.DataType.Boolean),

    # Signed integer types
    zelos_sdk.TraceEventFieldMetadata("temperature_offset", zelos_sdk.DataType.Int8, "celsius"),
    zelos_sdk.TraceEventFieldMetadata("pressure_delta", zelos_sdk.DataType.Int16, "Pa"),
    zelos_sdk.TraceEventFieldMetadata("error_code", zelos_sdk.DataType.Int32),
    zelos_sdk.TraceEventFieldMetadata("timestamp_offset", zelos_sdk.DataType.Int64, "ns"),

    # Unsigned integer types
    zelos_sdk.TraceEventFieldMetadata("sensor_id", zelos_sdk.DataType.UInt8),
    zelos_sdk.TraceEventFieldMetadata("sample_count", zelos_sdk.DataType.UInt16),
    zelos_sdk.TraceEventFieldMetadata("total_readings", zelos_sdk.DataType.UInt32),
    zelos_sdk.TraceEventFieldMetadata("device_serial", zelos_sdk.DataType.UInt64),

    # Floating point types
    zelos_sdk.TraceEventFieldMetadata("temperature", zelos_sdk.DataType.Float32, "celsius"),
    zelos_sdk.TraceEventFieldMetadata("pressure", zelos_sdk.DataType.Float64, "Pa"),
    zelos_sdk.TraceEventFieldMetadata("voltage", zelos_sdk.DataType.Float64, "V"),
    zelos_sdk.TraceEventFieldMetadata("current", zelos_sdk.DataType.Float64, "A"),

    # Timestamp
    zelos_sdk.TraceEventFieldMetadata("measurement_time", zelos_sdk.DataType.TimestampNs),

    # String and binary data
    zelos_sdk.TraceEventFieldMetadata("device_name", zelos_sdk.DataType.String),
    zelos_sdk.TraceEventFieldMetadata("calibration_data", zelos_sdk.DataType.Binary),
])

Mock data

import time
import zelos_sdk

zelos_sdk.init()

mocks = [
    zelos_sdk.mock.MockBmsSource(),
    zelos_sdk.mock.MockSensorSource(),
    zelos_sdk.mock.MockInverterSource(),
    zelos_sdk.mock.MockDcdcSource(),
]

while 1:
    for mock in mocks:
        mock.generate()
    time.sleep(0.1)

Example: Logging Thermal Zones on Linux

Linux supports reading temperature sensors via sysfs. On a linux system, install uv and save the file below as temp_linux.py and run uv run temp_linux.py to start streaming data live. uv run --script <script_name>.py

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.10"
# dependencies = [ "zelos-sdk" ]
# ///
from pathlib import Path
import time

import zelos_sdk

zelos_sdk.init()

# Find all of our thermal zones from sysfs
thermal_zones = list(Path("/sys/class/thermal").glob("thermal_zone*"))

# Once per second, read the zone info and log it
src = zelos_sdk.TraceSource("linux_thermal")
while True:
    for zone_path in thermal_zones:
        # Read the zone info from sysfs
        zone_type = (zone_path / "type").read_text().strip()
        zone_temp_c = int((zone_path / "temp").read_text().strip()) / 1000.0

        # Log to the zelos sdk
        src.log_dict(zone_path.name, {"type": zone_type, "temp_c": zone_temp_c})

    time.sleep(1)