Trace Cache

The Zelos SDK Trace Cache API provides a caching layer for trace events that allows you to store and retrieve the last logged values for any field, at a small performance cost.

This is particularly useful for applications that need to reference previously logged values, implement conditional logic based on cached data, or integrate with testing frameworks.

Core Components

TraceSourceCacheLast

A wrapper around TraceSource that automatically caches the last logged value for every field in every event.

import zelos_sdk

zelos_sdk.init()

# Create a cached trace source
source = zelos_sdk.TraceSourceCacheLast("motor_controller")

# Define event schema
source.add_event("sensor_data", [
    zelos_sdk.TraceEventFieldMetadata("temperature", zelos_sdk.DataType.Float64, "celsius"),
    zelos_sdk.TraceEventFieldMetadata("voltage", zelos_sdk.DataType.Float64, "V"),
    zelos_sdk.TraceEventFieldMetadata("current", zelos_sdk.DataType.Float64, "A")
])

# Log some data
source.sensor_data.log(temperature=45.5, voltage=48.2, current=12.5)

# Access cached values
print(f"Last temperature: {source.sensor_data.temperature.get()}")  # 45.5
print(f"Last voltage: {source.sensor_data.voltage.get()}")          # 48.2

TraceSourceCacheLastEvent

Represents a cached event that provides access to fields and nested submessages. Events are automatically created when you define them with add_event() or when you log to them dynamically.

# Access event and log new values
event = source.sensor_data
event.log(temperature=46.2, voltage=47.8, current=13.1)

# The cache is automatically updated
assert source.sensor_data.temperature.get() == 46.2

TraceSourceCacheLastField

Represents a cached field that stores the last logged value along with metadata about the field's data type and units.

# Get field reference
temp_field = source.sensor_data.temperature

# Access cached value
current_temp = temp_field.get()

# Access field metadata
print(f"Field name: {temp_field.name}")           # "sensor_data.temperature"
print(f"Data type: {temp_field.data_type}")       # DataType.Float64
print(f"Current value: {temp_field.get()}")       # 46.2

Access Patterns

Attribute Access

The most convenient way to access cached values is through attribute notation:

source = zelos_sdk.TraceSourceCacheLast("motor_controller")

# Define nested event structure
source.add_event("motor", [
    zelos_sdk.TraceEventFieldMetadata("rpm", zelos_sdk.DataType.Float64),
    zelos_sdk.TraceEventFieldMetadata("torque", zelos_sdk.DataType.Float64, "Nm")
])
source.add_event("motor/thermal", [
    zelos_sdk.TraceEventFieldMetadata("temperature", zelos_sdk.DataType.Float64, "celsius")
])

# Log data
source.log("motor", {"rpm": 3500.0, "torque": 42.8})
source.log("motor/thermal", {"temperature": 75.5})

# Access via attributes
motor_rpm = source.motor.rpm.get()                    # 3500.0
motor_temp = source.motor.thermal.temperature.get()   # 75.5

Dictionary Access

For dynamic access patterns or when field names contain special characters:

# Dictionary-style access for events
event = source["motor"]                    # Returns TraceSourceCacheLastEvent
temp_event = source["motor/thermal"]       # Returns nested TraceSourceCacheLastEvent

# Dictionary-style access for fields
rpm_value = source["motor/rpm"]            # Returns the cached value directly (3500.0)
temp_value = source["motor/thermal/temperature"]  # Returns the cached value directly (75.5)

# Get field objects for metadata access
rpm_field = source["motor"].rpm            # Returns TraceSourceCacheLastField
temp_field = source["motor/thermal"].temperature  # Returns TraceSourceCacheLastField

Handling Name Conflicts

When a submessage has the same name as a field in the parent event, use explicit methods to disambiguate:

source = zelos_sdk.TraceSourceCacheLast("sensor_system")

