Tracing¶

The top-level zelos_sdk module streams data to the Zelos agent and reads and writes trace files. The Agent API names that also import from the top level are on the Agent page.

`TraceSource(name, namespace=..., strict=...)` ¶

Central source for trace events in an application.

A TraceSource represents a single data source within your application (like a service or component) and manages the event schemas and transmission of events to the trace collection system.

Examples:

>>> client = TracePublishClient()
>>> source = TraceSource("motor_controller")
>>>
>>> # Define an event schema
>>> motor_event = source.add_event("motor_stats", [
...     TraceEventFieldMetadata("rpm", DataType.Float64),
...     TraceEventFieldMetadata("torque", DataType.Float64, "Nm"),
...     TraceEventFieldMetadata("temperature", DataType.Float64, "celsius"),
...     TraceEventFieldMetadata("voltage", DataType.Float64, "V"),
... ])
>>>
>>> # Log an event
>>> motor_event.log(**{
...     "rpm": 3500.0,
...     "torque": 42.8,
...     "temperature": 75.5,
...     "voltage": 48.2
... })

`add_event(name, schema, event_type=None)` ¶

Register an event schema.

Parameters:

Name	Type	Description	Default
`name`	`str`	The event name (e.g. `"log"`).	required
`schema`	`list[TraceEventFieldMetadata] \| type`	Either a `list[TraceEventFieldMetadata]` (ad-hoc fields) or a class with `FIELDS` and (optionally) `EVENT_TYPE` classvars — e.g. `zelos_sdk.schemas.Log`. When a class is passed, `FIELDS` is used and `EVENT_TYPE` populates the `event_type` argument if not explicitly set.	required
`event_type`	`Optional[str]`	Stable identifier for this event's schema (e.g. `"zelos.log.v1"`). Defaults to `schema.EVENT_TYPE` when `schema` is a class.	`None`

Returns:

Name	Type	Description
`TraceSourceEvent`	`TraceSourceEvent`	A handle to the newly registered event.

Raises:

Type	Description
`ValueError`	If registering the schema fails internally.

`add_value_table(name, field_name, data)` ¶

Add a value table to the trace source.

Register the event and field with add_event first: a key is typed by the field it labels, so there is nothing to type it against otherwise.

Parameters:

Name	Type	Description	Default
`name`	`str`	The name of the value table.	required
`data`	`dict`	A dictionary of values to add to the value table.	required

Returns:

Type	Description
`None`	None

Examples:

>>> source.add_value_table("motor_status", "state", {0: "stopped", 1: "running"})
>>> source.add_value_table("sensor_data", "sensor_id", {1: "temp_sensor", 2: "pressure_sensor"})

`flush()` ¶

Flush all buffered events as Arrow RecordBatches.

Events logged via log() are buffered internally and flushed automatically when the batch size or timeout threshold is reached. Call this method to force-flush any remaining buffered events.

Returns:

Type	Description
`None`	None

`get_event(name)` ¶

Get a handle to a previously registered event schema.

Parameters:

Name	Type	Description	Default
`name`	`str`	The name of the event schema.	required

Returns:

Name	Type	Description
`TraceSourceEvent`	`TraceSourceEvent`	A handle to the event.

Raises:

Type	Description
`KeyError`	If no event with the given name is registered.

Examples:

>>> # After defining an event schema
>>> event = source.get_event("motor_stats")

`log(name, data)` ¶

Log an event with a name and a dictionary of fields.

Parameters:

Name	Type	Description	Default
`name`	`str`	The name to log.	required
`data`	`dict`	A dictionary of fields to log.	required

Returns:

Type	Description
`None`	None

Examples:

>>> source.log("sensor_data", {"temperature": 25.0, "pressure": 101325})

`log_at(time_ns, name, data)` ¶

Log an event with a name and a dictionary of fields.

Parameters:

Name	Type	Description	Default
`time_ns`	`int`	The time to log the event at.	required
`name`	`str`	The name to log.	required
`data`	`dict`	A dictionary of fields to log.	required

Returns:

Type	Description
`None`	None

Examples:

>>> source.log_at(time.time_ns(), "sensor_data", {"temperature": 25.0})

`log_batch(event_name, data)` ¶

Log an Arrow RecordBatch directly (zero-copy from PyArrow).

The RecordBatch must have a time_ns column (TimestampNanosecond) as its first column. Schema is auto-registered on the first batch per event name.

Parameters:

Name	Type	Description	Default
`event_name`	`str`	The event name for this batch.	required
`data`	`Any`	A pyarrow.RecordBatch or pyarrow.Table.	required

Examples:

>>> import pyarrow as pa
>>> batch = pa.record_batch({"time_ns": pa.array([1, 2], type=pa.timestamp("ns", tz="UTC")), "value": [1.0, 2.0]})
>>> source.log_batch("sensor", batch)

`log_dict(name, data)` ¶

Log an event with a name and a dictionary of fields.

Parameters:

Name	Type	Description	Default
`name`	`str`	The name to log.	required
`data`	`dict`	A dictionary of fields to log.	required

Returns:

Type	Description
`None`	None

Examples:

>>> source.log_dict("sensor_data", {"temperature": 25.0, "pressure": 101325})

`log_many(events)` ¶

Log multiple events in a single call, minimizing Python↔Rust overhead.

Each entry is a (time_ns, event_name, signals_dict) tuple. Events are grouped by event name and emitted in bulk with the GIL released.

This is the recommended path for high-throughput sources like CAN codecs that decode many frames per cycle.

Parameters:

Name	Type	Description	Default
`events`	`Sequence[tuple[int, str, dict]]`	List of (time_ns: int, name: str, data: dict) tuples.	required

Examples:

>>> source.log_many([
...     (time.time_ns(), "sensor", {"temp": 25.0}),
...     (time.time_ns(), "sensor", {"temp": 25.1}),
...     (time.time_ns(), "status", {"mode": "active"}),
... ])

`TraceSourceEvent` ¶

`log(**kwargs)` ¶

Log an event with a dictionary of fields.

Parameters:

Name	Type	Description	Default
`kwargs`	`dict`	Keyword arguments to log.	`{}`

Returns:

Type	Description
`None`	None

Examples:

>>> event = source.get_event("motor_stats")
>>> event.log(rpm=3500, torque=42.8)

`log_at(time_ns, **kwargs)` ¶

Log an event with data provided as a dictionary at a specific time. Performs type checking based on the schema.

Parameters:

Name	Type	Description	Default
`time_ns`	`int`	Timestamp in nanoseconds since Unix epoch.	required
`kwargs`	`dict`	Field names and values to log.	`{}`

Raises:

Type	Description
`ValueError`	If a field is not in the schema.
`TypeError`	If a value's type doesn't match the schema.
`RuntimeError`	If sending the event fails internally.

Examples:

>>> event = source.get_event("motor_stats")
>>> # Log with custom timestamp
>>> event.log_at(1625097600000000000, rpm=3500.0, torque=42.8, temperature=75.5)

`TraceEventFieldMetadata(name, data_type, unit=...)` ¶

Metadata describing a field in a trace event schema.

This class defines the structure of a field within an event schema, including its name, data type, and optional unit of measurement.

Parameters:

Name	Type	Description	Default
`name`	`str`	The field name.	required
`data_type`	`DataType`	The data type for the field.	required
`unit`	`Optional[str]`	Optional unit of measurement.	`...`

Examples:

>>> # Define a field for HTTP status code
>>> status_field = TraceEventFieldMetadata("status_code", DataType.Int32)
>>>
>>> # Define a field with a unit of measurement
>>> duration_field = TraceEventFieldMetadata(
...     "duration_ms", DataType.Float64, "milliseconds")

`DataType` ¶

`TraceNamespace(name)` ¶

A namespace that manages and organizes TraceSources.

TraceNamespace provides a centralized registry for TraceSources with an isolated router. Each namespace has its own router, allowing complete isolation between different namespaces for testing or multi-tenant scenarios.

Examples:

>>> # Create an isolated namespace
>>> ns = TraceNamespace("my_app")
>>> source = TraceSource("service", namespace=ns)
>>> with TraceWriter("data.trz", namespace=ns) as writer:
...     source.log("event", value=42)

`drain()` ¶

Flush all sources, signal the router to drain, and block until every queued event has been delivered to subscribed sinks.

Call this before tearing down a writer (or at the end of a transcode) when you need at-least-once delivery — __del__ / atexit drain are best-effort and don't block on the caller's thread. Releases the GIL while waiting.

Idempotent and safe to call multiple times. Must NOT be called from inside the tokio runtime (e.g. from an async-Python task) because it blocks on the router's drain ack.

`source_count()` ¶

Get the number of registered sources.

Returns:

Name	Type	Description
`int`	`int`	Number of registered sources.

