How to Filter Events

Python Only

These APIs exist only in the Python SDK.

Log conditions reduce data volume. A TraceSourceCache evaluates them on every log() call and sends only the rows that pass.

The condition classes live in zelos_sdk.trace.conditions. Every example below imports them as conditions:

from zelos_sdk.trace import conditions

Quick Start

Set a default condition on the source:

import zelos_sdk
from zelos_sdk.trace import conditions

zelos_sdk.init()

source = zelos_sdk.TraceSourceCache("filtered")

# Log on change, or at least once per second
source.set_default_log_condition(conditions.DefaultLogCondition())

source.log("sensor", {"temperature": 25.0})  # Sent (first value)
source.log("sensor", {"temperature": 25.0})  # Skipped (no change)
source.log("sensor", {"temperature": 25.1})  # Sent (changed by 0.1)

set_default_log_condition() with no argument also sets DefaultLogCondition(). Pass None to remove the default.

How Filtering Decides

Filtering decides per row, not per field. Each log() call is one row.

Each field gets its condition from one of two places:

• Its own entry in the conditions dict of add_event().
• The source default from set_default_log_condition(), when the field has no entry.

A field with neither has no condition. A new TraceSourceCache has no default, so it sends every row until you set a condition.

The cache sends a row when at least one field in the row passes its condition. A field with no condition always passes. So one field without a condition turns filtering off for the whole event. A None entry, as in conditions={"status": None}, gives that field no condition.

Some more rules follow from the per-row decision:

• A sent row carries every value you passed, not only the values that changed.
• When a row is sent, every field in it records that value and time as its last logged value and time.
• A field missing from the row, or set to None, is not evaluated.
• A row with no values is always sent.
• The cache stores every value, even from a skipped row. .get() always returns the latest logged value.
• log_batch() does not evaluate conditions. It sends every row in the batch.

Put fields that should filter together in one event. Give a field its own event when it should filter on its own.

Each field gets its own copy of a condition object. You can pass one object for several fields; they do not share state.

Condition Types

ValueLogCondition

Logs when the value differs from the last logged value:

import zelos_sdk
from zelos_sdk.trace import conditions

zelos_sdk.init()
source = zelos_sdk.TraceSourceCache("state")

# Only log when the mode changes
event = source.add_event("machine", [
    zelos_sdk.TraceEventFieldMetadata("mode", zelos_sdk.DataType.String)
], conditions={
    "mode": conditions.ValueLogCondition()
})

source.log("machine", {"mode": "IDLE"})     # Sent (first value)
source.log("machine", {"mode": "IDLE"})     # Skipped (same)
source.log("machine", {"mode": "RUNNING"})  # Sent (changed)

DeltaLogCondition

Logs when a numeric value moves by at least delta from the last logged value. Small steps never add up to drift, because the reference only moves when a row is sent. A non-numeric value falls back to a plain change check.

# Log if position changes by at least 0.0001° (~11 meters)
event = source.add_event("gps", [
    zelos_sdk.TraceEventFieldMetadata("latitude", zelos_sdk.DataType.Float64, "deg")
], conditions={
    "latitude": conditions.DeltaLogCondition(delta=0.0001)
})

source.log("gps", {"latitude": 37.0})      # Sent (first value)
source.log("gps", {"latitude": 37.00005})  # Skipped (Δ=0.00005 from 37.0)
source.log("gps", {"latitude": 37.00008})  # Skipped (Δ=0.00008 from 37.0)
source.log("gps", {"latitude": 37.00011})  # Sent (Δ=0.00011 from 37.0)
# The new reference is 37.00011

EpsilonLogCondition

EpsilonLogCondition(epsilon) behaves exactly like DeltaLogCondition(delta=epsilon). Without an argument, epsilon is the machine epsilon, so almost any change logs.

# Only log if temperature changes by at least 0.5°C
event = source.add_event("sensor", [
    zelos_sdk.TraceEventFieldMetadata("temperature", zelos_sdk.DataType.Float64, "°C")
], conditions={
    "temperature": conditions.EpsilonLogCondition(epsilon=0.5)
})

source.log("sensor", {"temperature": 25.0})   # Sent (first value)
source.log("sensor", {"temperature": 25.3})   # Skipped (Δ=0.3 < 0.5)
source.log("sensor", {"temperature": 25.6})   # Sent (Δ=0.6 ≥ 0.5)

