How to Record to Files¶
Python Only
These APIs exist only in the Python SDK.
Record trace data to .trz files for offline analysis, debugging, or archival.
TraceWriter records every event that the sources in its namespace log between open() and close(). Events logged before open() are not in the file. Writing a file needs no running agent.
Basic Recording¶
Use TraceWriter to capture all events to a file:
from pathlib import Path
import random
import zelos_sdk
zelos_sdk.init()
source = zelos_sdk.TraceSource("experiment")
# TraceWriter refuses to open an existing file, so remove the last run's file
Path("output.trz").unlink(missing_ok=True)
# Record to file using context manager
with zelos_sdk.TraceWriter("output.trz") as writer:
# All events are automatically captured
for i in range(1000):
source.log("measurement", {
"index": i,
"value": random.random() * 100,
"valid": True
})
print("Data saved to output.trz")
Writer Options¶
TraceWriter(path, batch_size=1000, batch_timeout_ms=100, allow_existing=False, namespace=None)
| Parameter | Default | What it does |
|---|---|---|
path |
required | File to write. writer.path returns it. |
batch_size |
1000 |
Events per write to the file. |
batch_timeout_ms |
100 |
Longest wait before a partial batch is written. |
allow_existing |
False |
Open a file that already exists. See Existing Files. |
namespace |
None |
Record only the sources in this TraceNamespace. None records the global namespace. |
Existing Files¶
By default, open() raises RuntimeError when the file exists:
To make a script safe to run again, write a new file each time, as the patterns below do with timestamped names. To replace the file, delete it before you open the writer, as in Basic Recording.
allow_existing=True does not replace the file. It opens the file and appends a new data segment to it. The file's stored time range comes from its first recording. So TraceReader.time_range() can end before the appended data. Query appended data with explicit start and end bounds.
Record One Namespace¶
Pass a TraceNamespace to record only the sources created in it. Sources in the global namespace and in other namespaces stay out of the file.
from pathlib import Path
import zelos_sdk
bench = zelos_sdk.TraceNamespace("bench")
recorded = zelos_sdk.TraceSource("rig", namespace=bench)
ignored = zelos_sdk.TraceSource("ui") # Global namespace
Path("bench.trz").unlink(missing_ok=True)
with zelos_sdk.TraceWriter("bench.trz", namespace=bench):
recorded.log("status", {"voltage": 12.6}) # Written to bench.trz
ignored.log("status", {"clicks": 3}) # Not written
Recording Methods¶
Context Manager (Recommended)¶
Path("data.trz").unlink(missing_ok=True)
with zelos_sdk.TraceWriter("data.trz") as writer:
source.log("event", {"value": 123})
# File automatically closed
Manual Control¶
Path("data.trz").unlink(missing_ok=True)
writer = zelos_sdk.TraceWriter("data.trz")
try:
writer.open()
source.log("event", {"value": 123})
finally:
writer.close() # Always close!
Common Patterns¶
Timestamped Filenames¶
import datetime
import random
import time
import zelos_sdk
def record_session(duration_seconds=60):
"""Record with timestamp in filename"""
timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
filename = f"session_{timestamp}.trz"
zelos_sdk.init()
source = zelos_sdk.TraceSource("session")
with zelos_sdk.TraceWriter(filename) as writer:
start = time.time()
count = 0
while time.time() - start < duration_seconds:
source.log("data", {"value": random.random()})
count += 1
time.sleep(0.01)
print(f"Recorded {count} events to {filename}")
return filename
Conditional Recording¶
Start/stop recording based on conditions:
import datetime
import random
import time
import zelos_sdk
class ConditionalRecorder:
"""Record only when threshold exceeded"""
def __init__(self, threshold=80.0):
zelos_sdk.init()
self.source = zelos_sdk.TraceSource("monitor")
self.threshold = threshold
self.writer = None
self.recording = False
def update(self, value):
"""Check value and manage recording"""
# Always log for real-time viewing
self.source.log("measurement", {"value": value})
# Start recording if threshold exceeded
if value > self.threshold and not self.recording:
# Microseconds keep two triggers in one second apart
timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S_%f")
self.writer = zelos_sdk.TraceWriter(f"event_{timestamp}.trz")
self.writer.open()
self.recording = True
self.source.log("trigger", {
"threshold": self.threshold,
"value": value
})
print(f"Started recording: value {value} > {self.threshold}")
# Stop recording when value returns to normal
elif value < self.threshold * 0.8 and self.recording:
self.source.log("recovered", {"value": value})
self.writer.close()
self.recording = False
print(f"Stopped recording: value {value} back to normal")
def close(self):
"""Ensure file is closed"""
if self.recording and self.writer:
self.writer.close()
# Usage
recorder = ConditionalRecorder(threshold=75.0)
try:
for _ in range(1000):
# Simulate sensor with occasional spikes
value = 50 + random.gauss(0, 10)
if random.random() < 0.01: # 1% chance of spike
value += 40
recorder.update(value)
time.sleep(0.1)
finally:
recorder.close()
Ring Buffer Recording¶
Keep a rolling buffer and save on trigger:
import collections
import datetime
import random
import time
import zelos_sdk
class RingBufferRecorder:
"""Continuously buffer data, save on trigger"""
def __init__(self, buffer_seconds=30):
zelos_sdk.init()
self.source = zelos_sdk.TraceSource("buffer")
self.buffer_seconds = buffer_seconds
self.buffer = collections.deque()
def add_sample(self, data):
"""Add to ring buffer"""
timestamp = time.time()
# Add new sample
self.buffer.append((timestamp, data))
# Remove old samples
cutoff = timestamp - self.buffer_seconds
while self.buffer and self.buffer[0][0] < cutoff:
self.buffer.popleft()
# Always log for real-time
self.source.log("sample", data)
return len(self.buffer)
def save_buffer(self, filename=None):
"""Save buffer contents to file"""
if not filename:
# Microseconds keep two saves in one second apart
timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S_%f")
filename = f"buffer_{timestamp}.trz"
with zelos_sdk.TraceWriter(filename) as writer:
# Log metadata
self.source.log("buffer_saved", {
"samples": len(self.buffer),
"duration": self.buffer_seconds if self.buffer else 0
})
# Replay buffer with original timestamps
for timestamp, data in self.buffer:
timestamp_ns = int(timestamp * 1e9)
self.source.log_at(timestamp_ns, "sample", data)
print(f"Saved {len(self.buffer)} samples to {filename}")
return filename
# Usage
buffer = RingBufferRecorder(buffer_seconds=10)
for i in range(500):
data = {
"index": i,
"value": random.random() * 100,
"status": "OK" if random.random() > 0.1 else "ERROR"
}
buffer.add_sample(data)
# Save buffer on error
if data["status"] == "ERROR":
buffer.save_buffer()
time.sleep(0.1)
Multi-File Recording¶
Split large recordings across multiple files:
import datetime
import os
import random
import time
import zelos_sdk
class MultiFileRecorder:
"""Split recording across files by size or time"""
def __init__(self, prefix="trace", max_mb=100, max_seconds=3600):
zelos_sdk.init()
self.source = zelos_sdk.TraceSource("multi")
# A per-run timestamp keeps file names unique across runs
started = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
self.prefix = f"{prefix}_{started}"
self.max_bytes = max_mb * 1024 * 1024
self.max_seconds = max_seconds
self.file_index = 0
self.writer = None
self.current_file = None
self.file_start_time = None
def _rotate_file(self):
"""Start a new file"""
# Close current
if self.writer:
self.writer.close()
# Open new file
self.file_index += 1
self.current_file = f"{self.prefix}_{self.file_index:04d}.trz"
self.writer = zelos_sdk.TraceWriter(self.current_file)
self.writer.open()
self.file_start_time = time.time()
self.source.log("file_started", {
"filename": self.current_file,
"index": self.file_index
})
print(f"Recording to {self.current_file}")
def _should_rotate(self):
"""Check if rotation needed"""
if not self.current_file:
return True
# Check file size
if os.path.exists(self.current_file):
if os.path.getsize(self.current_file) >= self.max_bytes:
return True
# Check time limit
if time.time() - self.file_start_time >= self.max_seconds:
return True
return False
def log(self, event_name, data):
"""Log with automatic rotation"""
if self._should_rotate():
self._rotate_file()
self.source.log(event_name, data)
def close(self):
"""Close current file"""
if self.writer:
self.writer.close()
print(f"Closed {self.current_file}")
# Usage
recorder = MultiFileRecorder(prefix="long_test", max_mb=50, max_seconds=300)
try:
for i in range(100000):
recorder.log("data", {
"index": i,
"timestamp": time.time(),
"value": random.random()
})
time.sleep(0.01)
finally:
recorder.close()
Experiment Recording¶
Record experiments with metadata:
import datetime
import hashlib
import json
import math
import random
import time
import zelos_sdk
def record_experiment(config, run_function, duration_s=60):
"""
Record an experiment with metadata.
Args:
config: Dictionary of experiment parameters
run_function: Function that generates data
duration_s: Max duration in seconds
Returns:
Filename of recorded data
"""
# Generate unique filename
config_hash = hashlib.md5(
json.dumps(config, sort_keys=True).encode()
).hexdigest()[:8]
timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
filename = f"{config['name']}_{timestamp}_{config_hash}.trz"
zelos_sdk.init()
source = zelos_sdk.TraceSource("experiment")
with zelos_sdk.TraceWriter(filename) as writer:
# Log experiment metadata
source.log("metadata", {
"config": config,
"duration_planned": duration_s,
"start_time": time.time()
})
# Run experiment
start = time.time()
samples = 0
while time.time() - start < duration_s:
try:
data = run_function(time.time() - start)
if data is None: # Function signals completion
break
source.log("measurement", data)
samples += 1
except Exception as e:
source.log("error", {"message": str(e)})
break
# Log summary
actual_duration = time.time() - start
source.log("summary", {
"duration_actual": actual_duration,
"samples": samples,
"rate_hz": samples / actual_duration if actual_duration > 0 else 0
})
print(f"Experiment complete: {filename}")
print(f" Samples: {samples}")
print(f" Duration: {actual_duration:.1f}s")
print(f" Rate: {samples/actual_duration:.1f} Hz")
return filename
# Usage
def sine_wave_generator(elapsed_time):
"""Example data generator"""
if elapsed_time > 30: # Stop after 30 seconds
return None
return {
"time": elapsed_time,
"sine": math.sin(2 * math.pi * elapsed_time),
"cosine": math.cos(2 * math.pi * elapsed_time),
"noise": random.gauss(0, 0.1)
}
config = {
"name": "waveform_test",
"frequency": 1.0,
"amplitude": 1.0,
"noise_level": 0.1
}
filename = record_experiment(config, sine_wave_generator, duration_s=60)
Best Practices¶
1. Always Close Files¶
Use context managers or try/finally. close() writes the last buffered events and finalizes the file.
# Automatic cleanup
with zelos_sdk.TraceWriter(filename) as writer:
... # Your code
# Explicit cleanup
writer = zelos_sdk.TraceWriter(filename)
try:
writer.open()
... # Your code
finally:
writer.close()
2. Use Descriptive Filenames¶
Include timestamp and context. A timestamp also keeps a second run from failing on an existing file.
3. Log Metadata First¶
Always record context at the start:
# Context for analysis
source.log("metadata", {
"version": "1.2.3",
"config": experiment_config,
"operator": username,
"timestamp": time.time()
})
4. Handle Errors Gracefully¶
Don't let recording failures crash your app: