testtest/nd2reader/artificial.py

"""Functions to create artificial nd2 data for testing purposes
"""
import six
import numpy as np
import struct
from nd2reader.common import check_or_make_dir
from os import path

global_labels = ['image_attributes', 'image_text_info', 'image_metadata',
                 'image_metadata_sequence', 'image_calibration', 'x_data',
                 'y_data', 'z_data', 'roi_metadata', 'pfs_status', 'pfs_offset',
                 'guid', 'description', 'camera_exposure_time', 'camera_temp',
                 'acquisition_times', 'acquisition_times_2',
                 'acquisition_frames', 'lut_data', 'grabber_settings',
                 'custom_data', 'app_info', 'image_frame_0']

global_file_labels = ["ImageAttributesLV!", "ImageTextInfoLV!",
                      "ImageMetadataLV!", "ImageMetadataSeqLV|0!",
                      "ImageCalibrationLV|0!", "CustomData|X!", "CustomData|Y!",
                      "CustomData|Z!", "CustomData|RoiMetadata_v1!",
                      "CustomData|PFS_STATUS!", "CustomData|PFS_OFFSET!",
                      "CustomData|GUIDStore!", "CustomData|CustomDescriptionV1_0!",
                      "CustomData|Camera_ExposureTime1!", "CustomData|CameraTemp1!",
                      "CustomData|AcqTimesCache!", "CustomData|AcqTimes2Cache!",
                      "CustomData|AcqFramesCache!", "CustomDataVar|LUTDataV1_0!",
                      "CustomDataVar|GrabberCameraSettingsV1_0!",
                      "CustomDataVar|CustomDataV2_0!", "CustomDataVar|AppInfo_V1_0!",
                      "ImageDataSeq|0!"]