TimeLogCondition

Logs when at least time_threshold_s seconds have passed since the field was last logged, whatever the value:

# Log at most every 30 seconds
event = source.add_event("heartbeat", [
    zelos_sdk.TraceEventFieldMetadata("uptime", zelos_sdk.DataType.Float64, "s")
], conditions={
    "uptime": conditions.TimeLogCondition(time_threshold_s=30.0)
})

DefaultLogCondition

DefaultLogCondition(time_threshold_s=1.0, epsilon=0.1) logs when any of these is true:

• At least time_threshold_s seconds have passed since the last logged value.
• A float value moved by at least epsilon.
• A non-float value changed.

Composite Conditions

These conditions combine a time threshold with a value check. The field logs when either one passes. time_threshold_s defaults to 1.0.

# Log on change OR timeout (heartbeat)
conditions.ValueOrTimeLogCondition(time_threshold_s=60.0)

# Log on a change of at least epsilon OR timeout
conditions.EpsilonOrTimeLogCondition(epsilon=0.5, time_threshold_s=30.0)

# Log on a change of at least delta OR timeout
conditions.DeltaOrTimeLogCondition(delta=1.0, time_threshold_s=10.0)

CompositeLogCondition combines any list of conditions with the same OR logic:

conditions.CompositeLogCondition([
    conditions.TimeLogCondition(time_threshold_s=5.0),
    conditions.DeltaLogCondition(delta=2.0),
])

Custom Conditions

Subclass LogCondition and implement should_log(current_value, current_time_ns). The base class on_logged() stores last_logged_value and last_log_time_ns each time the field is logged. The cache copies the condition with copy.deepcopy, so keep it deep-copyable.

class RisingCondition(conditions.LogCondition):
    """Log only when the value rises above the last logged value."""

    def should_log(self, current_value, current_time_ns):
        return self.last_logged_value is None or current_value > self.last_logged_value

event = source.add_event("peak", [
    zelos_sdk.TraceEventFieldMetadata("level", zelos_sdk.DataType.Int32)
], conditions={
    "level": RisingCondition()
})

for level in [1, 3, 2, 5]:
    source.log("peak", {"level": level})  # Sends 1, 3 and 5

Common Patterns

High-Frequency Sensor Filtering

Reduce data from noisy sensors:

import time
import random
import math
import zelos_sdk
from zelos_sdk.trace import conditions

zelos_sdk.init()
source = zelos_sdk.TraceSourceCache("vibration")

# Aggressive filtering for a high-rate sensor
source.set_default_log_condition(
    conditions.DeltaOrTimeLogCondition(
        delta=0.1,            # Log on a change of at least 0.1 g
        time_threshold_s=1.0  # Heartbeat every second
    )
)

# All three axes use the default, so a row goes out when any axis passes
event = source.add_event("accel", [
    zelos_sdk.TraceEventFieldMetadata("x", zelos_sdk.DataType.Float32, "g"),
    zelos_sdk.TraceEventFieldMetadata("y", zelos_sdk.DataType.Float32, "g"),
    zelos_sdk.TraceEventFieldMetadata("z", zelos_sdk.DataType.Float32, "g")
])

# Simulate a 1 kHz sensor with noise
samples_sent = 0
start_time = time.time()

while time.time() - start_time < 10:  # Run for 10 seconds
    t = time.time() - start_time
    x = 0.5 * math.sin(2 * math.pi * 10 * t) + random.gauss(0, 0.05)
    y = 0.3 * math.cos(2 * math.pi * 10 * t) + random.gauss(0, 0.05)
    z = 9.8 + random.gauss(0, 0.05)

    source.log("accel", {"x": x, "y": y, "z": z})
    samples_sent += 1

    time.sleep(0.001)  # 1 kHz

print(f"Samples generated: {samples_sent}")

State Machine Monitoring

Log state transitions with a slow temperature heartbeat:

from enum import IntEnum
import random
import time
import zelos_sdk
from zelos_sdk.trace import conditions

class State(IntEnum):
    IDLE = 0
    RUNNING = 1
    ERROR = 2

zelos_sdk.init()
source = zelos_sdk.TraceSourceCache("machine")

