Python SDK Reference¶
This section is generated from the installed zelos_sdk package.
Use the sidebar to browse modules, or jump directly:
Package Overview¶
zelos_sdk
¶
ActionFailed
¶
Bases: AgentError
An action ran and reported failure.
agent.actions.execute itself does not raise this — a failed run
reports through ActionResult.ok / ActionResult.status instead.
Raised only by callers that choose to escalate a failed result.
Example: >>> result = agent.actions.execute('battery_test', params={'target_soc': 80}) >>> if not result.ok: ... raise ActionFailed(result.status)
ActionInfo
¶
Identifier carried by every entry from agent.actions.list().
Currently exposes a single name field; kept as its own type (rather
than a plain string) so future fields (description, category, …) can ship
without breaking for info in agent.actions.list(): info.name notebooks.
Example: >>> [info.name for info in agent.actions.list()] ['battery_test']
name
instance-attribute
¶
The action's registered name, the string execute() takes.
ActionResult
¶
Result of agent.actions.execute(...) — parsed value plus a non-optional status.
A missing status from the agent reads as ActionStatus.Done. Fail/error statuses come back
here rather than raising, so a caller branches on result.ok instead of try/except.
Example: >>> result = agent.actions.execute("battery_test", params={"target_soc": 80}) >>> result.ok True
ActionSchema
¶
Schema from agent.actions.schema(...) with pre-parsed JSON objects.
Schema blobs (action_schema, ui_schema) are pre-parsed dicts.
Example: >>> agent.actions.schema("battery_test").action_schema["properties"]
ActionStatus
¶
Typed action execution status: one of ActionStatus.Pass, .Fail, .Error, .Done.
Compares equal to its lowercase wire string ("pass" / "fail" / "error" / "done"),
like a str enum.
Example: >>> result = agent.actions.execute("battery_test") >>> result.status == "pass" True
Done
instance-attribute
¶
Typed action execution status: one of ActionStatus.Pass, .Fail, .Error, .Done.
Compares equal to its lowercase wire string ("pass" / "fail" / "error" / "done"),
like a str enum.
Example: >>> result = agent.actions.execute("battery_test") >>> result.status == "pass" True
Error
instance-attribute
¶
Typed action execution status: one of ActionStatus.Pass, .Fail, .Error, .Done.
Compares equal to its lowercase wire string ("pass" / "fail" / "error" / "done"),
like a str enum.
Example: >>> result = agent.actions.execute("battery_test") >>> result.status == "pass" True
Fail
instance-attribute
¶
Typed action execution status: one of ActionStatus.Pass, .Fail, .Error, .Done.
Compares equal to its lowercase wire string ("pass" / "fail" / "error" / "done"),
like a str enum.
Example: >>> result = agent.actions.execute("battery_test") >>> result.status == "pass" True
Pass
instance-attribute
¶
Typed action execution status: one of ActionStatus.Pass, .Fail, .Error, .Done.
Compares equal to its lowercase wire string ("pass" / "fail" / "error" / "done"),
like a str enum.
Example: >>> result = agent.actions.execute("battery_test") >>> result.status == "pass" True
value
instance-attribute
¶
Wire-string view ("pass" / "fail" / "error" / "done").
Agent
¶
A handle to a Zelos agent — the entry point for signals, queries and checks.
Construction is lazy: no network call happens until the first RPC, and the
channel reconnects on its own across agent restarts. Use
zelos_sdk.connect() when you want the connection validated up front.
Example: >>> import zelos_sdk >>> agent = zelos_sdk.Agent("http://localhost:2300") >>> agent.signals()[0].path 'bus0/BMS_message/status.pack_current'
actions
instance-attribute
¶
The action registry and executor for this agent.
check
instance-attribute
¶
Typed predicate Check API. Returns a Checks proxy bound to this
agent: agent.check.that(signal, op, rhs, **kwargs) builds a spec
and runs it through _run_check. Results auto-post to any active
pytest Checker board via the ContextVar in
zelos_sdk.pytest.checker.
extensions
instance-attribute
¶
Extension management (list, start, stop, logs) for this agent.
layouts
instance-attribute
¶
Console-backed layout CRUD (Agent.layouts).
Returns an AgentLayouts handle sharing this agent's transport.
list() / show(id) / create(name, data) / update(id, ...) /
delete(id) all round-trip to the agent's layout service, which
requires console login.
target
instance-attribute
¶
The resolved agent endpoint this handle talks to.
url
instance-attribute
¶
Alias of [Self::target], matching the url= constructor keyword.
__del__()
¶
Best-effort cleanup when the handle is GC'd without close() /
with. close() is idempotent, so explicit cleanup still wins.
__new__(target=None, *, url=None)
¶
Create and return a new object. See help(type) for accurate signature.
at(paths, time, *, min_time=None, producers=None, lookback=None)
¶
Per-signal value at (or before) a specific timestamp.
Returns a Snapshot mapping each resolved signal's path to its most
recent LatestValue whose timestamp is <= time. Every dict spelling
works on it. time and min_time accept
datetime, ISO / relative string ("-1m", "now"), or raw int epoch
ns. Use min_time to floor the search. producers = None (default)
covers every connected producer.
Args:
paths: A str, Signal, or sequence thereof.
time: Anchor timestamp (datetime, str, or epoch ns).
min_time: Optional lower bound; values strictly older than this
are skipped. Also floors the window wildcards resolve against.
producers: Producer addresses; None covers every connected producer.
Returns: Snapshot: One entry per matched signal, keyed by path.
Raises:
SignalNotFound: a wildcard expanded to zero matches.
ValueError: min_time > time.
close()
¶
Close the transport and drop the catalog cache.
Idempotent. Subsequent RPCs on this Agent (or any Trace/TraceSet
derived from it) raise AgentCancelled. Prefer with Agent(...).
close_traces(paths=[])
¶
Release one or more open trace files in a single RPC. Empty (default) closes all.
connect(timeout=5.0)
¶
export(path, *, start, end=None, signals=None, paths=None, producers=None, lookback=None, overwrite=False)
¶
Export signal data to a .trz file on disk.
signals selects the producers that own the matched signals, and each
of those producers writes its whole slice of the time window into the
same output file — the exported trace is wider than the filter. When
signals is None (default) every connected producer writes. The agent
does the actual writing;
on a localhost target with localhost producers an atomic tmp-rename
strategy guards against partial files when overwrite=True and the
destination already exists.
Cross-host targets reject overwrite=False outright (the gate
has no way to atomically replace a file on a remote host); pass
overwrite=True to opt in.
Args:
path: Filesystem destination for the .trz file.
start: Start of the time window.
end: End of the window; defaults to "now".
signals: Optional str, Signal, or sequence naming the producers
to export; None (default) exports every connected producer.
(paths= is the older spelling.)
producers: Producer addresses; None (default) covers every
connected producer.
overwrite: When True, replace an existing file. Required
for cross-host writes regardless of whether the path
exists.
Returns:
ExportResult: On-disk path plus per-producer success bits.
result.ok is only True if every producer succeeded.
Raises:
ValueError: overwrite=False on a cross-host target.
FileExistsError: overwrite=False and the destination
already exists.
health()
¶
Round-trip the agent's Health RPC and return its status string.
Example: >>> agent.health() 'OK'
health_check()
¶
info()
¶
Composite identity + settings + memory + dirs.
Health-check gates; the four sub-RPCs (settings/memory/logs/configs) fan out
concurrently and partial failures land in info.failures rather than raising.
is_connected(*, timeout=1.0)
¶
Round-trip reachability probe — runs HealthCheck with a short
deadline and returns True iff the agent answered SERVING.
This costs a network round-trip (~1 ms typical, ~timeout on
failure) — it is a question, not a property. Ask it when you need the
answer; there is no cached "connected" flag to read, because the lazy
channel reconnects on its own and any such flag would lie.
Every failure reads as False. Call health() when you want the
typed error instead.
latest(paths, *, lookback=60.0, producers=None)
¶
Latest value(s): one literal path → LatestValue; a wildcard pattern or a
sequence → Snapshot, a read-only Mapping[str, LatestValue] (missing keys omitted).
A single path or pattern: no catalog match → SignalNotFound; a literal that matched but
has no rows in lookback → NoData; two producers owning that literal → AmbiguousSignal.
Sequence: misses omit keys.
producers = None (default) covers every connected producer.
Args:
paths: A str, Signal, or sequence thereof.
lookback: How far back to look for a value (default 60 s).
producers: Producer addresses; None covers every connected producer.
query(paths, *, start, end=None, producers=None, lookback=None, max_rows=0, sort=None, sort_order=None, downsample=None)
¶
Time-bounded query → SignalFrame (Arrow IPC). Empty frame when the
signals resolved but no rows fall in the window.
A path that names no signal raises SignalNotFound rather than
returning a frame quietly missing that column — including a literal the
catalog lacks that no producer answers for.
paths accepts a str, Signal, or sequence. start/end accept
datetime, ISO 8601, or relative strings ("-1m", "now").
downsample = N runs the agent's M4 strategy with N buckets and is mutually
exclusive with max_rows / sort_order. max_rows = 0 means no cap.
producers = None (default) covers every connected producer. Wildcards
expand against the signals active inside [start, end]. Convert via
frame.to_pandas() or frame.series(name).
resolve_paths(paths, *, producers=None, lookback=60.0)
¶
segments(producers=None)
¶
Live data segments. producers = None (default) covers every connected
producer; pass a str or a sequence of addresses to narrow it.
Per-producer warnings (an unreachable producer, a stale segment) are
emitted as RuntimeWarning so partial results never silently hide
failures. Catch with warnings.catch_warnings() to inspect them.
signal(path)
¶
Path-only Signal reference. Shorthand for
Signal.from_path(path) — no catalog roundtrip, no
SignalNotFound. The path is verified server-side at the next
check / latest / query invocation, so the resolver picks
the freshest matching segment at evaluation time.
Use this in tests and extension consumers that need a typed signal handle before (or independent of) the catalog being populated.
signals(producers=None, lookback=60.0)
¶
Signal catalog. producers = None (default) covers every connected
producer; pass a str or a sequence of addresses to narrow it.
An explicit call always fetches — the short-lived cache behind
wildcard resolution would otherwise hand back a catalog from before the
producer you just started. Per-producer failures land in
catalog.warnings and are also emitted as RuntimeWarning.
Args:
producers: Producer addresses; None covers every connected producer.
lookback: Seconds since a signal last produced a sample (default 60).
trace(path)
¶
Open a trace file. Returns a Trace handle sharing this agent's transport.
Raises TraceNotFound if the agent doesn't have the file. The
returned Trace is a context manager — prefer
with agent.trace(path) as trace: over manual trace.close().
traces(paths)
¶
Open multiple traces as one queryable set; routes per-signal to the owning file.
paths is a sequence of files — the runs are then named by file stem —
or a {label: path} mapping when you want to name them yourself
(agent.traces({"baseline": "a.trz", "candidate": "b.trz"})).
Raises ValueError on empty paths, TraceNotFound if any file is unknown.
TraceSet is a context manager — prefer
with agent.traces([...]) as traces: over manual traces.close().
watch(paths=None, *, interval=1.0, lookback=60.0, producers=None, until=None, on_error='raise', max_consecutive_errors=5)
¶
Yield a snapshot of the latest values every interval until until.
producers = None (default) covers every connected producer.
window(paths, *, start, duration, producers=None, lookback=None, display_fps=0)
¶
Replay snapshot + change-stream over a fixed time window.
Returns a ReplayWindow with two parts: the snapshot (latest
value per signal at start) and the changes (every subsequent
update inside the window). start accepts datetime, str, or
int ns; duration accepts int/float seconds, a
datetime.timedelta, or a string like "5s". producers = None
(default) covers every connected producer. display_fps = 0 means
no thinning.
Args:
paths: A str, Signal, or sequence thereof.
start: Window start.
duration: Window length.
producers: Producer addresses; None covers every connected producer.
display_fps: Optional display-rate cap on the change stream.
Returns: ReplayWindow: Snapshot + changes for the window.
Raises:
SignalNotFound: a wildcard expanded to zero matches.
ValueError: duration <= 0 or display_fps < 0.
AgentActions
¶
Agent action API (Agent.actions): list, schema, execute over gRPC (no caching).
Callers pass Python dicts; they are JSON-encoded at the wire boundary.
Example: >>> [info.name for info in agent.actions.list()] ['battery_test'] >>> result = agent.actions.execute("battery_test", params={"target_soc": 80}) >>> result.ok True
execute(action, params=None, timeout=None)
¶
Run action with optional params. Status lives on ActionResult, not exceptions.
Args:
timeout: Seconds; None uses the agent default (typically 30 s).
Returns:
ActionResult: Always — a failed run reports through result.ok /
result.status.
Raises: Internal: the agent answered without a result payload.
Example: >>> result = agent.actions.execute("battery_test", params={"target_soc": 80}) >>> result.ok True
list()
¶
Registered actions from actions_list (order as on the agent; not cached).
Returns:
list[ActionInfo]: One entry per action, with at least a .name field.
Raises: AgentUnavailable: agent is unreachable.
schema(action, current_values=None)
¶
JSON-Schema and ui-schema for action, pre-parsed as dicts.
Args:
current_values: Partial form data for dynamic schemas; None omits from the wire.
Returns: ActionSchema: The action's form schema.
Raises: Internal: the agent answered without a schema payload.
Example: >>> agent.actions.schema("battery_test").action_schema["properties"]
AgentCancelled
¶
Bases: AgentError
The RPC was cancelled or its deadline expired.
Example: >>> try: ... agent.query(['bus0/BMS_message/status.pack_current'], start='-5m', timeout=0.001) ... except AgentCancelled: ... pass
AgentError
¶
Bases: Exception
Base class for every Agent SDK wire or RPC failure.
Every subclass carries suggestions, matches, and cause
attributes (empty/None unless the specific subclass fills them in),
so a broad handler can inspect any of them without an AttributeError.
Example: >>> try: ... agent.latest('bus0/typo.voltage') ... except AgentError as exc: ... exc.cause
AgentExtensions
¶
Sub-namespace exposing the agent's extension lifecycle API.
Reached via Agent.extensions. The handle holds a transport clone
plus a shared handle to the parent agent's signals catalog cache —
start(), stop(), and restart() invalidate that cache so the
next agent.signals() call sees any new producers an extension
brought online (or saw a producer go away).
list() enumerates installed extensions; info(id) and
readme(id) return descriptive metadata; config_schema(id) and
last_config(id) return the JSON-Schema and saved config (decoded
to dict). start(id, config={...}) and restart(id, config={...})
launch (or relaunch) the extension and return a typed
ExtensionStart carrying the resolved version + pid.
Example: >>> [e.id for e in agent.extensions.list()] ['zeloscloud.zelos-extension-can'] >>> agent.extensions.start( ... "zeloscloud.zelos-extension-can", ... config={"buses": [{"interface": "demo"}]}, ... ).pid 4821
config_schema(id, version=None)
¶
Returns the extension's JSON-Schema config, decoded into a dict.
None means "no schema declared"; an empty dict means
"configured but with no fields" (proto schema_json = "{}").
info(id, version=None)
¶
Fetch metadata for a specific extension.
Returns the extension's manifest data: id, version, description,
any actions/signals it provides, and so on. Useful for building
catalog UIs or verifying an extension is installed before
start()-ing it.
Args:
id: Extension identifier (e.g. "zeloscloud.zelos-extension-can").
version: Specific version to query; None resolves to the
currently-installed version.
Returns: ExtensionInfo: The extension's manifest.
Raises:
ExtensionError: nothing matching id is installed.
Internal: the agent answered without a manifest.
Example: >>> agent.extensions.info("zeloscloud.zelos-extension-can").version
last_config(id, version=None)
¶
Returns the last saved config, decoded into a dict. None means
"never configured"; an empty dict means "configured but with no
fields" (proto config_json = "{}").
list()
¶
List every extension installed in the agent.
The returned ExtensionEntry objects carry id, version, current
ExtensionState (Installed or Running), and pid (when running).
The list is not cached, so it always reflects the agent's
current view.
Returns: list[ExtensionEntry]: One entry per installed extension.
Raises: AgentUnavailable: agent is unreachable.
readme(id, version=None)
¶
Fetch the extension's README markdown.
Useful for building marketplace UIs or showing inline help in a notebook. Returns the raw markdown verbatim, or an empty string when the extension shipped no README — render it directly with whatever tooling fits the surface.
Args:
id: Extension identifier.
version: Specific version; None resolves to the
currently-installed version.
Returns:
str: Markdown text, or "" when the extension shipped no README.
restart(id, config=None)
¶
Stop then start an extension, applying config on the way back up.
See [Self::start] for config = None vs config = {} semantics.
Example: >>> agent.extensions.restart("zeloscloud.zelos-extension-can")
start(id, config=None)
¶
config = None means "no config" (transport sends nothing on
the wire). config = {} means "empty config" (transport sends
"{}" on the wire). The returned ExtensionStart already carries the
resolved version and pid, so no follow-up call is needed.
Example: >>> agent.extensions.start("zeloscloud.zelos-extension-can").version
stop(id)
¶
Stop a running extension.
Asks the agent to terminate the extension process; the
signals-catalog cache is invalidated so the next
agent.signals() call reflects the producer leaving. Idempotent
— stopping an already-stopped extension is a no-op.
Args: id: Extension identifier.
Raises: ExtensionError: agent reported an error tearing down the extension process.
AgentInfo
¶
Composite agent identity and runtime state.
Returned by Agent.info(). Bundles the connection target, latest
health-check result, and best-effort introspection: the agent's
AgentSettings, current memory usage, and the on-disk paths the
agent uses for logs and configs.
info() runs four optional sub-RPCs concurrently behind the scenes
(settings + memory + logs-dir + configs-dir); each of them may fail
independently of the others. Failed sub-RPC names appear in
failures (a stable list of strings — "SettingsGet",
"SystemGetMemory", "SystemGetDirLogs", "SystemGetDirConfigs" —
useful for grep-friendly notebook checks). Only the gating
health-check is fatal; everything else is fail-soft.
Example: >>> info = agent.info() >>> info.target, info.health ('http://127.0.0.1:2300', 'OK') >>> info.failures []
configs_dir
instance-attribute
¶
Directory the agent keeps its configs in; None when that sub-call failed.
failures
instance-attribute
¶
Names of the optional sub-calls that failed, e.g. ["SystemGetMemory"].
health
instance-attribute
¶
Result of the gating health check; "OK" when the agent answered.
logs_dir
instance-attribute
¶
Directory the agent writes its logs to; None when that sub-call failed.
memory_bytes
instance-attribute
¶
Memory the agent process is using, in bytes; None when that sub-call
failed.
settings
instance-attribute
¶
The agent's settings; None when that sub-call failed.
target
instance-attribute
¶
The agent endpoint this info was read from.
AgentLayouts
¶
Sub-namespace exposing the agent's layout CRUD API.
Reached via Agent.layouts. The handle holds a transport clone. Every
method round-trips to the agent's console-backed layout service, which
requires the agent to be logged in — unauthenticated calls surface as the
usual typed AgentError subclasses.
list() enumerates every layout visible to the logged-in user;
show(id) fetches one; create(name, data, *, is_personal=False) saves a
new layout and returns it; update(id, *, name=..., data=..., is_personal=...)
patches an existing layout (read-modify-write — omitted fields keep their
current values); delete(id) removes one.
Every update is also recorded in a Google Docs-style version history:
versions(id) lists a layout's versions (newest first, metadata only);
show_version(id, version) fetches one version with its data snapshot;
restore(id, version) restores the layout to a previous version
(appended to the history as a new version — no history is lost);
set_version_label(id, version, label) names a version (None clears —
labeled versions are exempt from retention pruning).
Example: >>> created = agent.layouts.create("My Dashboard", {"panels": []}) >>> agent.layouts.update(created.id, name="Renamed").name 'Renamed' >>> agent.layouts.delete(created.id)
create(name, data, *, is_personal=False)
¶
Create a new layout and return it.
Args:
name: Human-readable layout name.
data: Arbitrary JSON-serializable dict (panels, config, …).
is_personal: When True, scope the layout to the current user
rather than the team. Defaults to False.
Returns: Layout: The newly-created layout, with its server-assigned id.
Raises:
ValueError: data is not JSON-serializable.
AgentError: the agent is not logged in, or returned malformed data.
delete(id)
¶
Delete a layout by id.
Args: id: The layout's UUID (as a string).
Raises:
ValueError: id is not a valid UUID.
AgentError: the layout does not exist, or the agent is not logged in.
list()
¶
List every layout visible to the logged-in console user.
Returns: list[Layout]: One entry per saved layout (may be empty).
Raises: AgentUnavailable: agent is unreachable. AgentError: agent is not logged in, or returned malformed data.
restore(id, version)
¶
Restore a layout to a previous version.
The restore is appended to the history as a new version (Google Docs-style — no history is lost), so it can itself be undone by restoring an earlier version.
Args: id: The layout's UUID (as a string). version: The version number to restore (versions start at 1).
Returns: Layout: The layout after the restore.
Raises:
ValueError: id is not a valid UUID, or version is outside the
valid range (1 to 4294967295).
AgentError: the layout or version does not exist, or the agent is
not logged in.
set_version_label(id, version, label)
¶
Set or clear a layout version's label (name).
Labeled versions are kept forever — they are exempt from the
version-retention pruning. The label is trimmed before sending; pass
None (or a blank string) to clear it.
Args:
id: The layout's UUID (as a string).
version: The version number to label (versions start at 1).
label: The label to set, or None to clear the current label.
Returns: LayoutVersion: The updated version metadata.
Raises:
ValueError: id is not a valid UUID, or version is outside the
valid range (1 to 4294967295) — both raised before any network
call. Also raised when the server rejects the request: a label
longer than 100 characters, or naming a version when the
layout already has 40 named versions.
AgentError: the layout or version does not exist, or the agent is
not logged in.
show(id)
¶
Fetch a single layout by id.
Args: id: The layout's UUID (as a string).
Returns: Layout: The requested layout.
Raises:
ValueError: id is not a valid UUID.
AgentError: the layout does not exist, or the agent is not logged in.
show_version(id, version)
¶
Fetch a single layout version, including its data snapshot.
Args: id: The layout's UUID (as a string). version: The version number to fetch (versions start at 1).
Returns:
LayoutVersion: The requested version, with its data snapshot.
Raises:
ValueError: id is not a valid UUID, or version is outside the
valid range (1 to 4294967295).
OverflowError: version does not fit in a signed 64-bit integer.
AgentError: the layout or version does not exist, or the agent is
not logged in.
update(id, *, name=None, data=None, is_personal=None)
¶
Update an existing layout and return the updated version.
Any kwarg left as None keeps its current value. The full
(name, data, is_personal) triple is always sent on the wire, so any
omitted field is filled from the current layout via a show() round-trip
(skipped only when all three are provided). Sending the full triple keeps
updates correct against older bundled agents, which predate partial
updates and would read a missing is_personal as False (silently
un-sharing a personal layout). Pass at least one kwarg — an all-None
call raises ValueError without touching the network.
Args:
id: The layout's UUID (as a string).
name: New name, or None to keep the current name.
data: New payload dict, or None to keep the current payload.
is_personal: New personal/team scope, or None to keep the current
scope.
Returns: Layout: The updated layout.
Raises:
ValueError: id is not a valid UUID, data is not
JSON-serializable, or all three kwargs are None.
AgentError: the layout does not exist, or the agent is not logged in.
versions(id)
¶
List a layout's version history, newest first.
Version entries are metadata-only — their data snapshots come back
as None. Fetch a specific snapshot with show_version().
Args: id: The layout's UUID (as a string).
Returns: list[LayoutVersion]: One entry per saved version, newest first.
Raises:
ValueError: id is not a valid UUID.
AgentError: the layout does not exist, or the agent is not logged in.
AgentSettings
¶
Raw settings snapshot from SettingsGet. Wire fields cross verbatim;
the Python AgentSettings dataclass in agent/info.py re-exports
the same data (memory/disk strings stay strings — they are
human-readable like "25%" / "10GB" and are parsed against actual
host capacity at runtime).
Plot UX defaults and subscription / version-migration fields are omitted (not part of the SDK's public surface).
memory_limit / disk_limit are strings (e.g. "25%", "10GB"),
not u64 — the agent stores human-readable strings and parses them
against actual host capacity at runtime.
Example: >>> settings = agent.info().settings >>> settings.store_type, settings.memory_limit ('memory', '25%')
data_retention
instance-attribute
¶
datetime.timedelta; None means time-based pruning is disabled (or
the wire value would overflow chrono::Duration — purely defensive,
realistic retention values stay far below i64::MAX seconds). Call
.total_seconds() for raw seconds.
dev_mode
instance-attribute
¶
True when the agent runs with developer features enabled.
disk_limit
instance-attribute
¶
e.g. "10GB", "25%"; None means unlimited.
log_retention
instance-attribute
¶
See [Self::data_retention]; None means no limit.
memory_limit
instance-attribute
¶
e.g. "25%", "10GB"; None means unlimited.
store_path
instance-attribute
¶
None means platform-default data directory. Only meaningful when
store_type == "disk".
store_type
instance-attribute
¶
Lowercase "memory" (ArrowStore, default), "disk" (ParquetStore), or
"metadata" (catalog only, no data — headless agents) — matches the
StoreType serde encoding so notebook code lines up with CLI / app
JSON debugging.
AgentUnavailable
¶
Bases: AgentError
No agent answered at the target.
Covers a dead connection, a closed port, and an agent that was killed or restarted mid-RPC.
Example: >>> try: ... agent.latest('bus0/BMS_message/status.pack_current') ... except AgentUnavailable: ... reconnect()
AmbiguousSignal
¶
Bases: AgentError
A path or pattern matched more than one signal.
Carries the matching catalog entries as matches: list[Signal]; pass
one of them directly to disambiguate.
Example: >>> try: ... agent.latest('*/BMS_message/status.pack_current') ... except AmbiguousSignal as exc: ... [s.path for s in exc.matches] ['bus0/BMS_message/status.pack_current', 'bus1/BMS_message/status.pack_current']
CheckEvidence
¶
The single sample that decided a CheckResult's verdict: which signal,
at what time, with what value. None on result.evidence when the
check has no one deciding sample (e.g. a passing count check).
Example: >>> result = agent.check.that( ... "bus0/BMS_message/status.pack_current", "<", 100, temporal="always", last="5m" ... ) >>> result.evidence.signal 'bus0/BMS_message/status.pack_current'
arrow_value
instance-attribute
¶
Raw 1-row, 1-column Arrow IPC RecordBatch bytes (same convention as
LatestSignalValue.arrow_value). None when the cell is NULL or the
column type is unsupported.
rhs_value
instance-attribute
¶
Typed value of the right-hand signal at the evidence row, or None when the check compared against a literal.
signal
instance-attribute
¶
Clean path of the signal the evidence sample came from.
time_ns
instance-attribute
¶
When the evidence sample was recorded, epoch nanoseconds.
value
instance-attribute
¶
Typed sample value at the evidence row. Decoded once from the
underlying 1-row, 1-column Arrow IPC payload — returns Python
bool / int / float / str / bytes matching the column
type, or None for NULL / unsupported types.
CheckResult
¶
Outcome of running a CheckSpec against the agent.
bool(result) is True iff status == "pass". result.failed is
True for both fail (predicate violated, evidence available) and
error (precondition failed, no evaluation possible).
Example: >>> result = agent.check.that("bus0/BMS_message/status.pack_current", "<", 100) >>> result.passed True
count
instance-attribute
¶
Populated for temporal="count"; otherwise None.
evidence
instance-attribute
¶
The sample that decided the verdict — the extreme on a pass, the first
violation on a fail. None when the check could not be evaluated.
failed
instance-attribute
¶
True for Status::Fail (predicate violated) or Status::Error
(precondition failed). The reason and evidence distinguish them.
fired_at_ns
instance-attribute
¶
Wall-clock UTC nanoseconds the agent captured the moment this check
finished evaluating. Same field the checkerboard's timestamp
column renders for live, trace, and literal-only checks alike.
message
instance-attribute
¶
Free-form human message for error/type_mismatch/signal_not_found.
name
instance-attribute
¶
Name of the spec this result came from.
passed
instance-attribute
¶
True iff the predicate held over the entire range (or for
count, always — count results carry the count, not a verdict).
query_ms
instance-attribute
¶
Time spent in the executor's query plan / SQL execution, in milliseconds.
reason
instance-attribute
¶
Why the check ended as it did ("ok", "signal_not_found",
"type_mismatch", ...).
status
instance-attribute
¶
The verdict as a wire string: "pass", "fail", or "error".
wall_ms
instance-attribute
¶
Total wall-clock time the agent spent on the check, in milliseconds.
CheckResults
¶
Bases: list
A run's results: a plain list that also knows its own tally.
bool(results) is "nothing failed and nothing errored" — not
"non-empty" — so a suite reads as one verdict.
Example: >>> results = agent.check.suite("checks/battery.json") >>> len(results.failed) 0 >>> results.raise_if_failed()
errors
property
¶
The results that never got as far as a verdict.
failed
property
¶
The results whose predicate did not hold.
passed
property
¶
The results whose predicate held.
raise_if_failed()
¶
Raise AssertionError listing every check that did not pass —
each on its own line, with its predicate, window and evidence.
For a single check, assert agent.check.that(...): a result is
truthy when it passed.
CheckSpec
¶
A typed check specification — predicate + time range + temporal quantifier.
Constructed by the Python agent.check.that(...) / agent.check.count(...)
helpers. Field access via typed getters (signal, op, temporal,
start_ns, end_ns, rhs).
Example: >>> spec = CheckSpec( ... "pack_current_ok", ... signal="bus0/BMS_message/status.pack_current", ... op="<", ... rhs=100.0, ... ) >>> spec.op '<'
duration_ns
instance-attribute
¶
How long the predicate must hold for the check to pass, in nanoseconds;
None means a single sample is enough.
end_ns
instance-attribute
¶
End of the window the check evaluates, epoch nanoseconds; None for a
latest check.
lhs
instance-attribute
¶
("literal", <python_value>) or ("signal", <signal_path>).
Symmetric with .rhs so Python wrappers render either side
uniformly on the rich-table board.
lookback
instance-attribute
¶
Seconds of history a latest check looks back over when it resolves its signal.
name
instance-attribute
¶
The check's name, echoed on every CheckResult it produces.
op
instance-attribute
¶
The comparison, in its canonical spelling ("<", "is_close", ...).
producers
instance-attribute
¶
Producer addresses the signal resolves against; empty means every connected producer.
rhs
instance-attribute
¶
("literal", <python_value>) or ("signal", <signal_path>) for
binary specs; None for unary specs (is_empty, is_positive,
...). The Python wrapper uses this to render the RHS on the
rich-table board — the None case renders as an empty cell.
start_ns
instance-attribute
¶
Start of the window the check evaluates, epoch nanoseconds; None for a
latest check.
strict_resolution
instance-attribute
¶
Whether the resolver should opt out of newest-segment-wins
disambiguation for this spec. See agent.check.that(..., strict=...).
temporal
instance-attribute
¶
How samples are quantified: "latest", "always", "ever", "never",
or "count".
tolerance
instance-attribute
¶
Tolerance for is_close / is_approximately; None for any
other op. Returned as a 3-tuple (rel_tol, abs_tol, nan_ok).
__new__(name, signal=None, op=None, *, lhs_value=None, lhs_is_null=False, rhs=None, rhs_signal=None, rhs_is_null=False, temporal='latest', start_ns=None, end_ns=None, lookback=60.0, tolerance=None, duration_ns=None, producers=[], strict=False)
¶
Create and return a new object. See help(type) for accurate signature.
from_json(_cls, s)
classmethod
¶
Parse and validate a CheckSpec from a JSON string.
The same loader the CLI uses for suite files, so a spec built as a dict
round-trips through CheckSpec.from_json(json.dumps(spec)) and is held
to exactly the same schema.
Checks(parent)
¶
Proxy returned by Agent.check and Trace.check: .that() / .count() / .run() / .suite().
Example: >>> result = agent.check.that("bus0/BMS_message/status.pack_current", "<", 100) >>> result.passed True
count(lhs, op='is_true', rhs=None, *, producers=None, last=None, start=None, end=None, name=None, lookback=DEFAULT_LOOKBACK_S, rel_tol=None, abs_tol=None, nan_ok=False, strict=False)
¶
Count the satisfying samples. Posts to any active Checker board.
Returns the :class:CheckResult, not a bare number: int(result)
is the count, and it raises rather than reading 0 off a result
that never got as far as counting. Requires at least one signal
operand (a literal-only count is degenerate — there are no
samples to enumerate). Shape-changes on producers the same
way as :meth:that.
strict matches :meth:that — pass True to require a
unique segment per signal path (opt out of newest-segment-wins
disambiguation).
run(spec, *, producers=None, interval_s=DEFAULT_DURATION_INTERVAL_S, display_lhs=None, display_rhs=None, pure_literal=None)
¶
Run a pre-built :class:CheckSpec and post the result(s).
Lower-level seam: bypasses the kwargs ergonomics. Useful for replaying specs loaded from JSON suites or constructed programmatically.
Same shape-change as :meth:that — scalar for a single
producer, dict for fan-out. Per-producer transport failures
synthesize :class:CheckResult (status="error",
reason="internal_error") entries so the caller always sees
one entry per requested producer.
temporal="for_duration" / "within_duration" specs
dispatch into a client-side polling loop (cadence
interval_s) — the wire-level spec records the original
temporal so the artifact and any future streaming RPC stay
identical; only the execution path differs.
Routing: signal-bearing specs go through the agent's
gRPC executor (point-in-time or polling). Literal-only specs
(pure literals or TraceSourceCacheLastField operands whose
values were captured at spec-build time) are evaluated
in-process — no gRPC round-trip, no agent dependency. Same CheckResult
shape either way; same Checker board / artifact output.
Emission: while a Checker session is active, each
posted result also publishes a zelos.check.result.v1 event
through a shared TraceSource("checks"). Outside a session
there is no sink, no emission, and the literal-only path stays
pure in-process.
suite(path, *, producers=None, last=None, start=None, end=None, lookback=DEFAULT_LOOKBACK_S, strict=False)
¶
Run a JSON check suite and post each result to the Checker board.
path may be a single .json file, a directory (every
*.json inside, lex order), a glob pattern, or a list of any
of the above.
Each JSON file holds an array of spec dicts. An operand is
{"signal": "<path>"} for a catalog lookup, or a bare JSON scalar
for a literal. range is required for every temporal but
latest, and last= / start=+end= fill it in when the
file omits it. One entry reads:
[
{
"name": "pack current under load",
"lhs": {"signal": "bus0/BMS_message/status.pack_current"},
"op": "<",
"rhs": 120.0,
"temporal": "always",
"range": {"start_ns": 1000, "end_ns": 9000}
}
]
Element shape follows :meth:run: each spec yields a scalar
:class:CheckResult for single-producer (the default), or a
{address: CheckResult} dict for fan-out.
last / start+end (mutually exclusive) inject a
runtime window into specs whose JSON omits range. Same JSON
suite ports between live (with e.g. last=30.0 or
start="-30s", end="now") and trace (TraceChecks.suite
fills the trace's full extent automatically). lookback
sets the resolution-freshness default that flows onto every
spec on the wire.
strict is the suite-level default for
strict_resolution — applied only to entries whose JSON
omits the field (entries that set it explicitly win). Mirrors
the lookback injection semantics.
The signal-bearing specs are batched into one LiveCheck
RPC against the agent (per-agent customization: each producer
in producers gets the same spec list, sent in one call).
Literal-only specs (pure literals or captured
TraceSourceCacheLastField references) evaluate in-process,
with no RPC. Each result posts to the active Checker board in
input order regardless of which path it took.
that(lhs, op='is_true', rhs=None, *, producers=None, last=None, start=None, end=None, temporal=None, name=None, lookback=DEFAULT_LOOKBACK_S, rel_tol=None, abs_tol=None, nan_ok=False, duration_s=None, interval_s=DEFAULT_DURATION_INTERVAL_S, nonblocking=False, strict=False)
¶
Assert a typed predicate lhs op rhs (or unary lhs op).
Both lhs and rhs accept the same shapes:
bool/int/float/str→ literal.Signalhandle (fromagent.signals().by_path(...)ortrace.signals().by_path(...)) → signal reference; the agent fetches the underlying samples (full range foralways/ever/never/count, latest forlatest) and evaluates server-side.TraceSourceCacheLastField→ the current cached value is captured via.get()at spec build time and sent on the wire as a literal — the agent stores the check as an event with the source path preserved for naming / replay.SignalSeries(anlhscolumn of a queried frame) → check the thing you just queried. A signal-backed series resolves to its signal and to its own time extent, so the window is the one you queried; passinglast=/start=/end=alongside it is an error. A derived series (cell_a - cell_b) is evaluated in-process over its own samples.
op defaults to "is_true" so any Python predicate
result becomes a checked assertion routed through the agent
— check.that(isinstance(foo, int)),
check.that(x is None), check.that(0 < y < 10) all
work without needing an explicit op. Encourages the rich
check artifact over plain assert.
rhs is omitted for unary ops (is_empty,
is_positive / is_not_positive, is_negative /
is_not_negative, is_true, is_false).
rel_tol / abs_tol / nan_ok set the tolerance for
the tolerance ops (is_close / is_around,
is_approximately / ~=). Each defaults to None /
False — left unset, the op's stdlib default applies
(is_close → math.isclose: rel_tol=1e-9,
abs_tol=0; is_approximately → pytest.approx:
rel_tol=1e-6, abs_tol=1e-12). Override any subset.
All combinations route through the agent CheckMulti RPC —
the agent is the single eval/storage point.
Shape-changes by the producers argument:
producers=None(default, the local store) or a single address → one :class:CheckResult.producers=()or a multi-address list → fan-out; returns{address: CheckResult}.
None is the local store, not every connected producer as it
is on :meth:Agent.query.
With no last / start / end, defaults to
temporal="latest" (most recent sample). When a range is
given, defaults to temporal="always" (every sample must
satisfy). last= reads the duration grammar — 120,
"2m", timedelta(minutes=2).
Mistakes raise here rather than becoming an error row:
comparing two incomparable literals is a TypeError, and a
window on a check with no samples to quantify over is a
ValueError.
temporal="for_duration" + duration_s=... asserts the
predicate holds continuously from now until the duration
elapses; temporal="within_duration" asserts the predicate
becomes true at least once before the duration elapses. Both
run as a client-side polling loop over temporal="latest"
with cadence interval_s (default
:data:DEFAULT_DURATION_INTERVAL_S). On :class:TraceChecks
the same temporals desugar to Always / Ever over
[trace.start_ns, trace.start_ns + duration_s*1e9] — the
wire spec is identical, the executor differs.
ColumnMetadata
¶
Per-column metadata attached to a query result: which signal a column holds, its producer, segment, and the time range it covers.
signal keeps its wire name (not name) because name collides with
__name__-style lookups when used as a pandas/pyarrow column header.
Example: >>> frame = agent.query(["bus0/BMS_message/status.pack_current"], start="-5m") >>> frame.columns[0].signal 'pack_current'
data_segment_id
instance-attribute
¶
The data segment this column belongs to, None when segmentation is
not in play.
end_time_s
instance-attribute
¶
End of the covered time range, in seconds, None when unknown.
message
instance-attribute
¶
The column's message ({source}/{message}.{signal}).
path
instance-attribute
¶
Clean signal path: "{source}/{message}.{signal}". Producer / segment
disambiguation lives on the matching column name and on the
producer / data_segment_id fields; this getter is just the
human-readable label for display and notebook lookups.
producer
instance-attribute
¶
The producer this column came from, None when the query carries no
producer metadata.
signal
instance-attribute
¶
The column's wire signal name (see the class docstring for why this
isn't called name).
source
instance-attribute
¶
The column's source ({source}/{message}.{signal}).
start_time_s
instance-attribute
¶
Start of the covered time range, in seconds, None when unknown.
trace_path
instance-attribute
¶
The trace file this column was read from, None for a live query.
ConnectionTargetError
¶
Bases: AgentError, ValueError
The agent target string could not be parsed into a URL.
Also a ValueError, so except ValueError: catches it too.
Example: >>> try: ... zelos_sdk.connect('not a url') ... except ConnectionTargetError as exc: ... str(exc) "could not parse target 'not a url'"
DataSegment
¶
Python wrapper for a data segment.
Represents a segment of trace data with metadata about its time range and producer.
ExitInfo
¶
Exit information from a terminated extension process.
Both fields are optional because some platforms only report one (POSIX signal vs. exit code).
Example: >>> entry = agent.extensions.list()[0] >>> entry.last_exit.code if entry.last_exit else None 0
ExportProducerResult
¶
One producer's outcome from Agent.export(...) / Trace.export(...), keyed
by producer address in the parent ExportResult.results dict.
Example: >>> result = agent.export("bms.trz", start="-1m", paths=["can/Battery.*"]) >>> for producer, r in result.results.items(): ... if not r.ok: ... print(producer, r.error)
ExportResult
¶
Result of Agent.export(...) / Trace.export(...).
Bundles the on-disk path the agent (or local tmp-rename strategy)
wrote to with a results dict keyed by producer address. Each entry
is an ExportProducerResult carrying that producer's per-shard
success bit and any error message.
ok is True only when every producer shard succeeded — partial
failures surface as ok=False with per-producer detail in results
rather than raising. This lets callers branch on partial-success
outcomes without a try/except.
Example: >>> result = agent.export("bms.trz", start="-1m", paths=["can/Battery.*"]) >>> result.ok True
format
instance-attribute
¶
The trace format the agent wrote ("trz2" / "legacy"), None when
it did not report one.
ok
instance-attribute
¶
True only when every producer's shard succeeded.
path
instance-attribute
¶
Path of the trace file the export wrote.
results
instance-attribute
¶
Per-producer results keyed by producer address. Builds the dict
directly from &self.results — cloning the inner HashMap first
would double the allocation work for no benefit.
open()
¶
Open the file this export wrote as a Trace, through the same agent.
Example: with agent.export("run.trz", start="-5m").open() as trace: frame = trace.query("bus0/BMS.voltage")
ExtensionEntry
¶
Catalog row returned by agent.extensions.list().
host_type and app_contribution_kind are exposed as their canonical
lowercase wire strings (e.g. "agent", "web_app").
Example: >>> entry = agent.extensions.list()[0] >>> entry.id, entry.version, entry.state.value ('zeloscloud.zelos-extension-can', '0.1.10', 'running')
app_contribution_kind
instance-attribute
¶
e.g. "web_app"; None for non-app extensions.
author
instance-attribute
¶
Manifest author; None when the manifest omits it.
categories
instance-attribute
¶
Manifest categories, used for grouping in the app.
description
instance-attribute
¶
One-line summary from the manifest.
dev_mode
instance-attribute
¶
True when the extension is loaded from a local development checkout.
entry
instance-attribute
¶
Entry point the host runs; None for an extension with nothing to run.
homepage
instance-attribute
¶
Project homepage URL; None when the manifest omits it.
host_type
instance-attribute
¶
"agent" or "app".
icon_path
instance-attribute
¶
Absolute path to the extension's icon file; None when it ships none.
id
instance-attribute
¶
Fully qualified extension id, e.g. "zeloscloud.zelos-extension-can".
keywords
instance-attribute
¶
Manifest keywords, used for search.
last_exit
instance-attribute
¶
How the extension process last terminated; None when it has never run.
name
instance-attribute
¶
Display name from the extension's manifest.
pid
instance-attribute
¶
Process id while the extension runs; None when it is not running.
repository
instance-attribute
¶
Source repository URL; None when the manifest omits it.
state
instance-attribute
¶
Whether the extension is installed or currently running.
version
instance-attribute
¶
Installed version.
zelos_version
instance-attribute
¶
Zelos version range the extension declares it works with.
ExtensionError
¶
Bases: AgentError
An extension lifecycle call failed.
Example: >>> try: ... agent.extensions.start('missing-extension') ... except ExtensionError: ... pass
ExtensionInfo
¶
Rich info returned by agent.extensions.info(id).
Compared to ExtensionEntry, this carries install_path, readme_path,
etc. but not author or last_exit (those are list-time concerns).
Example: >>> agent.extensions.info("zeloscloud.zelos-extension-can").install_path '/opt/zelos/extensions/zeloscloud.zelos-extension-can/0.1.10'
app_contribution_kind
instance-attribute
¶
What an app extension contributes, e.g. "web_app"; None for an agent
extension.
categories
instance-attribute
¶
Manifest categories, used for grouping in the app.
description
instance-attribute
¶
One-line summary from the manifest.
dev_mode
instance-attribute
¶
True when the extension is loaded from a local development checkout.
entry
instance-attribute
¶
Entry point the host runs; None for an extension with nothing to run.
homepage
instance-attribute
¶
Project homepage URL; None when the manifest omits it.
host_type
instance-attribute
¶
Which host runs the extension: "agent" or "app".
icon_path
instance-attribute
¶
Absolute path to the extension's icon file; None when it ships none.
id
instance-attribute
¶
Fully qualified extension id, e.g. "zeloscloud.zelos-extension-can".
install_path
instance-attribute
¶
Directory the extension is installed in.
keywords
instance-attribute
¶
Manifest keywords, used for search.
name
instance-attribute
¶
Display name from the extension's manifest.
pid
instance-attribute
¶
Process id while the extension runs; None when it is not running.
readme_path
instance-attribute
¶
Absolute path to the extension's README; None when it ships none.
repository
instance-attribute
¶
Source repository URL; None when the manifest omits it.
state
instance-attribute
¶
Whether the extension is installed or currently running.
version
instance-attribute
¶
Installed version.
zelos_version
instance-attribute
¶
Zelos version range the extension declares it works with.
ExtensionStart
¶
Result of agent.extensions.start(id, ...) / restart(id, ...).
Carries the caller-known id alongside the resolved pid and
version the agent actually launched — no follow-up
agent.extensions.list() call is needed to learn either.
Example: >>> started = agent.extensions.start("zeloscloud.zelos-extension-can") >>> started.pid, started.version (4821, '0.1.10')
ExtensionState
¶
Runtime state of an extension.
Two-member enum-like class: ExtensionState.Installed and
ExtensionState.Running. Wire form is the lowercase string
("installed" / "running"); access via state.value, or compare
directly to the member or its wire string.
Example: >>> entry = agent.extensions.list()[0] >>> entry.state == ExtensionState.Running True >>> entry.state.value 'running'
Installed
instance-attribute
¶
Runtime state of an extension.
Two-member enum-like class: ExtensionState.Installed and
ExtensionState.Running. Wire form is the lowercase string
("installed" / "running"); access via state.value, or compare
directly to the member or its wire string.
Example: >>> entry = agent.extensions.list()[0] >>> entry.state == ExtensionState.Running True >>> entry.state.value 'running'
Running
instance-attribute
¶
Runtime state of an extension.
Two-member enum-like class: ExtensionState.Installed and
ExtensionState.Running. Wire form is the lowercase string
("installed" / "running"); access via state.value, or compare
directly to the member or its wire string.
Example: >>> entry = agent.extensions.list()[0] >>> entry.state == ExtensionState.Running True >>> entry.state.value 'running'
value
instance-attribute
¶
The state as its wire string: "installed" or "running".
Internal
¶
Bases: RuntimeError
An SDK bug marker.
Deliberately outside AgentError so a broad except AgentError
handler cannot swallow it — the agent answered, but the SDK could not
make sense of the reply. If you hit this, it is a bug in the SDK, not
a caller mistake.
Example: >>> try: ... agent.actions.execute('battery_test') ... except Internal: ... pass
LatestValue
¶
One row from latest() / at().
Optional Arrow bytes plus a typed scalar; decoding uses catalog metadata when signal is known.
Clone only bumps refcounts / copies small buffers (no GIL).
Example: >>> value = agent.latest("bus0/BMS_message/status.pack_current") >>> value.value 12.4
arrow_array
instance-attribute
¶
Lazily decode arrow_value into a 1-element pyarrow.Array.
arrow_value
instance-attribute
¶
Raw 1-row Arrow IPC bytes when the agent provided a typed
payload, else None.
data_segment_id
instance-attribute
¶
Id of the data segment the sample came from; None when unknown.
display_value
instance-attribute
¶
Agent-provided display string. Always populated; the typed value
getter is preferred for arithmetic / formatting.
label
instance-attribute
¶
signal.value_table[int(value)] lookup. Returns None when no
value_table is present or the value isn't int-coercible.
message
instance-attribute
¶
Message the signal belongs to.
name
instance-attribute
¶
Checkable contract — alias for [Self::path].
path
instance-attribute
¶
Fully-qualified path: "{source}/{message}.{signal}".
producer
instance-attribute
¶
Address of the producer that reported the sample; None for trace data.
raw
instance-attribute
¶
Alias for display_value (matches pre-rewrite SDK).
signal
instance-attribute
¶
Resolved catalog Signal, when known. Populated by
PyLatestRow::with_signal.
signal_name
instance-attribute
¶
Signal name within the message — the part after the final ..
source
instance-attribute
¶
Source name — the first segment of the signal's path.
time
instance-attribute
¶
time_ns as a UTC datetime, or None when the agent did not
timestamp this row. chrono::DateTime::from_timestamp_nanos is
total over i64 ns (every ns since the Unix epoch is in range).
time_ns
instance-attribute
¶
When the sample was recorded, epoch nanoseconds; None when the agent
reported no timestamp.
trace_path
instance-attribute
¶
Trace file the sample came from; None for live data.
value
instance-attribute
¶
Typed Python value (bool / int / float / str / bytes /
None) decoded once at construction. Falls back to the
agent-provided display_value string when no typed payload was
decodable.
__repr__()
¶
Return repr(self).
formatted(digits=2)
¶
Display string: enum label, numeric+unit, bool, hex bytes, em-dash for null, etc.
get()
¶
Checkable contract — typed value, mirrors the value property.
Layout
¶
A single saved layout, as returned by agent.layouts.list() / show() /
create() / update().
Mirrors the console API LayoutData envelope. Fields are parsed once from
the wire JSON; data is the arbitrary panel/config payload decoded to a
Python dict on access. team_slug, user_id, email, and
updated_by_email are optional and come back as None when the wire
omits them.
Example: >>> layout = agent.layouts.list()[0] >>> layout.name, layout.is_personal ('My Dashboard', False)
created_at
instance-attribute
¶
When the layout was created, as an RFC 3339 string.
data
instance-attribute
¶
The layout's stored panel/config payload, decoded into a dict.
The wire data is validated to be a JSON object when the layout is
parsed, so this getter never fails on a well-formed layout.
email
instance-attribute
¶
Email of the layout's owner; None when the wire omits it.
id
instance-attribute
¶
Layout id — the handle every other agent.layouts call takes.
is_personal
instance-attribute
¶
True for a private layout, False for one shared with the team.
name
instance-attribute
¶
Display name.
team_id
instance-attribute
¶
Id of the team the layout belongs to.
team_slug
instance-attribute
¶
URL slug of that team; None when the wire omits it.
updated_at
instance-attribute
¶
When the layout was last saved, as an RFC 3339 string.
updated_by_email
instance-attribute
¶
Email of whoever saved the layout last; None when the wire omits it.
user_id
instance-attribute
¶
Id of the owning user; None for a team layout.
LayoutVersion
¶
A single entry in a layout's version history, as returned by
agent.layouts.versions() / show_version().
Mirrors the console API LayoutVersionData envelope. Fields are parsed
once from the wire JSON; data is the layout payload snapshot at this
version — present on show_version(), None on versions() list
entries (the list is metadata-only). label is the user-given version
name set via set_version_label() (None when unlabeled),
restored_from_version is set when the version was created by a
restore(), and created_by_email comes back as None when the wire
omits it.
Example: >>> history = agent.layouts.versions(layout.id) >>> history[0].version, history[0].label (3, None)
created_at
instance-attribute
¶
When the version was saved, as an RFC 3339 string.
created_by_email
instance-attribute
¶
Email of whoever saved the version; None when the wire omits it.
data
instance-attribute
¶
The layout payload snapshot at this version, decoded into a dict.
None for entries from versions() (the list is metadata-only);
always populated on show_version().
id
instance-attribute
¶
Version id.
label
instance-attribute
¶
User-given version name; labeled versions are never pruned.
layout_id
instance-attribute
¶
Id of the layout this version belongs to.
name
instance-attribute
¶
The layout's display name when this version was saved.
restored_from_version
instance-attribute
¶
Version this one was restored from; None for an ordinary save.
version
instance-attribute
¶
Version number, counting up from 1.
NamedScalar
¶
Bases: float
float carrying the aggregation's name and unit (Checkable contract).
str and repr round to four significant digits and keep the unit;
float(scalar) is the exact value.
Attributes:
name: How the value was produced, e.g. max(bus0/BMS.voltage).
unit: The unit it inherits from the series, None when dimensionless.
Example: >>> series = frame["bus0/BMS_message/status.pack_current"] >>> peak = series.max() >>> peak.name, peak.unit ('max(bus0/BMS_message/status.pack_current)', 'A')
get()
¶
The bare float, dropping the name and unit.
NoData
¶
Bases: AgentError
The signal exists but produced no value inside the window.
Example: >>> try: ... agent.latest('bus0/BMS_message/status.pack_current', lookback=0.001) ... except NoData: ... pass
QueryRangeTooLarge
¶
Bases: AgentError
The requested range exceeds what the agent will serve at once.
No agent build caps a query range today, so nothing raises this; it is reserved
for the gRPC ResourceExhausted status, which a storage quota or a rate limit
on a console-backed call can also carry.
Example: >>> try: ... agent.query(['bus0/BMS_message/status.pack_current'], start='-30d', end='now') ... except QueryRangeTooLarge: ... pass
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: 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()
QueryType
¶
Raw / M4 — selects the agent-side query strategy. Wire encoding
mirrors the proto's QueryType enum (Raw=0, M4=1); the proto's
MinMax=2 is internal-only and not exposed through the SDK surface.
Example: >>> QueryType.M4.value 'm4'
M4
instance-attribute
¶
Raw / M4 — selects the agent-side query strategy. Wire encoding
mirrors the proto's QueryType enum (Raw=0, M4=1); the proto's
MinMax=2 is internal-only and not exposed through the SDK surface.
Example: >>> QueryType.M4.value 'm4'
Raw
instance-attribute
¶
Raw / M4 — selects the agent-side query strategy. Wire encoding
mirrors the proto's QueryType enum (Raw=0, M4=1); the proto's
MinMax=2 is internal-only and not exposed through the SDK surface.
Example: >>> QueryType.M4.value 'm4'
value
instance-attribute
¶
Wire-string view (matches the proto enum's lower-case form).
ReplayWindow
¶
Replay-window response: I-frame snapshot + P-frame changes + window bounds.
Rows use the same typed LatestValue shape as latest() and at().
Example: >>> window = agent.window(["bus0/BMS_message/status.pack_current"], start="-10s", duration=5) >>> [row.value for row in window.changes]
changes
instance-attribute
¶
duration
instance-attribute
¶
How long the window spans, as a timedelta.
end
instance-attribute
¶
Window end as a datetime.
end_ns
instance-attribute
¶
Alias for [Self::window_end_ns] (matches pre-rewrite SDK).
snapshot
property
¶
Opening values at start, one entry per signal — the same Snapshot
at() returns.
start
instance-attribute
¶
Window start as a datetime.
start_ns
instance-attribute
¶
Alias for [Self::window_start_ns] (matches pre-rewrite SDK).
window_end_ns
instance-attribute
¶
Authoritative window end, epoch nanoseconds (clamped like the start).
window_start_ns
instance-attribute
¶
Authoritative window start (may differ from request if clamped).
Segment
¶
Metadata for a single data segment.
Both bounds are optional because a segment that is still recording does
not have a final end time. connection is set for live segments, trace_path
for trace-file segments — they're mutually exclusive in practice.
Example: >>> segments = agent.segments() >>> segments[0].producer, segments[0].sources ('bus0', ['bus0/BMS_message', 'bus0/inverter_status'])
connection
instance-attribute
¶
Live-connection address (None for trace segments).
duration
instance-attribute
¶
None if the segment is still recording (no end bound).
end
instance-attribute
¶
Last sample time in the segment; None while it is still recording.
id
instance-attribute
¶
Data segment id.
producer
instance-attribute
¶
Producer that owns the segment.
sources
instance-attribute
¶
Source names that appear in the segment.
start
instance-attribute
¶
First sample time in the segment; None when the agent reported none.
trace_path
instance-attribute
¶
Trace file path (None for live segments).
SeriesByRun(runs, frame=None)
¶
Bases: Mapping
One :class:SignalSeries per run of a :class:~zelos_sdk.TraceSet, keyed by run label.
A read-only mapping, so dict(by_run), by_run["baseline"] and
by_run.items() all work.
Example: >>> by_run = runs.series(frame, "bus0/BMS_message/status.pack_current") >>> list(by_run) ['baseline', 'retest']
Signal
¶
Selector + metadata for a single signal: source, message, name, data type, unit, and (for enum-like signals) the value table.
The wire field signal is exposed as the Pythonic name (signal.signal
is awkward in Python); data_segment_id is exposed as a string.
Example: >>> sig = agent.signals()["bus0/BMS_message/status.pack_current"] >>> sig.unit 'A' >>> sig.path 'bus0/BMS_message/status.pack_current'
data_segment_id
instance-attribute
¶
The data segment this signal belongs to, "" when segmentation is
not in play.
data_type
instance-attribute
¶
None on a path-only handle — no catalog entry has described it yet.
message
instance-attribute
¶
The signal's message ({source}/{message}.{name}).
name
instance-attribute
¶
The signal's own name — the trailing component of its path
({source}/{message}.{name}).
path
instance-attribute
¶
Fully-qualified path: "{source}/{message}.{name}".
producer
instance-attribute
¶
The producer this signal came from, None when the frame carries no
producer metadata.
source
instance-attribute
¶
The signal's source ({source}/{message}.{name}).
trace_path
instance-attribute
¶
The trace file this signal was read from, None for a live signal.
unit
instance-attribute
¶
The signal's engineering unit (e.g. "V", "A"), None when unknown.
value_table
instance-attribute
¶
{raw value: label} for an enum-like signal, None otherwise.
__new__(source, message, name, data_type, data_segment_id=None, producer=None, trace_path=None, unit=None, value_table=None)
¶
Create and return a new object. See help(type) for accurate signature.
from_path(path, data_type=None)
classmethod
¶
Construct a path-only Signal handle. No catalog roundtrip —
the path is verified server-side at the next check / latest
/ query invocation.
Use this when the producer hasn't emitted yet (cold-start tests, fixture order races, extension consumers) or when you want a long-lived handle that re-resolves to the freshest segment on every call rather than binding to a specific one.
data_type is informational only — only the path crosses the
wire (SignalOperand(path)); the agent uses its own resolved
schema for evaluation. Left unset, data_type and unit read
back as None: nothing has verified them yet.
Accepts bare paths ("src/foo.bar"), wildcard-segment paths
("*/src/foo.bar"), and fully-qualified segment paths
("{uuid}/src/foo.bar") — the last pins resolution to a
specific segment.
SignalCatalog
¶
Immutable signal catalog with fast search and exact-path lookup, returned
by agent.signals().
Example: >>> catalog = agent.signals() >>> catalog["bus0/BMS_message/status.pack_current"].unit 'A'
paths
instance-attribute
¶
Every canonical path in catalog order.
warnings
instance-attribute
¶
Per-producer / per-trace warnings collected while building the catalog.
__contains__(key)
¶
Return bool(key in self).
__repr__()
¶
Return repr(self).
by_path(path)
¶
Deprecated alias for catalog[path].
get(path, default=None)
¶
Exact-path lookup that returns default instead of raising when the
catalog has no such signal. Ambiguity still raises AmbiguousSignal.
match(pattern)
¶
Filter the catalog by a glob over the full canonical path.
* and ? span / and ., so "bus0/*", "*.cell_*" and
"bus0/BMS_message/*.cell_0" all work. The same grammar every
paths= argument uses. No match returns an empty catalog.
search(query)
¶
Filter the catalog by case-insensitive substring match.
Each signal is checked across five fields: full path, source, message, signal name, and unit. Matching is case-insensitive; the empty query returns the catalog unchanged.
Args: query: Substring to search for.
Returns: SignalCatalog: A new catalog containing only the matched signals; warnings are inherited from the parent.
to_list()
¶
Deprecated alias for list(catalog).
to_pandas()
¶
Render the catalog as a pandas DataFrame (one row per signal).
Columns: path, source, message, name, data_type, unit,
producer, trace_path, data_segment_id. Raises ImportError
when pandas is not installed.
SignalFrame
¶
Result of a tabular query.
Arrow IPC bytes cross verbatim and are decoded by to_arrow().
arrow_ipc_data is held so Clone is cheap on the data
path; normalized_table_cache resets to empty on clone (it's a perf aid,
not identity, and sliced frames need a fresh table anyway).
Example: >>> frame = agent.query(["bus0/BMS_message/status.pack_current"], start="-5m") >>> frame.to_pandas()["bus0/BMS_message/status.pack_current"].mean()
arrow_ipc_data
instance-attribute
¶
Decode with pyarrow.ipc.open_stream. A fresh bytes is allocated
per access; users typically call to_arrow() once per
response so the cost is negligible.
column_names
instance-attribute
¶
Every column label except the time column — the same labels as keys()
and the signal columns of to_pandas().
columns
instance-attribute
¶
Wire metadata for the data columns, in Arrow order. The time column is not a signal and has no metadata worth reporting, so it is not here.
downsampled
instance-attribute
¶
True when the agent ran an M4 / downsampled query rather than raw.
meta
instance-attribute
¶
Alias for columns.
query_duration_s
instance-attribute
¶
How long the agent took to answer the query, in seconds.
requested_range
instance-attribute
¶
The range the caller ASKED for. time_range reports the data's own
extent; this is what the request said.
signals
instance-attribute
¶
Resolved catalog signals in column order (excludes the time column).
time_origin
instance-attribute
¶
Where the frame's time axis starts: "epoch" for wall-clock time,
"run_start" for a TraceSet frame aligned at each run's own start,
None for a frame built locally with no wire metadata.
time_range
instance-attribute
¶
The extent of the DATA: (first, last) timestamp in the frame, None
when it has no rows. The window the caller asked for is
requested_range.
truncated
instance-attribute
¶
True when max_rows > 0 and the result hit the cap.
warnings
instance-attribute
¶
Per-producer / per-trace warnings collected during the query.
__contains__(name)
¶
Return bool(key in self).
__getitem__(key)
¶
Return self[key].
__iter__()
¶
Implement iter(self).
__repr__()
¶
Return repr(self).
between(start=None, end=None)
¶
Rows inside [start, end], both anchored on the frame's own extent:
"+10s" counts from the first sample, "-10s" back from the last, and
"start" / "end" are the bounds themselves. An omitted bound is the
matching end of the frame.
Example: frame.between("+30s", "+40s") # the second half-minute of the run
describe()
¶
pandas.DataFrame.describe() over the data columns, with each column's
unit as the first row.
Example: frame.describe() # unit / count / mean / std / min / quartiles / max
dropna(how='any')
¶
Drop rows with nulls: how="any" (default) keeps only rows where every
column has a sample, how="all" drops only the empty rows.
Example: frame.dropna() # rows both messages landed on
ffill()
¶
Sample-and-hold every column: each null takes the last value before it.
One row per timestamp means a two-message frame is mostly nulls; this is how you get a dense frame to do arithmetic on.
Example: power = frame.ffill()["bus0/BMS.voltage"] * frame.ffill()["bus0/BMS.current"]
from_arrow_ipc(data, columns=[], signals=[])
staticmethod
¶
staticmethod(function) -> method
Convert a function to be a static method.
A static method does not receive an implicit first argument. To declare a static method, use this idiom:
class C:
@staticmethod
def f(arg1, arg2, argN):
...
It can be called either on the class (e.g. C.f()) or on an instance (e.g. C().f()). Both the class and the instance are ignored, and neither is passed implicitly as the first argument to the method.
Static methods in Python are similar to those found in Java or C++. For a more advanced concept, see the classmethod builtin.
head(n=5)
¶
First n rows as a new SignalFrame (default 5).
items()
¶
(label, series) tuples for every non-time column.
keys()
¶
Column labels excluding the leading time column. Pandas-style
iteration, and the same labels to_pandas() puts on its columns.
plot(**kwargs)
¶
Render with zelos_sdk.agent.plot.plot_frame. Keyword arguments
(title=, height=) forward verbatim.
rename(mapping)
¶
Relabel columns from a {old: new} mapping or a callable over the
current labels. Unknown keys keep their label; two columns landing on
one label raise ValueError.
Example: frame.rename({"bus0/BMS_message/status.pack_current": "current"})
resample(every, how='mean')
¶
Aggregate onto a fixed time grid: one row per every, combined with
how ("mean", "min", "max", "first", "last").
This is the analysis verb pandas users reach for; downsample= on
query() is a chart thinning by bucket count, not a period.
Example: frame.resample("1s").to_pandas()
series(key)
¶
Project one column as a Python SignalSeries (pandas/pyarrow ergonomics).
key accepts a str (a column label, a clean catalog path, or the raw
wire column name) or a Signal from the frame's catalog. Ties across
producers / segments raise [errors::AmbiguousSignal] with a hint to
pass a Signal, except on a TraceSet frame, where one path really
does name one column per run: that returns a SeriesByRun mapping.
short_names()
¶
Label each column by its signal name alone, falling back to
message.name where a bare name repeats in this frame. A run frame
shortens within each run and keeps the run: baseline · cell_0.
Example: frame.short_names().to_pandas() # columns "cell_0", "pack_current"
tail(n=5)
¶
Last n rows as a new SignalFrame (default 5).
to_arrow()
¶
The frame's rows as a pyarrow.Table, decoded and normalized.
The returned object is a pyarrow.Table with a canonical
time: timestamp[ns, tz=UTC] column and signal columns labeled with
the catalog path source/message.signal. The name mirrors
SignalSeries.to_arrow() (a column accessor) so notebook users have one
symbol to reach for at both the frame and column level.
Cached after the first call; every caller gets a handle to the same immutable Table.
to_pandas(*, index=True)
¶
Notebook one-liner: frame.to_pandas(), indexed by a UTC
DatetimeIndex named time. A run-aligned TraceSet frame
(time_origin == "run_start") is indexed by a TimedeltaIndex of
offsets instead. Pass index=False for the older shape, where time
is an ordinary column.
The index is what df.resample("1s"), df.loc["...":] and df.plot()
all read, and it matches SignalSeries.to_pandas(). Columns are labeled
the way the catalog labels signals — source/message.signal — taking a
producer:: prefix only when two columns in one result share a path.
Example: frame.to_pandas().resample("1s").mean()
Raises ImportError with an actionable hint if pandas is not
installed; we deliberately do not list it in the SDK's hard
requirements so notebook users opt-in via zelos-sdk[notebook].
values()
¶
One SignalSeries per non-time column (parallel to keys()).
SignalNotFound
¶
Bases: AgentError, KeyError
No signal matched the requested path or pattern.
Carries a computed suggestions: list[str] of the closest catalog
paths, so a caller can surface a "did you mean" hint.
Example: >>> try: ... agent.latest('bus0/BMS_message/status.pack_currnt') ... except SignalNotFound as exc: ... exc.suggestions ['bus0/BMS_message/status.pack_current']
SignalSeries(name, values, *, signal=None, unit=None, time=None)
¶
One Arrow column with pyarrow/pandas helpers and a unit-aware math surface.
Unit composes through * and /; + and - require matching
units. Scalars, Arrow arrays and plain sequences are unit-blind operands;
pandas and numpy objects are refused. Values are always a
pa.ChunkedArray; time is the frame's shared time column, so two
series from one frame align by identity.
Attributes:
name: The signal's path, a caller's rename(), or None.
values: The samples as a pa.ChunkedArray; a null is a missing sample.
signal: The catalog Signal a frame column came from; None once
the series is derived.
time: The shared time axis, None on a series built without one.
Example: >>> frame = agent.query(["bus0/BMS_message/status.pack_current"], start="-5m") >>> series = frame["bus0/BMS_message/status.pack_current"] >>> series.mean().unit 'A'
legend
property
¶
Display label as "name (unit)", falling back to whichever exists.
null_count
property
¶
Number of null samples in the series.
unit
property
¶
The series' unit as a display string, None when unknown.
abs()
¶
Magnitude of every sample, unit kept.
clip(lower=None, upper=None)
¶
Clamp values to [lower, upper]. At least one bound is required.
count()
¶
Number of non-null values, as an int — not a NamedScalar.
cumsum()
¶
Running total over the samples, unit kept.
derivative()
¶
Numerical derivative over time (centered, np.gradient).
derive(values=None, *, label=None, unit=None)
¶
Deprecated alias of :meth:from_arrow on this series' time axis.
diff()
¶
Delta to the previous sample; the first value is null.
dropna()
¶
Drop null samples, keeping time aligned with the values that remain.
ffill()
¶
from_arrow(values, *, name=None, unit=None, time=None)
classmethod
¶
Series from Arrow-compatible values, checked against time.
from_pandas(series, *, unit=None, name=None)
classmethod
¶
Series from pandas; a DatetimeIndex becomes the UTC time axis.
get()
¶
Deprecated alias of :meth:to_pandas.
head(n=5)
¶
The first n samples, time kept alongside.
integrate()
¶
Cumulative trapezoidal integration over time; first sample is 0.
isna()
¶
Boolean series, True wherever the sample is missing.
max()
¶
Largest non-null sample; nan when there is none.
mean()
¶
Arithmetic mean over the non-null samples.
median()
¶
Exact median of the non-null samples.
min()
¶
Smallest non-null sample; nan when there is none.
plot(**kwargs)
¶
Render with :func:zelos_sdk.agent.plot.plot_series. Keyword
arguments (title=, height=) forward verbatim.
rename(name)
¶
Same samples under a new name — what a derived series wants.
std(ddof=1)
¶
Sample standard deviation (ddof=1, as in pandas).
sum()
¶
Total of the non-null samples.
tail(n=5)
¶
The last n samples, time kept alongside.
time_arrow()
¶
Deprecated alias of the :attr:time attribute.
to_arrow()
¶
The samples as a pa.ChunkedArray, without copying.
to_numpy()
¶
float64 numpy array; nulls become NaN. The fast path for bulk math.
to_pandas()
¶
pandas.Series (UTC DatetimeIndex when timed, unit in attrs).
where(mask)
¶
Keep values where mask is True; the rest become nulls.
with_unit(unit)
¶
Assert a unit (no conversion).
Snapshot
¶
The values of several signals as of one moment: what latest(paths),
at(paths, time) and each watch() tick hand back.
A read-only Mapping[str, LatestValue] keyed by clean path, so every
spelling that worked against the plain dict still works — snapshot[path],
in, len, .get, .keys(), .values(), .items(), iteration, and
dict(snapshot). time is the newest sample time in the set.
Example: >>> snapshot = agent.latest(["bus0/BMS_message/cells.cell_0", "bus0/BMS_message/status.pack_current"]) >>> snapshot["bus0/BMS_message/status.pack_current"].value 12.4
SortOrder
¶
Asc / Desc — drives row ordering for Agent.query() /
Trace.query(). Wire encoding mirrors the proto's SortOrder enum
(Asc=0, Desc=1).
Example: >>> agent.query(["bus0/BMS_message/status.pack_current"], start="-5m", sort_order=SortOrder.Desc) >>> SortOrder.Asc.value 'asc'
Asc
instance-attribute
¶
Asc / Desc — drives row ordering for Agent.query() /
Trace.query(). Wire encoding mirrors the proto's SortOrder enum
(Asc=0, Desc=1).
Example: >>> agent.query(["bus0/BMS_message/status.pack_current"], start="-5m", sort_order=SortOrder.Desc) >>> SortOrder.Asc.value 'asc'
Desc
instance-attribute
¶
Asc / Desc — drives row ordering for Agent.query() /
Trace.query(). Wire encoding mirrors the proto's SortOrder enum
(Asc=0, Desc=1).
Example: >>> agent.query(["bus0/BMS_message/status.pack_current"], start="-5m", sort_order=SortOrder.Desc) >>> SortOrder.Asc.value 'asc'
value
instance-attribute
¶
Wire-string view (matches the proto enum's lower-case form).
TimeMode
¶
Relative / Absolute — selects how trace-side query bounds are
interpreted. Wire encoding mirrors MultiTraceTimeMode
(Relative=0, Absolute=1).
Example: >>> TimeMode.Relative.value 'relative'
Absolute
instance-attribute
¶
Relative / Absolute — selects how trace-side query bounds are
interpreted. Wire encoding mirrors MultiTraceTimeMode
(Relative=0, Absolute=1).
Example: >>> TimeMode.Relative.value 'relative'
Relative
instance-attribute
¶
Relative / Absolute — selects how trace-side query bounds are
interpreted. Wire encoding mirrors MultiTraceTimeMode
(Relative=0, Absolute=1).
Example: >>> TimeMode.Relative.value 'relative'
value
instance-attribute
¶
Wire-string view (matches the proto enum's lower-case form).
TimeRange
¶
Inclusive datetime interval [start, end].
Example: >>> trace = agent.trace("run.trz") >>> trace.time_range.start, trace.time_range.end (datetime.datetime(2026, 8, 1, 12, 0, tzinfo=datetime.timezone.utc), datetime.datetime(2026, 8, 1, 12, 5, tzinfo=datetime.timezone.utc))
duration
instance-attribute
¶
end - start, as a timedelta.
end
instance-attribute
¶
Last instant in the interval.
end_ns
instance-attribute
¶
end as epoch nanoseconds, the precision a datetime cannot hold.
start
instance-attribute
¶
First instant in the interval.
start_ns
instance-attribute
¶
Epoch nanoseconds. Prefer this over .start / .end when precision
matters: a Python datetime only holds microseconds.
TimeRangeMulti
¶
Multi-trace time range response (overlay / absolute mode): the union
extent across every run, plus each run's own TraceTiming.
Example: >>> timing = TraceTiming("a.trz", start=trace.time_range.start, duration=trace.time_range.duration) >>> multi = TimeRangeMulti(trace.time_range.start, trace.time_range.end, max_duration=trace.time_range.duration, traces=[timing]) >>> multi.max_duration, len(multi.traces) (datetime.timedelta(seconds=300), 1)
end
instance-attribute
¶
Latest end across every run.
max_duration
instance-attribute
¶
The longest run's duration, as a timedelta.
max_duration_s
instance-attribute
¶
The longest run's duration, in seconds.
start
instance-attribute
¶
Earliest start across every run.
traces
instance-attribute
¶
One entry per run, in the order the set was opened.
Trace
¶
Handle to a trace file opened through Agent.trace.
Owns an Arc<AgentTransport> cloned from its parent Agent, so the
trace shares the agent's connection lifecycle: calling agent.close()
invalidates this handle too, and any subsequent method on it raises
AgentCancelled.
Time arguments anchor on this file, not on wall-clock now: start="-30s"
is the last 30 seconds OF THE FILE. The frame's time axis is always epoch.
Example: >>> trace = agent.trace("run.trz") >>> df = trace.query(["bus0/BMS_message/status.pack_current"], start="-30s").to_pandas() >>> trace.close()
check
instance-attribute
¶
Typed predicate Check API bound to this trace file.
Returns a TraceChecks proxy whose default is temporal="always"
over the trace's full [start_ns, end_ns] — matches the natural
reading of trace.check.that('voltage', '<', 42) ("is voltage < 42
across this recording?"). Override with last= (anchored at the
trace's end_ns, not wall-clock), start_ns=+end_ns=, or
temporal="latest".
path
instance-attribute
¶
The trace file's path on disk.
time_range
instance-attribute
¶
The trace file's own time extent.
__enter__()
¶
Enable with agent.trace(path) as trace: — returns self.
__exit__(_exc_type=None, _exc_value=None, _traceback=None)
¶
Close the trace handle on context exit. Mirrors the try/finally
trace.close() pattern; idempotent on multiple closes.
at(paths, time=None, *, min_time=None)
¶
Per-signal value at (or before) a moment inside the trace.
time and min_time use the same source-anchored grammar as query
and default to the file's end. One literal path returns a LatestValue;
a wildcard or a sequence returns a Snapshot keyed by signal path.
Raises SignalNotFound on catalog miss; ValueError if min_time > time.
close()
¶
Release the agent's hold on this trace file.
A hint, not a shutdown: it tells the agent it may drop the catalog and
indices it loaded, and the agent reopens the file on demand the next
time this handle is used. Idempotent; the handle stays usable, and only
a file that has since left the disk raises TraceNotFound. The
per-handle catalog cache is kept — a trace file cannot change.
export(path, *, start=None, end=None, overwrite=False, format=None, relative_start=None, relative_end=None)
¶
Export this trace (or a slice of it) to a new .trz file.
start / end use the same source-anchored grammar as query; omitting
both exports the full trace. format defaults to the agent's release
default (legacy DuckDB), except over a TRZ2 input, which re-exports as
TRZ2.
Args:
path: Destination path for the new .trz file.
start: Slice start; defaults to the file's start.
end: Slice end; defaults to the file's end.
overwrite: Replace path if it exists.
format: "trz2", "legacy", or None for the agent's default.
Returns: ExportResult: On-disk path plus per-producer success bits.
info()
¶
Aggregate trace inspection — one round-trip for file size, time
range, segments, per-table row counts, and total data rows.
Same shape as zelos trace info.
latest(paths)
¶
The file's last value for each path — at(paths, "end").
One literal path returns a LatestValue, anything else a Snapshot.
query(paths, *, start=None, end=None, downsample=None, max_rows=0, sort=None, sort_order=None, relative_start=None, relative_end=None)
¶
Time-bounded query against this trace file.
Resolves paths against the trace's catalog (trace mode is
authoritative — literal-path misses raise SignalNotFound).
start / end anchor on THIS FILE: "-30s" is 30 s before its end,
"+30s" is 30 s after its start, "start" / "end" / "now" are its
bounds, and a datetime, ISO 8601 string or epoch-ns int is taken
literally. An omitted bound is the matching end of the file. A window
entirely outside the file raises ValueError naming the file's extent;
one that only overhangs is clamped.
Args:
paths: A str, Signal, or sequence thereof.
start: Window start; defaults to the file's start.
end: Window end; defaults to the file's end.
downsample: Optional M4 bucket count — same semantics as
Agent.query(downsample=N). The agent returns at most
four points per bucket. Mutually exclusive with
max_rows / sort.
max_rows: Cap on returned rows; 0 (default) means no cap.
sort: "asc" (default) or "desc".
Returns: SignalFrame: Arrow-backed frame on an epoch time axis.
Raises: SignalNotFound: catalog miss for a literal path. ValueError: an unparsable or out-of-range window.
segments()
¶
List the trace file's data segments.
A segment is a contiguous chunk inside the trace bounded by a
TraceSegmentStart / TraceSegmentEnd pair. Per-trace read
warnings are emitted as RuntimeWarning, matching Agent.segments;
use segments_with_warnings() to receive them as a list instead.
segments_with_warnings()
¶
Same as segments() but returns (segments, warnings).
signals()
¶
Fetch the trace's signal catalog.
Cached per-handle: the first call hits the wire, every subsequent
call returns the same SignalCatalog (trace files are immutable
— the catalog cannot drift). Supports search(), by_path(),
and slicing — same shape as the live agent catalog.
Returns: SignalCatalog: All signals present in this trace.
window(paths, *, start, duration, display_fps=0)
¶
Replay snapshot + change-stream over a window inside this trace.
Same semantics as Agent.window() scoped to this trace. start uses the
source-anchored grammar; duration accepts int/float seconds, a
datetime.timedelta, or a string like "30s". display_fps=0 means
"no thinning". Raises ValueError on duration <= 0 or display_fps < 0.
TraceEventFieldMetadata
¶
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.
Args: name (str): The field name. data_type (DataType): The data type for the field. 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")
TraceMetadata
¶
Python wrapper for trace metadata.
Contains information about a complete trace including its time range, producer, and associated data segments.
TraceNamespace
¶
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)
__del__()
¶
Python destructor — only shuts down non-global namespaces. The global namespace is cleaned up via atexit.
__repr__()
¶
String representation of the namespace.
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: int: Number of registered sources.
TraceNotFound
¶
Bases: AgentError
The agent has no trace file at that path.
Example: >>> try: ... agent.trace('missing.trz') ... except TraceNotFound: ... pass
TracePublishClient
¶
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)
__del__()
¶
Python destructor - automatically shutdown background tasks
__repr__()
¶
String representation of the client.
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
¶
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)
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.
TraceReadSource
¶
Python wrapper for a trace read source.
Represents a source (e.g., "can") containing multiple events.
TraceReader
¶
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")
__repr__()
¶
String representation of the reader.
close()
¶
Close the trace reader.
This method closes the trace file and releases resources. It's automatically called when exiting the context manager.
Returns: None
get_value_table(data_segment_id, field_path)
¶
Get the value table (enum mapping) for a specific field.
Args: data_segment_id (str): Data segment ID to query. field_path (str): Field path in format "source/event.field" (without the "*/" prefix).
Returns: dict: Mapping of integer keys to string values, or None if no value table exists.
Raises: 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: List[DataSegment]: List of data segment metadata.
Raises: 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.
Args: start (str): Start of time range (inclusive), ISO 8601 / RFC 3339. end (str): End of time range (inclusive), ISO 8601 / RFC 3339.
Returns: List[DataSegment]: List of data segments overlapping the time range.
Raises: 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.
Args: data_segment_id (str, optional): Specific data segment ID to query. If None, queries all segments.
Returns: List[TraceReadSource]: List of sources, each containing events and fields.
Raises: 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: List[TraceMetadata]: List of trace metadata.
Raises:
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: None
Raises: 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.
Args:
data_segment_ids (List[str]): List of data segment IDs to query.
An empty list selects nothing.
fields (List[str]): List of field selectors
(e.g., "*/bus0/msg1.sig1", or "
Returns:
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: 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: TimeRange: Object containing start and end timestamps.
Raises: 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}")
TraceSender
¶
Communication channel for sending trace events.
This class is typically obtained from a TracePublishClient and passed to a TraceSource during creation. It handles the underlying message transport.
Note: Users generally don't need to interact with this class directly; it's used internally to connect TraceSource to TracePublishClient.
TraceSet
¶
Handle to a set of trace files opened together for multi-trace queries.
Returned by Agent.traces([...]). Like Trace, holds an
Arc<AgentTransport> cloned from its parent Agent and shares the
agent's connection lifecycle.
Each member file is a RUN, named by runs (the file stems, deduplicated).
len(traces), traces[label], and iteration over (label, Trace) address
them individually.
Bounds given as source-anchored offsets (or omitted) align the runs at
t = 0 and the frame's axis is seconds from each run's start; bounds given
as absolute instants apply once and the axis is epoch.
Example: >>> ts = agent.traces(["a.trz", "b.trz"]) >>> df = ts.query(["bus1/inverter_status.rpm"], end="+30s").to_pandas() >>> ts.close()
check
instance-attribute
¶
Typed predicate Check API bound to this set. Each assertion returns a
dict[run label, CheckResult].
max_duration
instance-attribute
¶
The longest run's duration — what time_range.max_duration returned
when time_range was a TimeRangeMulti.
paths
instance-attribute
¶
The trace files' paths on disk, in the order the set was opened.
runs
instance-attribute
¶
The run labels, in the order the set was opened.
time_range
instance-attribute
¶
The union extent across every run.
traces
instance-attribute
¶
Per-run timing entries, one per run — what time_range.traces
returned when time_range was a TimeRangeMulti.
__enter__()
¶
Enable with agent.traces(paths) as traces: — returns self.
__exit__(_exc_type=None, _exc_value=None, _traceback=None)
¶
Close every trace in the set on context exit. Mirrors the
try/finally traces.close() pattern; idempotent on multiple closes.
at(paths, time=None, *, min_time=None)
¶
Per-signal value at (or before) a moment, across all member runs.
time and min_time resolve against the set's union extent and default
to its end. One literal path returns a LatestValue; a wildcard or a
sequence returns a Snapshot keyed by signal path.
close()
¶
Release the agent's hold on every trace in the set.
Idempotent. Same semantics as Trace.close() but issued for
every member trace in one RPC.
export(path, *, start=None, end=None, overwrite=False, format=None, time_mode=None, relative_start=None, relative_end=None)
¶
Export the union of all member runs to a single new .trz file.
Same bounds contract as TraceSet.query(): offset bounds slice each run
from its own start, absolute bounds slice once on wall-clock.
latest(paths)
¶
The last value for each path across the set — at(paths, "end").
One literal path returns a LatestValue, anything else a Snapshot.
query(paths, *, start=None, end=None, downsample=None, max_rows=0, sort=None, sort_order=None, time_mode=None, relative_start=None, relative_end=None)
¶
Time-bounded query joining results across every member run.
start / end take the same grammar as Trace.query. Offset bounds
("+30s", "-30s", "start", "end") — or no bounds at all — align
the runs at t = 0 and the frame's axis is seconds from each run's own
start; a datetime or ISO instant selects the epoch axis instead.
Args:
paths: A str, Signal, or sequence thereof. Wildcards
expand against the union catalog.
start: Window start; defaults to each run's start.
end: Window end; defaults to the longest run's end.
downsample: Optional M4 bucket count; mutually exclusive with
max_rows / sort.
max_rows: Cap on returned rows; 0 (default) means no cap.
sort: "asc" (default) or "desc".
Returns:
SignalFrame: Arrow-backed frame; one column per (run, signal).
segments()
¶
List per-trace segments across every member trace.
Segment entries carry their owning trace_path so callers can
group them. Read warnings surface as RuntimeWarning; use
segments_with_warnings() to receive them as a list instead.
segments_with_warnings()
¶
Same as segments() but returns (segments, warnings).
series(frame, path)
¶
One SignalSeries per run for path, keyed by run label.
A set frame lays out one column per (run, signal); this is the run
dimension a caller can name, resolved through each column's trace_path.
frame["path"] cannot do it on its own — the frame does not know the
set's labels.
signals()
¶
Fetch the union signal catalog across every member trace.
Cached per handle. Each signal carries its source file in
signal.trace_path, so a later query routes to the right one.
Returns: SignalCatalog: Union of all member-trace catalogs.
window(paths, *, start, duration, display_fps=0)
¶
Replay snapshot + change-stream over a window across all member runs.
start resolves against the set's union extent; duration accepts
int/float seconds, a datetime.timedelta, or a string like "30s".
TraceSource
¶
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 ... })
__repr__()
¶
String representation of the source.
add_event(name, schema, event_type=...)
¶
Register an event schema.
Args:
name (str): The event name (e.g. "log").
schema: 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.
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.
Returns: TraceSourceEvent: A handle to the newly registered event.
Raises: 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.
Args: name (str): The name of the value table. data (dict): A dictionary of values to add to the value table.
Returns: 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: None
get_event(name)
¶
Get a handle to a previously registered event schema.
Args: name (str): The name of the event schema.
Returns: TraceSourceEvent: A handle to the event.
Raises: 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.
Args: name (str): The name to log. data (dict): A dictionary of fields to log.
Returns: 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.
Args: time_ns (int): The time to log the event at. name (str): The name to log. data (dict): A dictionary of fields to log.
Returns: 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.
Args: event_name (str): The event name for this batch. data: A pyarrow.RecordBatch or pyarrow.Table.
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.
Args: name (str): The name to log. data (dict): A dictionary of fields to log. prefix (str): A prefix to add to the event name.
Returns: 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.
Args: events: List of (time_ns: int, name: str, data: dict) tuples.
Examples: >>> source.log_many([ ... (time.time_ns(), "sensor", {"temp": 25.0}), ... (time.time_ns(), "sensor", {"temp": 25.1}), ... (time.time_ns(), "status", {"mode": "active"}), ... ])
TraceSourceCache(name, namespace=None)
¶
A TraceSource wrapper that caches the last value of each field.
Uses a Rust core for cache storage, condition evaluation, and emit decisions. Python layer provides Pythonic attribute navigation.
Example: source = TraceSourceCache("motor_controller") source.add_event("motor_stats", [ TraceEventFieldMetadata("rpm", DataType.Float64), TraceEventFieldMetadata("torque", DataType.Float64, "Nm") ]) source.log("motor_stats", {"rpm": 3500.0, "torque": 42.8}) assert source.motor_stats.rpm.get() == 3500.0
add_event(name, schema, conditions=None)
¶
Register an event schema.
add_value_table(name, field_name, data)
¶
Register a value table (enum mapping).
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.
get_source()
¶
Not supported — use the cache API directly.
log(name, data)
¶
Log data and update cache. Auto-registers event if not yet registered.
log_at(time_ns, name, data)
¶
Log data at a specific timestamp.
set_default_log_condition(condition=None)
¶
Set default log condition. Pass a Python LogCondition or None.
TraceSourceCacheLastEvent(name, cache, source, conditions=None)
¶
A cached event with attribute access to fields and submessages.
Example: event = source.motor_stats event.rpm.get() # field access event.thermal.temp.get() # submessage access event.log(rpm=3500) # log via event
time_ns
property
¶
Wall-clock timestamp of the last log call for this event, in
nanoseconds since the Unix epoch. None if the event hasn't
been logged yet. Symmetric with 
meth:TraceSourceCacheLastField.get.
TraceSourceCacheLastField(name, full_path, data_type, cache, event_name, condition=None, uses_default=False)
¶
A cached field that stores the last logged value.
Example: field = event.rpm field.get() # Get cached value field.name # Full path: "motor_stats.rpm"
TraceSourceEvent
¶
__repr__()
¶
String representation of the event.
log(**kwargs)
¶
Log an event with a dictionary of fields.
Args: kwargs (dict): Keyword arguments to log.
Returns: None
Examples: >>> event = source.add_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.
Args: time_ns (int): Timestamp in nanoseconds since Unix epoch. fields (dict[str, Any]): Dictionary of field names to values.
Raises: 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 ... })
TraceStdout
¶
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
__repr__()
¶
String representation of the sink.
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: 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: None
Raises: RuntimeError: If the sink cannot be initialized.
TraceTiming
¶
One run's timing entry inside a TimeRangeMulti (per-trace start and duration).
Example: >>> timing = TraceTiming("a.trz", start=trace.time_range.start, duration=trace.time_range.duration) >>> timing.path, timing.start, timing.duration ('a.trz', datetime.datetime(2026, 8, 1, 12, 0, tzinfo=datetime.timezone.utc), datetime.timedelta(seconds=300))
duration
instance-attribute
¶
How long the run spans, as a timedelta.
duration_s
instance-attribute
¶
How long the run spans, in seconds.
path
instance-attribute
¶
Path of the trace file this timing describes.
start
instance-attribute
¶
When the run starts, as a datetime.
start_s
instance-attribute
¶
When the run starts, as epoch seconds.
TraceWriter
¶
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
__repr__()
¶
String representation of the writer.
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: 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: None
Raises: RuntimeError: If the writer cannot be initialized.
Note: This method is called automatically by enter when using the context manager pattern.
UnsupportedAgentService
¶
Bases: AgentError
The agent is reachable but does not implement the RPC.
Usually means the SDK is newer than the running agent build.
Example: >>> try: ... agent.actions.list() ... except UnsupportedAgentService: ... pass
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.
connect(target=None, *, timeout=5.0)
¶
Connect to an agent, checking it answers before returning.
The canonical entry point: from zelos_sdk import connect. Prefer the lazy
Agent(target) for long-running scripts and pytest fixtures — its channel
reconnects on its own and RPC errors surface at the call site.
Args:
target: Agent endpoint; None resolves ZELOS_AGENT_URL, then
http://localhost:2300.
timeout: Seconds to wait for the health check.
Returns: Agent: A handle whose agent answered.
Raises:
AgentUnavailable: nothing answered before timeout.
UnsupportedAgentService: something answered but is not a Zelos agent.
Example: >>> agent = connect("http://localhost:2300") >>> agent.signals()[0].path 'bus0/BMS_message/status.pack_current'
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".
Args: log_level (Optional[str]): The log level to use. Valid values: "trace", "debug", "info", "warn", "error". Defaults to "info" if not specified.
Returns: 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
get_global_router_sender()
¶
Get the global default trace router sender (from global namespace)
Returns: TraceSender: The global namespace's router sender
Examples: >>> sender = get_global_router_sender()
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.
Args:
name: Application identifier; defaults to "python".
url: 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.
client_config: Configuration for the TracePublishClient
(batch_size, batch_timeout_ms).
log_level: Logging level to enable, None leaves logging untouched.
trace: Initialize the trace system. Defaults to True.
actions: Initialize the actions system. Defaults to False.
block: Block the current thread until interrupted (useful for
actions-only programs).
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)
init_global_actions_client(name, url=..., actions_registry=...)
¶
Initialize the global actions client with a background task
This creates a global actions client that runs in the background without blocking or interfering with Python's signal handling. The client will automatically reconnect if the connection is lost and can be cleanly shut down via atexit handlers.
Args: name (str): Name for this service url (str): Server URL (e.g., "grpc://localhost:2300") actions_registry (ActionsRegistry): The actions registry to serve
init_global_actions_registry()
¶
Initialize the global actions registry
init_global_client(url=None, config=None)
¶
Initialize the global client with custom settings
Args:
url (str): The URL of the trace publish service. Defaults to
ZELOS_AGENT_URL (legacy: ZELOS_TRACE_FORWARD_URL), then
http://localhost:2300.
config (TracePublishClientConfig): Configuration for the client.
If None, default settings will be used.
Returns: 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.
Args:
name (Optional[str]): Source name for the first call. Defaults to
"python".
Returns: TraceSource: The global source.
Examples: >>> source = init_global_source("my_app")
parse_connect_target(target=None)
¶
Normalize a connect-target string into the URL connect() would use,
without actually connecting — for pre-flight config validation in CI.
None falls back to ZELOS_AGENT_URL, then http://127.0.0.1:2300.
host:port and bare host get an http:// scheme and, for a bare
host, the default port; grpc:// normalizes to http://.
Raises:
ConnectionTargetError: the string has whitespace, more than one
://, or an otherwise malformed port.
Example: >>> parse_connect_target('localhost:2300') 'http://localhost:2300'
parse_duration(value, *, allow_zero=True)
¶
Non-negative seconds.
Accepts int/float seconds, timedelta, or the unsigned
duration-string grammar ("30s", "2m", "1.5h", "500ms").
Rejects bool, negative values, NaN/inf, and (optionally) zero. A
signed string ("-30s") is a time offset, not a duration — rejected
with a hint to use :func:parse_time instead.
Example: >>> parse_duration("1.5h") 5400.0
parse_time(value)
¶
Normalize to a UTC-aware datetime.
Accepts a datetime (naive → assumed UTC), "now", ISO 8601, or a
relative offset ("-30s", "-2m", "-1.5h", "-1d", "+30s").
Relative forms resolve against the wall clock.
Example: >>> parse_time("-30s") datetime.datetime(2026, 9, 2, 11, 59, 30, tzinfo=datetime.timezone.utc)
parse_time_ns(value)
¶
Coerce a flexible time argument to epoch nanoseconds.
Same surface as :func:parse_time plus integer / float / None
pass-through:
None→None(caller decides what unset means).int→ epoch ns, but only if unambiguous (>= 1e15, i.e. after 1970-01-12); a smaller int is almost certainly seconds by mistake and raises with a hint.float→ epoch seconds, unless it is already too large to be seconds (>= 1e15, year 31 million), in which case it is read as epoch ns.datetime/str→ :func:parse_timethen converted.
Used at any agent SDK boundary that takes start_ns / end_ns
on the wire — the Check API, custom replay helpers, suite runners.
Example: >>> parse_time_ns("now") 1788609600000000000 >>> parse_time_ns(None)
parse_until(value)
¶
Absolute deadline from datetime/ISO-string, or now + N seconds.
A float means "this many real seconds from now". A string in the
unsigned duration grammar ("30s", "1.5h") means the same thing.
Other string forms go through :func:parse_time and resolve the same
way, against the wall clock.
Example: >>> parse_until("30s") datetime.datetime(2026, 9, 2, 12, 0, 30, tzinfo=datetime.timezone.utc)
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).
Args:
name (str): The raw, externally-sourced name.
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".
Returns: str: A name that always passes registration for that kind.
Raises:
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.