Trace Cache

The Trace Cache keeps the last value of every field while still emitting normal trace events. Read a cached value back through attribute or item access.

Use it to read back a previously logged value, branch on cached data, or assert against live state in a test.

The cache can also filter what it emits. See Log Conditions.

Backward-compatible alias

TraceSourceCacheLast is an alias for TraceSourceCache. Existing code that uses the old name will continue to work.

Core Components

TraceSourceCache

A TraceSource-like API backed by the Rust trace cache. It 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.TraceSourceCache("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 and exposes its full path plus data type.

# 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

Behavior to rely on

• add_event() registers a schema immediately and returns a cached event wrapper.
• log() and log_at() can also auto-register new events on first use.
• TraceSourceCacheLastField.set() is kept for backward compatibility, but it does not mutate the cache. The cache updates when you log through the source or event API.
• get_source() is intentionally unsupported on the cache wrapper. If you need a plain TraceSource, create one separately.

Access Patterns

Attribute Access

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

source = zelos_sdk.TraceSourceCache("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 returns the field object
rpm_field = source["motor/rpm"]            # Returns TraceSourceCacheLastField
temp_field = source["motor/thermal/temperature"]  # Returns TraceSourceCacheLastField

# Get cached values via .get()
rpm_value = rpm_field.get()                 # 3500.0
temp_value = temp_field.get()               # 75.5

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.TraceSourceCache("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 Rust-backed cache supports dynamic event creation. You can log to events that have not been explicitly defined, and the Python wrapper becomes available after the first log:

source = zelos_sdk.TraceSourceCache("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

Pass a cached field straight to the checker:

Basic Checker Usage

import pytest
import zelos_sdk

def test_motor_control(check):
    source = zelos_sdk.TraceSourceCache("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, ">", 10.0)
    check.that(source.motor.torque, "<", 20.0)
    check.that(source.motor.temperature, ">", 30.0)
    check.that(source.motor.temperature, "<", 40.0)

Time-based Checker Assertions

def test_motor_warmup_sequence(check):
    import threading
    import time

    source = zelos_sdk.TraceSourceCache("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,
               temporal="within_duration", 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,
               temporal="for_duration", duration_s=1.0)

    warmup_thread.join()

Available Checker Operators

Cached fields work with the full operator vocabulary:

def test_checker_operators(check):
    source = zelos_sdk.TraceSourceCache("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, ">", 46.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")

Error Handling

The cache provides clear error messages for common mistakes:

source = zelos_sdk.TraceSourceCache("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.TraceSourceCache("motor_controller")
motor_controller.add_event("thermal_monitoring", [...])
motor_controller.add_event("speed_control", [...])

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

3. Handle None Values

Cached fields return None until first logged to:

source = zelos_sdk.TraceSourceCache("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.TraceSourceCache("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()