AWG70K._write_wfmx()

Complexity

Conditions

16

Size

Total Lines

74

Duplication

Lines	0
Ratio	0 %

Importance

Changes	1
Bugs	0	Features	0

# -*- coding: utf-8 -*-
    def _write_wfmx(self, filename, analog_samples, marker_bytes, is_first_chunk, is_last_chunk,
                    total_number_of_samples):
        """
        Appends a sampled chunk of a whole waveform to a wfmx-file. Create the file
        if it is the first chunk.
        If both flags (is_first_chunk, is_last_chunk) are set to TRUE it means
        that the whole ensemble is written as a whole in one big chunk.

        @param name: string, represents the name of the sampled ensemble
        @param analog_samples: dict containing float32 numpy ndarrays, contains the
                                       samples for the analog channels that
                                       are to be written by this function call.
        @param marker_bytes: np.ndarray containing bool numpy ndarrays, contains the samples
                                      for the digital channels that
                                      are to be written by this function call.
        @param total_number_of_samples: int, The total number of samples in the
                                        entire waveform. Has to be known in advance.
        @param is_first_chunk: bool, indicates if the current chunk is the
                               first write to this file.
        @param is_last_chunk: bool, indicates if the current chunk is the last
                              write to this file.

        @return list: the list contains the string names of the created files for the passed
                      presampled arrays
        """
        # The memory overhead of the tmp file write/read process in bytes. Only used if wfmx file is
        # written in chunks in order to avoid excessive memory usage.
        tmp_bytes_overhead = 16777216  # 16 MB

        if not filename.endswith('.wfmx'):
            filename += '.wfmx'
        wfmx_path = os.path.join(self._tmp_work_dir, filename)
        tmp_path = os.path.join(self._tmp_work_dir, 'digital_tmp.bin')

        # if it is the first chunk, create the .WFMX file with header.
        if is_first_chunk:
            # create header
            header = self._create_xml_header(total_number_of_samples, marker_bytes is not None)
            # write header
            with open(wfmx_path, 'wb') as wfmxfile:
                wfmxfile.write(header.encode('utf8'))
            # Check if a tmp digital samples file is present and delete it if necessary.
            if os.path.isfile(tmp_path):
                os.remove(tmp_path)

        # append analog samples to the .WFMX file.
        # Write digital samples in temporary file if not the entire samples are passed at once.
        with open(wfmx_path, 'ab') as wfmxfile:
            # append analog samples in binary format. One sample is 4 bytes (np.float32).
            wfmxfile.write(analog_samples)

        # Write digital samples to tmp file if chunkwise writing is used and it's not the last chunk
        if not is_last_chunk and marker_bytes is not None:
            with open(tmp_path, 'ab') as tmp_file:
                tmp_file.write(marker_bytes)

        # If this is the last chunk, write digital samples from tmp file to wfmx file (if present)
        # and also append the currently passed digital samples to wfmx file.
        # Read from tmp file in chunks of tmp_bytes_overhead in order to avoid too much memory
        # overhead.
        if is_last_chunk and marker_bytes is not None:
            with open(wfmx_path, 'ab') as wfmxfile:
                # Copy over digital samples from tmp file. Delete tmp file afterwards.
                if os.path.isfile(tmp_path):
                    with open(tmp_path, 'rb') as tmp_file:
                        while True:
                            tmp = tmp_file.read(tmp_bytes_overhead)
                            if not tmp:
                                break
                            wfmxfile.write(tmp)
                    os.remove(tmp_path)
                # Append current digital samples array to wfmx file
                wfmxfile.write(marker_bytes)
        return