`TracePublishClient(url=..., config=...)` ¶

Client for publishing trace events to a Zelos Cloud service.

This client manages the connection to a remote trace service and provides the communication channel needed by TraceSource objects to transmit events. It handles batching, retries, and connection management.

Examples:

>>> # Create a client with default settings
>>> client = TracePublishClient()
>>>
>>> # Create a client with custom configuration
>>> config = TracePublishClientConfig(url="grpc://localhost:2300")
>>> client = TracePublishClient(config)

`shutdown()` ¶

Shutdown the client and all background tasks.

This will cancel the background tasks and wait for them to complete. After calling this method, the client should not be used further.

`TracePublishClientConfig(batch_size=..., batch_timeout_ms=...)` ¶

Configuration for the PyTracePublishClient.

This class allows customizing the behavior of trace publishing, including: - Batch size: Number of events to batch before sending - Batch timeout: Maximum time to wait before sending a partial batch

Examples:

>>> config = TracePublishClientConfig(
...     batch_size=500,
...     batch_timeout_ms=2000,
... )
>>> client = TracePublishClient(config)

`set_batch_size(size)` ¶

Set the configured batch size.

`set_batch_timeout_ms(ms)` ¶

Set the configured batch timeout in milliseconds.

`TraceWriter(path, batch_size=..., batch_timeout_ms=..., allow_existing=..., namespace=...)` ¶

Python wrapper for the TraceWriter.

This writer manages writing trace events to a local file, with support for batching and buffering. It can be used with a TraceSource to capture events for later analysis.

The writer uses context management and should be used with a with statement to ensure proper resource cleanup and automatic start/stop of trace capture.

Examples:

>>> # Basic usage with default settings
>>> with TraceWriter("my_trace.trz") as writer:
...     # Trace events will be captured automatically
...     pass
>>>
>>> # Custom batch configuration
>>> with TraceWriter("my_trace.trz", batch_size=500, batch_timeout_ms=2000) as writer:
...     # Trace events will be captured with custom batch settings
...     pass

`close()` ¶

Stop the trace writer and finalize trace capture.

This method gracefully shuts down the writer, cancels background tasks, and ensures all buffered events are written to the trace file. It's automatically called when exiting the context manager.

Returns:

Type	Description
`None`	None

Note

This method is called automatically by exit when using the context manager pattern.

`open()` ¶

Start the trace writer and begin capturing events.

This method initializes the writer and starts background tasks for batching and writing trace events. It's automatically called when entering the context manager (with statement).

Returns:

Type	Description
`None`	None

Raises:

Type	Description
`RuntimeError`	If the writer cannot be initialized.

Note

This method is called automatically by enter when using the context manager pattern.

`TraceReader(path)` ¶

Python wrapper for the TraceReader.

This reader provides read-only access to trace files, allowing you to query metadata and retrieve trace data programmatically. It supports listing data segments, querying time ranges, and retrieving raw or downsampled data.

The reader uses context management and should be used with a with statement to ensure proper resource cleanup.

Complete End-to-End Workflow¶

This example demonstrates opening a trace file, discovering available fields, and querying specific data:

import zelos_sdk
import pyarrow as pa

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

    # Discover available fields hierarchically
    sources = reader.list_fields()
    assert len(sources) > 0

    # Navigate hierarchy: source → event → field
    can_source = next(s for s in sources if s.name == "can")
    msg_event = next(e for e in can_source.events if e.name == "VehicleSpeed")
    speed_field = next(f for f in msg_event.fields if f.name == "speed")

    # Query discovered field
    time_range = reader.time_range()
    result = reader.query(
        data_segment_ids=[s.id for s in segments],
        fields=[speed_field.path],  # "*/can/VehicleSpeed.speed"
        start=time_range.start,
        end=time_range.end,
    )

    # Verify data received. Columns are labeled with the field path,
    # so the queried selector minus its `*/` prefix indexes the table.
    arrow_reader = pa.ipc.open_stream(result.to_arrow())
    table = arrow_reader.read_all()
    assert table.num_rows > 0
    assert table.column("can/VehicleSpeed.speed")

`close()` ¶

Close the trace reader.

This method closes the trace file and releases resources. It's automatically called when exiting the context manager.

Returns:

Type	Description
`None`	None

`get_value_table(data_segment_id, field_path)` ¶

Get the value table (enum mapping) for a specific field.

Parameters:

Name	Type	Description	Default
`data_segment_id`	`str`	Data segment ID to query.	required
`field_path`	`str`	Field path in format "source/event.field" (without the "*/" prefix).	required

Returns:

Name	Type	Description
`dict`	`Optional[dict[int, str]]`	Mapping of integer keys to string values, or None if no value table exists.

Raises:

Type	Description
`RuntimeError`	If the reader is not open or query fails.

Examples:

>>> with TraceReader("my_trace.trz") as reader:
...     segments = reader.list_data_segments()
...     # Get enum mapping for a status field
...     status_map = reader.get_value_table(
...         segments[0].id,
...         "controller/state.status"
...     )
...     if status_map:
...         print(status_map)  # {0: "IDLE", 1: "RUNNING", 2: "ERROR"}

`list_data_segments()` ¶

List all data segments in the trace.

Returns:

Type	Description
`list[DataSegment]`	List[DataSegment]: List of data segment metadata.

Raises:

Type	Description
`RuntimeError`	If the reader is not open or query fails.

Examples:

>>> with TraceReader("my_trace.trz") as reader:
...     segments = reader.list_data_segments()
...     for seg in segments:
...         print(f"Segment {seg.id}: {seg.producer}")

`list_data_segments_in_time_range(start, end)` ¶

List data segments within a specific time range.

Parameters:

Name	Type	Description	Default
`start`	`str`	Start of time range (inclusive), ISO 8601 / RFC 3339.	required
`end`	`str`	End of time range (inclusive), ISO 8601 / RFC 3339.	required

Returns:

Type	Description
`list[DataSegment]`	List[DataSegment]: List of data segments overlapping the time range.

Raises:

Type	Description
`RuntimeError`	If the reader is not open, a timestamp does not parse, or the query fails.

Examples:

>>> with TraceReader("my_trace.trz") as reader:
...     time_range = reader.time_range()
...     segments = reader.list_data_segments_in_time_range(
...         time_range.start, time_range.end
...     )

`list_fields(data_segment_id=None)` ¶

List all fields in the trace organized by source and event.

This method discovers all available fields in the trace by querying the database schema and organizing them hierarchically by source and event.

Parameters:

Name	Type	Description	Default
`data_segment_id`	`str`	Specific data segment ID to query. If None, queries all segments.	`None`

Returns:

Type	Description
`list[TraceReadSource]`	List[TraceReadSource]: List of sources, each containing events and fields.

Raises:

Type	Description
`RuntimeError`	If the reader is not open or query fails.

Example: Discover and Query Fields¶

with TraceReader("recording.trz") as reader:
    # Discover all available fields
    sources = reader.list_fields()
    assert len(sources) > 0

    # Navigate the hierarchy
    for source in sources:
        for event in source.events:
            for field in event.fields:
                # field.path is the field path for queries (e.g., "*/can/VehicleSpeed.speed")
                assert field.path.startswith("*/") and "." in field.path

`list_traces()` ¶

List all traces in the trace file.

Returns:

Type	Description
`list[TraceMetadata]`	List[TraceMetadata]: List of trace metadata.

Raises:

Type	Description
`RuntimeError`	If the reader is not open or query fails.
`ValueError`	If the trace has no such registry. Use `time_range()` and `list_data_segments()` instead.

Examples:

>>> with TraceReader("my_trace.trz") as reader:
...     traces = reader.list_traces()
...     for trace in traces:
...         print(f"Trace {trace.name}: {trace.start_date} to {trace.end_date}")

`open()` ¶

Open the trace file for reading.

This method initializes the reader and opens the trace file in read-only mode. It's automatically called when entering the context manager (with statement).

Returns:

Type	Description
`None`	None

Raises:

Type	Description
`ValueError`	If the path is not a trace this build can read — unrecognized content, a container written by a newer Zelos, or a directory that is not a sealed trace.
`RuntimeError`	If the trace file cannot be opened (including a missing file and any other I/O failure).

`query(data_segment_ids, fields, start, end)` ¶

Query data for specified fields within a time range.

This returns raw, unsampled data for the requested fields.

Columns are labeled with the field path source/event.field, so the selector you queried indexes the result directly. Two producers emitting the same path into one file take a producer:: prefix.

Parameters:

Name	Type	Description	Default
`data_segment_ids`	`List[str]`	List of data segment IDs to query. An empty list selects nothing.	required
`fields`	`List[str]`	List of field selectors (e.g., "*/bus0/msg1.sig1", or "/bus0/msg1.sig1" for one specific segment).	required
`start`	`str`	Start of time range (inclusive), ISO 8601 / RFC 3339.	required
`end`	`str`	End of time range (inclusive), ISO 8601 / RFC 3339.	required