# Create events with conflicting names
source.add_event("battery", [
    zelos_sdk.TraceEventFieldMetadata("status", zelos_sdk.DataType.Int32)
])
source.add_event("battery/status", [
    zelos_sdk.TraceEventFieldMetadata("code", zelos_sdk.DataType.Int32)
])

# Log data
source.log("battery", {"status": 1})
source.log("battery/status", {"code": 200})

# Access field explicitly
status_field_value = source.battery.get_field("status").get()    # 1

# Access submessage explicitly
status_submessage = source.battery.get_submessage("status")      # TraceSourceCacheLastEvent
status_code = status_submessage.code.get()                       # 200

# Or use attribute access (defaults to submessage if both exist)
status_event = source.battery.status        # Returns TraceSourceCacheLastEvent
status_code = source.battery.status.code.get()  # 200

Dynamic Event Creation

The cache supports dynamic event creation - you can log to events that haven't been explicitly defined:

source = zelos_sdk.TraceSourceCacheLast("vehicle_system")

# Log to undefined events - they will be created dynamically
source.log("engine_stats", {"rpm": 2500, "load": 65.5})
source.log("engine_stats/emissions", {"nox": 45.2, "particulates": 12.1})

# Access the dynamically created cached values
rpm = source.engine_stats.rpm.get()           # 2500
nox_level = source.engine_stats.emissions.nox.get()  # 45.2

Time-based Logging

Cache supports logging with specific timestamps:

import time

# Log with current timestamp
source.log("sensor_data", {"temperature": 45.0})

# Log with specific timestamp
timestamp_ns = int(time.time_ns())
source.log_at(timestamp_ns, "sensor_data", {"temperature": 46.0})

# Log via event object with specific timestamp
source.sensor_data.log_at(timestamp_ns + 1000000000, temperature=47.0)

Checker Integration

The trace cache integrates seamlessly with the Zelos testing framework for automated verification:

Basic Checker Usage

import pytest
import zelos_sdk

def test_motor_control(check):
    source = zelos_sdk.TraceSourceCacheLast("motor_test")

    # Define motor control events
    source.add_event("motor", [
        zelos_sdk.TraceEventFieldMetadata("rpm", zelos_sdk.DataType.Float64),
        zelos_sdk.TraceEventFieldMetadata("torque", zelos_sdk.DataType.Float64, "Nm"),
        zelos_sdk.TraceEventFieldMetadata("temperature", zelos_sdk.DataType.Float64, "celsius")
    ])

    # Log initial motor state
    source.log("motor", {"rpm": 0.0, "torque": 0.0, "temperature": 20.0})

    # Verify initial state
    check.that(source.motor.rpm, "==", 0.0)
    check.that(source.motor.temperature, "<", 25.0)

    # Simulate motor startup
    source.log("motor", {"rpm": 1500.0, "torque": 15.5, "temperature": 35.0})

    # Verify running state
    check.that(source.motor.rpm, ">", 1000.0)
    check.that(source.motor.torque, "between", 10.0, 20.0)
    check.that(source.motor.temperature, "between", 30.0, 40.0)

Time-based Checker Assertions

def test_motor_warmup_sequence(check):
    import threading
    import time

    source = zelos_sdk.TraceSourceCacheLast("motor_warmup")

    source.add_event("motor", [
        zelos_sdk.TraceEventFieldMetadata("temperature", zelos_sdk.DataType.Float64, "celsius"),
        zelos_sdk.TraceEventFieldMetadata("ready", zelos_sdk.DataType.Boolean)
    ])

    # Start with cold motor
    source.log("motor", {"temperature": 20.0, "ready": False})

    # Simulate gradual warmup in background thread
    def warmup_motor():
        for temp in range(20, 61, 5):  # Heat up from 20°C to 60°C
            source.log("motor", {
                "temperature": float(temp),
                "ready": temp >= 50.0  # Ready when temperature reaches 50°C
            })
            time.sleep(0.2)

    warmup_thread = threading.Thread(target=warmup_motor)
    warmup_thread.start()

    # Check that motor becomes ready within 3 seconds
    check.that(source.motor.ready, "==", True, within_duration_s=3.0)

    # Check that temperature stays above 45°C for at least 1 second once reached
    check.that(source.motor.temperature, ">", 45.0, for_duration_s=1.0)

    warmup_thread.join()