# A row goes out when any of the three fields passes its condition
status = source.add_event("status", [
    zelos_sdk.TraceEventFieldMetadata("state", zelos_sdk.DataType.UInt8),
    zelos_sdk.TraceEventFieldMetadata("cycles", zelos_sdk.DataType.UInt32),
    zelos_sdk.TraceEventFieldMetadata("temperature", zelos_sdk.DataType.Float32, "°C")
], conditions={
    "state": conditions.ValueLogCondition(),   # Any state change
    "cycles": conditions.ValueLogCondition(),  # Every new cycle
    "temperature": conditions.EpsilonOrTimeLogCondition(
        epsilon=2.0,           # Temperature change of at least 2°C
        time_threshold_s=60.0  # Or once a minute
    )
})

# Add readable state names
source.add_value_table("status", "state", {
    0: "IDLE", 1: "RUNNING", 2: "ERROR"
})

# Simulate the machine
current_state = State.IDLE
cycles = 0
temp = 25.0

for i in range(1000):
    # Occasional state changes
    if random.random() < 0.01:
        current_state = random.choice(list(State))
        if current_state == State.RUNNING:
            cycles += 1

    # Temperature fluctuates slowly
    temp += random.uniform(-0.3, 0.3)
    temp = max(20, min(80, temp))  # Clamp to range

    # Log (filtered by conditions)
    source.log("status", {
        "state": current_state.value,
        "cycles": cycles,
        "temperature": temp
    })

    time.sleep(0.1)

print(f"Final state: {current_state.name}, Cycles: {cycles}, Temp: {temp:.1f}°C")

Battery Monitoring

Give each field its own sensitivity. The row still goes out as a whole: when any field passes, the row carries all five values. The most active field sets the row rate. Move a field to its own event if it should not drive the rate of the others.

import random
import time
import zelos_sdk
from zelos_sdk.trace import conditions

zelos_sdk.init()
source = zelos_sdk.TraceSourceCache("battery")

battery = source.add_event("status", [
    zelos_sdk.TraceEventFieldMetadata("voltage", zelos_sdk.DataType.Float32, "V"),
    zelos_sdk.TraceEventFieldMetadata("current", zelos_sdk.DataType.Float32, "A"),
    zelos_sdk.TraceEventFieldMetadata("soc", zelos_sdk.DataType.Float32, "%"),
    zelos_sdk.TraceEventFieldMetadata("temperature", zelos_sdk.DataType.Float32, "°C"),
    zelos_sdk.TraceEventFieldMetadata("charging", zelos_sdk.DataType.Boolean)
], conditions={
    # Voltage: 0.1 V changes or every 10 s
    "voltage": conditions.EpsilonOrTimeLogCondition(epsilon=0.1, time_threshold_s=10.0),

    # Current: 1 A changes or every 10 s
    "current": conditions.DeltaOrTimeLogCondition(delta=1.0, time_threshold_s=10.0),

    # SOC: 1% changes
    "soc": conditions.DeltaLogCondition(delta=1.0),

    # Temperature: 0.5°C changes or every minute
    "temperature": conditions.EpsilonOrTimeLogCondition(epsilon=0.5, time_threshold_s=60.0),

    # Charging: state changes only
    "charging": conditions.ValueLogCondition()
})

# Simulate the battery
voltage = 12.6
current = 0.0
soc = 100.0
temp = 25.0
charging = False

for _ in range(100):
    # Simulate discharge/charge
    if charging:
        current = random.uniform(1, 5)  # Charging current
        soc = min(100, soc + 0.1)
        voltage = 12.0 + (soc / 100) * 2.6
        if soc >= 100:
            charging = False
    else:
        current = random.uniform(-10, -1)  # Discharge current
        soc = max(0, soc - 0.05)
        voltage = 11.0 + (soc / 100) * 2.6
        if soc <= 20:
            charging = True

    # Add noise
    voltage += random.gauss(0, 0.05)
    current += random.gauss(0, 0.2)
    temp += random.uniform(-0.1, 0.1)

    # Log (filtered)
    source.log("status", {
        "voltage": voltage,
        "current": current,
        "soc": soc,
        "temperature": temp,
        "charging": charging
    })

    time.sleep(0.1)

print(f"Final SOC: {soc:.1f}%, Charging: {charging}")

Adaptive Filtering

Change the default condition at run time. set_default_log_condition() also replaces the condition on every existing field that uses the default.

import zelos_sdk
from zelos_sdk.trace import conditions

