How to Define Schemas

Schemas provide type safety, documentation, and validation for your trace events. Source, event and field names follow the name rules.

Dynamic vs Structured

Dynamic Events (Python only)

Python allows logging without pre-defined schemas:

import zelos_sdk

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

# Types are inferred from values
source.log("sensor", {
    "temperature": 25.5,    # → Float64
    "pressure": 1013,       # → Int64
    "status": "OK",         # → String
    "valid": True          # → Boolean
})

Use for: Quick prototyping, exploration Avoid for: Production code, APIs, shared data

Structured Events (recommended)

Define schemas for type safety and documentation:

PythonRustGo

import zelos_sdk

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

# Define schema with types and units
sensor_event = source.add_event("sensor", [
    zelos_sdk.TraceEventFieldMetadata("temperature", zelos_sdk.DataType.Float64, "°C"),
    zelos_sdk.TraceEventFieldMetadata("pressure", zelos_sdk.DataType.Int32, "Pa"),
    zelos_sdk.TraceEventFieldMetadata("status", zelos_sdk.DataType.String),
    zelos_sdk.TraceEventFieldMetadata("valid", zelos_sdk.DataType.Boolean)
])

# Type-safe logging
sensor_event.log(
    temperature=25.5,
    pressure=1013,
    status="OK",
    valid=True
)

// Rust requires schema definition
let sensor_event = source
    .build_event("sensor")
    .add_f64_field("temperature", Some("°C".to_string()))
    .add_i32_field("pressure", Some("Pa".to_string()))
    .add_string_field("status", None)
    .add_bool_field("valid", None)
    .build()?;

// Type-safe logging
sensor_event.build()
    .try_insert_f64("temperature", 25.5)?
    .try_insert_i32("pressure", 1013)?
    .try_insert_string("status", "OK".to_string())?
    .try_insert_bool("valid", true)?
    .emit()?;

// Go requires schema definition
unitC := "°C"
unitPa := "Pa"
sensorEvent, _ := source.BuildEvent("sensor").
    AddFloat64Field("temperature", &unitC).
    AddInt32Field("pressure", &unitPa).
    AddStringField("status", nil).
    AddBooleanField("valid", nil).
    Build()

// Type-safe logging
builder, _ := sensorEvent.Build().TryInsertFloat64("temperature", 25.5)
builder, _ = builder.TryInsertInt32("pressure", 1013)
builder, _ = builder.TryInsertString("status", "OK")
builder, _ = builder.TryInsertBoolean("valid", true)
builder.Emit()

Well-Known Event Schemas

Use a well-known schema when an event has standard Zelos semantics. In addition to defining the fields, the schema stamps a stable event-type identifier into the trace. Zelos uses that identifier to recognize logs, spans, check results, and annotations without relying on your source or event names.

import zelos_sdk
from zelos_sdk import schemas

zelos_sdk.init()
source = zelos_sdk.TraceSource("web_service")
application_log = source.add_event("application", schemas.Log)

application_log.log(
    level="info",
    message="health check passed",
    name="web_service",
    file="server.py",
    line=42,
)

application is the event name chosen by this producer; zelos.log.v1 is the schema identity supplied by schemas.Log. The class exposes the identifier as EVENT_TYPE and its field list as FIELDS.

The agent recognizes these well-known event types, and built-in workflows produce them. A ? marks a nullable field:

Event type	Purpose and fields
zelos.event.v1	Point event: name, status, data
zelos.span.v1	Duration: name, status, start_ns, end_ns, data
zelos.log.v1	Log record: level, message, name, file, line
zelos.check.result.v1	Check result: name, status, reason, op, temporal, lhs?, rhs?, message?, wall_ms, query_ms, count?, evidence_signal?, evidence_time_ns?, evidence_value?, range_start_ns?, range_end_ns?, duration_ns?
zelos.annotation.comment.v1	Annotation: name, status?, author?, text?, target_path?, range_start_ns?, range_end_ns?, tags?
zelos.can.frame.v1	Raw CAN frame: arbitration_id, is_extended, is_fd, is_rx?, dlc, data, is_remote_frame

Use the check and annotation APIs to create their event types. If you implement another compatible producer, register the exact canonical fields and pass the matching event_type= to add_event().

Well-known schemas are versioned contracts:

• The event name is application-defined and can differ between producers.
• time_ns is implicit and is not part of the declared fields.
• data and tags are strings. Encode structured content as JSON when needed.
• Do not change the fields behind an existing event-type identifier. Use a new version such as acme.motor.state.v2 for an incompatible schema change.

You can give a custom schema the same stable identity behavior:

class MotorState:
    EVENT_TYPE = "acme.motor.state.v1"
    FIELDS = [
        zelos_sdk.TraceEventFieldMetadata("state", zelos_sdk.DataType.String),
        zelos_sdk.TraceEventFieldMetadata("rpm", zelos_sdk.DataType.Float64, "rpm"),
    ]


motor_state = source.add_event("state", MotorState)
motor_state.log(state="running", rpm=3200.0)

