How to Read Trace Files¶

Python Only

These APIs exist only in the Python SDK.

Read .trz trace files for offline analysis, debugging, or post-processing. TraceReader opens the file directly and needs no running agent.

PyArrow

The TraceReader API returns data as PyArrow tables. Use pyarrow.compute for analysis operations.

Analysis through the agent

When the Zelos agent is running, zelos_sdk.connect().trace(path) opens the same file with the agent query API. It returns pandas frames, takes time bounds relative to the file, and runs checks. See Open Trace Files. The agent reads path from its own filesystem.

The examples on this page read output.trz from Basic Recording.

Basic Reading¶

Use TraceReader to query metadata and data from trace files:

import zelos_sdk
import pyarrow as pa

# Read trace file using context manager
with zelos_sdk.TraceReader("output.trz") as reader:
    # Get time range
    time_range = reader.time_range()
    print(f"Trace from {time_range.start} to {time_range.end}")

    # List data segments
    segments = reader.list_data_segments()
    print(f"Found {len(segments)} data segments")

    # Query data (field format: */source/event.field)
    result = reader.query(
        data_segment_ids=[s.id for s in segments],
        fields=["*/experiment/measurement.value"],
        start=time_range.start,
        end=time_range.end,
    )

    # Convert to PyArrow table
    arrow_reader = pa.ipc.open_stream(result.to_arrow())
    table = arrow_reader.read_all()
    print(f"Retrieved {table.num_rows} rows with columns: {table.column_names}")

    # Index by the path you queried
    values = table.column("experiment/measurement.value")

time_range() returns a TimeRange with start and end as RFC 3339 strings. query() takes the same format for start and end; both bounds are inclusive. It returns raw rows, not a downsampled series.

Field Selectors¶

A field selector has the form */source/event.field. The * matches every data segment in data_segment_ids. Replace it with a segment ID to read one segment only: <segment-id>/experiment/measurement.value.

Column Labels¶

The first column is always time_s (epoch seconds). Every other column is labeled with the field path — source/event.field — so you index it with the same string you queried and get_value_table() takes:

table.column("experiment/measurement.value")

One trace file can hold the same path in two data segments. Only those columns get a longer label:

Segments from two producers take a producer:: prefix, as in rig-a::can/Battery.voltage.
Segments from the same producer take a segment ID prefix, as in <segment-id>/can/Battery.voltage.

Unambiguous columns stay short. result.fields always lists exactly what the table is labeled with.

Empty Results¶

When no rows match, result.to_arrow() returns empty bytes, and pa.ipc.open_stream() raises on them. Check before you decode:

data = result.to_arrow()
table = pa.ipc.open_stream(data).read_all() if data else None

A selector that matches no field is not an error. It returns an empty result with result.fields == ["time_s"].

Reading Methods¶

Context Manager (Recommended)¶

with zelos_sdk.TraceReader("output.trz") as reader:
    segments = reader.list_data_segments()
    # File automatically closed

Manual Control¶

reader = zelos_sdk.TraceReader("output.trz")
try:
    reader.open()
    segments = reader.list_data_segments()
finally:
    reader.close()  # Always close!

open() raises RuntimeError when the file is missing or cannot be read. It raises ValueError when the content is not a trace this SDK version can read. Any method on a reader that is not open raises RuntimeError.

Common Patterns¶

Field Discovery¶

Discover available fields before querying:

import zelos_sdk
import pyarrow as pa

# 1. Open trace file for reading
with zelos_sdk.TraceReader("output.trz") as reader:
    # 2. Discover available segments
    segments = reader.list_data_segments()
    assert len(segments) > 0

    # 3. Discover all available fields hierarchically
    sources = reader.list_fields()

    # Navigate hierarchy: source → event → field
    for source in sources:
        print(f"Source: {source.name}")
        for event in source.events:
            print(f"  Event: {event.name}")
            for field in event.fields:
                print(f"    Field: {field.name} → {field.path}")

    # 4. Query discovered field
    experiment = next(s for s in sources if s.name == "experiment")
    measurement = next(e for e in experiment.events if e.name == "measurement")
    value_field = next(f for f in measurement.fields if f.name == "value")

    time_range = reader.time_range()
    result = reader.query(
        data_segment_ids=[s.id for s in segments],
        fields=[value_field.path],  # "*/experiment/measurement.value"
        start=time_range.start,
        end=time_range.end,
    )

    # Convert to Arrow table
    arrow_reader = pa.ipc.open_stream(result.to_arrow())
    table = arrow_reader.read_all()
    print(f"Retrieved {table.num_rows} rows")

list_fields() covers every segment. Pass a segment ID, as in list_fields(segments[0].id), to list one segment's fields. Each event also has a path and an event_type; event_type is None for an untyped event.

Query and Analyze Data¶

Query trace data and perform analysis using PyArrow:

import zelos_sdk
import pyarrow as pa
import pyarrow.compute as pc