Available Checker Operators

The checker supports various operators when working with cached fields:

def test_checker_operators(check):
    source = zelos_sdk.TraceSourceCacheLast("battery_monitor")

    source.add_event("battery", [
        zelos_sdk.TraceEventFieldMetadata("voltage", zelos_sdk.DataType.Float64, "V"),
        zelos_sdk.TraceEventFieldMetadata("state", zelos_sdk.DataType.String),
        zelos_sdk.TraceEventFieldMetadata("charging", zelos_sdk.DataType.Boolean)
    ])

    source.log("battery", {"voltage": 48.2, "state": "nominal", "charging": True})

    # Equality checks
    check.that(source.battery.voltage, "==", 48.2)
    check.that(source.battery.state, "==", "nominal")

    # Comparison checks
    check.that(source.battery.voltage, ">", 45.0)
    check.that(source.battery.voltage, "<=", 50.0)
    check.that(source.battery.voltage, "between", 46.0, 50.0)

    # Boolean checks
    check.that(source.battery.charging, "is true")
    check.that(source.battery.charging, "==", True)

    # String checks
    check.that(source.battery.state, "contains", "nom")
    check.that(source.battery.state, "starts with", "nom")

    # Type checks
    check.that(source.battery.voltage, "is instance of", float)

Error Handling

The cache provides clear error messages for common mistakes:

source = zelos_sdk.TraceSourceCacheLast("error_demo")

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

try:
    # This will raise AttributeError - field doesn't exist
    value = source.motor.nonexistent_field.get()
except AttributeError as e:
    print(f"Field access error: {e}")

try:
    # This will raise KeyError - event doesn't exist
    event = source["nonexistent_event"]
except KeyError as e:
    print(f"Event access error: {e}")

try:
    # This will raise KeyError - field path doesn't exist
    value = source["motor/nonexistent_field"]
except KeyError as e:
    print(f"Field path error: {e}")

Best Practices

1. Initialize Fields Early

Define your event schemas early in your application to get proper type checking and metadata:

# Good: Define schema upfront
source.add_event("sensor", [
    zelos_sdk.TraceEventFieldMetadata("temperature", zelos_sdk.DataType.Float64, "celsius")
])

# Less ideal: Relying on dynamic creation
source.log("sensor", {"temperature": 25.0})  # Schema inferred from data

2. Use Meaningful Names

Choose descriptive names for your trace sources and events:

# Good: Descriptive names
motor_controller = zelos_sdk.TraceSourceCacheLast("motor_controller")
motor_controller.add_event("thermal_monitoring", [...])
motor_controller.add_event("speed_control", [...])

# Less ideal: Generic names
source = zelos_sdk.TraceSourceCacheLast("app")
source.add_event("data", [...])

3. Handle None Values

Cached fields return None until first logged to:

source = zelos_sdk.TraceSourceCacheLast("safety_check")
source.add_event("sensor", [
    zelos_sdk.TraceEventFieldMetadata("temperature", zelos_sdk.DataType.Float64)
])

# Check for None before using cached values
temp = source.sensor.temperature.get()
if temp is not None:
    if temp > 80.0:
        print("Temperature warning!")
else:
    print("No temperature data logged yet")

4. Organize Hierarchical Data

Use nested events to organize related data:

source = zelos_sdk.TraceSourceCacheLast("vehicle_system")

# Organize by subsystem
source.add_event("powertrain/engine", [...])
source.add_event("powertrain/transmission", [...])
source.add_event("chassis/suspension", [...])
source.add_event("chassis/brakes", [...])

# Access hierarchically
engine_rpm = source.powertrain.engine.rpm.get()
brake_pressure = source.chassis.brakes.pressure.get()