You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
415 lines
14 KiB
415 lines
14 KiB
import re
|
|
|
|
from nd2reader.common import read_chunk, read_array, read_metadata, parse_date
|
|
import xmltodict
|
|
import six
|
|
import numpy as np
|
|
|
|
|
|
def ignore_missing(func):
|
|
def wrapper(*args, **kwargs):
|
|
try:
|
|
return func(*args, **kwargs)
|
|
except:
|
|
return None
|
|
|
|
return wrapper
|
|
|
|
|
|
class RawMetadata(object):
|
|
def __init__(self, fh, label_map):
|
|
self._fh = fh
|
|
self._label_map = label_map
|
|
self._metadata_parsed = None
|
|
|
|
@property
|
|
def __dict__(self):
|
|
"""
|
|
Returns the parsed metadata in dictionary form
|
|
:return:
|
|
"""
|
|
return self.get_parsed_metadata()
|
|
|
|
def get_parsed_metadata(self):
|
|
"""
|
|
Returns the parsed metadata in dictionary form
|
|
:return:
|
|
"""
|
|
|
|
if self._metadata_parsed is not None:
|
|
return self._metadata_parsed
|
|
|
|
self._metadata_parsed = {
|
|
"height": self.image_attributes[six.b('SLxImageAttributes')][six.b('uiHeight')],
|
|
"width": self.image_attributes[six.b('SLxImageAttributes')][six.b('uiWidth')],
|
|
"date": parse_date(self.image_text_info[six.b('SLxImageTextInfo')]),
|
|
"fields_of_view": self._parse_fields_of_view(),
|
|
"frames": self._parse_frames(),
|
|
"z_levels": self._parse_z_levels(),
|
|
"total_images_per_channel": self._parse_total_images_per_channel(),
|
|
"channels": self._parse_channels(),
|
|
"pixel_microns": self.image_calibration.get(six.b('SLxCalibration'), {}).get(six.b('dCalibration')),
|
|
}
|
|
|
|
self._metadata_parsed['num_frames'] = len(self._metadata_parsed['frames'])
|
|
|
|
self._parse_roi_metadata()
|
|
self._parse_experiment_metadata()
|
|
|
|
return self._metadata_parsed
|
|
|
|
def _parse_channels(self):
|
|
"""
|
|
These are labels created by the NIS Elements user. Typically they may a short description of the filter cube
|
|
used (e.g. "bright field", "GFP", etc.)
|
|
|
|
:rtype: list
|
|
|
|
"""
|
|
channels = []
|
|
metadata = self.image_metadata_sequence[six.b('SLxPictureMetadata')][six.b('sPicturePlanes')]
|
|
try:
|
|
validity = self.image_metadata[six.b('SLxExperiment')][six.b('ppNextLevelEx')][six.b('')][0][
|
|
six.b('ppNextLevelEx')][six.b('')][0][six.b('pItemValid')]
|
|
except (KeyError, TypeError):
|
|
# If none of the channels have been deleted, there is no validity list, so we just make one
|
|
validity = [True for _ in metadata]
|
|
# Channel information is contained in dictionaries with the keys a0, a1...an where the number
|
|
# indicates the order in which the channel is stored. So by sorting the dicts alphabetically
|
|
# we get the correct order.
|
|
for (label, chan), valid in zip(sorted(metadata[six.b('sPlaneNew')].items()), validity):
|
|
if not valid:
|
|
continue
|
|
channels.append(chan[six.b('sDescription')].decode("utf8"))
|
|
return channels
|
|
|
|
def _parse_fields_of_view(self):
|
|
"""
|
|
The metadata contains information about fields of view, but it contains it even if some fields
|
|
of view were cropped. We can't find anything that states which fields of view are actually
|
|
in the image data, so we have to calculate it. There probably is something somewhere, since
|
|
NIS Elements can figure it out, but we haven't found it yet.
|
|
|
|
:rtype: list
|
|
|
|
"""
|
|
return self._parse_dimension(r""".*?XY\((\d+)\).*?""")
|
|
|
|
def _parse_frames(self):
|
|
"""
|
|
The number of cycles.
|
|
|
|
:rtype: list
|
|
|
|
"""
|
|
return self._parse_dimension(r""".*?T'?\((\d+)\).*?""")
|
|
|
|
def _parse_z_levels(self):
|
|
"""
|
|
The different levels in the Z-plane. Just a sequence from 0 to n.
|
|
|
|
:rtype: list
|
|
|
|
"""
|
|
return self._parse_dimension(r""".*?Z\((\d+)\).*?""")
|
|
|
|
def _parse_dimension_text(self):
|
|
"""
|
|
While there are metadata values that represent a lot of what we want to capture, they seem to be unreliable.
|
|
Sometimes certain elements don't exist, or change their data type randomly. However, the human-readable text
|
|
is always there and in the same exact format, so we just parse that instead.
|
|
|
|
:rtype: str
|
|
|
|
"""
|
|
dimension_text = six.b("")
|
|
textinfo = self.image_text_info[six.b('SLxImageTextInfo')].values()
|
|
|
|
for line in textinfo:
|
|
if six.b("Dimensions:") in line:
|
|
entries = line.split(six.b("\r\n"))
|
|
for entry in entries:
|
|
if entry.startswith(six.b("Dimensions:")):
|
|
return entry
|
|
|
|
return dimension_text
|
|
|
|
def _parse_dimension(self, pattern):
|
|
"""
|
|
:param pattern: a valid regex pattern
|
|
:type pattern: str
|
|
|
|
:rtype: list of int
|
|
|
|
"""
|
|
dimension_text = self._parse_dimension_text()
|
|
if six.PY3:
|
|
dimension_text = dimension_text.decode("utf8")
|
|
match = re.match(pattern, dimension_text)
|
|
if not match:
|
|
return [0]
|
|
count = int(match.group(1))
|
|
return list(range(count))
|
|
|
|
def _parse_total_images_per_channel(self):
|
|
"""
|
|
The total number of images per channel. Warning: this may be inaccurate as it includes "gap" images.
|
|
|
|
:rtype: int
|
|
|
|
"""
|
|
return self.image_attributes[six.b('SLxImageAttributes')][six.b('uiSequenceCount')]
|
|
|
|
def _parse_roi_metadata(self):
|
|
"""
|
|
Parse the raw ROI metadata.
|
|
:return:
|
|
"""
|
|
if self.roi_metadata is None or not six.b('RoiMetadata_v1') in self.roi_metadata:
|
|
return
|
|
|
|
raw_roi_data = self.roi_metadata[six.b('RoiMetadata_v1')]
|
|
|
|
number_of_rois = raw_roi_data[six.b('m_vectGlobal_Size')]
|
|
|
|
roi_objects = []
|
|
for i in range(number_of_rois):
|
|
current_roi = raw_roi_data[six.b('m_vectGlobal_%d' % i)]
|
|
roi_objects.append(self._parse_roi(current_roi))
|
|
|
|
self._metadata_parsed['rois'] = roi_objects
|
|
|
|
def _parse_roi(self, raw_roi_dict):
|
|
"""
|
|
Extract the vector animation parameters from the ROI.
|
|
This includes the position and size at the given timepoints.
|
|
:param raw_roi_dict:
|
|
:return:
|
|
"""
|
|
number_of_timepoints = raw_roi_dict[six.b('m_vectAnimParams_Size')]
|
|
|
|
roi_dict = {
|
|
"timepoints": [],
|
|
"positions": [],
|
|
"sizes": [],
|
|
"shape": self._parse_roi_shape(raw_roi_dict[six.b('m_sInfo')][six.b('m_uiShapeType')]),
|
|
"type": self._parse_roi_type(raw_roi_dict[six.b('m_sInfo')][six.b('m_uiInterpType')])
|
|
}
|
|
for i in range(number_of_timepoints):
|
|
roi_dict = self._parse_vect_anim(roi_dict, raw_roi_dict[six.b('m_vectAnimParams_%d') % i])
|
|
|
|
# convert to NumPy arrays
|
|
roi_dict["timepoints"] = np.array(roi_dict["timepoints"], dtype=np.float)
|
|
roi_dict["positions"] = np.array(roi_dict["positions"], dtype=np.float)
|
|
roi_dict["sizes"] = np.array(roi_dict["sizes"], dtype=np.float)
|
|
|
|
return roi_dict
|
|
|
|
@staticmethod
|
|
def _parse_roi_shape(shape):
|
|
if shape == 3:
|
|
return 'rectangle'
|
|
elif shape == 9:
|
|
return 'circle'
|
|
|
|
return None
|
|
|
|
@staticmethod
|
|
def _parse_roi_type(type_no):
|
|
if type_no == 4:
|
|
return 'stimulation'
|
|
elif type_no == 3:
|
|
return 'reference'
|
|
elif type_no == 2:
|
|
return 'background'
|
|
|
|
return None
|
|
|
|
def _parse_vect_anim(self, roi_dict, animation_dict):
|
|
"""
|
|
Parses a ROI vector animation object and adds it to the global list of timepoints and positions.
|
|
:param animation_dict:
|
|
:return:
|
|
"""
|
|
roi_dict["timepoints"].append(animation_dict[six.b('m_dTimeMs')])
|
|
|
|
image_width = self._metadata_parsed["width"] * self._metadata_parsed["pixel_microns"]
|
|
image_height = self._metadata_parsed["height"] * self._metadata_parsed["pixel_microns"]
|
|
|
|
# positions are taken from the center of the image as a fraction of the half width/height of the image
|
|
position = np.array((0.5 * image_width * (1 + animation_dict[six.b('m_dCenterX')]),
|
|
0.5 * image_height * (1 + animation_dict[six.b('m_dCenterY')]),
|
|
animation_dict[six.b('m_dCenterZ')]))
|
|
roi_dict["positions"].append(position)
|
|
|
|
size_dict = animation_dict[six.b('m_sBoxShape')]
|
|
|
|
# sizes are fractions of the half width/height of the image
|
|
roi_dict["sizes"].append((size_dict[six.b('m_dSizeX')] * 0.25 * image_width,
|
|
size_dict[six.b('m_dSizeY')] * 0.25 * image_height,
|
|
size_dict[six.b('m_dSizeZ')]))
|
|
return roi_dict
|
|
|
|
def _parse_experiment_metadata(self):
|
|
"""
|
|
Parse the metadata of the ND experiment
|
|
:return:
|
|
"""
|
|
if not six.b('SLxExperiment') in self.image_metadata:
|
|
return
|
|
|
|
raw_data = self.image_metadata[six.b('SLxExperiment')]
|
|
|
|
experimental_data = {
|
|
'description': 'unknown',
|
|
'loops': []
|
|
}
|
|
|
|
if six.b('wsApplicationDesc') in raw_data:
|
|
experimental_data['description'] = raw_data[six.b('wsApplicationDesc')].decode('utf8')
|
|
|
|
if six.b('uLoopPars') in raw_data:
|
|
experimental_data['loops'] = self._parse_loop_data(raw_data[six.b('uLoopPars')])
|
|
|
|
self._metadata_parsed['experiment'] = experimental_data
|
|
|
|
def _parse_loop_data(self, loop_data):
|
|
"""
|
|
Parse the experimental loop data
|
|
:param loop_data:
|
|
:return:
|
|
"""
|
|
if six.b('uiPeriodCount') not in loop_data or loop_data[six.b('uiPeriodCount')] == 0:
|
|
return []
|
|
|
|
if six.b('pPeriod') not in loop_data:
|
|
return []
|
|
|
|
# take the first dictionary element, it contains all loop data
|
|
loops = loop_data[six.b('pPeriod')][list(loop_data[six.b('pPeriod')].keys())[0]]
|
|
|
|
# take into account the absolute time in ms
|
|
time_offset = 0
|
|
|
|
parsed_loops = []
|
|
|
|
for loop in loops:
|
|
# duration of this loop
|
|
duration = loop[six.b('dDuration')]
|
|
|
|
# uiLoopType == 6 is a stimulation loop
|
|
is_stimulation = loop[six.b('uiLoopType')] == 6
|
|
|
|
# sampling interval in ms
|
|
interval = loop[six.b('dAvgPeriodDiff')]
|
|
|
|
parsed_loop = {
|
|
'start': time_offset,
|
|
'duration': duration,
|
|
'stimulation': is_stimulation,
|
|
'sampling_interval': interval
|
|
}
|
|
|
|
parsed_loops.append(parsed_loop)
|
|
|
|
# increase the time offset
|
|
time_offset += duration
|
|
|
|
return parsed_loops
|
|
|
|
@property
|
|
@ignore_missing
|
|
def image_text_info(self):
|
|
return read_metadata(read_chunk(self._fh, self._label_map.image_text_info), 1)
|
|
|
|
@property
|
|
@ignore_missing
|
|
def image_metadata_sequence(self):
|
|
return read_metadata(read_chunk(self._fh, self._label_map.image_metadata_sequence), 1)
|
|
|
|
@property
|
|
@ignore_missing
|
|
def image_calibration(self):
|
|
return read_metadata(read_chunk(self._fh, self._label_map.image_calibration), 1)
|
|
|
|
@property
|
|
@ignore_missing
|
|
def image_attributes(self):
|
|
return read_metadata(read_chunk(self._fh, self._label_map.image_attributes), 1)
|
|
|
|
@property
|
|
@ignore_missing
|
|
def x_data(self):
|
|
return read_array(self._fh, 'double', self._label_map.x_data)
|
|
|
|
@property
|
|
@ignore_missing
|
|
def y_data(self):
|
|
return read_array(self._fh, 'double', self._label_map.y_data)
|
|
|
|
@property
|
|
@ignore_missing
|
|
def z_data(self):
|
|
return read_array(self._fh, 'double', self._label_map.z_data)
|
|
|
|
@property
|
|
@ignore_missing
|
|
def roi_metadata(self):
|
|
return read_metadata(read_chunk(self._fh, self._label_map.roi_metadata), 1)
|
|
|
|
@property
|
|
@ignore_missing
|
|
def pfs_status(self):
|
|
return read_array(self._fh, 'int', self._label_map.pfs_status)
|
|
|
|
@property
|
|
@ignore_missing
|
|
def pfs_offset(self):
|
|
return read_array(self._fh, 'int', self._label_map.pfs_offset)
|
|
|
|
@property
|
|
@ignore_missing
|
|
def camera_exposure_time(self):
|
|
return read_array(self._fh, 'double', self._label_map.camera_exposure_time)
|
|
|
|
@property
|
|
@ignore_missing
|
|
def lut_data(self):
|
|
return xmltodict.parse(read_chunk(self._fh, self._label_map.lut_data))
|
|
|
|
@property
|
|
@ignore_missing
|
|
def grabber_settings(self):
|
|
return xmltodict.parse(read_chunk(self._fh, self._label_map.grabber_settings))
|
|
|
|
@property
|
|
@ignore_missing
|
|
def custom_data(self):
|
|
return xmltodict.parse(read_chunk(self._fh, self._label_map.custom_data))
|
|
|
|
@property
|
|
@ignore_missing
|
|
def app_info(self):
|
|
return xmltodict.parse(read_chunk(self._fh, self._label_map.app_info))
|
|
|
|
@property
|
|
@ignore_missing
|
|
def camera_temp(self):
|
|
camera_temp = read_array(self._fh, 'double', self._label_map.camera_temp)
|
|
if camera_temp:
|
|
for temp in map(lambda x: round(x * 100.0, 2), camera_temp):
|
|
yield temp
|
|
|
|
@property
|
|
@ignore_missing
|
|
def acquisition_times(self):
|
|
acquisition_times = read_array(self._fh, 'double', self._label_map.acquisition_times)
|
|
if acquisition_times:
|
|
for acquisition_time in map(lambda x: x / 1000.0, acquisition_times):
|
|
yield acquisition_time
|
|
|
|
@property
|
|
@ignore_missing
|
|
def image_metadata(self):
|
|
if self._label_map.image_metadata:
|
|
return read_metadata(read_chunk(self._fh, self._label_map.image_metadata), 1)
|