BasicPulseExtractor.gated_conv_deriv() - Code Metrics - Inspection of "Pulsed 3.0" - Ulm-IQO/qudi - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#384)

unknown

created 2018-06-06 09:41 UTC

BasicPulseExtractor.gated_conv_deriv() B

↳ Parent: BasicPulseExtractor

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	2
Bugs	0	Features	0

Metric	Value
cc	4
c	2
b	0
f	0
dl	0
loc	49
rs	8.7972

# -*- coding: utf-8 -*-
"""
This file contains basic pulse extraction methods for Qudi.

Qudi is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.

Qudi is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
GNU General Public License for more details.

You should have received a copy of the GNU General Public License
along with Qudi. If not, see <http://www.gnu.org/licenses/>.

Copyright (c) the Qudi Developers. See the COPYRIGHT.txt file at the
top-level directory of this distribution and at <https://github.com/Ulm-IQO/qudi/>
"""

import numpy as np
from scipy import ndimage

from logic.pulsed.pulse_extractor import PulseExtractorBase


class BasicPulseExtractor(PulseExtractorBase):
    """

    """
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

    def gated_conv_deriv(self, count_data, conv_std_dev=20.0):
        """
        Detects the rising flank in the gated timetrace data and extracts just the laser pulses.
        The flank detection is based on an image edge detection technique performed in 1D.
        There is no information about the data needed.
        Only the gaussian filter width to reduce shot noise can be given as parameter.

        @param 2D numpy.ndarray count_data: the raw timetrace data from a gated fast counter
                                            dim 0: gate number; dim 1: time bin
        @param float conv_std_dev: The standard deviation of the gaussian filter used for smoothing

        @return dict: The extracted laser pulses of the timetrace as well as the indices for rising
                      and falling flanks.
        """
        # Create return dictionary
        return_dict = {'laser_counts_arr': np.zeros(0, dtype='int64'),
                       'laser_indices_rising': -1,
                       'laser_indices_falling': -1}

        # sum up all gated timetraces to ease flank detection
        timetrace_sum = np.sum(count_data, 0)

        # apply gaussian filter to remove noise and compute the gradient of the timetrace sum
        try:
            conv = ndimage.filters.gaussian_filter1d(timetrace_sum.astype(float), conv_std_dev)
        except:
            conv = np.zeros(timetrace_sum.size)
        try:
            conv_deriv = np.gradient(conv)
        except:
            conv_deriv = np.zeros(conv.size)

        # get indices of rising and falling flank
        rising_ind = conv_deriv.argmax()
        falling_ind = conv_deriv.argmin()

        # If gaussian smoothing or derivative failed, the returned array only contains zeros.
        # Check for that and return also only zeros to indicate a failed pulse extraction.
        if len(conv_deriv.nonzero()[0]) == 0:
            laser_arr = np.zeros(count_data.shape, dtype='int64')
        else:
            # slice the data array to cut off anything but laser pulses
            laser_arr = count_data[:, rising_ind:falling_ind]

        return_dict['laser_counts_arr'] = laser_arr.astype('int64')
        return_dict['laser_indices_rising'] = rising_ind
        return_dict['laser_indices_falling'] = falling_ind

        return return_dict

    def ungated_conv_deriv(self, count_data, conv_std_dev=20.0):
        """ Detects the laser pulses in the ungated timetrace data and extracts
            them.

        @param numpy.ndarray count_data: 1D array the raw timetrace data from an ungated fast counter
        @param dict measurement_settings: The measurement settings of the currently running measurement.
        @param float conv_std_dev: The standard deviation of the gaussian used for smoothing

        @return 2D numpy.ndarray:   2D array, the extracted laser pulses of the timetrace.
                                    dimensions: 0: laser number, 1: time bin

        Procedure:
            Edge Detection:
            ---------------

            The count_data array with the laser pulses is smoothed with a
            gaussian filter (convolution), which used a defined standard
            deviation of 10 entries (bins). Then the derivation of the convolved
            time trace is taken to obtain the maxima and minima, which
            corresponds to the rising and falling edge of the pulses.

            The convolution with a gaussian removes nasty peaks due to count
            fluctuation within a laser pulse and at the same time ensures a
            clear distinction of the maxima and minima in the derived convolved
            trace.

            The maxima and minima are not found sequentially, pulse by pulse,
            but are rather globally obtained. I.e. the convolved and derived
            array is searched iteratively for a maximum and a minimum, and after
            finding those the array entries within the 4 times
            self.conv_std_dev (2*self.conv_std_dev to the left and
            2*self.conv_std_dev) are set to zero.

            The crucial part is the knowledge of the number of laser pulses and
            the choice of the appropriate std_dev for the gauss filter.

            To ensure a good performance of the edge detection, you have to
            ensure a steep rising and falling edge of the laser pulse! Be also
            careful in choosing a large conv_std_dev value and using a small
            laser pulse (rule of thumb: conv_std_dev < laser_length/10).
        """
        # Create return dictionary
        return_dict = {'laser_counts_arr': np.empty(0, dtype='int64'),
                       'laser_indices_rising': np.empty(0, dtype='int64'),
                       'laser_indices_falling': np.empty(0, dtype='int64')}

        number_of_lasers = self.measurement_settings.get('number_of_lasers')
        if not isinstance(number_of_lasers, int):
            return return_dict

        # apply gaussian filter to remove noise and compute the gradient of the timetrace sum
        try:
            conv = ndimage.filters.gaussian_filter1d(count_data.astype(float), conv_std_dev)
        except:
            conv = np.zeros(count_data.size)
        try:
            conv_deriv = np.gradient(conv)
        except:
            conv_deriv = np.zeros(conv.size)

        # if gaussian smoothing or derivative failed, the returned array only contains zeros.
        # Check for that and return also only zeros to indicate a failed pulse extraction.
        if len(conv_deriv.nonzero()[0]) == 0:
            return_dict['laser_counts_arr'] = np.zeros((number_of_lasers, 10), dtype='int64')
            return return_dict

        # use a reference for array, because the exact position of the peaks or dips
        # (i.e. maxima or minima, which are the inflection points in the pulse) are distorted by
        # a large conv_std_dev value.
        try:
            conv = ndimage.filters.gaussian_filter1d(count_data.astype(float), 10)
        except:
            conv = np.zeros(count_data.size)
        try:
            conv_deriv_ref = np.gradient(conv)
        except:
            conv_deriv_ref = np.zeros(conv.size)

        # initialize arrays to contain indices for all rising and falling
        # flanks, respectively
        rising_ind = np.empty(number_of_lasers, dtype='int64')
        falling_ind = np.empty(number_of_lasers, dtype='int64')

        # Find as many rising and falling flanks as there are laser pulses in
        # the trace:
        for i in range(number_of_lasers):
            # save the index of the absolute maximum of the derived time trace
            # as rising edge position
            rising_ind[i] = np.argmax(conv_deriv)

            # refine the rising edge detection, by using a small and fixed
            # conv_std_dev parameter to find the inflection point more precise
            start_ind = int(rising_ind[i] - conv_std_dev)
            if start_ind < 0:
                start_ind = 0

            stop_ind = int(rising_ind[i] + conv_std_dev)
            if stop_ind > len(conv_deriv):
                stop_ind = len(conv_deriv)

            if start_ind == stop_ind:
                stop_ind = start_ind + 1

            rising_ind[i] = start_ind + np.argmax(conv_deriv_ref[start_ind:stop_ind])

            # set this position and the surrounding of the saved edge to 0 to
            # avoid a second detection
            if rising_ind[i] < 2 * conv_std_dev:
                del_ind_start = 0
            else:
                del_ind_start = rising_ind[i] - int(2 * conv_std_dev)
            if (conv_deriv.size - rising_ind[i]) < 2 * conv_std_dev:
                del_ind_stop = conv_deriv.size - 1
            else:
                del_ind_stop = rising_ind[i] + int(2 * conv_std_dev)
                conv_deriv[del_ind_start:del_ind_stop] = 0

            # save the index of the absolute minimum of the derived time trace
            # as falling edge position
            falling_ind[i] = np.argmin(conv_deriv)

            # refine the falling edge detection, by using a small and fixed
            # conv_std_dev parameter to find the inflection point more precise
            start_ind = int(falling_ind[i] - conv_std_dev)
            if start_ind < 0:
                start_ind = 0

            stop_ind = int(falling_ind[i] + conv_std_dev)
            if stop_ind > len(conv_deriv):
                stop_ind = len(conv_deriv)

            if start_ind == stop_ind:
                stop_ind = start_ind + 1

            falling_ind[i] = start_ind + np.argmin(conv_deriv_ref[start_ind:stop_ind])

            # set this position and the sourrounding of the saved flank to 0 to
            #  avoid a second detection
            if falling_ind[i] < 2 * conv_std_dev:
                del_ind_start = 0
            else:
                del_ind_start = falling_ind[i] - int(2 * conv_std_dev)
            if (conv_deriv.size - falling_ind[i]) < 2 * conv_std_dev:
                del_ind_stop = conv_deriv.size - 1
            else:
                del_ind_stop = falling_ind[i] + int(2 * conv_std_dev)
            conv_deriv[del_ind_start:del_ind_stop] = 0

        # sort all indices of rising and falling flanks
        rising_ind.sort()
        falling_ind.sort()

        # find the maximum laser length to use as size for the laser array
        laser_length = np.max(falling_ind - rising_ind)

        # initialize the empty output array
        laser_arr = np.zeros((number_of_lasers, laser_length), dtype='int64')
        # slice the detected laser pulses of the timetrace and save them in the
        # output array according to the found rising edge
        for i in range(number_of_lasers):
            if rising_ind[i] + laser_length > count_data.size:
                lenarr = count_data[rising_ind[i]:].size
                laser_arr[i, 0:lenarr] = count_data[rising_ind[i]:]
            else:
                laser_arr[i] = count_data[rising_ind[i]:rising_ind[i] + laser_length]

        return_dict['laser_counts_arr'] = laser_arr.astype('int64')
        return_dict['laser_indices_rising'] = rising_ind
        return_dict['laser_indices_falling'] = falling_ind
        return return_dict

    def ungated_threshold(self, count_data, count_threshold=10, min_laser_length=200e-9,
                          threshold_tolerance=20e-9):
        """

        @param count_data:

        @return:
        """
        """
        Detects the laser pulses in the ungated timetrace data and extracts them.
    
        @param numpy.ndarray count_data: 1D array the raw timetrace data from an ungated fast counter
        @param measurement_settings: 
        @param fast_counter_settings: 
        @param count_threshold: 
        @param min_laser_length: 
        @param threshold_tolerance: 
        
        @return 2D numpy.ndarray:   2D array, the extracted laser pulses of the timetrace.
                                    dimensions: 0: laser number, 1: time bin
    
        Procedure:
            Threshold detection:
            ---------------
    
            All count data from the time trace is compared to a threshold value.
            Values above the threshold are considered to belong to a laser pulse.
            If the length of a pulse would be below the minimum length the pulse is discarded.
            If a number of bins which are below the threshold is smaller than the number of bins making the
            threshold_tolerance then they are still considered to belong to a laser pulse.
        """
        return_dict = dict()

        number_of_lasers = self.measurement_settings.get('number_of_lasers')
        counter_bin_width = self.fast_counter_settings.get('bin_width')

        if not isinstance(number_of_lasers, int):
            return_dict['laser_indices_rising'] = np.zeros(1, dtype='int64')
            return_dict['laser_indices_falling'] = np.zeros(1, dtype='int64')
            return_dict['laser_counts_arr'] = np.zeros((1, 3000), dtype='int64')
            return return_dict
        else:
            return_dict['laser_indices_rising'] = np.zeros(number_of_lasers, dtype='int64')
            return_dict['laser_indices_falling'] = np.zeros(number_of_lasers, dtype='int64')
            return_dict['laser_counts_arr'] = np.zeros((number_of_lasers, 3000), dtype='int64')

        # Convert length in seconds into length in time bins
        threshold_tolerance = round(threshold_tolerance / counter_bin_width)
        min_laser_length = round(min_laser_length / counter_bin_width)

        # get all bin indices with counts > threshold value
        bigger_indices = np.where(count_data >= count_threshold)[0]

        # get all indices with consecutive numbering (bin chains not interrupted by values < threshold
        index_list = np.split(bigger_indices,
                              np.where(np.diff(bigger_indices) >= threshold_tolerance)[0] + 1)
        for i, index_group in enumerate(index_list):
            if index_group.size > 0:
                start, end = index_list[i][0], index_list[i][-1]
                index_list[i] = np.arange(start, end + 1)
        consecutive_indices_unfiltered = index_list

        # sort out all groups shorter than minimum laser length
        consecutive_indices = [item for item in consecutive_indices_unfiltered if
                               len(item) > min_laser_length]

        # Check if the number of lasers matches the number of remaining index groups
        if number_of_lasers != len(consecutive_indices):
            return return_dict

        # determine max length of laser pulse and initialize laser array
        max_laser_length = max([index_array.size for index_array in consecutive_indices])
        return_dict['laser_counts_arr'] = np.zeros((number_of_lasers, max_laser_length), dtype='int64')

        # fill laser array with slices of raw data array. Also populate the rising/falling index arrays
        for i, index_group in enumerate(consecutive_indices):
            return_dict['laser_indices_rising'][i] = index_group[0]
            return_dict['laser_indices_falling'][i] = index_group[-1]
            return_dict['laser_counts_arr'][i, :index_group.size] = count_data[index_group]

        return return_dict


1			# -- coding: utf-8 --
2			"""
3			This file contains basic pulse extraction methods for Qudi.
4
5			Qudi is free software: you can redistribute it and/or modify
6			it under the terms of the GNU General Public License as published by
7			the Free Software Foundation, either version 3 of the License, or
8			(at your option) any later version.
9
10			Qudi is distributed in the hope that it will be useful,
11			but WITHOUT ANY WARRANTY; without even the implied warranty of
12			MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13			GNU General Public License for more details.
14
15			You should have received a copy of the GNU General Public License
16			along with Qudi. If not, see <http://www.gnu.org/licenses/>.
17
18			Copyright (c) the Qudi Developers. See the COPYRIGHT.txt file at the
19			top-level directory of this distribution and at <https://github.com/Ulm-IQO/qudi/>
20			"""
21
22			import numpy as np
23			from scipy import ndimage
24
25			from logic.pulsed.pulse_extractor import PulseExtractorBase
26
27
28			class BasicPulseExtractor(PulseExtractorBase):
29			"""
30
31			"""
32			def __init__(self, args, *kwargs):
33			super().__init__(args, *kwargs)
34
35			def gated_conv_deriv(self, count_data, conv_std_dev=20.0):
36			"""
37			Detects the rising flank in the gated timetrace data and extracts just the laser pulses.
38			The flank detection is based on an image edge detection technique performed in 1D.
39			There is no information about the data needed.
40			Only the gaussian filter width to reduce shot noise can be given as parameter.
41
42			@param 2D numpy.ndarray count_data: the raw timetrace data from a gated fast counter
43			dim 0: gate number; dim 1: time bin
44			@param float conv_std_dev: The standard deviation of the gaussian filter used for smoothing
45
46			@return dict: The extracted laser pulses of the timetrace as well as the indices for rising
47			and falling flanks.
48			"""
49			# Create return dictionary
50			return_dict = {'laser_counts_arr': np.zeros(0, dtype='int64'),
51			'laser_indices_rising': -1,
52			'laser_indices_falling': -1}
53
54			# sum up all gated timetraces to ease flank detection
55			timetrace_sum = np.sum(count_data, 0)
56
57			# apply gaussian filter to remove noise and compute the gradient of the timetrace sum
58			try:
59			conv = ndimage.filters.gaussian_filter1d(timetrace_sum.astype(float), conv_std_dev)
60			except:
61			conv = np.zeros(timetrace_sum.size)
62			try:
63			conv_deriv = np.gradient(conv)
64			except:
65			conv_deriv = np.zeros(conv.size)
66
67			# get indices of rising and falling flank
68			rising_ind = conv_deriv.argmax()
69			falling_ind = conv_deriv.argmin()
70
71			# If gaussian smoothing or derivative failed, the returned array only contains zeros.
72			# Check for that and return also only zeros to indicate a failed pulse extraction.
73			if len(conv_deriv.nonzero()[0]) == 0:
74			laser_arr = np.zeros(count_data.shape, dtype='int64')
75			else:
76			# slice the data array to cut off anything but laser pulses
77			laser_arr = count_data[:, rising_ind:falling_ind]
78
79			return_dict['laser_counts_arr'] = laser_arr.astype('int64')
80			return_dict['laser_indices_rising'] = rising_ind
81			return_dict['laser_indices_falling'] = falling_ind
82
83			return return_dict
84
85			def ungated_conv_deriv(self, count_data, conv_std_dev=20.0):
86			""" Detects the laser pulses in the ungated timetrace data and extracts
87			them.
88
89			@param numpy.ndarray count_data: 1D array the raw timetrace data from an ungated fast counter
90			@param dict measurement_settings: The measurement settings of the currently running measurement.
91			@param float conv_std_dev: The standard deviation of the gaussian used for smoothing
92
93			@return 2D numpy.ndarray: 2D array, the extracted laser pulses of the timetrace.
94			dimensions: 0: laser number, 1: time bin
95
96			Procedure:
97			Edge Detection:
98			---------------
99
100			The count_data array with the laser pulses is smoothed with a
101			gaussian filter (convolution), which used a defined standard
102			deviation of 10 entries (bins). Then the derivation of the convolved
103			time trace is taken to obtain the maxima and minima, which
104			corresponds to the rising and falling edge of the pulses.
105
106			The convolution with a gaussian removes nasty peaks due to count
107			fluctuation within a laser pulse and at the same time ensures a
108			clear distinction of the maxima and minima in the derived convolved
109			trace.
110
111			The maxima and minima are not found sequentially, pulse by pulse,
112			but are rather globally obtained. I.e. the convolved and derived
113			array is searched iteratively for a maximum and a minimum, and after
114			finding those the array entries within the 4 times
115			self.conv_std_dev (2*self.conv_std_dev to the left and
116			2*self.conv_std_dev) are set to zero.
117
118			The crucial part is the knowledge of the number of laser pulses and
119			the choice of the appropriate std_dev for the gauss filter.
120
121			To ensure a good performance of the edge detection, you have to
122			ensure a steep rising and falling edge of the laser pulse! Be also
123			careful in choosing a large conv_std_dev value and using a small
124			laser pulse (rule of thumb: conv_std_dev < laser_length/10).
125			"""
126			# Create return dictionary
127			return_dict = {'laser_counts_arr': np.empty(0, dtype='int64'),
128			'laser_indices_rising': np.empty(0, dtype='int64'),
129			'laser_indices_falling': np.empty(0, dtype='int64')}
130
131			number_of_lasers = self.measurement_settings.get('number_of_lasers')
132			if not isinstance(number_of_lasers, int):
133			return return_dict
134
135			# apply gaussian filter to remove noise and compute the gradient of the timetrace sum
136			try:
137			conv = ndimage.filters.gaussian_filter1d(count_data.astype(float), conv_std_dev)
138			except:
139			conv = np.zeros(count_data.size)
140			try:
141			conv_deriv = np.gradient(conv)
142			except:
143			conv_deriv = np.zeros(conv.size)
144
145			# if gaussian smoothing or derivative failed, the returned array only contains zeros.
146			# Check for that and return also only zeros to indicate a failed pulse extraction.
147			if len(conv_deriv.nonzero()[0]) == 0:
148			return_dict['laser_counts_arr'] = np.zeros((number_of_lasers, 10), dtype='int64')
149			return return_dict
150
151			# use a reference for array, because the exact position of the peaks or dips
152			# (i.e. maxima or minima, which are the inflection points in the pulse) are distorted by
153			# a large conv_std_dev value.
154			try:
155			conv = ndimage.filters.gaussian_filter1d(count_data.astype(float), 10)
156			except:
157			conv = np.zeros(count_data.size)
158			try:
159			conv_deriv_ref = np.gradient(conv)
160			except:
161			conv_deriv_ref = np.zeros(conv.size)
162
163			# initialize arrays to contain indices for all rising and falling
164			# flanks, respectively
165			rising_ind = np.empty(number_of_lasers, dtype='int64')
166			falling_ind = np.empty(number_of_lasers, dtype='int64')
167
168			# Find as many rising and falling flanks as there are laser pulses in
169			# the trace:
170			for i in range(number_of_lasers):
171			# save the index of the absolute maximum of the derived time trace
172			# as rising edge position
173			rising_ind[i] = np.argmax(conv_deriv)
174
175			# refine the rising edge detection, by using a small and fixed
176			# conv_std_dev parameter to find the inflection point more precise
177			start_ind = int(rising_ind[i] - conv_std_dev)
178			if start_ind < 0:
179			start_ind = 0
180
181			stop_ind = int(rising_ind[i] + conv_std_dev)
182			if stop_ind > len(conv_deriv):
183			stop_ind = len(conv_deriv)
184
185			if start_ind == stop_ind:
186			stop_ind = start_ind + 1
187
188			rising_ind[i] = start_ind + np.argmax(conv_deriv_ref[start_ind:stop_ind])
189
190			# set this position and the surrounding of the saved edge to 0 to
191			# avoid a second detection
192			if rising_ind[i] < 2 * conv_std_dev:
193			del_ind_start = 0
194			else:
195			del_ind_start = rising_ind[i] - int(2 * conv_std_dev)
196			if (conv_deriv.size - rising_ind[i]) < 2 * conv_std_dev:
197			del_ind_stop = conv_deriv.size - 1
198			else:
199			del_ind_stop = rising_ind[i] + int(2 * conv_std_dev)
200			conv_deriv[del_ind_start:del_ind_stop] = 0
201
202			# save the index of the absolute minimum of the derived time trace
203			# as falling edge position
204			falling_ind[i] = np.argmin(conv_deriv)
205
206			# refine the falling edge detection, by using a small and fixed
207			# conv_std_dev parameter to find the inflection point more precise
208			start_ind = int(falling_ind[i] - conv_std_dev)
209			if start_ind < 0:
210			start_ind = 0
211
212			stop_ind = int(falling_ind[i] + conv_std_dev)
213			if stop_ind > len(conv_deriv):
214			stop_ind = len(conv_deriv)
215
216			if start_ind == stop_ind:
217			stop_ind = start_ind + 1
218
219			falling_ind[i] = start_ind + np.argmin(conv_deriv_ref[start_ind:stop_ind])
220
221			# set this position and the sourrounding of the saved flank to 0 to
222			# avoid a second detection
223			if falling_ind[i] < 2 * conv_std_dev:
224			del_ind_start = 0
225			else:
226			del_ind_start = falling_ind[i] - int(2 * conv_std_dev)
227			if (conv_deriv.size - falling_ind[i]) < 2 * conv_std_dev:
228			del_ind_stop = conv_deriv.size - 1
229			else:
230			del_ind_stop = falling_ind[i] + int(2 * conv_std_dev)
231			conv_deriv[del_ind_start:del_ind_stop] = 0
232
233			# sort all indices of rising and falling flanks
234			rising_ind.sort()
235			falling_ind.sort()
236
237			# find the maximum laser length to use as size for the laser array
238			laser_length = np.max(falling_ind - rising_ind)
239
240			# initialize the empty output array
241			laser_arr = np.zeros((number_of_lasers, laser_length), dtype='int64')
242			# slice the detected laser pulses of the timetrace and save them in the
243			# output array according to the found rising edge
244			for i in range(number_of_lasers):
245			if rising_ind[i] + laser_length > count_data.size:
246			lenarr = count_data[rising_ind[i]:].size
247			laser_arr[i, 0:lenarr] = count_data[rising_ind[i]:]
248			else:
249			laser_arr[i] = count_data[rising_ind[i]:rising_ind[i] + laser_length]
250
251			return_dict['laser_counts_arr'] = laser_arr.astype('int64')
252			return_dict['laser_indices_rising'] = rising_ind
253			return_dict['laser_indices_falling'] = falling_ind
254			return return_dict
255
256			def ungated_threshold(self, count_data, count_threshold=10, min_laser_length=200e-9,
257			threshold_tolerance=20e-9):
258			"""
259
260			@param count_data:
261
262			@return:
263			"""
264			"""
265			Detects the laser pulses in the ungated timetrace data and extracts them.
266
267			@param numpy.ndarray count_data: 1D array the raw timetrace data from an ungated fast counter
268			@param measurement_settings:
269			@param fast_counter_settings:
270			@param count_threshold:
271			@param min_laser_length:
272			@param threshold_tolerance:
273
274			@return 2D numpy.ndarray: 2D array, the extracted laser pulses of the timetrace.
275			dimensions: 0: laser number, 1: time bin
276
277			Procedure:
278			Threshold detection:
279			---------------
280
281			All count data from the time trace is compared to a threshold value.
282			Values above the threshold are considered to belong to a laser pulse.
283			If the length of a pulse would be below the minimum length the pulse is discarded.
284			If a number of bins which are below the threshold is smaller than the number of bins making the
285			threshold_tolerance then they are still considered to belong to a laser pulse.
286			"""
287			return_dict = dict()
288
289			number_of_lasers = self.measurement_settings.get('number_of_lasers')
290			counter_bin_width = self.fast_counter_settings.get('bin_width')
291
292			if not isinstance(number_of_lasers, int):
293			return_dict['laser_indices_rising'] = np.zeros(1, dtype='int64')
294			return_dict['laser_indices_falling'] = np.zeros(1, dtype='int64')
295			return_dict['laser_counts_arr'] = np.zeros((1, 3000), dtype='int64')
296			return return_dict
297			else:
298			return_dict['laser_indices_rising'] = np.zeros(number_of_lasers, dtype='int64')
299			return_dict['laser_indices_falling'] = np.zeros(number_of_lasers, dtype='int64')
300			return_dict['laser_counts_arr'] = np.zeros((number_of_lasers, 3000), dtype='int64')
301
302			# Convert length in seconds into length in time bins
303			threshold_tolerance = round(threshold_tolerance / counter_bin_width)
304			min_laser_length = round(min_laser_length / counter_bin_width)
305
306			# get all bin indices with counts > threshold value
307			bigger_indices = np.where(count_data >= count_threshold)[0]
308
309			# get all indices with consecutive numbering (bin chains not interrupted by values < threshold
310			index_list = np.split(bigger_indices,
311			np.where(np.diff(bigger_indices) >= threshold_tolerance)[0] + 1)
312			for i, index_group in enumerate(index_list):
313			if index_group.size > 0:
314			start, end = index_list[i][0], index_list[i][-1]
315			index_list[i] = np.arange(start, end + 1)
316			consecutive_indices_unfiltered = index_list
317
318			# sort out all groups shorter than minimum laser length
319			consecutive_indices = [item for item in consecutive_indices_unfiltered if
320			len(item) > min_laser_length]
321
322			# Check if the number of lasers matches the number of remaining index groups
323			if number_of_lasers != len(consecutive_indices):
324			return return_dict
325
326			# determine max length of laser pulse and initialize laser array
327			max_laser_length = max([index_array.size for index_array in consecutive_indices])
328			return_dict['laser_counts_arr'] = np.zeros((number_of_lasers, max_laser_length), dtype='int64')
329
330			# fill laser array with slices of raw data array. Also populate the rising/falling index arrays
331			for i, index_group in enumerate(consecutive_indices):
332			return_dict['laser_indices_rising'][i] = index_group[0]
333			return_dict['laser_indices_falling'][i] = index_group[-1]
334			return_dict['laser_counts_arr'][i, :index_group.size] = count_data[index_group]
335
336			return return_dict
337

Ulm-IQO / qudi

Pull Request — master (#384)

BasicPulseExtractor.gated_conv_deriv() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like