class AdaptiveMonitor:
    """Switch between normal and high-precision modes"""

    def __init__(self):
        zelos_sdk.init()
        self.source = zelos_sdk.TraceSourceCache("adaptive")
        self.set_precision(high=False)

        # Define schema once
        self.event = self.source.add_event("data", [
            zelos_sdk.TraceEventFieldMetadata("value", zelos_sdk.DataType.Float64),
            zelos_sdk.TraceEventFieldMetadata("quality", zelos_sdk.DataType.Float32)
        ])

    def set_precision(self, high=False):
        """Toggle between filtering modes"""
        if high:
            # Minimal filtering for critical operations
            condition = conditions.EpsilonOrTimeLogCondition(
                epsilon=0.01,
                time_threshold_s=0.1
            )
            print("High precision mode: minimal filtering")
        else:
            # Aggressive filtering for normal operation
            condition = conditions.DeltaOrTimeLogCondition(
                delta=1.0,
                time_threshold_s=10.0
            )
            print("Normal mode: aggressive filtering")

        self.source.set_default_log_condition(condition)

    def log(self, value, quality):
        """Log data with current filter settings"""
        self.source.log("data", {
            "value": value,
            "quality": quality
        })

# Usage
monitor = AdaptiveMonitor()

# Normal operation
monitor.set_precision(high=False)
for i in range(100):
    monitor.log(value=i * 0.1, quality=0.95)

# Critical measurement
monitor.set_precision(high=True)
for i in range(100):
    monitor.log(value=50 + i * 0.01, quality=0.99)

# Back to normal
monitor.set_precision(high=False)

Using Cached Values

Read back previously logged values. This source has no conditions, so it sends every row.

import random
import time
import zelos_sdk

zelos_sdk.init()
source = zelos_sdk.TraceSourceCache("control")

# Define PID controller data
pid = source.add_event("pid", [
    zelos_sdk.TraceEventFieldMetadata("setpoint", zelos_sdk.DataType.Float64),
    zelos_sdk.TraceEventFieldMetadata("measured", zelos_sdk.DataType.Float64),
    zelos_sdk.TraceEventFieldMetadata("error", zelos_sdk.DataType.Float64),
    zelos_sdk.TraceEventFieldMetadata("output", zelos_sdk.DataType.Float64)
])

# Simple PID simulation
kp, ki, kd = 0.5, 0.1, 0.05
integral = 0.0
setpoint = 50.0
measured = 30.0

for _ in range(100):
    # Calculate error
    error = setpoint - measured

    # The previous error is None until the first log
    prev_error = source.pid.error.get()
    if prev_error is None:
        prev_error = 0.0

    # PID calculation
    integral += error * 0.01  # dt = 0.01
    derivative = (error - prev_error) / 0.01
    output = kp * error + ki * integral + kd * derivative

    # Log current values
    source.log("pid", {
        "setpoint": setpoint,
        "measured": measured,
        "error": error,
        "output": output
    })

    # Simulate system response
    measured += output * 0.1  # System gain
    measured += random.gauss(0, 0.5)  # Noise

    time.sleep(0.01)

print(f"Final: Setpoint={setpoint:.1f}, Measured={measured:.1f}, Error={error:.1f}")

Best Practices

1. Match Conditions to Data

# Match each condition to how the field behaves
per_field = {
    "discrete_state": conditions.ValueLogCondition(),     # Discrete values
    "analog_sensor": conditions.DeltaLogCondition(0.5),   # Threshold above the noise
    "counter": conditions.ValueLogCondition(),            # Every change
}

2. Give Every Field a Condition

One field without a condition sends every row of its event. Set a source default, or list every field in conditions.

3. Always Include Heartbeats

# Ensure liveness
condition = conditions.DeltaOrTimeLogCondition(
    delta=1.0,
    time_threshold_s=30.0  # At most 30 s between rows
)

4. Document Filter Behavior

event = source.add_event("temp", [
    zelos_sdk.TraceEventFieldMetadata("value", zelos_sdk.DataType.Float64, "°C")
], conditions={
    # Logs if temperature changes by at least 0.5°C from the last logged value
    # Also sends a heartbeat every 60 s regardless
    "value": conditions.EpsilonOrTimeLogCondition(
        epsilon=0.5,
        time_threshold_s=60.0
    )
})