Returns:

Name	Type	Description
`QueryResult`	`QueryResult`	Query results with Arrow data. `fields` reflects the REQUEST: a requested field is listed even when it has no row in the window. A request that resolves to nothing is the empty result (`fields == ["time_s"]`, no data), not an error. `sql` is empty on TRZ2 traces — a DataFusion plan has no SQL text — and carries the executed query on legacy `.trz` files.

Raises:

Type	Description
`RuntimeError`	If the reader is not open, an argument does not parse, or the query fails.

Examples:

>>> with TraceReader("my_trace.trz") as reader:
...     segments = reader.list_data_segments()
...     time_range = reader.time_range()
...     result = reader.query(
...         data_segment_ids=[s.id for s in segments],
...         fields=["*/bus0/msg1.sig1", "*/bus0/msg2.sig3"],
...         start=time_range.start,
...         end=time_range.end,
...     )
...     # Convert to PyArrow table
...     import pyarrow as pa
...     arrow_reader = pa.ipc.open_stream(result.to_arrow())
...     table = arrow_reader.read_all()
...     sig1 = table.column("bus0/msg1.sig1")
...     df = table.to_pandas()

`time_range()` ¶

Get the time range covered by the trace.

Returns:

Name	Type	Description
`TimeRange`	`TimeRange`	Object containing start and end timestamps.

Raises:

Type	Description
`RuntimeError`	If the reader is not open or query fails.

Examples:

>>> with TraceReader("my_trace.trz") as reader:
...     time_range = reader.time_range()
...     print(f"Start: {time_range.start}")
...     print(f"End: {time_range.end}")

`TraceMetadata` ¶

Python wrapper for trace metadata.

Contains information about a complete trace including its time range, producer, and associated data segments.

`DataSegment` ¶

Python wrapper for a data segment.

Represents a segment of trace data with metadata about its time range and producer.

`QueryResult` ¶

Python wrapper for query results.

Contains the results of a trace data query, including column labels, the raw Arrow data as Python bytes, and the SQL query that was executed.

fields[0] is always time_s; every other column is labeled with its field path source/event.field, so table.column("source/event.field") indexes the decoded Arrow table directly.

sql carries the executed query on legacy .trz files and is EMPTY on TRZ2 traces — those are served by a DataFusion plan, which has no SQL text.

`to_arrow()` ¶

Convert the Arrow data to a Python object that can be read by PyArrow.

Returns:

Name	Type	Description
`bytes`	`bytes`	Arrow IPC stream data

Examples:

>>> import pyarrow as pa
>>> result = reader.query(...)
>>> arrow_bytes = result.to_arrow()
>>> reader = pa.ipc.open_stream(arrow_bytes)
>>> table = reader.read_all()

`TraceReadSource` ¶

Python wrapper for a trace read source.

Represents a source (e.g., "can") containing multiple events.

`TraceReadEvent` ¶

Python wrapper for a trace read event.

Represents an event containing multiple fields.

`TraceReadEventField` ¶

Python wrapper for a trace read event field.

Represents a single field within an event.

`TraceStdout(log_level=..., batch_size=..., batch_timeout_ms=...)` ¶

Python wrapper for the stdout trace sink.

This sink outputs trace events to stdout with configurable log levels. It subscribes to all trace events from the router and formats them as structured log messages.

The sink uses context management and should be used with a with statement to ensure proper resource cleanup and automatic start/stop of trace capture.

Examples:

>>> # Basic usage with default settings (info level)
>>> with TraceStdout() as sink:
...     # Trace events will be logged to stdout
...     pass
>>>
>>> # Custom log level and batch configuration
>>> with TraceStdout(log_level="debug", batch_size=500, batch_timeout_ms=2000) as sink:
...     # Trace events will be logged with custom settings
...     pass

`close()` ¶

Stop the stdout sink and finalize trace capture.

This method gracefully shuts down the sink and cancels background tasks. It's automatically called when exiting the context manager.

Returns:

Type	Description
`None`	None

`open()` ¶

Start the stdout sink and begin capturing events.

This method subscribes to the trace router and starts a background task to process and output trace events to stdout. It's automatically called when entering the context manager (with statement).

Returns:

Type	Description
`None`	None

Raises:

Type	Description
`RuntimeError`	If the sink cannot be initialized.

`init(name=None, *, url=None, client_config=None, log_level=None, trace=True, actions=False, block=False)` ¶

Initialize the Zelos SDK tracing and actions systems.

Parameters:

Name	Type	Description	Default
`name`	`str \| None`	Application identifier; defaults to `"python"`.	`None`
`url`	`str \| None`	Agent endpoint (e.g. `"http://host:port"`). Forwarded to the trace publish client and the actions client. Falls back to `ZELOS_AGENT_URL` and finally `http://localhost:2300`.	`None`
`client_config`	`TracePublishClientConfig \| None`	Configuration for the `TracePublishClient` (`batch_size`, `batch_timeout_ms`).	`None`
`log_level`	`str \| None`	Logging level to enable, `None` leaves logging untouched.	`None`
`trace`	`bool`	Initialize the trace system. Defaults to `True`.	`True`
`actions`	`bool`	Initialize the actions system. Defaults to `False`.	`False`
`block`	`bool`	Block the current thread until interrupted (useful for actions-only programs).	`False`

Notes

When ZELOS_STANDALONE is set — the standalone action harness sets it, as does the agent when it spawns a one-shot run — no agent connection is opened. The extension is stopped in that context, and connecting anyway would briefly register it as live. The global trace source is still created, so a module that logs through it keeps working; the events go nowhere.

Examples:

>>> init()
>>> init("my_app", url="grpc://localhost:2300", log_level="debug")
>>> init(log_level="debug", trace=False)  # logging only
>>> init(actions=True, block=True)

`initialized()` ¶

`enable_logging(log_level=None)` ¶

Enable logging for the Zelos SDK native module.

This function initializes the tracing system with the specified log level. If no log level is provided, it defaults to "info".

Parameters:

Name	Type	Description	Default
`log_level`	`Optional[str]`	The log level to use. Valid values: "trace", "debug", "info", "warn", "error". Defaults to "info" if not specified.	`None`

Returns:

Type	Description
`None`	None

Examples:

>>> enable_logging("debug")  # Set log level to debug
>>> enable_logging("info")   # Set log level to info
>>> enable_logging()         # Set log level to info

`agent_url()` ¶

The agent URL init() settled on, or None before init().

Explicit argument, then ZELOS_AGENT_URL, then ZELOS_TRACE_FORWARD_URL, then http://localhost:2300: the same resolution the publish client used, latched at the moment it was made.

`init_global_client(url=None, config=None)` ¶

Initialize the global client with custom settings

Parameters:

Name	Type	Description	Default
`url`	`str`	The URL of the trace publish service. Defaults to `ZELOS_AGENT_URL` (legacy: `ZELOS_TRACE_FORWARD_URL`), then `http://localhost:2300`.	`None`
`config`	`TracePublishClientConfig`	Configuration for the client. If None, default settings will be used.	`None`

Returns:

Name	Type	Description
`TracePublishClient`	`TracePublishClient`	The global client instance

Examples:

>>> # Initialize with default settings
>>> client = init_global_client()
>>>
>>> # Initialize with custom settings
>>> config = TracePublishClientConfig(batch_size=500)
>>> client = init_global_client(url="grpc://localhost:2300", config=config)

`init_global_source(name=None)` ¶

Get the global TraceSource, creating it on the first call.

Idempotent: later calls return the source that already exists and ignore name. zelos_sdk.init() calls this for you.

Parameters:

Name	Type	Description	Default
`name`	`Optional[str]`	Source name for the first call. Defaults to `"python"`.	`None`

Returns:

Name	Type	Description
`TraceSource`	`TraceSource`	The global source.

Examples:

>>> source = init_global_source("my_app")

`log(name, data, source=...)` ¶

Log one event on a source, now.

Parameters:

Name	Type	Description	Default
`name`	`str`	The event name. A new name registers its schema from `data`.	required
`data`	`dict`	Field names and values.	required
`source`	`TraceSource \| None`	The source to log on. Defaults to the global source that `init()` creates.	`...`

Examples:

>>> zelos_sdk.init()
>>> zelos_sdk.log("motor", {"rpm": 3500.0, "torque": 42.8})

`sanitize_name(name, *, kind='source')` ¶

Rewrite an externally-sourced name into one that registers cleanly.

