Query Live Data¶
Everything the agent knows how to answer, in one API: what signals exist, what they were doing over a window, what they are right now, and what changed in between. The same calls work on a saved .trz file — see Open trace files.
connect() checks that something answers before it returns. With no argument it resolves ZELOS_AGENT_URL, then http://localhost:2300. Pass a target to be explicit:
For a long-running script or a pytest fixture, prefer the lazy Agent(target): its channel reconnects on its own and errors surface at the call site rather than at construction.
Browse the catalog¶
signals() returns a SignalCatalog of everything that produced data recently. It renders itself — path, type, unit, producer — so a bare catalog in a notebook cell is the listing:
A catalog is a sequence and a mapping:
catalog[0] # by position
catalog["bus0/BMS_message/status.pack_current"] # by path; a miss raises SignalNotFound, and names the path a bare "pack_current" meant
catalog.get("maybe/missing.signal") # by path, None when absent
"bus0/BMS_message/cells.cell_0" in catalog
len(catalog)
match() filters by glob and returns another catalog; search() is the loose, human spelling that scores substrings across path, message and unit:
The glob grammar¶
One grammar, used everywhere a path is accepted — match(), query(), latest(), at(), window(), export():
| Pattern | Matches |
|---|---|
* |
any run of characters, including / and . |
? |
exactly one character |
bus0/* |
every signal on the bus0 source |
bus0/BMS_message/* |
every signal on every BMS_message event |
bus0/BMS_message/cells.* |
every field of the cells event |
*.cell_* |
every signal whose field name starts with cell_, on any source |
*/status.pack_current |
that field, whichever source carries it |
A pattern that matches nothing raises SignalNotFound in query(), latest(), at(), window() and export(), rather than quietly returning a frame with a column missing. match() is the exception: filtering a catalog down to nothing returns an empty catalog. In a query, wildcards expand against the signals active inside the window you asked for.
Producers¶
An agent can serve several producers. Every read takes producers=, and the default — None — covers every connected one:
agent.signals() # all producers
agent.signals(producers="192.168.1.50:2300") # one
agent.signals(producers=["a:2300", "b:2300"]) # several
lookback= bounds how recently a signal must have produced a sample to appear. It applies to signals() and latest(), the two calls that ask "what is live right now"; a query is bounded by its own window instead.
Segments¶
A segment is one contiguous run of data from one producer. segments() is how you see where the gaps are:
Time and duration grammar¶
Two grammars, and the sign tells them apart.
A time is a point on the clock. start=, end=, at()'s cursor and min_time= take one:
| Form | Means |
|---|---|
"-30s", "-2m", "-1.5h", "-1d" |
that long before now |
"+30s" |
that long after now |
"now" |
now |
"2026-09-02T21:30:00Z" |
ISO 8601 |
datetime(...) |
a Python datetime; a naive one is read as UTC |
A duration is a length. duration=, last=, until= and resample()'s interval take one, unsigned:
| Form | Means |
|---|---|
"500ms", "30s", "2m" or "2min", "1.5h", "1d" |
that much time |
timedelta(minutes=2) |
the same |
120 |
seconds, as a number |
Passing a signed string where a duration belongs is an error that says so, and so is passing a bare integer where a time belongs — 1757000000 is far more often seconds-by-mistake than an epoch in nanoseconds. Write "-30s".
Run a query¶
start is required on a live agent; end defaults to now. paths takes a string, a Signal, or a sequence of either. The result is a SignalFrame — Arrow columns plus the metadata that says what you actually got — and it renders itself: row count, time range, and one line per column with its unit, type and non-null count.
How a frame joins signals¶
A frame is one row per distinct timestamp. Every requested signal is a column, and a cell is null wherever that signal had no sample at that instant. Nothing is interpolated, resampled or forward-filled on the way in: what you get back is what was recorded, joined on time.
So a frame over two messages that tick at different rates is sparse, and the per-column non-null count in its repr is how you see that:
mixed = agent.query(
["bus0/BMS_message/status.pack_current", "bus1/inverter_status.battery_power"],
start="-1m",
)
mixed
# SignalFrame(rows=647, signals=2, 2026-09-04 17:05:42.063 → 17:06:41.737)
# bus0/BMS_message/status.pack_current A Float32 59 non-null
# bus1/inverter_status.battery_power W Float64 590 non-null
Alignment is explicit. ffill() carries each column's last value forward — sample-and-hold, the right reading for a signal that only reports on change — and dropna() drops the rows that still have nothing:
Write those two calls whenever you combine columns from different messages. A check on a column of the aligned frame evaluates the aligned samples, the same way a check on any derived series does. integrate() and derivative() refuse a series with nulls rather than guessing across a gap, and their error names this pair as the fix.
Signals from the same message share a timestamp by construction, so a frame over one message is always dense.
Downsample and bound the result¶
downsample=N runs the agent's M4 strategy and returns about N points per signal, preserving the shape of the trace — the right tool for charting a long window:
max_rows=N caps the row count instead; 0 (the default) means no cap. The two are mutually exclusive. Either way frame.downsampled and frame.truncated tell you whether the agent had to reduce what it sent, and the frame's repr adds a flags: line naming whichever is set.
Read a frame¶
frame.columns # column paths; list(frame) and frame.keys() too
frame["bus0/BMS_message/cells.cell_0"] # one column, as a SignalSeries
frame.signals["bus0/BMS_message/cells.cell_0"].unit # the frame's own SignalCatalog
frame.items() # (path, series) pairs
len(frame) # rows; frame.shape is (rows, columns)
frame.head(5) # first rows, still a frame
frame[10:20] # rows by position; frame.iloc[10:20] too
frame["+10s":"+20s"] # rows by time, in the between() grammar; frame.loc too
frame.iloc[-1] # one row, a pandas Series by column; frame.loc[timestamp] too
frame.iloc[0, 2] # one value; frame.loc[timestamp, path] too
frame.loc[mask, ["bus0/BMS_message/cells.cell_0"]] # rows and columns at once
frame[frame.index >= t] # a boolean list or array keeps rows too
frame.short_names() # drop the shared path prefix from column names
frame.rename(columns=str.upper) # a mapping or a callable, alone or as columns=
frame.between(start, end) # narrow to a sub-window, in memory
frame.resample("10s", "mean") # re-bucket on a fixed interval
frame.describe() # count/mean/std/quartiles per column, with units
frame.corr() # pandas' correlation matrix of the columns
frame.ffill(limit=4) # sample-and-hold across gaps of up to four rows
frame.fillna(0.0) # every gap takes a value
frame.dropna(subset=["bus0/BMS_message/cells.cell_0"]) # rows where that column has a sample
describe() is the fastest way to see a whole frame's shape at once, and it returns a pandas DataFrame, as corr() does. A frame prints its schema and then its first and last rows.
resample() aggregates each bucket with how: "mean", "min", "max", "median", "sum", "std", "count", "first" or "last". An empty bucket is missing, never a made-up zero, and a state or enum column takes its last value where averaging would invent a code that does not exist. The spread of temperature readings is a ΔdegC, as std() reports it.
Add a column¶
Set a column the way pandas does, in place or on a copy. A series brings its unit, so the new column reads in the right unit everywhere the frame shows one:
frame["power"] = (frame["bus0/BMS.voltage"] * frame["bus0/BMS.current"]).to("kW")
frame.assign(temp_f=lambda f: f["bus0/BMS.temp"].to("degF")) # a copy, as pandas' assign
A value is a series on the frame's time axis, a scalar that fills every row, or one value per row. A frame with a computed column is derived: a check on one of its columns evaluates the samples the frame holds.
Hand off to pandas / Arrow¶
You do not need these to see a frame — every object here renders itself. Hand a SignalFrame to pandas as a DataFrame, or to Arrow as a Table, when you want to do something the SDK does not do:
frame.to_pandas() # UTC DatetimeIndex, one column per signal
frame.to_arrow() # the underlying pyarrow.Table
to_pandas() indexes by a timezone-aware UTC DatetimeIndex. Pass index=False to get time as an ordinary column. frame.index and frame.dtypes answer without converting.
frame.groupby(by) hands a grouping to pandas directly, with by a column label or a series on the frame's time axis.
numpy, Arrow and polars¶
np.asarray(frame) and frame.to_numpy() give the data columns as one 2-D array, and pd.DataFrame.from_arrow(frame), pa.table(frame) and polars.DataFrame(frame) read the frame's Arrow stream. pd.DataFrame(frame) reads only the numbers, without labels or time; use to_pandas().
Back from pandas¶
SignalFrame.from_pandas(df, units={...}) brings a whole DataFrame back as a frame, labels and nanosecond times intact, and SignalSeries.from_pandas(s, unit=...) brings back one column. A unit comes back only when you pass it. df.attrs["units"] lists each column's unit for reading, but pandas carries attrs through arithmetic unchanged, so a unit found there after df["v"] * df["i"] would be wrong:
df = frame.to_pandas()
df["power"] = df["voltage"] * df["current"]
frame = SignalFrame.from_pandas(df, units={**df.attrs["units"], "power": "W"})
units= also takes a frame's catalog, units=frame.signals, for columns you did not change.
Chart a frame¶
plot() emits Vega-Lite. The app renders it inline, an HTML export renders it offline, and nothing has to be added to dependencies:
The legend reads the frame's column labels, as a pandas plot does, so it carries full paths unless you rename() the columns or call short_names() first. On a TraceSet frame the legend reads baseline::bus0/... until you do. A single series plots under its own name:
To chart some columns and not others, name them, the same keys frame[[...]] takes:
The y-axis fits the data, so four cells between 3.84 V and 3.95 V show the 45 mV that separates them. Pass zero=True when the distance from zero is the point:
Compute on series with units¶
Every column of a frame is a SignalSeries: one Arrow column that knows its path, its unit, and the frame's time axis. Two series from the same frame share that axis by identity, so they compose without an alignment step.
Units are an exponent map over the tokens in the catalog string, so they compose through arithmetic and cancel where they should:
pack = agent.query(
["bus0/DCDC_message/status.input_voltage", "bus0/BMS_message/status.pack_current"],
start="-1m",
).ffill().dropna()
voltage = pack["bus0/DCDC_message/status.input_voltage"] # V
current = pack["bus0/BMS_message/status.pack_current"] # A
power = voltage * current # W
energy = power.integrate() # W·s, rendered J
slew = current.derivative() # A/s
(power / current).unit # 'V' — the A cancels
to() converts a series or a single value to another unit of the same kind. It knows the SI prefixes and the units vehicle and battery catalogs use, including Wh, Ah, rpm, kph, mph, bar, psi, % and the temperature scales:
power.to("kW") # the same samples in kW
energy.to("Wh").max() # the peak, in Wh
speed.to("rad/s") # rpm × 2π/60
energy.max().to("kWh") # a single value converts too
A derived series is named after the operands it came from, so a chart legend and a check row say where the numbers came from:
power.name
# 'bus0/DCDC_message/status.input_voltage × bus0/BMS_message/status.pack_current'
energy
# SignalSeries 'integrate(bus0/DCDC_message/status.input_voltage × bus0/BMS_message/status.pack_current)' (J) len=59 nulls=0 2026-09-04 17:05:42.760 → 17:06:40.831
# [0, 1199, 1744, …, 68132]
* and / compose units. +, - and comparisons put both sides on the left side's scale, so voltage + millivolts reads in V, and an aggregate keeps its unit as an operand, so current - current.mean() stays in A. Units that measure different things raise ValueError:
voltage + current
# ValueError: cannot add 'V' and 'A': they measure different things. .to() converts
# between units that measure the same thing; .with_unit() asserts a unit you know
Temperature follows the physics. A degC reading converts to degF with the offset, but the difference of two readings is a ΔdegC, which converts by scale alone:
rise = outlet - inlet # ΔdegC, not degC
rise.to("ΔdegF") # × 1.8, no +32
outlet.to("degF") # a reading: × 1.8 + 32
diff(), std() and a rolling std() of a reading are differences too, and so is temp - temp.mean(). A reading on an offset scale does not scale: twice 20 °C is not 40 °C of anything. So temp * 2, -temp, abs(temp), inlet + outlet, temp.sum(), temp.integrate(), and temp - 25.0 (where 25.0 could be a reading or a difference) keep their numbers and lose their unit rather than carry a wrong one. A bare number added to a reading is a difference, so temp + 5 stays in degC. Write a difference as ΔdegC, or delta_degC in ASCII.
An aggregate keeps its unit through arithmetic too: voltage.max() - voltage.min() is a spread in V named max(...) − min(...), and a value in mV compares with one in V on one scale.
Two series on different time axes align by time for arithmetic, as pandas aligns them: coolant_out.dropna() - coolant_in.dropna() spans both axes and is missing where either had no sample. Comparisons and masks need one axis, so compare the difference: (a - b) > 0. Signals sampled on different clocks share no timestamps at all, which warns; hold one on the other's axis first:
power = speed.to("rad/s") * torque.reindex(speed.index, method="ffill") # rad·Nm/s
power.to("kW")
torque.asof("+12.5s") # the last torque sample at or before that time
A number is a factor, so power * 0.001 keeps W: use power.to("kW") to change the unit. A number joins the derived name, so (power * 0.001).max() reads max(power × 0.001), never max(power). Arrow arrays, numpy arrays and plain sequences pair by position. Adding one keeps the series' unit, but multiplying or dividing by one leaves no unit, since the array may hold a quantity of its own. A pandas Series on the series' own time axis combines the same way; any other pandas operand is refused, because pandas pairs by index:
series * pandas_series
# TypeError: cannot combine a SignalSeries with a pandas Series on another index; do the
# arithmetic in pandas after SignalSeries.to_pandas()
with_unit() asserts a unit you know to be right; rename() relabels a derived series.
Aggregations¶
An aggregation returns a NamedScalar — a float that also carries the name of what it measured and its unit, and prints to four significant digits:
series.min() # 3.629 V
series.max() series.mean() series.sum()
series.median() series.std() # ddof=1, the sample standard deviation
series.var() series.corr(other) # variance in V², correlation aligned by time; method="spearman" too
series.quantile(0.95) # p95(...), interpolated as in pandas
series.idxmax() # the timestamp of the largest sample
series.describe() # pandas' summary, with the unit as its first row
state.value_counts() counts each value of a state or fault code, and unique() and nunique() list and count them.
float(series.min()) is the exact value. count() is the one aggregation that is not a NamedScalar: it returns a plain int.
median is exact, not interpolated. Nulls are skipped by every aggregation.
Integrate, differentiate, mask¶
power.integrate() # cumulative trapezoid over time; first sample 0
power.where(current > 100).fillna(0.0).integrate() # only while the condition holds
current.rate() # change per second between neighbors, A/s
current.derivative() # centered np.gradient over time: smoother, lower peaks
power.where(current > 25.0) # keep matching samples, null the rest
current.abs() current.diff() current.cumsum()
current.clip(0.0, 30.0) current.clip(upper=limit) # a bound can be a series
series.ffill() series.ffill(limit=2) series.fillna(0.0) series.dropna()
series.replace([np.inf, -np.inf], np.nan) # pandas' replace; a NaN is a missing sample
series.isna() series.notna() series.shift(1) series.shift(freq="1s") # the times move
current.where(current > 0, 0.0) # pandas' where: the rest become 0.0
speed.fillna(gps_speed) # each gap takes the other series' sample at that time
current.add(other, fill_value=0) # add/sub/mul/div, filling a gap on either side first
where(), fillna() and clip() take a series as the other value. It aligns by time, as pandas aligns it, and converts to this series' unit, so a km/h fill for an m/s speed lands in m/s.
Comparison operators on a series produce a boolean series, which is what where() masks with. An aggregation over a mask answers the narrower question: power.where(current > 25.0).mean() is the mean power under load. (current > 25.0).any() and .all() answer whether any or every sample was under load.
A comparison treats a missing sample the way pandas treats NaN: unequal to everything, and neither above nor below anything. So (temp > 25).mean() is the fraction of all samples above 25, and (s != s.shift()).cumsum() numbers runs from the first sample. A NaN is a missing sample too. Masks combine with &, |, ^ and ~; on a boolean signal with gaps, a gap stays unknown, so False & missing is False and True & missing is missing. A mask indexes a series or a whole frame, sums to a count, counts as 0 and 1 in arithmetic, reads as a numpy boolean array from to_numpy(), and numbers its runs:
under_load = (current > 25.0) & ~(voltage < 3.2)
current[under_load] # the samples under load, time kept
pack[under_load] # the frame's rows under load
under_load.sum() # how many samples, as an int
under_load.runs() # one row per run: start, end, duration, samples
(speed == 0).runs(min_duration="2s") # only the stops of two seconds or more
fault.isin([3, 7]) # a mask from a set of values
state == "CHARGING" # a string compares sample by sample
soc.between(20, 80) # a value range, as pandas' Series.between
runs() is how events come out of a signal. A missing sample compares as False and so ends a run; on a sparse signal, compare the sample-and-held series, (temp.ffill() > 20).runs(). A run lasts until the sample that ends it, and complete is False for a run cut off by either end of the capture, whose duration is only what was recorded. min_samples= and min_duration= keep only the runs that long; the table is a DataFrame, so filter it further as one.
For the pandas run-numbering idiom, under_load.ne(under_load.shift(fill_value=False)).cumsum() works as written, and power.groupby(state).mean() hands a grouping to pandas.
Moving windows¶
rolling() takes a sample count or a duration and returns a window to aggregate: mean(), min(), max(), sum(), std(), median(), quantile(q) or count(). The result is a series on the same time axis, and it keeps the unit:
speed.rolling("1s").mean() # one-second moving average, rpm
speed.rolling(10).max() # the largest of each ten samples
resample() puts one series on a fixed time grid, as it does a frame. It takes pandas' spelling, speed.resample("1s").mean(), or the frame's, speed.resample("1s", how="mean"), and aggregates each bucket the same ways.
Escape hatch and back¶
When the built-in surface runs out, take the data to pandas or numpy and come back:
series.to_pandas() # pandas Series, UTC DatetimeIndex
series.to_numpy() # float64, nulls as NaN; to_numpy(dtype=...) casts first
series.to_arrow()
SignalSeries.from_pandas(smoothed, unit="rpm") # a unit comes back only when you name it
One-operand numpy functions map onto the same time axis, so np.sqrt(series) and np.isfinite(series) return series, and np.sqrt of m² is in m. np.mean(series), np.maximum(series, 0), np.clip(series, 0, 30) and series ** 2 all work, and reductions take pandas' skipna=. A one-dimensional array pairs by position, as a list does. A pandas Series combines only when it sits on the series' own time axis, as current.index.to_series().diff() does.
A series and a frame answer to pandas' names for what they do natively: current.gt(25.0) is current > 25.0, series.index is the time axis as a pandas DatetimeIndex, and astype(), shape, size and empty mean what they mean in pandas. Iterating a frame yields its column paths, and frame.columns lists them. A pandas name they do not have raises an AttributeError naming the call that does:
current.ewm(span=5)
# AttributeError: 'SignalSeries' object has no attribute 'ewm'; pandas has it:
# call .to_pandas().ewm(...), and SignalSeries.from_pandas() to come back
Read latest values¶
latest() answers "what is it right now", with no window:
agent.latest("bus0/BMS_message/status.pack_current")
# pack_current = 30.59 A @ 2026-09-04 17:06:40.831 (bus0/BMS_message/status, localhost)
A single literal path returns a LatestValue carrying the value, its unit, its enum label if it has one, and its timestamp. A pattern or a sequence returns a Snapshot, a read-only Mapping[str, LatestValue] — snapshot[path], in, .items(), len() all work, and it renders as a table:
agent.latest("bus0/BMS_message/cells.*")
# Snapshot @ 2026-09-04 17:06:40.831 · 8 signals
# bus0/BMS_message/cells.cell_0 3.653 V
# bus0/BMS_message/cells.cell_1 3.658 V
# bus0/BMS_message/cells.cell_2 3.661 V
# bus0/BMS_message/cells.cell_3 3.084 V
# bus0/BMS_message/cells.cell_4 3.664 V
# bus0/BMS_message/cells.cell_5 3.669 V
# bus0/BMS_message/cells.cell_6 3.676 V
# bus0/BMS_message/cells.cell_7 3.685 V
For one path: no catalog match raises SignalNotFound, and a match with no sample inside lookback raises NoData. For a pattern or a sequence, misses simply omit their key.
Read at a cursor time¶
at() is the freeze-frame: the value of every requested signal at or before one instant.
It returns a Snapshot, keyed by path. min_time= floors the search so a stale, unrelated sample cannot pose as context — anchor it to the incident rather than to the clock:
Replay a window of changes¶
window() is built for states and events, where a query would return thousands of rows repeating one string. It returns a ReplayWindow: the opening snapshot at start, plus changes, one entry per update after it.
replay = agent.window("bus1/inverter_status.mode", start="-1m", duration="1m")
replay
for change in replay.changes:
print(f"{change.time:%H:%M:%S} {change.value}")
start takes the time grammar, duration the duration grammar. display_fps= thins the change stream for a live display; the default, 0, thins nothing.
Watch live values¶
watch() polls and yields a Snapshot per tick. until= bounds it with the duration grammar, which is what makes a watching notebook finish on its own:
for tick in agent.watch(["bus1/inverter_status.mode"], interval=1.0, until="30s"):
value = tick["bus1/inverter_status.mode"]
print(f"{value.time:%H:%M:%S} {value.value}")
Without until the loop runs until you interrupt it. on_error="skip" tolerates transient failures rather than raising, up to max_consecutive_errors.
Polling can miss a transition between ticks; window() cannot. Use watch() for a dashboard and window() for evidence.
Errors¶
Every failure is a typed subclass of AgentError, so you can catch the class or the family:
| Error | When |
|---|---|
AgentUnavailable |
nothing answered before the timeout |
SignalNotFound |
a path or pattern matched nothing |
AmbiguousSignal |
a path resolves to more than one signal — two producers, traces or segments own it, or it matches several columns of a frame |
NoData |
the signal resolved but had no sample in range |
QueryRangeTooLarge |
the window asks for more than the agent will send |
ConnectionTargetError |
the target string isn't a usable endpoint |
from zelos_sdk import errors
try:
agent.latest("bus0/BMS_message/status.pack_current")
except errors.NoData:
...
What's next¶
-
The same calls against a saved
.trz— one file, or several runs on a shared clock. -
Turn a queried series into a rule with a result and the evidence behind it.
-
Put all of this in a markdown file the agent runs.
-
Call into a producer instead of reading from it.
-
Every class, method and parameter, generated from the package.