Triggering stream data
Author: Philipp Schreiner
Created: Aug. 28, 2025
Last updated: Jun. 29, 2026 by Philipp Schreiner
Note
In previous cait versions, different approaches for triggering existed, depending on the hardware/file format that was used to record the stream data (e.g. dh.include_bin_triggers, dh.include_dac_triggers, dh.include_csmpl_triggers, etc.). The current version aims to unify the approach and we STRONGLY encourage you to use the approach detailed below!
This tutorial will probably mark the starting point of many analyses: Triggering a raw stream file to obtain particle pulses, testpulses, etc. to start the analysis. You will probably need this right away, because to build SEVs and OFs, you need voltage traces which can only be obtained after triggering. Once you have built the OF(s), you will most likely do a second round of triggering with this filter and an optimised threshold.
import os
import numpy as np
import cait as ai
import cait.versatile as vai
Hardware overview
Depending on how the stream data was recorded (with which hardware), this first step will vary slighlty. We will create a Stream object which handles the details of the respective hardware/file format internally. Afterwards, the process is exactly the same, independent of the hardware.
Hardware
csmpl:
This format consists of one file per recorded channel (.csmpl-files) and a configuration file (.par-file). Optionally, files containing testpulse information (.test_stamps- and.dig_stamps-files) can also be included (and are needed if you want to separate particle pulses from testpulses). Acsmplstream object is constructed like this:
s = vai.Stream(
hardware="csmpl",
src=[
"par_file.par",
"stream_Ch0.csmpl",
"stream_Ch1.csmpl",
"optional.test_stamps",
"optional.dig_stamps",
]
)
Hardware
vdaq2:
This format consists of a single file (.bin-file) which contains multiple detector channels (ADC-channels) and testpulse channels (DAC-channels). To obtain the testpulse timestamps and testpulse amplitudes (TPAs), the DAC-channels need to be triggered first (happens internally in theStreamobject). Avdaq2stream object is constructed like this:
s = vai.Stream(hardware="vdaq2", src="path/to/file.bin")
Hardware
vdaq3:
This format consists of one file per recorded channel (.bin-files) which contains the testpulse information for the respective channel as well. Avdaq3stream object is constructed like this:
s = vai.Stream(hardware="vdaq3", src=["first_channel.bin", "second_channel.bin"])
New Hardware:
Implementing a new hardware is easier than you think. A ‘hardware’ doesn’t even have to be an actual hardware or file format. It can also just be something that you put together to be treated as if it was a stream object. See New Hardware for details.
Triggering procedure
Since the particular hardware used is not important for the remaining process, we will use mock-up data for this tutorial. For that, a special stream ‘hardware’ vai.MockStream exists:
# One hour of stream data with two channels:
# Here, we set a random seed so that the tutorial looks the same for you.
# If you want to trigger actual data, you want to construct either of the
# stream objects explained above, depending on the hardware used.
stream = vai.MockStream(seed=137, rate_Hz=2)
# You can check which channels are present in the stream ...
print(f"Available channels: {stream.keys}")
# ... which testpulse channels are available ...
print(f"Available TP channels: {stream.tp_keys}")
# ... and which TPAs are available:
print(f"Available TPAs:", {k: np.unique(v) for k, v in stream.tpas.items()})
# Just putting the stream object at the end of a cell gives you
# an information overview. E.g. the timebase 'dt_us' and the
# measuring time in hours 'measuring_time_h'.
stream
Available channels: ['Ch0', 'Ch1']
Available TP channels: ['TP0', 'TP1']
Available TPAs: {'TP0': array([ 1., 2., 3., 4., 100.]), 'TP1': array([ 1., 2., 3., 4., 100.])}
MockStream(start_us=1426321613000000, dt_us=10, length=360000000, keys=['Ch0', 'Ch1'], tp_keys=['TP0', 'TP1'], calp_keys=[], measuring_time_h=1.00)
# You can also check what the stream data looks like.
# Check out the remaining arguments of vai.StreamViewer
# for more interesting features.
vai.StreamViewer(
stream,
mark_timestamps=stream.tp_timestamps,
backend="plotly",
);
# NOTE: the controls don't work on the tutorial page
Before we start triggering, we have to make some decisions:
How long do we want the record window to be? Usually, you would choose this such that a pulse fits nicely into the record window. Using the
vai.StreamViewerabove and thedt_usproperty ofstream, you can get a good idea on what’s reasonable.Which channels do we want to trigger? In most cases, you want to trigger the phonon channel (usually the first one). It is also common to read a second channel (e.g. the light channel) in coincidence (i.e. it is not triggered itself).
Which testpulses are controlpulses? Usually, those are the largest TPAs. In our case, they are 100 (see above).
# Configure trigger
record_length = 2**14
# Basic configuration
trigger_config = {
"trigger_channels": ["Ch0"], # those will be triggered, e.g. phonon channel
"passive_channels": ["Ch1"], # those will be read in coincidence, e.g. light channel
"testpulse_channels": ["TP0", "TP1"], # the testpulse channels corresponding to all trigger/passive channels
"controlpulses_above": [9., 9.], # controlpulses have TPA=10 (see above)
"f_noise": 1000, # we will sample 1000 random noise traces per hour (later needed for NPS creation)
"copy_events": True, # set this to False if you don't want to copy the raw data of the stream to the HDF5 file (to save disk space)
}
Tip
Setting copy_events=False does not physically copy the data. Nevertheless, a reference to the original data is saved (i.e. you can still access the traces, it might just be slower because cait has to find them in their original location). Whether or not you set this parameter to True depends on your situation: If you store the HDF5 file in a location where you have limited disk space, setting it to False might be the way to go. If you have unlimited storage space and want fast computations, set it to True. If you want to have fast computation first and small file size later, use True and when you’re done with the calculations, use dh.drop("events", "event", repackage=True) to delete the copied traces (only the reference to the original data will remain, i.e. the access will still work but will be slower).
We need to save the results of our trigger somewhere. In cait this is done in a DataHandler instance, which we initialize here:
# Path to where we want to save the HDF5 file
fdirh5 = "tutorial_output"
hdf5_name = "my_first_trigger"
os.makedirs(fdirh5, exist_ok=True)
dh = ai.DataHandler(record_length=record_length,
nmbr_channels=len(trigger_config["trigger_channels"]) + len(trigger_config["passive_channels"]),
sample_frequency=stream.sample_frequency)
dh.set_filepath(path_h5=fdirh5,
fname=hdf5_name,
appendix=False)
dh.init_empty()
print(dh)
DataHandler Instance created.
tutorial_output/my_first_trigger.h5 does not exist. Use dh.init_empty() if you mean to initialize a new (empty) file.
DataHandler linked to HDF5 file 'tutorial_output/my_first_trigger.h5'
HDF5 file size on disk: 800.0 B
Groups in file: [].
Now that everything is set up, triggering is quite simple. Note that we use a running z-score trigger here which (by default) triggers on 5-sigma-deviations. Most of the time, this performs decently for the first round of triggering. However, you will later repeat the triggering procedure with an optimum filter, for which this line will change to dh.trigger_of(...).
dh.trigger_zscore(stream, **trigger_config)
# DONE. That's it. Now you can harvest the fruits of your hard work:
# E.g. you can check if the trigger did what you expect by marking the trigger
# timestamps in the StreamViewer:
vai.StreamViewer(
stream,
keys="Ch0",
mark_timestamps={"Events": dh["events/timestamps"], "Testpulses": dh["testpulses/timestamps"]},
backend="plotly",
);
# NOTE: the controls don't work on the tutorial page
# Above, we already accessed some of the data that was stored in the DataHandler using
# the [] notation. To see all the available groups/datasets, you can run
dh.content()
controlpulses
event (2, 143, 16384) float32
hours (143,) float64
testpulseamplitude (2, 143) float32
time_mus (143,) int32
time_s (143,) int32
|timestamps (143,)
event_building-z-score
cp_ts (143,) int64
cpas (2, 143) float32
event_timestamps (5318,) int64
noise_ts (1000,) int64
tp_ts (576,) int64
tpas (2, 576) float32
trigger_flag (2, 5318) bool
trigger_phs (2, 5318) float32
trigger_timestamps (2, 5318) int64
events
event (2, 5318, 16384) float32
hours (5318,) float64
time_mus (5318,) int32
time_s (5318,) int32
|timestamps (5318,)
trigger_flag (2, 5318) bool
trigger_phs (2, 5318) float32
trigger_timestamps (2, 5318) int64
noise
event (2, 1000, 16384) float32
hours (1000,) float64
time_mus (1000,) int32
time_s (1000,) int32
|timestamps (1000,)
testpulses
event (2, 576, 16384) float32
hours (576,) float64
testpulseamplitude (2, 576) float32
time_mus (576,) int32
time_s (576,) int32
|timestamps (576,)
triggers-z-score
ph_Ch0 (6007,) float32
tp_ts_TP0 (719,) int64
tp_ts_TP1 (719,) int64
tpas_TP0 (719,) float32
tpas_TP1 (719,) float32
ts_Ch0 (6007,) int64
# Something that can also be immediatly interesting is looking at the triggered events:
vai.Preview(dh.get_event_iterator("events"), backend="plotly");
# NOTE: the controls don't work on the tutorial page
Main Parameters
One of the first things one usually does after triggering, is calculate some pulse shape parameters, the main parameters. They will allow you to define quality cuts later (see upcoming tutorials).
for group in ["events", "testpulses", "noise", "controlpulses"]:
dh.cmp(group)
# Now you can e.g. look at the pulse heights of the phonon channel
vai.Histogram(dh["events/pulse_height", 0], bins=200, xlabel="Pulse Height (V)", backend="plotly");
# NOTE: the controls don't work on the tutorial page
Advanced
Using cait.versatile functions directly
Almost always, the triggering functions implemented on the DataHandler will cover your needs. If you need more flexibility, however, you can use the underlying cait.versatile triggering functions as well. We illustrate it with the stream object that we constructed above and the vai.trigger_zscore function (works the same for vai.trigger_of).
# Perform triggering (use context to keep stream file opened)
with stream:
trigger_inds, amplitudes = vai.trigger_zscore(
stream["Ch0"],
record_length=record_length,
# Here, especially the 'apply_first' argument might be useful.
# Have a look at the documentation of vai.trigger_zscore
)
# Get trigger timestamps from trigger indices
# by slicing the stream's time object with the indices
timestamps = stream.time[trigger_inds]
# Have a look at the amplitudes
vai.Histogram(amplitudes, xlabel="Z-score trigger amplitudes (V)", bins=200, backend="plotly");
The code above only performed the triggering. No event building was done. In particular, no identifications of the testpulses happend, which is why the testpulses show up as peaks in the spectrum above. To tag them, we can use vai.timestamp_coincidence and the testpulse timestamps that the stream object provides:
tp_timestamps = stream.tp_timestamps["TP0"]
# Take the testpulse timestamps and check which trigger timestamps fall within
# a quarter record window before or after that timestamp.
in_coinc, _, not_in_coinc = vai.timestamp_coincidence(
tp_timestamps,
timestamps,
(-stream.dt_us * record_length//4, stream.dt_us * record_length//4),
)
# Plot separate spectra
vai.Histogram(
{
"Particle pulses": amplitudes[not_in_coinc],
"Testpulses": amplitudes[in_coinc],
},
bins=(0, 0.2, 400),
xlabel="Z-score trigger amplitudes (V)",
backend="plotly",
);
Nice! We have separated the testpulses. If you want the corresponding event traces, you can request event iterators from the stream:
tp_timestamps = timestamps[in_coinc]
event_timstamps = timestamps[not_in_coinc]
events_it = stream.get_event_iterator("Ch0", record_length, timestamps=event_timstamps)
tp_it = stream.get_event_iterator("Ch0", record_length, timestamps=tp_timestamps)
New hardware
Above, we showed you several ‘stream sources’ like the vai.MockStream and vai.Stream which implements different file formats (.csmpl, .bin). For illustration purposes, we show you here how a new stream source could be implemented. For that, we choose a very simple example: We want a stream object that just returns the inverted stream.
Note
This is not necessarily a useful implementation and serves as an illustration only. E.g. if you just want to trigger on the inverted stream (which can sometimes be insightful), you can set apply_first=lambda x: -x in the triggering functions directly.
The two things that you have to care about when implementing a new stream object are the __init__ function where you load the actual data, e.g. using BinaryFile, and the get_trace method, which defines how the stream samples for a given stream key (channel) and a given location (slice) are returned. Everything else is handled by the StreamBaseClass.
from cait.versatile.datasources.stream.streambase import StreamBaseClass
class InvertedStream(StreamBaseClass):
"""
Stream object that returns the voltage inverted stream object that it is constructed from.
:param stream: The stream which we want to invert (all channels).
:type stream: StreamBaseClass
"""
def __init__(self, stream: StreamBaseClass):
# Construct super class with all inputs to this class
super().__init__(stream=stream)
# ... space for input validation
self._stream = stream
def __len__(self):
# len(stream) is forwarded to the underlying stream object.
# In a more real-life scenario this would probably be something
# like the file size.
return len(self._stream)
def __enter__(self):
# Forwarded to underlying stream object.
# Usually this handles how a file is opened on disk.
self._stream.__enter__()
return self
def __exit__(self, typ, val, tb):
# Forwarded to underlying stream object.
# Usually this handles how a file is closed on disk.
self._stream.__exit__(typ, val, tb)
def get_trace(self, key: str, where: slice, voltage: bool = True):
# THIS IS THE IMPORTANT PART!
# Here, we again just return the samples from the underlying stream
# object but with the difference that we INVERT them.
# In a real-life scenario, this would be something where you slice
# numpy arrays or similar.
# The 'voltage' argument decides whether the raw ADC values are returned
# or if they are converted to voltage values.
return -self._stream.get_trace(key, where, voltage=voltage)
@property
def start_us(self):
# The microsecond timestamp of the the beginning of the stream.
return self._stream.start_us
@property
def dt_us(self):
# The microsecond timebase (sample length) of the stream.
return self._stream.dt_us
@property
def keys(self):
# The available keys (channels) in the stream.
return self._stream.keys
@property
def tp_keys(self):
# The available testpulse keys (channels) in the stream.
return self._stream.tp_keys
@property
def tpas(self):
# The testpulse amplitudes in the stream (if available).
return self._stream.tpas
@property
def tp_timestamps(self):
# The testpulse timestamps in the stream (if available).
return self._stream.tp_timestamps
@property
def calp_keys(self):
# The available calibration keys (channels) in the stream.
return self._stream.calp_keys
@property
def calpas(self):
# The calibration amplitudes in the stream (if available).
return self._stream.calpas
@property
def calp_timestamps(self):
# The calibration timestamps in the stream (if available).
return self._stream.calp_timestamps
# Now, we can just use the InvertedStream like any other stream object.
# E.g., we can give it to the StreamViewer:
vai.StreamViewer(InvertedStream(stream), backend="plotly");
For more examples of how such a stream class could be implemented have a look at the existing implementations for Stream_VDAQ2, Stream_VDAQ3, and Stream_CSMPL, all of which are collected in the Stream class for convenience.