Trace names are an allow-list — letters, digits, spaces, _ and -, plus / for event names — enforced when a name is registered: TraceSource(...), add_event(...) and the first log_batch(...) raise ValueError on a name that violates it. Names lifted out of an external artifact (DBC signals, scope channel labels, packet decoder schemas) routinely do. This is the one blessed way to fix them up; do not hand-roll a sanitizer.

Every disallowed character becomes _ (a run of them collapses to a single _), edge spaces and underscores are trimmed, the result is capped at 128 bytes on a character boundary, and a name with nothing left becomes "unnamed".

Lossy. Distinct inputs can collapse to the same name (a.b, a..b and a:b all become a_b). De-duplicate before registering if the upstream artifact can produce collisions — the trace layer treats two identical names as one signal.

Idempotent. sanitize_name(sanitize_name(x)) == sanitize_name(x).

Parameters:

Name	Type	Description	Default
`name`	`str`	The raw, externally-sourced name.	required
`kind`	`str`	Which name this is: `"source"`, `"event"`, `"field"` or `"value_table"`. Only event and value-table names may contain `/`; for the others it is substituted. Defaults to `"source"`.	`'source'`

Returns:

Name	Type	Description
`str`	`str`	A name that always passes registration for that kind.

Raises:

Type	Description
`ValueError`	If `kind` is not one of the four listed above.

Examples:

>>> sanitize_name("Engine.RPM[0]", kind="field")
'Engine_RPM_0'
>>> sanitize_name("bus:can0")
'bus_can0'
>>> sanitize_name("battery/status", kind="event")
'battery/status'
>>> sanitize_name("...")
'unnamed'

`well_known_event_schema(event_type)` ¶

(EVENT_TYPE, FIELDS) for one well-known event type id, read straight from the canonical Rust zelos-event-types::field_schemas(). The Python schemas.* wrappers populate their FIELDS/EVENT_TYPE from this, so the two languages cannot drift.

`all_well_known_event_schemas()` ¶

(EVENT_TYPE, FIELDS) for every well-known event type, in canonical order — the single enumeration point the Python schema wrappers and the drift test consume. Sourced from the canonical Rust zelos-event-types.

Predefined event schemas¶

`zelos_sdk.schemas` ¶

Predefined schemas for well-known typed events.

Pass a schema class directly to source.add_event(name, schema)::

from zelos_sdk import schemas

source = zelos_sdk.TraceSource("my_app")
log = source.add_event("log", schemas.Log)
log.log(level="info", message="hello", name="my_app", file="main.py", line=42)

Each schema class exposes

EVENT_TYPE: the stable identifier (e.g. "zelos.log.v1") the agent and frontend use to dispatch typed routing.
FIELDS: the field list.

User-defined schemas: any class with FIELDS and (optionally) EVENT_TYPE classvars works the same way — no base class required.

`AnnotationComment` ¶

Schema for zelos.annotation.comment.v1 — point/range annotations.

Fields: name (label), status (color hint), author, text (body), target_path (optional signal/panel context), range_start_ns / range_end_ns (optional duration), tags (optional comma-separated or JSON string).

`CanFrame` ¶

Schema for zelos.can.frame.v1 — undecoded CAN frames.

Used with source.add_event(name, CanFrame) — the wire event_type is set to EVENT_TYPE and the fields are emitted as FIELDS.

`CheckResult` ¶

Schema for zelos.check.result.v1 — check evaluation results.

Contains the predicate details (op, temporal, lhs, rhs), the verdict (status, reason), timing stats, optional evidence, and the full spec as JSON for provenance replay.

`Event` ¶

Schema for zelos.event.v1 — generic point events.

Fields: name (label), status (for coloring), data (JSON payload).

`Log` ¶

Schema for zelos.log.v1 — text log events.

Used with source.add_event(name, Log) — the wire event_type is set to EVENT_TYPE and the fields are emitted as FIELDS.

`Span` ¶

Schema for zelos.span.v1 — generic duration spans.

Fields: name, status, start_ns, end_ns, data. The timeline renders a horizontal bar from start_ns to end_ns.

Tracing¶

TraceSource(name, namespace=..., strict=...) ¶

add_event(name, schema, event_type=None) ¶

add_value_table(name, field_name, data) ¶

flush() ¶

get_event(name) ¶

log(name, data) ¶

log_at(time_ns, name, data) ¶

log_batch(event_name, data) ¶

log_dict(name, data) ¶

log_many(events) ¶

TraceSourceEvent ¶

log(**kwargs) ¶

log_at(time_ns, **kwargs) ¶