class ArtificialND2(object):
    """Artificial ND2 class (for testing purposes)
    """
    header = 0xabeceda
    relative_offset = 0
    data_types = {'unsigned_char': 1,
                  'unsigned_int': 2,
                  'unsigned_int_2': 3,
                  'unsigned_long': 5,
                  'double': 6,
                  'string': 8,
                  'char_array': 9,
                  'metadata_item': 11,
                  }

    def __init__(self, file, version=(3, 0), skip_blocks=None):
        self.version = version
        self.raw_text, self.locations, self.data = b'', None, None
        check_or_make_dir(path.dirname(file))
        self._fh = open(file, 'w+b', 0)
        self.write_file(skip_blocks)

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        self.close()

    @property
    def file_handle(self):
        """The file handle to the binary file

        Returns:
            file: the file handle

        """
        return self._fh

    def close(self):
        """Correctly close the file handle
        """
        if self._fh is not None:
            self._fh.close()

    def write_file(self, skip_blocks=None):
        if skip_blocks is None:
            skip_blocks = []

        if 'version' not in skip_blocks:
            # write version header at start of the file
            self.write_version()

        if 'label_map' not in skip_blocks:
            # write label map + data in the center
            self.locations, self.data = self.write_label_map()

        if 'label_map_marker' not in skip_blocks:
            # write start position of label map at the end of the file
            self.write_label_map_info()

        # write all to file
        self._fh.write(self.raw_text)

    def write_version(self):
        """Write file header
        """
        # write 16 empty bytes
        self.raw_text += bytearray(16)

        # write version info
        self.raw_text += self._get_version_string()

    def write_label_map_info(self):
        """Write the location of the start of the label map at the end of the file
        """
        location = self._get_version_byte_length()
        self.raw_text += struct.pack("Q", location)

    def _get_version_string(self):
        return six.b('ND2 FILE SIGNATURE CHUNK NAME01!Ver%s.%s' % self.version)

    def _get_version_byte_length(self):
        return 16 + len(self._get_version_string())

    def write_label_map(self):
        raw_text, locations, data = self.create_label_map_bytes()
        self.raw_text += raw_text

        return locations, data

    def create_label_map_bytes(self):
        """Construct a binary label map

        Returns:
            tuple: (binary data, dictionary data)

        """
        raw_text = six.b('')
        labels = global_labels
        file_labels = global_file_labels

        file_data, file_data_dict = self._get_file_data(labels)

        locations = {}

        # generate random positions and lengths
        version_length = self._get_version_byte_length()

        # calculate data length
        label_length = np.sum([len(six.b(l)) + 16 for l in file_labels])

        # write label map
        cur_pos = version_length + label_length
        for label, file_label, data in zip(labels, file_labels, file_data):
            raw_text += six.b(file_label)
            data_length = len(data)
            raw_text += struct.pack('QQ', cur_pos, data_length)
            locations[label] = (cur_pos, data_length)
            cur_pos += data_length

        # write data
        raw_text += six.b('').join(file_data)

        return raw_text, locations, file_data_dict

    def _pack_data_with_metadata(self, data):
        packed_data = self._pack_raw_data_with_metadata(data)

        raw_data = struct.pack("IIQ", self.header, self.relative_offset, len(packed_data))
        raw_data += packed_data

        return raw_data

    def _pack_raw_data_with_metadata(self, data):
        raw_data = b''

        if isinstance(data, dict):
            raw_data = self._pack_dict_with_metadata(data)
        elif isinstance(data, int):
            raw_data = struct.pack('I', data)
        elif isinstance(data, float):
            raw_data = struct.pack('d', data)
        elif isinstance(data, str):
            raw_data = self._str_to_padded_bytes(data)

        return raw_data

    def _get_data_type(self, data):
        if isinstance(data, dict):
            return self.data_types['metadata_item']
        elif isinstance(data, int):
            return self.data_types['unsigned_int']
        elif isinstance(data, str):
            return self.data_types['string']
        else:
            return self.data_types['double']

    @staticmethod
    def _str_to_padded_bytes(data):
        return six.b('').join([struct.pack('cx', six.b(s)) for s in data]) + struct.pack('xx')

    def _pack_dict_with_metadata(self, data):
        raw_data = b''

        for data_key in data.keys():
            # names have always one character extra and are padded in zero bytes???
            b_data_key = self._str_to_padded_bytes(data_key)

            # header consists of data type and length of key name, it is represented by 2 unsigned chars
            raw_data += struct.pack('BB', self._get_data_type(data[data_key]), len(data_key) + 1)
            raw_data += b_data_key

            sub_data = self._pack_raw_data_with_metadata(data[data_key])

            if isinstance(data[data_key], dict):
                # Pack: the number of keys and the length of raw data until now, sub data
                # and the 12 bytes that we add now
                raw_data += struct.pack("<IQ", len(data[data_key].keys()), len(sub_data) + len(raw_data) + 12)

            raw_data += sub_data

            if isinstance(data[data_key], dict):
                # apparently there is also a huge empty space
                raw_data += b''.join([struct.pack('x')] * len(data[data_key].keys()) * 8)

        return raw_data

    @staticmethod
    def _get_slx_img_attrib():
        return {'uiWidth': 128,
                'uiWidthBytes': 256,
                'uiHeight': 128,
                'uiComp': 1,
                'uiBpcInMemory': 16,
                'uiBpcSignificant': 12,
                'uiSequenceCount': 70,
                'uiTileWidth': 128,
                'uiTileHeight': 128,
                'eCompression': 2,
                'dCompressionParam': -1.0,
                'ePixelType': 1,
                'uiVirtualComponents': 1
                }

    @staticmethod
    def _get_slx_picture_metadata():
        return {'sPicturePlanes':
                {
                    'sPlaneNew': {
                        # channels are numbered a0, a1, ..., aN
                        'a0': {
                            'sDescription': 'TRITC'
                            }
                        }
                    }
                }

    def _get_file_data(self, labels):
        file_data = [
            {'SLxImageAttributes': self._get_slx_img_attrib()},  # ImageAttributesLV!",
            7,  # ImageTextInfoLV!",
            7,  # ImageMetadataLV!",
            {'SLxPictureMetadata': self._get_slx_picture_metadata()},  # ImageMetadataSeqLV|0!",
            7,  # ImageCalibrationLV|0!",
            7,  # CustomData|X!",
            7,  # CustomData|Y!",
            7,  # CustomData|Z!",
            7,  # CustomData|RoiMetadata_v1!",
            7,  # CustomData|PFS_STATUS!",
            7,  # CustomData|PFS_OFFSET!",
            7,  # CustomData|GUIDStore!",
            7,  # CustomData|CustomDescriptionV1_0!",
            7,  # CustomData|Camera_ExposureTime1!",
            7,  # CustomData|CameraTemp1!",
            [0],  # CustomData|AcqTimesCache!",
            [0],  # CustomData|AcqTimes2Cache!",
            [0],  # CustomData|AcqFramesCache!",
            7,  # CustomDataVar|LUTDataV1_0!",
            7,  # CustomDataVar|GrabberCameraSettingsV1_0!",
            7,  # CustomDataVar|CustomDataV2_0!",
            7,  # CustomDataVar|AppInfo_V1_0!",
            7,  # ImageDataSeq|0!"
        ]

        file_data_dict = {l: d for l, d in zip(labels, file_data)}

        # convert to bytes
        file_data = [self._pack_data_with_metadata(d) for d in file_data]

        return file_data, file_data_dict