Passing a class to add_event() reads its FIELDS and optional EVENT_TYPE class variables. A field-list schema remains untyped unless you pass event_type= explicitly.

Discover Typed Event Tables

List event tables by exact event type or by a trailing-* prefix:

zelos live events --event-type 'zelos.span.v1' --latest
zelos live events --event-type 'zelos.check.*'
zelos trace events capture.trz --event-type 'zelos.annotation.*'

zelos live events searches the live catalog with a 60-second lookback by default; change it with --lookback 5m. zelos trace events always searches the entire bounded trace and has no lookback option. Only a trailing * is a wildcard, so zelos.*.v1 does not match.

For the generic agent-tool endpoint, event_list has the same distinction: scope: {"kind": "live", "durationSeconds": 60} searches a live lookback, while scope: {"kind": "trace", "tracePaths": ["capture.trz"]} lists the whole trace.

Data Types

Choosing the Right Type

Type	Range	Use For	Memory
Int8	-128 to 127	State codes, small counters	1 byte
Int16	±32K	Sensor ADC values	2 bytes
Int32	±2.1B	Timestamps (seconds), IDs	4 bytes
Int64	±9.2×10¹⁸	Nanosecond timestamps	8 bytes
UInt8	0 to 255	Percentages, byte values	1 byte
UInt16	0 to 65K	Port numbers, ADC values	2 bytes
UInt32	0 to 4.2B	Counters, CAN IDs	4 bytes
UInt64	0 to 1.8×10¹⁹	Large counters, UUIDs	8 bytes
Float32	±3.4×10³⁸	Sensor data (7 digits)	4 bytes
Float64	±1.7×10³⁰⁸	Precise measurements (15 digits)	8 bytes
Boolean	true/false	Flags, states	1 byte
String	UTF-8 text	Text, identifiers	Variable
Binary	Raw bytes	Payloads, images	Variable
TimestampNs	Nanoseconds	Time points	8 bytes

Complete Example

PythonRustGo

# One field per data type
event = source.add_event("all_types", [
    # Choose smallest type that fits your data
    zelos_sdk.TraceEventFieldMetadata("state", zelos_sdk.DataType.UInt8),        # 0-255
    zelos_sdk.TraceEventFieldMetadata("temperature", zelos_sdk.DataType.Float32, "°C"),  # ±0.1°C precision
    zelos_sdk.TraceEventFieldMetadata("timestamp", zelos_sdk.DataType.TimestampNs),
    zelos_sdk.TraceEventFieldMetadata("message", zelos_sdk.DataType.String),
    zelos_sdk.TraceEventFieldMetadata("payload", zelos_sdk.DataType.Binary),
])

import time
event.log(
    state=3,
    temperature=25.5,
    timestamp=time.time_ns(),
    message="Sensor reading",
    payload=bytes([0xDE, 0xAD, 0xBE, 0xEF])
)

use zelos::trace::time::now_time_ns;

let event = source
    .build_event("all_types")
    .add_u8_field("state", None)
    .add_f32_field("temperature", Some("°C".to_string()))
    .add_timestamp_ns_field("timestamp", None)
    .add_string_field("message", None)
    .add_binary_field("payload", None)
    .build()?;

event.build()
    .try_insert_u8("state", 3)?
    .try_insert_f32("temperature", 25.5)?
    .try_insert_timestamp_ns("timestamp", now_time_ns())?
    .try_insert_string("message", "Sensor reading".to_string())?
    .try_insert_binary("payload", vec![0xDE, 0xAD, 0xBE, 0xEF])?
    .emit()?;

unitC := "°C"
event, _ := source.BuildEvent("all_types").
    AddUint8Field("state", nil).
    AddFloat32Field("temperature", &unitC).
    AddTimestampNsField("timestamp", nil).
    AddStringField("message", nil).
    AddBinaryField("payload", nil).
    Build()

builder, _ := event.Build().TryInsertUint8("state", 3)
builder, _ = builder.TryInsertFloat32("temperature", 25.5)
builder, _ = builder.TryInsertTimestampNs("timestamp", time.Now().UnixNano())
builder, _ = builder.TryInsertString("message", "Sensor reading")
builder, _ = builder.TryInsertBinary("payload", []byte{0xDE, 0xAD, 0xBE, 0xEF})
builder.Emit()

Units

Always specify units for physical quantities:

# Common units by domain
telemetry = source.add_event("telemetry", [
    # Mechanical
    zelos_sdk.TraceEventFieldMetadata("speed", zelos_sdk.DataType.Float64, "m/s"),
    zelos_sdk.TraceEventFieldMetadata("acceleration", zelos_sdk.DataType.Float64, "m/s²"),
    zelos_sdk.TraceEventFieldMetadata("torque", zelos_sdk.DataType.Float64, "Nm"),

    # Electrical
    zelos_sdk.TraceEventFieldMetadata("voltage", zelos_sdk.DataType.Float32, "V"),
    zelos_sdk.TraceEventFieldMetadata("current", zelos_sdk.DataType.Float32, "A"),
    zelos_sdk.TraceEventFieldMetadata("power", zelos_sdk.DataType.Float32, "W"),

    # Thermal
    zelos_sdk.TraceEventFieldMetadata("temperature", zelos_sdk.DataType.Float32, "°C"),

    # Dimensionless
    zelos_sdk.TraceEventFieldMetadata("efficiency", zelos_sdk.DataType.Float32, "%"),
    zelos_sdk.TraceEventFieldMetadata("gear", zelos_sdk.DataType.Int8),  # No unit
])

