TriggerCollectionMixin

class cait.mixins.TriggerCollectionMixin[source]

Bases: object

A mixin class with convenience functions concerning triggering and event building.

trigger_of(stream: StreamBaseClass, trigger_channels: List[Tuple[str] | str], of: ndarray, thresholds: List[float], passive_channels: List[str] = None, testpulse_channels: List[str] = None, controlpulses_above: List[float | Tuple[float]] = None, calibration_channels: List[str] = None, copy_events: bool = False, reuse_triggers: bool = False, interval: Tuple[float] = None, f_noise: float = 0, **kwargs)[source]

Trigger stream channels from arbitrary hardware using a moving optimum filter trigger and build events from trigger timestamps (of multiple channels) and exclude testpulses if the respective information is provided.

The stream channels specified by trigger_channels are triggered. If some channels are not triggered but read out in coincidence (i.e. as ‘passive’ channels), you can specify their channel names using passive_channels.

If you want to perform a 2D optimum filter trigger, you specify the respective channels (which are filtered together) as tuples in the trigger_channels list (e.g. [('ch0', 'ch1'), 'ch2'] will apply a 2D optimum filter to the first two channels. This will result in A SINGLE filtered trace which is triggered. The last channel is triggered regularly). Note that the of you pass to this function has to have as many channels as total trigger channels. I.e. in the example above, it would need to have 3 channels. Likewise, the testpulse_channels and controlpulses_above lists have to have length 3 here. HOWEVER, you only need 2 (!) trigger threshold (one for the combined channel, one for the single channel).

Events are built as follows: Starting from the first channel’s trigger timestamps, the remaining channels’ triggers are checked to be in coincidence with already existing timestamps. The default coincidence window (if interval=None), is -+dt_us*record_length//4 but can be adapted as needed.

If you provide testpulse_channels, triggers within a record window of a testpulse are treated as testpulses. Note that you have to provide testpulse information for all channels INCLUDING ‘passive’ channels.

Parameters:

  • stream (StreamBaseClass) – The stream object including the channels that you want to trigger.

  • trigger_channels (List[Union[Tuple[str], str]]) – The list of channel names to be triggered. Have to be present in stream.keys. If you pass a tuple of channel names, the 2D optimum filter is applied to this combination. See explanation above.

  • of (np.ndarray) – The optimum filter to use for triggering (Has to have one for each channel in trigger_channels. If either of the trigger channels is a tuple, i.e. treated using the 2D optimum filter, they count as multiple channels. E.g. if you 2D optimum filter the first two channels together and the third channel with the regular optimum filter, of would have to have shape (3, kernel_length)).

  • thresholds (Union[float, List[float]]) – A list of trigger thresholds (in V) for each entry in trigger_channels. If only one is specified, the respective threshold is used for all channels. Note that a tuple in trigger_channels, i.e. a channel combination treated by a 2D optimum filter, needs only one threshold.

  • passive_channels (List[str], optional) – A list of channel names to be read out as ‘passives’. Have to be present in stream.keys. Defaults to None

  • testpulse_channels (List[str], optional) – A list of channel names to be used as testpulses. Have to be present in stream.tp_keys. Defaults to None

  • controlpulses_above (List[Union[float, Tuple[float]]], optional) – If specified, all testpulses with testpulse amplitudes above this value are considered to be controlpulses (i.e. they are saved in their own group in the DataHandler). You have to specify as many values as in ‘testpulse_channels’ (as a list). If you want to enable this feature for only one channel, just set the values for the other channels to some which cannot be exceeded, e.g. 1000. If you want to enforce an upper limit as well (i.e. count testpulses as controlpulses for testpulse amplitudes between a and b), you can do so by passing a tuple (a, b). Defaults to None, i.e. no testpulse is counted as controlpulse.

  • calibration_channels (List[str], optional) – A list of channel names to be used as calibration channels. Have to be present in stream.calp_keys. Defaults to None

  • copy_events (bool, optional) – If True, the voltage traces of the events which were built are saved in the DataHandler (i.e. copied from the stream files). If False, only a reference to the original data is saved. Defaults to False.

  • reuse_triggers (bool, optional) – If true, the triggers from a previous call of this function (which were saved in the DataHandler) are reused and only the event building is performed again (possibly with a different coincidence interval). Defaults to False.

  • interval (Tuple[float], optional) – The coincidence interval for event building in microseconds, i.e. if a trigger lies within the specified interval around a trigger of another channel, they are collected to represent one event. Defaults to -+dt_us*record_length//4.

  • f_noise (float, optional) – The frequency (in events per hour) of empty noise traces to include. Defaults to 0, i.e. no noise is included.

  • kwargs (Any) – Additional keyword arguments forwarded to cait.versatile.trigger_of().

Example:

import cait as ai
import cait.versatile as vai

# Construct stream object
stream = vai.Stream(hardware="vdaq2", src="path/to/stream_file.bin")

print(f"Available channels: {stream.keys}")
print(f"Available TP channels: {stream.tp_keys}")

