Skip to content

frib_event Module

Defines the event structure from the FRIBDAQ data.

FribEvent

Represents an event from the FRIBDAQ system.

Contains traces from the SIS3300 module, typically encompassing the ion chamber, silicon detector, and mesh signals.

Parameters:

Name Type Description Default
trace_array ndarray

The hdf5 dataset containing the FRIBDAQ traces

 required
coincidence_array ndarray

The hdf5 dataset containing the FRIBDAQ coincidence bits

 required
event_number int

The event number

 required
params FribParameters

The configuration parameters for FRIBDAQ signal analysis

 required

Attributes:

Name Type Description
event_number int

The event number
trace_data list[FribTrace]

The traces in the event

Methods:

Name Description
FribEvent

Construct the FribEvent and process the traces
get_ic_trace

Get the ion chamber trace for this event
get_si_trace

Get the silicon trace for this event
get_mesh_trace

Get the mesh trace for this event
get_good_ic_peak

Attempts to retrieve the "good" ion chamber signal.
correct_ic_time

Calculate the correction to the GET time buckets using the ion chamber signal
Source code in src/spyral/trace/frib_event.py 
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
class FribEvent:
    """Represents an event from the FRIBDAQ system.

    Contains traces from the SIS3300 module, typically encompassing the ion chamber,
    silicon detector, and mesh signals.

    Parameters
    ----------
    trace_array: h5py.Dataset
        The hdf5 dataset containing the FRIBDAQ traces
    coincidence_array: h5py.Dataset
        The hdf5 dataset containing the FRIBDAQ coincidence bits
    event_number: int
        The event number
    params: FribParameters
        The configuration parameters for FRIBDAQ signal analysis

    Attributes
    ----------
    event_number: int
        The event number
    trace_data: list[FribTrace]
        The traces in the event

    Methods
    -------
    FribEvent(self, trace_array, coincidence_array, event_number, params)
        Construct the FribEvent and process the traces
    get_ic_trace() -> FribTrace
        Get the ion chamber trace for this event
    get_si_trace() -> FribTrace
        Get the silicon trace for this event
    get_mesh_trace() -> FribTrace
        Get the mesh trace for this event
    get_good_ic_peak(params) -> Peak | None
        Attempts to retrieve the "good" ion chamber signal.
    correct_ic_time(good_peak, get_frequency) -> float
        Calculate the correction to the GET time buckets using the ion chamber signal
    """

    def __init__(
        self,
        trace_array: np.ndarray,
        coincidence_array: np.ndarray,
        event_number: int,
        params: FribParameters,
    ):
        self.event_number = event_number
        trace_data = preprocess_frib_traces(
            trace_array[:].copy(), params.baseline_window_scale
        )
        self.trigger = TriggerType(int(coincidence_array[0]))  # bleh
        self.traces = [FribTrace(column, params) for column in trace_data.T]

    def get_ic_trace(self) -> FribTrace:
        """Get the ion chamber trace for this event

        Returns
        -------
        FribTrace
            The ion chamber trace
        """
        return self.traces[IC_COLUMN]

    def get_si_trace(self) -> FribTrace:
        """Get the silicon trace for this event

        Returns
        -------
        FribTrace
            The silicon trace
        """
        return self.traces[SI_COLUMN]

    def get_mesh_trace(self) -> FribTrace:
        """Get the mesh trace for this event

        Returns
        -------
        FribTrace
            The mesh trace
        """
        return self.traces[MESH_COLUMN]

    def get_triggering_ic_peak(self, params: FribParameters) -> Peak | None:
        """Attempts to retrieve the triggering ion chamber peak

        The ion chamber signal is delayed relative to the mesh signal. This means that the signal
        recorded in the FRIBDAQ contains times (time buckets) from before the event window is actually open.
        As such, we cannot rely on the first peak of the ion chamber being the triggering peak. To correct for
        this we have a parameter `ic_delay_time_bucket` which essentially sets a threshold on peak centroid.
        All peaks before this time bucket are ignored, and the first peak after the time bucket is the triggering
        peak.

        Parameters
        ----------
        params: FribParameters
            Configuration parameters taht control the algorithm

        Returns
        -------
        Peak | None
            Returns the triggering peak or None if there is no triggering peak found (not good)
        """

        ic_peaks = self.get_ic_trace().get_peaks()
        if len(ic_peaks) == 0:
            return None

        # Sort the peaks in ascending order, so that the first one we find
        # greater than our delay is the trigger.
        ic_sorted_peaks = sorted(ic_peaks, key=lambda x: x.centroid)
        for peak in ic_sorted_peaks:
            if peak.centroid > params.ic_delay_time_bucket:
                return peak
        return None

    def get_ic_multiplicity(self, params: FribParameters) -> int:
        """Calculates the IC multiplicity, accounting for the ion chamber delay

        The ion chamber signal is delayed relative to the mesh signal. This means that the signal
        recorded in the FRIBDAQ contains times (time buckets) from before the event window is actually open.
        As such, we cannot rely on the first peak of the ion chamber being the triggering peak. To correct for
        this we have a parameter `ic_delay_time_bucket` which essentially sets a threshold on peak centroid.
        All peaks before this time bucket are ignored, and the first peak after the time bucket is the triggering
        peak. This method counts the number of peaks after the delay.

        Parameters
        ----------
        params: FribParameters
            Configuration parameters taht control the algorithm

        Returns
        -------
        int
            The number of IC peaks after the delay
        """

        ic_peaks = self.get_ic_trace().get_peaks()
        if len(ic_peaks) == 0:
            return 0

        count = 0
        for peak in ic_peaks:
            if peak.centroid > params.ic_delay_time_bucket:
                count += 1
        return count

    def get_good_ic_peak(self, params: FribParameters) -> tuple[int, Peak] | None:
        """Attempts to retrieve the "good" ion chamber signal.

        There are several issues with determining the "good" ion chamber signal.

        First, we must identify the triggering peak, which is handled through the `get_triggering_ic_peak` method.

        Then, in many cases, the ion chamber has multiple peaks due to bunches of beam being delivered to the AT-TPC and
        the AT-TPC's long event window. This poses an issue for the analysis; we must now deterimine which ion chamber signal to use,
        if any, for this event. The algorithm below uses the silicon signal to veto ion chamber signals which correspond
        to un-reacted beam particles and effectively reduce the multiplicity. The user then has control over how many ion chamber
        signals are allowed for a good event. In general, AT-TPC would only accept multiplicty one type events.

        Parameters
        ----------
        params: FribParameters
            Configuration parameters that control the algorithm

        Returns
        -------
        tuple[int, Peak] | None
            Returns None on failure to find any good ion chamber signal; otherwise returns the good ion chamber multiplicity and the first Peak associated with a good signal
        """

        trigger = self.get_triggering_ic_peak(params)
        # No trigger, exit
        if trigger is None:
            return None
        ic_peaks = self.get_ic_trace().get_peaks()
        si_peaks = self.get_si_trace().get_peaks()

        if len(ic_peaks) == 0:
            return None
        elif len(si_peaks) == 0:
            if len(ic_peaks) == 1:
                return (1, ic_peaks[0])
            else:
                return None

        good_ic_count = 0
        good_ic_index = -1
        si_coinc = False
        for idx, ic in enumerate(ic_peaks):
            # Ignore peaks before the trigger
            if ic.centroid < trigger.centroid:
                continue
            si_coinc = False
            for si in si_peaks:
                if abs(ic.centroid - si.centroid) < 50.0:
                    si_coinc = True
                    break
            if not si_coinc:
                good_ic_count += 1
                good_ic_index = idx

        if good_ic_count > params.ic_multiplicity or good_ic_count == 0:
            return None
        else:
            return (good_ic_count, ic_peaks[good_ic_index])

    def correct_ic_time(
        self, good_peak: Peak, params: FribParameters, get_frequency: float
    ) -> float:
        """Calculate the correction to the GET time buckets using the ion chamber signal

        In a case where the ion chamber fired multiple times, somtimes the wrong
        beam particle starts the event. That is: we trigger on an unreacted beam particle,
        but still capture a good event within the trigger window. After the good ion
        chamber peak is identified using the get_good_ic_peak method, a correction
        for this time walk effect can be calculated by taking the difference between the
        centroids in the good peak and the earliest peak in the ion chamber trace.
        The difference can then be converted to GET time buckets by multiplying by the
        ratio of the sampling frequency of the GET system and the SIS3300 module.

        Parameters
        ----------
        good_peak: Peak
            The good peak in the ion chamber trace, typically found using the get_good_ic_peak method
        params: FribParameters
            Configuration parameters that control the algorithm
        get_frequency: float
            The sampling frequency of the GET electronics

        Returns
        -------
        float
            The correction to the GET time in units of GET time buckets.
        """

        trigger = self.get_triggering_ic_peak(params)
        if trigger is None:
            return 0.0
        return (
            (good_peak.centroid - trigger.centroid) * get_frequency / SAMPLING_FREQUENCY
        )