with zelos_sdk.TraceReader("output.trz") as reader:
    segments = reader.list_data_segments()
    time_range = reader.time_range()

    # Query field data
    result = reader.query(
        data_segment_ids=[s.id for s in segments],
        fields=["*/experiment/measurement.value"],
        start=time_range.start,
        end=time_range.end
    )

    # Convert to Arrow table
    arrow_reader = pa.ipc.open_stream(result.to_arrow())
    table = arrow_reader.read_all()

    # Analyze with PyArrow
    values = table.column("experiment/measurement.value")

    print(f"Mean: {pc.mean(values).as_py():.2f}")
    print(f"Min: {pc.min(values).as_py():.2f}")
    print(f"Max: {pc.max(values).as_py():.2f}")

Metadata Inspection¶

Query trace metadata without loading data:

import zelos_sdk

def inspect_trace(filename):
    """Display trace file information"""
    with zelos_sdk.TraceReader(filename) as reader:
        # Time coverage
        time_range = reader.time_range()
        print(f"Time Range: {time_range.start} to {time_range.end}")

        # Data segments
        segments = reader.list_data_segments()
        print(f"\nData Segments ({len(segments)}):")
        for seg in segments:
            print(f"  {seg.id}")
            print(f"    Producer: {seg.producer}")
            if seg.start_date:
                print(f"    Time: {seg.start_date} to {seg.end_date}")

# Usage
inspect_trace("output.trz")

A segment's start_date and end_date are None when the segment holds no rows.

list_traces() returns the named recordings in files that carry a trace registry. Files from TraceWriter carry none, so the call raises there. Use time_range() and list_data_segments() instead.

Basic Analysis¶

Analyze trace data using PyArrow:

import zelos_sdk
import pyarrow as pa
import pyarrow.compute as pc

def load_fields(filename, field_patterns):
    """
    Load fields from trace file as Arrow table.

    Args:
        filename: Path to .trz file
        field_patterns: List of field paths (e.g., ["*/source/event.field"])

    Returns:
        PyArrow Table with time_s and one column per field, labeled
        source/event.field
    """
    with zelos_sdk.TraceReader(filename) as reader:
        time_range = reader.time_range()
        segments = reader.list_data_segments()

        # Query data
        result = reader.query(
            data_segment_ids=[s.id for s in segments],
            fields=field_patterns,
            start=time_range.start,
            end=time_range.end,
        )

        # Convert to Arrow table
        arrow_reader = pa.ipc.open_stream(result.to_arrow())
        table = arrow_reader.read_all()

        return table

# Usage
table = load_fields("output.trz", [
    "*/experiment/measurement.value",
    "*/experiment/measurement.index"
])

# Compute statistics using PyArrow
column = table.column("experiment/measurement.value")
print(f"Mean: {pc.mean(column).as_py()}")
print(f"Max: {pc.max(column).as_py()}")
print(f"Min: {pc.min(column).as_py()}")
print(f"Rows: {table.num_rows}")

Time Range Filtering¶

Query specific time windows:

import zelos_sdk
import pyarrow as pa

def load_time_window(filename, start_time, end_time, fields):
    """Load data from a specific time range"""
    with zelos_sdk.TraceReader(filename) as reader:
        # Get segments in time range
        segments = reader.list_data_segments_in_time_range(start_time, end_time)

        if not segments:
            print("No data in specified time range")
            return None

        # Query the window
        result = reader.query(
            data_segment_ids=[s.id for s in segments],
            fields=fields,
            start=start_time,
            end=end_time,
        )

        # No rows in the window
        data = result.to_arrow()
        if not data:
            return None

        # Convert to Arrow table
        return pa.ipc.open_stream(data).read_all()

# Usage - query one hour of data
table = load_time_window(
    "long_recording.trz",
    start_time="2024-01-15T14:00:00Z",
    end_time="2024-01-15T15:00:00Z",
    fields=["*/sensor/temperature.value"]
)

list_data_segments_in_time_range() returns the segments that overlap the window. Both bounds are inclusive.

Querying Enumeration Data¶

Query enum fields and convert integers to strings using stored value tables:

import zelos_sdk
import pyarrow as pa

with zelos_sdk.TraceReader("controller_log.trz") as reader:
    segments = reader.list_data_segments()
    time_range = reader.time_range()

    # Query enum field (returns integer values)
    result = reader.query(
        data_segment_ids=[s.id for s in segments],
        fields=["*/controller/state.status"],
        start=time_range.start,
        end=time_range.end,
    )

    # Extract integer values — the column label is the field path
    arrow_reader = pa.ipc.open_stream(result.to_arrow())
    table = arrow_reader.read_all()
    status_ints = table.column("controller/state.status").to_pylist()  # [0, 1, 2]

    # Value tables are stored per segment; use the first segment that has one
    status_map = None
    for segment in segments:
        status_map = reader.get_value_table(segment.id, "controller/state.status")
        if status_map:
            break

    if status_map:
        status_strs = [status_map.get(v, str(v)) for v in status_ints]
        print(status_strs)  # ["IDLE", "RUNNING", "ERROR"]

get_value_table() takes the field path without the */ prefix. It returns None when the segment has no value table for the field.