SequenceGeneratorLogic.analyze_block_ensemble()

Complexity

Conditions

16

Size

Total Lines

116

Duplication

Lines	9
Ratio	7.76 %

Importance

Changes

0

# -*- coding: utf-8 -*-
    def analyze_block_ensemble(self, ensemble):
        """
        This helper method runs through each element of a PulseBlockEnsemble object and extracts
        important information about the Waveform that can be created out of this object.
        Especially the discretization due to the set self.sample_rate is taken into account.
        The positions in time (as integer time bins) of the PulseBlockElement transitions are
        determined here (all the "rounding-to-best-match-value").
        Additional information like the total number of samples, total number of PulseBlockElements
        and the timebins for digital channel low-to-high transitions get returned as well.

        This method assumes that sanity checking has been already performed on the
        PulseBlockEnsemble (via _sampling_ensemble_sanity_check). Meaning it assumes that all
        PulseBlocks are actually present in saved blocks and the channel activation matches the
        current pulse settings.

        @param ensemble: A PulseBlockEnsemble object (see logic.pulse_objects.py)
        @return: number_of_samples (int): The total number of samples in a Waveform provided the
                                              current sample_rate and PulseBlockEnsemble object.
                 total_elements (int): The total number of PulseBlockElements (incl. repetitions) in
                                       the provided PulseBlockEnsemble.
                 elements_length_bins (1D numpy.ndarray[int]): Array of number of timebins for each
                                                               PulseBlockElement in chronological
                                                               order (incl. repetitions).
                 digital_rising_bins (dict): Dictionary with keys being the digital channel
                                             descriptor string and items being arrays of
                                             chronological low-to-high transition positions
                                             (in timebins; incl. repetitions) for each digital
                                             channel.
        """
        # memorize the channel state of the previous element
        tmp_digital_high = dict()
        # Set of used analog and digital channels
        digital_channels = set()
        analog_channels = set()
        # check for active channels and initialize tmp_digital_high with the state of the very last
        # element in the ensemble
        if len(ensemble.block_list) > 0:

This code seems to be duplicated in your project.

            block = self.get_block(ensemble.block_list[0][0])
            digital_channels = block.digital_channels
            analog_channels = block.analog_channels
            block = self.get_block(ensemble.block_list[-1][0])
            if len(block.element_list) > 0:
                tmp_digital_high = block.element_list[-1].digital_high.copy()
            else:
                tmp_digital_high = {chnl: False for chnl in digital_channels}

        # dicts containing the bins where the digital channels are rising/falling
        digital_rising_bins = {chnl: list() for chnl in digital_channels}
        digital_falling_bins = {chnl: list() for chnl in digital_channels}

        # Array to store the length in bins for all elements including repetitions in the order
        # they are occuring in the waveform later on. (Must be int64 or it will overflow eventually)
        elements_length_bins = np.empty(0, dtype='int64')

        # variables to keep track of the current timeframe
        current_end_time = 0.0
        current_start_bin = 0

        # Loop through all blocks in the ensemble block_list
        for block_name, reps in ensemble.block_list:
            # Get the stored PulseBlock instance
            block = self.get_block(block_name)

            # Temporary array to hold the length in bins for all elements in the block (incl. reps)
            tmp_length_bins = np.zeros((reps + 1) * len(block.element_list), dtype='int64')

            # Iterate over all repetitions of the current block while keeping track of the
            # current element index
            unrolled_element_index = 0
            for rep_no in range(reps + 1):
                # Iterate over the Block_Elements inside the current block
                for element in block.element_list:
                    # save bin position if a transition from low to high or vice versa has occured
                    # in a digital channel
                    if tmp_digital_high != element.digital_high:
                        for chnl, state in element.digital_high.items():
                            if not tmp_digital_high[chnl] and state:
                                digital_rising_bins[chnl].append(current_start_bin)
                            elif tmp_digital_high[chnl] and not state:
                                digital_falling_bins[chnl].append(current_start_bin)
                        tmp_digital_high = element.digital_high.copy()

                    # Calculate length of the current element with current repetition count in sec
                    # and add this to the ideal end time for the sequence up until this point.
                    current_end_time += element.init_length_s + rep_no * element.increment_s

                    # Nearest possible match including the discretization in bins
                    current_end_bin = int(np.rint(current_end_time * self.__sample_rate))

                    # append current element length in discrete bins to temporary array
                    tmp_length_bins[unrolled_element_index] = current_end_bin - current_start_bin

                    # advance bin offset for next element
                    current_start_bin = current_end_bin
                    # increment element counter
                    unrolled_element_index += 1

            # append element lengths (in bins) for this block to array
            elements_length_bins = np.append(elements_length_bins, tmp_length_bins)

        # convert digital rising/falling indices to numpy.ndarrays
        for chnl in digital_channels:
            digital_rising_bins[chnl] = np.array(digital_rising_bins[chnl], dtype='int64')
            digital_falling_bins[chnl] = np.array(digital_falling_bins[chnl], dtype='int64')

        return_dict = dict()
        return_dict['number_of_samples'] = np.sum(elements_length_bins)
        return_dict['number_of_elements'] = len(elements_length_bins)
        return_dict['elements_length_bins'] = elements_length_bins
        return_dict['digital_rising_bins'] = digital_rising_bins
        return_dict['digital_falling_bins'] = digital_falling_bins
        return_dict['analog_channels'] = analog_channels
        return_dict['digital_channels'] = digital_channels
        return_dict['channel_set'] = analog_channels.union(digital_channels)
        return_dict['generation_parameters'] = self.generation_parameters.copy()
        return return_dict