AWG70K.write_waveform()

Complexity

Conditions

15

Size

Total Lines

108

Duplication

Lines	0
Ratio	0 %

Importance

Changes	5
Bugs	0	Features	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
        """
        waveforms = list()

        # Sanity checks
        if len(analog_samples) == 0:
            self.log.error('No analog samples passed to write_waveform method in awg70k.')
            return -1, waveforms

        min_samples = int(self.query('WLIS:WAV:LMIN?'))
        if total_number_of_samples < min_samples:
            self.log.error('Unable to write waveform.\nNumber of samples to write ({0:d}) is '
                           'smaller than the allowed minimum waveform length ({1:d}).'
                           ''.format(total_number_of_samples, min_samples))
            return -1, waveforms

        # determine active channels
        activation_dict = self.get_active_channels()
        active_channels = {chnl for chnl in activation_dict if activation_dict[chnl]}
        active_analog = sorted(chnl for chnl in active_channels if chnl.startswith('a'))

        # Sanity check of channel numbers
        if active_channels != set(analog_samples.keys()).union(set(digital_samples.keys())):
            self.log.error('Mismatch of channel activation and sample array dimensions for '
                           'waveform creation.\nChannel activation is: {0}\nSample arrays have: '
                           ''.format(active_channels,
                                     set(analog_samples.keys()).union(set(digital_samples.keys()))))
            return -1, waveforms

        # Write waveforms. One for each analog channel.
        for a_ch in active_analog:
            # Get the integer analog channel number
            a_ch_num = int(a_ch.split('ch')[-1])
            # Get the digital channel specifiers belonging to this analog channel markers
            mrk_ch_1 = 'd_ch{0:d}'.format(a_ch_num * 2 - 1)
            mrk_ch_2 = 'd_ch{0:d}'.format(a_ch_num * 2)

            start = time.time()
            # Encode marker information in an array of bytes (uint8). Avoid intermediate copies!!!
            if mrk_ch_1 in digital_samples and mrk_ch_2 in digital_samples:
                mrk_bytes = digital_samples[mrk_ch_2].view('uint8')
                tmp_bytes = digital_samples[mrk_ch_1].view('uint8')
                np.left_shift(mrk_bytes, 7, out=mrk_bytes)
                np.left_shift(tmp_bytes, 6, out=tmp_bytes)
                np.add(mrk_bytes, tmp_bytes, out=mrk_bytes)
            elif mrk_ch_1 in digital_samples:
                mrk_bytes = digital_samples[mrk_ch_1].view('uint8')
                np.left_shift(mrk_bytes, 6, out=mrk_bytes)
            else:
                mrk_bytes = None
            print('Prepare digital channel data: {0}'.format(time.time()-start))

            # Create waveform name string
            wfm_name = '{0}_ch{1:d}'.format(name, a_ch_num)

            # Check if waveform already exists and delete if necessary.
            if wfm_name in self.get_waveform_names():
                self.delete_waveform(wfm_name)

            # Write WFMX file for waveform
            start = time.time()
            self._write_wfmx(filename=wfm_name,
                             analog_samples=analog_samples[a_ch],
                             digital_samples=mrk_bytes,
                             is_first_chunk=is_first_chunk,
                             is_last_chunk=is_last_chunk,
                             total_number_of_samples=total_number_of_samples)
            print('Write WFMX file: {0}'.format(time.time() - start))

            # transfer waveform to AWG and load into workspace
            start = time.time()
            self._send_file(filename=wfm_name + '.wfmx')
            print('Send WFMX file: {0}'.format(time.time() - start))

            start = time.time()
            self.write('MMEM:OPEN "{0}"'.format(os.path.join(self._ftp_path, wfm_name + '.wfmx')))
            # Wait for everything to complete
            while int(self.query('*OPC?')) != 1:
                time.sleep(0.25)
            # Just to make sure
            while wfm_name not in self.get_waveform_names():
                time.sleep(0.25)
            print('Load WFMX file into workspace: {0}'.format(time.time() - start))

            # Append created waveform name to waveform list
            waveforms.append(wfm_name)
        return total_number_of_samples, waveforms