TraceEventFieldMetadata(name, data_type, unit=...) ¶

DataType ¶

TraceNamespace(name) ¶

drain() ¶

source_count() ¶

TracePublishClient(url=..., config=...) ¶

shutdown() ¶

TracePublishClientConfig(batch_size=..., batch_timeout_ms=...) ¶

set_batch_size(size) ¶

set_batch_timeout_ms(ms) ¶

TraceWriter(path, batch_size=..., batch_timeout_ms=..., allow_existing=..., namespace=...) ¶

close() ¶

open() ¶

TraceReader(path) ¶

Complete End-to-End Workflow¶

close() ¶

get_value_table(data_segment_id, field_path) ¶

list_data_segments() ¶

list_data_segments_in_time_range(start, end) ¶

list_fields(data_segment_id=None) ¶

Example: Discover and Query Fields¶

list_traces() ¶

open() ¶

query(data_segment_ids, fields, start, end) ¶

time_range() ¶

TraceMetadata ¶

DataSegment ¶

QueryResult ¶

to_arrow() ¶

TraceReadSource ¶

TraceReadEvent ¶

TraceReadEventField ¶

TraceStdout(log_level=..., batch_size=..., batch_timeout_ms=...) ¶

close() ¶

open() ¶

init(name=None, *, url=None, client_config=None, log_level=None, trace=True, actions=False, block=False) ¶

initialized() ¶

enable_logging(log_level=None) ¶

agent_url() ¶

init_global_client(url=None, config=None) ¶

init_global_source(name=None) ¶

log(name, data, source=...) ¶

sanitize_name(name, *, kind='source') ¶

well_known_event_schema(event_type) ¶

all_well_known_event_schemas() ¶

Predefined event schemas¶

zelos_sdk.schemas ¶

AnnotationComment ¶

CanFrame ¶

CheckResult ¶

Event ¶

Log ¶

Span ¶

`TraceSource(name, namespace=..., strict=...)` ¶

`add_event(name, schema, event_type=None)` ¶

`add_value_table(name, field_name, data)` ¶

`flush()` ¶

`get_event(name)` ¶

`log(name, data)` ¶

`log_at(time_ns, name, data)` ¶

`log_batch(event_name, data)` ¶

`log_dict(name, data)` ¶

`log_many(events)` ¶

`TraceSourceEvent` ¶

`log(**kwargs)` ¶

`log_at(time_ns, **kwargs)` ¶

`TraceEventFieldMetadata(name, data_type, unit=...)` ¶

`DataType` ¶

`TraceNamespace(name)` ¶

`drain()` ¶

`source_count()` ¶

`TracePublishClient(url=..., config=...)` ¶

`shutdown()` ¶

`TracePublishClientConfig(batch_size=..., batch_timeout_ms=...)` ¶

`set_batch_size(size)` ¶

`set_batch_timeout_ms(ms)` ¶

`TraceWriter(path, batch_size=..., batch_timeout_ms=..., allow_existing=..., namespace=...)` ¶

`close()` ¶

`open()` ¶

`TraceReader(path)` ¶

`close()` ¶

`get_value_table(data_segment_id, field_path)` ¶

`list_data_segments()` ¶

`list_data_segments_in_time_range(start, end)` ¶

`list_fields(data_segment_id=None)` ¶

`list_traces()` ¶

`open()` ¶

`query(data_segment_ids, fields, start, end)` ¶

`time_range()` ¶

`TraceMetadata` ¶

`DataSegment` ¶

`QueryResult` ¶

`to_arrow()` ¶

`TraceReadSource` ¶

`TraceReadEvent` ¶

`TraceReadEventField` ¶

`TraceStdout(log_level=..., batch_size=..., batch_timeout_ms=...)` ¶

`close()` ¶

`open()` ¶

`init(name=None, *, url=None, client_config=None, log_level=None, trace=True, actions=False, block=False)` ¶

`initialized()` ¶

`enable_logging(log_level=None)` ¶

`agent_url()` ¶

`init_global_client(url=None, config=None)` ¶

`init_global_source(name=None)` ¶

`log(name, data, source=...)` ¶

`sanitize_name(name, *, kind='source')` ¶

`well_known_event_schema(event_type)` ¶

`all_well_known_event_schemas()` ¶

`zelos_sdk.schemas` ¶

`AnnotationComment` ¶

`CanFrame` ¶

`CheckResult` ¶

`Event` ¶

`Log` ¶

`Span` ¶