import numpy as np
import xarray as xr
def prepare_export_metadata(
dataset: xr.Dataset,
image_data=None,
dims: tuple[str, ...] | None = None,
*,
isel_dict: dict[str, int | slice] | None = None,
additional_vars: dict | None = None,
) -> xr.Dataset:
"""Prepare metadata for export by dropping spatial variables and attaching image data.
This helper consolidates the pattern used in stitch_frame.py and macroview.py
for preparing metadata before calling _save_tiff_with_metadata or _save_png_with_metadata.
Parameters
----------
dataset : xr.Dataset
The source dataset containing metadata.
image_data : array-like, optional
The rendered/stitched image data to attach. If None, 'images' is not added.
dims : tuple[str, ...], optional
Dimension names for the image data (e.g., ('y', 'x', 'rgb')).
Required if image_data is provided.
isel_dict : dict, optional
Dictionary of {dim: index} pairs to slice the dataset.
Used to collapse dimensions for scalar metadata.
additional_vars : dict, optional
Additional variables to add to the metadata (e.g., {'pixel_width': 1.0}).
Returns
-------
xr.Dataset
- Prepared metadata with 'images' attached and spatial variables dropped.
"""
vars_to_drop = []
for key, value in dataset.variables.items():
if 'y' in value.dims or 'x' in value.dims:
vars_to_drop.append(key)
coords_to_drop = tuple(c for c in ('y', 'x') if c in dataset.coords)
metadata = dataset.drop_vars(tuple(vars_to_drop) + coords_to_drop, errors='ignore')
if isel_dict:
metadata = metadata.isel(isel_dict)
if additional_vars:
for key, value in additional_vars.items():
metadata[key] = value
if image_data is not None:
if dims is None:
raise ValueError("dims must be provided when image_data is provided")
metadata['images'] = (dims, image_data)
return metadata
def get_latest_selection(dataset):
latest_acquisition_index = dataset.latest_acquisition_index.data
acquisition_index = dataset.acquisition_index.data
return acquisition_index == latest_acquisition_index
[docs]
def get_stack_dims(dataset, *, key='images'):
"""Can return one or more dimensions in which the stack was acquired."""
images_dims = dataset[key].dims
stack_dims = tuple(
dim
for dim in images_dims
if dim not in (
"image_y", "image_x",
"y", "x",
"rgb", "rgba",
)
)
return stack_dims
def get_stack_shape(dataset, *, key='images'):
stack_dims = get_stack_dims(dataset)
stack_shape = tuple(
dataset[key].shape[dataset.images.dims.index(dim)]
for dim in stack_dims
)
return stack_shape
[docs]
def get_stack_dimension(dataset):
stack_dims = get_stack_dims(dataset)
if len(stack_dims) == 0:
raise ValueError("Provided dataset does not contain a stack dimension.")
if len(stack_dims) > 1:
raise ValueError(
f"We found the potential stack dimensions {stack_dims} "
"but we do not support more than 1 stack dimension"
)
return stack_dims[0]
def get_dataset_stack_type(dataset):
if 'frame_number' not in dataset.images.dims and 'channel' not in dataset.images.dims:
raise ValueError("Dataset is not a stack does not contain a "
"frame_number or channel dimension.")
leading_dims = [
('z_stage', 'z_stack'),
('exposure', 'exposure_stack'),
]
for dim, dim_type in leading_dims:
if dim in dataset:
if len(dataset[dim].shape) == 3:
return dim_type
return 'timeseries_stack'
[docs]
def get_software_frame_rate(dataset, *,
frame_number_coordinate='frame_number',
):
"""Compute the average frame rate of the data acquired in the dataset.
Using the software timestamp, compute the frame rate of the acquired
dataset.
Parameters
----------
dataset: mcam_dataset
Dataset containing ``mcam_data`` and ``software_timestamp``.
frame_number_coordinate: str
Coordinate over which to compute the frame rate. The result is averaged
over all other coordinates.
Returns
-------
frame_rate: float
Average frame rate over all cameras in seconds.
"""
frame_time_delta = get_software_frame_time_difference(
dataset=dataset,
frame_number_coordinate=frame_number_coordinate
)
frame_rate = 1 / frame_time_delta
return frame_rate
[docs]
def get_software_frame_time_difference(dataset, *,
frame_number_coordinate='frame_number',
):
"""Compute the average time difference between subsequent frames.
Using the software timestamp, compute the frame rate of the acquired
dataset.
Parameters
----------
dataset: mcam_dataset
Dataset containing ``mcam_data`` and ``software_timestamp``.
frame_number_coordinate: str
Coordinate over which to compute the frame rate. The result is averaged
over all other coordinates.
Returns
-------
time_difference: float
Average time difference over all cameras in seconds between each
subsequent frame.
"""
if 'software_timestamp' not in dataset:
raise ValueError("dataset does not contain the software_timestamp key")
software_timestamp = dataset.software_timestamp
frame_number_axis = software_timestamp.get_axis_num(frame_number_coordinate)
n_frames = software_timestamp.shape[frame_number_axis]
if n_frames <= 1:
raise ValueError("Frame rate computation requires more than 1 frame to have been acquired.")
start_times = np.take(software_timestamp.data, 0, axis=frame_number_axis)
end_times = np.take(software_timestamp.data, -1, axis=frame_number_axis)
frame_time_delta = (end_times - start_times) / (n_frames - 1)
# convert to seconds
frame_time_delta = frame_time_delta / np.timedelta64(1, 's')
frame_time_delta = frame_time_delta.mean()
return frame_time_delta
[docs]
def get_bayer_pattern(dataset, default=None):
"""Return the bayer pattern of the underlying sensors.
Parameters
----------
dataset: mcam_data
A dataset containing MCAM data. The bayer pattern is expected to be found in
a key called ``bayer_pattern``.
default:
The default value in the case that the ``bayer_pattern`` key is not found
in the dataset.
Returns
-------
bayer_pattern: str
The bayer pattern as a string. Typical bayer patterns include
``"rggb", "bggr", "grbg", "gbrg"``.
"""
bayer_pattern = dataset.get('bayer_pattern', None)
if bayer_pattern is None:
return default
else:
values = bayer_pattern.values.flat
bayer_pattern = str(values[0])
if not (values == bayer_pattern).all():
raise ValueError(
"We only support a single bayer pattern for all cameras.")
return bayer_pattern
[docs]
def get_offset(dataset):
"""Return the offset of the dataset
Parameters
----------
dataset: mcam_data
A dataset containing MCAM data. The pixel information is expected
to be contained in the coordinates ``'y'`` and ``'x'``.
Returns
-------
bin_mode: tuple(int, int)
A integer for the offset in the y and x direction.
"""
y_offset = dataset.y.data[0]
x_offset = dataset.x.data[0]
return y_offset, x_offset
[docs]
def get_bin_mode(dataset):
"""Return the binning mode of the dataset
Parameters
----------
dataset: mcam_data
A dataset containing MCAM data. The pixel information is expected
to be contained in the coordinates ``'y'`` and ``'x'``.
Returns
-------
bin_mode: int
A single integer for the bin mode.
Note
----
The binning mode is assumed to be symmetric in both the row (``y``) and
column (``x``) dimension. This function uses the information in the
``y`` coordinate to extract the bin mode.
"""
# This likely triggers a load of the data for lazy loaded arrays
y = dataset.y.data
y_diff = y[1] - y[0]
# Explicitly cast to a python integer using the `int` method.
return int(abs(y_diff))
[docs]
def is_well_dataset(dataset):
return 'well_id' in dataset
[docs]
def using_field_id(dataset):
"""Check if a dataset is using field IDs (i.e., has non-zero field IDs).
Parameters
----------
dataset : xarray.Dataset
The dataset to check.
Returns
-------
bool
True if the dataset has field IDs and at least one non-zero field ID.
False otherwise.
"""
if 'field_id' not in dataset:
return False
return bool(dataset.field_id.max() > 0)
def is_square_well_dataset(dataset):
if not is_well_dataset(dataset):
return False
# assume if no tag for square wells that these are circle wells
if 'square_wells' not in dataset:
return False
return bool(dataset.square_wells)
[docs]
def get_chroma(dataset):
# This function aims to help alleviate some of the problems with
# https://gitlab.com/ramonaoptics/mcam/python-owl/-/issues/137
dims = dataset.images.dims
if "rgb" in dims:
return "rgb"
elif "rgba" in dims:
return "rgba"
elif "sensor_chroma" in dataset:
sensor_chroma = str(dataset["sensor_chroma"].data)
if sensor_chroma == "monochrome":
return "monochrome"
elif "bayer_pattern" in dataset:
return "bayer"
else:
return "monochrome"
elif 'bayer_pattern' not in dataset:
return "monochrome"
else:
# By "default" we assume the images are from bayer patterned
# sensors unless specified otherwise.
return "bayer"
[docs]
def is_stack_dataset(dataset):
return len(get_stack_dims(dataset)) > 0
[docs]
def get_groupings(dataset, key, trim_invalid_data=True):
"""Extracts groupings of images based on a given key from a dataset.
Parameters
----------
dataset: xarray
The MCAM dataset containing image data.
key: str
The key in the dataset to use for grouping images.
trim_invalid_data: bool
Flag to indicate whether to exclude invalid data when generating
groupings.
Returns
-------
groupings: dict
A dictionary where keys are group identifiers and values are lists
of tuples containing image indices within the dataset.
"""
if key not in dataset:
raise ValueError(f"Dataset does not contain the key {key}")
ids = dataset[key].data
groupings = {}
if trim_invalid_data and 'valid_data' in dataset:
valid_data = dataset.valid_data.compute().data
for group_id in np.unique(ids):
group_wells = ids == group_id
if trim_invalid_data and 'valid_data' in dataset:
group_wells = np.logical_and(group_wells, valid_data)
if not np.any(group_wells):
continue
image_indices = np.argwhere(group_wells)
groupings[group_id] = [tuple(index) for index in image_indices.tolist()]
return groupings
def get_group_shape(groupings):
# helper function to get the shape of each individual group after using 'get_groupings'
# assumes all groups in the dictionary have the same shape
any_group = next(iter(groupings.values()))
coords = np.array(any_group)
rows = coords[:, 0]
cols = coords[:, 1]
group_shape = (rows.max() - rows.min() + 1, cols.max() - cols.min() + 1)
return group_shape
[docs]
def group_data(dataset, key, groupings, func=np.mean):
"""This function groups data from a dataset based on specified
groupings and calculates a summary statistic for each group.
Parameters
----------
dataset: xarray
The MCAM dataset containing the data to be grouped.
key: str
The key column in the dataset based on which the data will be grouped.
groupings: dict
A dictionary where the keys represent group IDs and the values are
lists of image indices for each group.
func: function
The function to apply to the grouped data to calculate the
summary statistic.
Returns
-------
grouping_values: dict
A dictionary where the keys are group IDs and the values are the
calculated summary statistics for each group.
"""
if key not in dataset:
raise ValueError(f"Dataset does not contain the key {key}")
key_data = dataset[key].data
return group_data_array(key_data, groupings, func)
def group_data_array(data_array, groupings, func):
dtype = data_array.dtype
grouping_values = {}
for group_id, indices in groupings.items():
np_indices = [i[0] for i in indices], [i[1] for i in indices]
group_data = data_array[np_indices]
grouping_values[group_id] = np.asarray(func(group_data), dtype)
return grouping_values
def get_pixel_width(dataset):
if 'pixel_width' not in dataset:
return None
pixel_width = float(np.asarray(dataset.pixel_width).flat[0])
if pixel_width == 0.0 or np.isnan(pixel_width):
return None
return pixel_width
def get_binned_pixel_width(dataset):
pixel_width = get_pixel_width(dataset)
if pixel_width is None:
return None
binning = dataset.y.data[1] - dataset.y.data[0]
return pixel_width * binning
[docs]
def get_analysis_outputs(dataset):
if 'analysis' not in dataset.attrs:
return None
analysis_key_outputs = dataset.attrs['analysis']
if isinstance(analysis_key_outputs, str):
analysis_key_outputs = [analysis_key_outputs]
analysis_outputs = {}
for key_output in analysis_key_outputs:
if key_output not in dataset:
continue
outputs = dataset[key_output].attrs.get('additional_outputs', [])
if isinstance(outputs, str):
outputs = [outputs]
curr_analysis_outputs = {}
for output in outputs:
# check to ensure this output exists in the dataset in the correct format
if output not in dataset:
continue
if '__owl_settings_type__' not in dataset[output].attrs:
continue
output_type = dataset[output].attrs['__owl_settings_type__']
if output_type not in curr_analysis_outputs:
curr_analysis_outputs[output_type] = {}
curr_analysis_outputs[output_type][output] = {}
for attr in dataset[output].attrs:
curr_analysis_outputs[output_type][output][attr] = dataset[output].attrs[attr]
analysis_outputs[key_output] = curr_analysis_outputs
return analysis_outputs