correct_ic_time(good_peak, params, get_frequency)

Calculate the correction to the GET time buckets using the ion chamber signal

In a case where the ion chamber fired multiple times, somtimes the wrong beam particle starts the event. That is: we trigger on an unreacted beam particle, but still capture a good event within the trigger window. After the good ion chamber peak is identified using the get_good_ic_peak method, a correction for this time walk effect can be calculated by taking the difference between the centroids in the good peak and the earliest peak in the ion chamber trace. The difference can then be converted to GET time buckets by multiplying by the ratio of the sampling frequency of the GET system and the SIS3300 module.

Parameters:

Name Type Description Default
good_peak Peak

The good peak in the ion chamber trace, typically found using the get_good_ic_peak method

 required
params FribParameters

Configuration parameters that control the algorithm

 required
get_frequency float

The sampling frequency of the GET electronics

 required

Returns:

Type Description
float

The correction to the GET time in units of GET time buckets.
Source code in src/spyral/trace/frib_event.py 
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
def correct_ic_time(
    self, good_peak: Peak, params: FribParameters, get_frequency: float
) -> float:
    """Calculate the correction to the GET time buckets using the ion chamber signal

    In a case where the ion chamber fired multiple times, somtimes the wrong
    beam particle starts the event. That is: we trigger on an unreacted beam particle,
    but still capture a good event within the trigger window. After the good ion
    chamber peak is identified using the get_good_ic_peak method, a correction
    for this time walk effect can be calculated by taking the difference between the
    centroids in the good peak and the earliest peak in the ion chamber trace.
    The difference can then be converted to GET time buckets by multiplying by the
    ratio of the sampling frequency of the GET system and the SIS3300 module.

    Parameters
    ----------
    good_peak: Peak
        The good peak in the ion chamber trace, typically found using the get_good_ic_peak method
    params: FribParameters
        Configuration parameters that control the algorithm
    get_frequency: float
        The sampling frequency of the GET electronics

    Returns
    -------
    float
        The correction to the GET time in units of GET time buckets.
    """

    trigger = self.get_triggering_ic_peak(params)
    if trigger is None:
        return 0.0
    return (
        (good_peak.centroid - trigger.centroid) * get_frequency / SAMPLING_FREQUENCY
    )

get_good_ic_peak(params)

Attempts to retrieve the "good" ion chamber signal.

There are several issues with determining the "good" ion chamber signal.

First, we must identify the triggering peak, which is handled through the get_triggering_ic_peak method.

Then, in many cases, the ion chamber has multiple peaks due to bunches of beam being delivered to the AT-TPC and the AT-TPC's long event window. This poses an issue for the analysis; we must now deterimine which ion chamber signal to use, if any, for this event. The algorithm below uses the silicon signal to veto ion chamber signals which correspond to un-reacted beam particles and effectively reduce the multiplicity. The user then has control over how many ion chamber signals are allowed for a good event. In general, AT-TPC would only accept multiplicty one type events.

Parameters:

Name Type Description Default
params FribParameters

Configuration parameters that control the algorithm

 required

Returns:

Type Description
tuple[int, Peak] | None

Returns None on failure to find any good ion chamber signal; otherwise returns the good ion chamber multiplicity and the first Peak associated with a good signal
Source code in src/spyral/trace/frib_event.py 
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
def get_good_ic_peak(self, params: FribParameters) -> tuple[int, Peak] | None:
    """Attempts to retrieve the "good" ion chamber signal.

    There are several issues with determining the "good" ion chamber signal.

    First, we must identify the triggering peak, which is handled through the `get_triggering_ic_peak` method.

    Then, in many cases, the ion chamber has multiple peaks due to bunches of beam being delivered to the AT-TPC and
    the AT-TPC's long event window. This poses an issue for the analysis; we must now deterimine which ion chamber signal to use,
    if any, for this event. The algorithm below uses the silicon signal to veto ion chamber signals which correspond
    to un-reacted beam particles and effectively reduce the multiplicity. The user then has control over how many ion chamber
    signals are allowed for a good event. In general, AT-TPC would only accept multiplicty one type events.

    Parameters
    ----------
    params: FribParameters
        Configuration parameters that control the algorithm

    Returns
    -------
    tuple[int, Peak] | None
        Returns None on failure to find any good ion chamber signal; otherwise returns the good ion chamber multiplicity and the first Peak associated with a good signal
    """

    trigger = self.get_triggering_ic_peak(params)
    # No trigger, exit
    if trigger is None:
        return None
    ic_peaks = self.get_ic_trace().get_peaks()
    si_peaks = self.get_si_trace().get_peaks()

    if len(ic_peaks) == 0:
        return None
    elif len(si_peaks) == 0:
        if len(ic_peaks) == 1:
            return (1, ic_peaks[0])
        else:
            return None

    good_ic_count = 0
    good_ic_index = -1
    si_coinc = False
    for idx, ic in enumerate(ic_peaks):
        # Ignore peaks before the trigger
        if ic.centroid < trigger.centroid:
            continue
        si_coinc = False
        for si in si_peaks:
            if abs(ic.centroid - si.centroid) < 50.0:
                si_coinc = True
                break
        if not si_coinc:
            good_ic_count += 1
            good_ic_index = idx

    if good_ic_count > params.ic_multiplicity or good_ic_count == 0:
        return None
    else:
        return (good_ic_count, ic_peaks[good_ic_index])

get_ic_multiplicity(params)

Calculates the IC multiplicity, accounting for the ion chamber delay

The ion chamber signal is delayed relative to the mesh signal. This means that the signal recorded in the FRIBDAQ contains times (time buckets) from before the event window is actually open. As such, we cannot rely on the first peak of the ion chamber being the triggering peak. To correct for this we have a parameter ic_delay_time_bucket which essentially sets a threshold on peak centroid. All peaks before this time bucket are ignored, and the first peak after the time bucket is the triggering peak. This method counts the number of peaks after the delay.

Parameters:

Name Type Description Default
params FribParameters

Configuration parameters taht control the algorithm

 required

Returns:

Type Description
int

The number of IC peaks after the delay
Source code in src/spyral/trace/frib_event.py 
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
def get_ic_multiplicity(self, params: FribParameters) -> int:
    """Calculates the IC multiplicity, accounting for the ion chamber delay

    The ion chamber signal is delayed relative to the mesh signal. This means that the signal
    recorded in the FRIBDAQ contains times (time buckets) from before the event window is actually open.
    As such, we cannot rely on the first peak of the ion chamber being the triggering peak. To correct for
    this we have a parameter `ic_delay_time_bucket` which essentially sets a threshold on peak centroid.
    All peaks before this time bucket are ignored, and the first peak after the time bucket is the triggering
    peak. This method counts the number of peaks after the delay.

    Parameters
    ----------
    params: FribParameters
        Configuration parameters taht control the algorithm

    Returns
    -------
    int
        The number of IC peaks after the delay
    """

    ic_peaks = self.get_ic_trace().get_peaks()
    if len(ic_peaks) == 0:
        return 0

    count = 0
    for peak in ic_peaks:
        if peak.centroid > params.ic_delay_time_bucket:
            count += 1
    return count

get_ic_trace()

Get the ion chamber trace for this event

Returns:

Type Description
FribTrace

The ion chamber trace
Source code in src/spyral/trace/frib_event.py 
77
78
79
80
81
82
83
84
85
def get_ic_trace(self) -> FribTrace:
    """Get the ion chamber trace for this event

    Returns
    -------
    FribTrace
        The ion chamber trace
    """
    return self.traces[IC_COLUMN]

get_mesh_trace()

Get the mesh trace for this event

Returns:

Type Description
FribTrace

The mesh trace
Source code in src/spyral/trace/frib_event.py 
 97
 98
 99
100
101
102
103
104
105
def get_mesh_trace(self) -> FribTrace:
    """Get the mesh trace for this event

    Returns
    -------
    FribTrace
        The mesh trace
    """
    return self.traces[MESH_COLUMN]

get_si_trace()

Get the silicon trace for this event

Returns:

Type Description
FribTrace

The silicon trace
Source code in src/spyral/trace/frib_event.py 
87
88
89
90
91
92
93
94
95
def get_si_trace(self) -> FribTrace:
    """Get the silicon trace for this event

    Returns
    -------
    FribTrace
        The silicon trace
    """
    return self.traces[SI_COLUMN]

get_triggering_ic_peak(params)

Attempts to retrieve the triggering ion chamber peak

The ion chamber signal is delayed relative to the mesh signal. This means that the signal recorded in the FRIBDAQ contains times (time buckets) from before the event window is actually open. As such, we cannot rely on the first peak of the ion chamber being the triggering peak. To correct for this we have a parameter ic_delay_time_bucket which essentially sets a threshold on peak centroid. All peaks before this time bucket are ignored, and the first peak after the time bucket is the triggering peak.

Parameters:

Name Type Description Default
params FribParameters

Configuration parameters taht control the algorithm

 required

Returns:

Type Description
Peak | None

Returns the triggering peak or None if there is no triggering peak found (not good)
Source code in src/spyral/trace/frib_event.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
def get_triggering_ic_peak(self, params: FribParameters) -> Peak | None:
    """Attempts to retrieve the triggering ion chamber peak

    The ion chamber signal is delayed relative to the mesh signal. This means that the signal
    recorded in the FRIBDAQ contains times (time buckets) from before the event window is actually open.
    As such, we cannot rely on the first peak of the ion chamber being the triggering peak. To correct for
    this we have a parameter `ic_delay_time_bucket` which essentially sets a threshold on peak centroid.
    All peaks before this time bucket are ignored, and the first peak after the time bucket is the triggering
    peak.

    Parameters
    ----------
    params: FribParameters
        Configuration parameters taht control the algorithm

    Returns
    -------
    Peak | None
        Returns the triggering peak or None if there is no triggering peak found (not good)
    """

    ic_peaks = self.get_ic_trace().get_peaks()
    if len(ic_peaks) == 0:
        return None

    # Sort the peaks in ascending order, so that the first one we find
    # greater than our delay is the trigger.
    ic_sorted_peaks = sorted(ic_peaks, key=lambda x: x.centroid)
    for peak in ic_sorted_peaks:
        if peak.centroid > params.ic_delay_time_bucket:
            return peak
    return None

preprocess_frib_traces(traces, baseline_window_scale)

JIT-ed Method for pre-cleaning the trace data in bulk before doing trace analysis

These methods are more suited to operating on the entire dataset rather than on a trace by trace basis It includes

  • Removal of edge effects in traces (first and last time buckets can be noisy)
  • Baseline removal via fourier transform method (see J. Bradt thesis, pytpc library)

Parameters:

Name Type Description Default
traces ndarray

A (2048, n) matrix where n is the number of traces and each column corresponds to a trace. This should be a copied array, not a reference to an array in an hdf5 file

 required
baseline_window_scale float

The scale of the baseline filter used to perform a moving average over the basline

 required

Returns:

Type Description
ndarray

A new (2048, n) matrix which contains the traces with their baselines removed and edges smoothed
Source code in src/spyral/trace/frib_event.py 
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
@njit
def preprocess_frib_traces(
    traces: np.ndarray, baseline_window_scale: float
) -> np.ndarray:
    """JIT-ed Method for pre-cleaning the trace data in bulk before doing trace analysis

    These methods are more suited to operating on the entire dataset rather than on a trace by trace basis
    It includes

    - Removal of edge effects in traces (first and last time buckets can be noisy)
    - Baseline removal via fourier transform method (see J. Bradt thesis, pytpc library)

    Parameters
    ----------
    traces: ndarray
        A (2048, n) matrix where n is the number of traces and each column corresponds to a trace. This should be a copied
        array, not a reference to an array in an hdf5 file
    baseline_window_scale: float
        The scale of the baseline filter used to perform a moving average over the basline

    Returns
    -------
    ndarray
        A new (2048, n) matrix which contains the traces with their baselines removed and edges smoothed
    """
    # transpose cause its easier to process
    traces = traces.T

    # Smooth out the edges of the traces
    traces[:, 0] = traces[:, 1]
    traces[:, -1] = traces[:, -2]

    # Remove peaks from baselines and replace with average
    bases: np.ndarray = traces.copy()
    for row in bases:
        mean = np.mean(row)
        sigma = np.std(row)
        mask = row - mean > sigma * 1.5
        row[mask] = np.mean(row[~mask])

    # Create the filter
    window = np.arange(-1024.0, 1024.0, 1.0)
    fil = np.fft.ifftshift(np.sinc(window / baseline_window_scale))
    transformed = np.fft.fft2(bases, axes=(1,))
    result = np.real(
        np.fft.ifft2(transformed * fil, axes=(1,))
    )  # Apply the filter -> multiply in Fourier = convolve in normal

    # Make sure to transpose back into og format
    return (traces - result).T