# Construct DataHandler
dh = ai.DataHandler(record_length=2**13, nmbr_channels=2, sample_frequency=stream.sample_frequency)
dh.set_filepath(path_h5="folder_name/", fname="z-score-triggered", appendix=False)
dh.init_empty()

# Load OF (saved in text file), has one channel
# If you have some other DataHandler with the OF saved, you can use .from_dh instead
of = vai.OF.from_file("path/to/OF_file")

# Trigger ADC1 (phonon channel) and read ADC2 in coincidence (light channel)
dh.trigger_zscore(stream,
                  trigger_channels=["ADC1"],
                  of=of,
                  thresholds=[1e-3],
                  passive_channels=["ADC2"],
                  testpulse_channels=["DAC1", "DAC3"],
                  copy_events=True)
trigger_zscore(stream: StreamBaseClass, trigger_channels: List[str], thresholds: float | List[float] = 5, passive_channels: List[str] = None, testpulse_channels: List[str] = None, controlpulses_above: List[float | Tuple[float]] = None, calibration_channels: List[str] = None, copy_events: bool = False, reuse_triggers: bool = False, interval: Tuple[float] = None, f_noise: float = 0, **kwargs)[source]

Trigger stream channels from arbitrary hardware using a moving z-score trigger and build events from trigger timestamps (of multiple channels) and exclude testpulses if the respective information is provided.

The stream channels specified by trigger_channels are triggered. If some channels are not triggered but read out in coincidence (i.e. as ‘passive’ channels), you can specify their channel names using passive_channels.

Events are built as follows: Starting from the first channel’s trigger timestamps, the remaining channels’ triggers are checked to be in coincidence with already existing timestamps. The default coincidence window (if interval=None), is -+dt_us*record_length//4 but can be adapted as needed.

If you provide testpulse_channels, triggers within a record window of a testpulse are treated as testpulses. Note that you have to provide testpulse information for all channels INCLUDING ‘passive’ channels.

Parameters:

  • stream (StreamBaseClass) – The stream object including the channels that you want to trigger.

  • trigger_channels (List[str]) – The list of channel names to be triggered. Have to be present in stream.keys.

  • thresholds (Union[float, List[float]], optional) – A list of trigger thresholds (in sigmas) for each channel. If only a float is provided, it is used for all channels. Defaults to 5 sigmas

  • passive_channels (List[str], optional) – A list of channel names to be read out as ‘passives’. Have to be present in stream.keys. Defaults to None

  • testpulse_channels (List[str], optional) – A list of channel names to be used as testpulses. Have to be present in stream.tp_keys. Defaults to None

  • controlpulses_above (List[Union[float, Tuple[float]]], optional) – If specified, all testpulses with testpulse amplitudes above this value are considered to be controlpulses (i.e. they are saved in their own group in the DataHandler). You have to specify as many values as in ‘testpulse_channels’ (as a list). If you want to enable this feature for only one channel, just set the values for the other channels to some which cannot be exceeded, e.g. 1000. If you want to enforce an upper limit as well (i.e. count testpulses as controlpulses for testpulse amplitudes between a and b), you can do so by passing a tuple (a, b). Defaults to None, i.e. no testpulse is counted as controlpulse.

  • calibration_channels (List[str], optional) – A list of channel names to be used as calibration channels. Have to be present in stream.calp_keys. Defaults to None

  • copy_events (bool, optional) – If True, the voltage traces of the events which were built are saved in the DataHandler (i.e. copied from the stream files). If False, only a reference to the original data is saved. Defaults to False.

  • reuse_triggers (bool, optional) – If true, the triggers from a previous call of this function (which were saved in the DataHandler) are reused and only the event building is performed again (possibly with a different coincidence interval). Defaults to False.

  • interval (Tuple[float], optional) – The coincidence interval for event building in microseconds, i.e. if a trigger lies within the specified interval around a trigger of another channel, they are collected to represent one event. Defaults to -+dt_us*record_length//4.

  • f_noise (float, optional) – The frequency (in events per hour) of empty noise traces to include. Defaults to 0, i.e. no noise is included.

  • kwargs (Any) – Additional keyword arguments forwarded to cait.versatile.trigger_zscore().

Example:

import cait as ai
import cait.versatile as vai

# Construct stream object
stream = vai.Stream(hardware="vdaq2", src="path/to/stream_file.bin")

print(f"Available channels: {stream.keys}")
print(f"Available TP channels: {stream.tp_keys}")

# Construct DataHandler
dh = ai.DataHandler(record_length=2**13, nmbr_channels=2, sample_frequency=stream.sample_frequency)
dh.set_filepath(path_h5="folder_name/", fname="z-score-triggered", appendix=False)
dh.init_empty()

# Trigger ADC1 (phonon channel) and read ADC2 in coincidence (light channel)
dh.trigger_zscore(stream,
                  trigger_channels=["ADC1"],
                  passive_channels=["ADC2"],
                  testpulse_channels=["DAC1", "DAC3"],
                  copy_events=True)