Enumerations

Map numeric values to readable strings using value tables. Register the event first: a value table attaches to a field that already exists. In Python, add_value_table() raises KeyError when the event is not registered.

PythonRustGo

from enum import IntEnum

class State(IntEnum):
    IDLE = 0
    RUNNING = 1
    ERROR = 2

# Define event
status = source.add_event("status", [
    zelos_sdk.TraceEventFieldMetadata("state", zelos_sdk.DataType.UInt8),
])

# Map values to strings for visualization
source.add_value_table("status", "state", {
    0: "IDLE",
    1: "RUNNING",
    2: "ERROR"
})

# Log using enum
status.log(state=State.RUNNING)

Value comes from the zelos-trace-types crate. Add it with cargo add zelos-trace-types.

use zelos_trace_types::Value;

#[repr(u8)]
enum State {
    Idle = 0,
    Running = 1,
    Error = 2,
}

// Define event
let status = source
    .build_event("status")
    .add_u8_field("state", None)
    .build()?;

// Add value table
source.add_value_table("status", "state", vec![
    (Value::UInt8(0), "IDLE".to_string()),
    (Value::UInt8(1), "RUNNING".to_string()),
    (Value::UInt8(2), "ERROR".to_string()),
].into_iter())?;

// Log using enum
status.build()
    .try_insert_u8("state", State::Running as u8)?
    .emit()?;

const (
    StateIdle = iota
    StateRunning
    StateError
)

// Define event
status, _ := source.BuildEvent("status").
    AddUint8Field("state", nil).
    Build()

// Add value table
source.AddValueTable("status", "state", map[*zelos.Value]string{
    zelos.NewUint8Value(0): "IDLE",
    zelos.NewUint8Value(1): "RUNNING",
    zelos.NewUint8Value(2): "ERROR",
})

// Log using enum
builder, _ := status.Build().TryInsertUint8("state", StateRunning)
builder.Emit()

Nested Events

Organize related events using path notation:

PythonRustGo

# Parent event
source.add_event("motor", [
    zelos_sdk.TraceEventFieldMetadata("rpm", zelos_sdk.DataType.Float64, "rpm"),
])

# Child events use "/" separator
source.add_event("motor/thermal", [
    zelos_sdk.TraceEventFieldMetadata("temperature", zelos_sdk.DataType.Float32, "°C"),
])

# Access a top-level event as an attribute (Python only)
source.motor.log(rpm=3500)

# Access a child event by its path
source.get_event("motor/thermal").log(temperature=85.5)

# Or log by path
source.log("motor", {"rpm": 3500})
source.log("motor/thermal", {"temperature": 85.5})

# A nested dict logs to a child event: this logs "vehicle" and "vehicle/engine"
source.log("vehicle", {"speed": 12.5, "engine": {"rpm": 3500}})

// Use naming convention with "/"
let motor = source.build_event("motor")
    .add_f64_field("rpm", Some("rpm".to_string()))
    .build()?;

let motor_thermal = source.build_event("motor/thermal")
    .add_f32_field("temperature", Some("°C".to_string()))
    .build()?;

// Use naming convention with "/"
unitRpm := "rpm"
motor, _ := source.BuildEvent("motor").
    AddFloat64Field("rpm", &unitRpm).
    Build()

unitC2 := "°C"
motorThermal, _ := source.BuildEvent("motor/thermal").
    AddFloat32Field("temperature", &unitC2).
    Build()

Best Practices

The examples below use these imports:

from zelos_sdk import DataType
from zelos_sdk import TraceEventFieldMetadata as field

1. Define Once

Register each event once and keep the handle. A second add_event() with the same name raises ValueError ("Event=data already exists").

# Good
event = source.add_event("data", [field("value", DataType.Float64)])
for _ in range(1000000):
    event.log(value=read_value())

# Bad - raises ValueError on the second iteration
for _ in range(1000000):
    source.add_event("data", [field("value", DataType.Float64)])

2. Choose Minimal Types

# Use smallest type that fits
field("percentage", DataType.UInt8)     # 0-100 fits in UInt8
field("temperature", DataType.Float32)  # 0.1°C precision is enough

# Don't oversize
field("percentage", DataType.Float64)   # Wasteful for 0-100

3. Group Related Fields

# Single event preserves correlation
source.add_event("position", [
    field("x", DataType.Float64, "m"),
    field("y", DataType.Float64, "m"),
    field("z", DataType.Float64, "m"),
])

# Separate events lose correlation
source.add_event("pos_x", [field("value", DataType.Float64, "m")])
source.add_event("pos_y", [field("value", DataType.Float64, "m")])