Triggering stream data

Author: Philipp Schreiner
Created: Aug. 28, 2025
Last updated: Jan. 21, 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!

Warning

This tutorial is still work in progress!

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). A csmpl stream 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 the Stream object). A vdaq2 stream 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. A vdaq3 stream 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'], 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.StreamViewer above and the dt_us property of stream, 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)
}

Note

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)

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

Triggering stream data

Hardware overview

Triggering procedure

Main Parameters

Advanced triggering (individualize workflow, new hardware, etc.)

Using `cait.versatile` functions directly

New hardware

Triggering stream data

Hardware overview

Triggering procedure

Main Parameters

Advanced triggering (individualize workflow, new hardware, etc.)

Using cait.versatile functions directly

New hardware

Using `cait.versatile` functions directly