OkFpgaPulser.write_waveform()

Complexity

Conditions

7

Size

Total Lines

74

Duplication

Lines	0
Ratio	0 %

Importance

Changes

0

# -*- coding: utf-8 -*-
    def write_waveform(self, name, analog_samples, digital_samples, is_first_chunk, is_last_chunk,
                       total_number_of_samples):
        """
        Write a new waveform or append samples to an already existing waveform on the device memory.
        The flags is_first_chunk and is_last_chunk can be used as indicator if a new waveform should
        be created or if the write process to a waveform should be terminated.

        @param name: str, the name of the waveform to be created/append to
        @param analog_samples: numpy.ndarray of type float32 containing the voltage samples
        @param digital_samples: numpy.ndarray of type bool containing the marker states
                                (if analog channels are active, this must be the same length as
                                analog_samples)
        @param is_first_chunk: bool, flag indicating if it is the first chunk to write.
                                     If True this method will create a new empty wavveform.
                                     If False the samples are appended to the existing waveform.
        @param is_last_chunk: bool, flag indicating if it is the last chunk to write.
                                    Some devices may need to know when to close the appending wfm.
        @param total_number_of_samples: int, The number of sample points for the entire waveform
                                        (not only the currently written chunk)

        @return: (int, list) number of samples written (-1 indicates failed process) and list of
                             created waveform names
        """
        if analog_samples:
            self.log.error('FPGA pulse generator is purely digital and does not support waveform '
                           'generation with analog samples.')
            return -1, list()
        if not digital_samples:
            if total_number_of_samples > 0:
                self.log.warning('No samples handed over for waveform generation.')
                return -1, list()
            else:
                self.__current_waveform = bytearray([0])
                self.__current_waveform_name = ''
                return 0, list()

        # Initialize waveform array if this is the first chunk to write
        # Also append zero-timebins to waveform if the length is no integer multiple of 32
        if is_first_chunk:
            self.__samples_written = 0
            self.__current_waveform_name = name
            if total_number_of_samples % 32 != 0:
                number_of_zeros = 32 - (total_number_of_samples % 32)
                self.__current_waveform = np.zeros(total_number_of_samples + number_of_zeros,
                                                   dtype='uint8')
                self.log.warning('FPGA pulse sequence length is no integer multiple of 32 samples.'
                                 '\nAppending {0:d} zero-samples to the sequence.'
                                 ''.format(number_of_zeros))
            else:
                self.__current_waveform = np.zeros(total_number_of_samples, dtype='uint8')

        # Determine which part of the waveform array should be written
        chunk_length = len(digital_samples[list(digital_samples)[0]])
        write_end_index = self.__samples_written + chunk_length

        # Encode samples for each channel in bit mask and create waveform array
        for chnl, samples in digital_samples.items():
            # get channel index in range 0..7
            chnl_ind = int(chnl.rsplit('ch', 1)[1]) - 1
            # Represent bool values as np.uint8
            uint8_samples = samples.view('uint8')
            # left shift 0/1 values to bit position corresponding to channel index
            np.left_shift(uint8_samples, chnl_ind, out=uint8_samples)
            # Add samples to waveform array
            np.add(self.__current_waveform[self.__samples_written:write_end_index],
                   uint8_samples,
                   out=self.__current_waveform[self.__samples_written:write_end_index])

        # Convert numpy array to bytearray
        self.__current_waveform = bytearray(self.__current_waveform.tobytes())

        # increment the current write index
        self.__samples_written += chunk_length
        return chunk_length, [self.__current_waveform_name]