_registration_utils

Utils functions for registration

add_zero_translation_columns(ad_table)

Add three zero-filled columns (translation_{x,y,z}) to an AnnData table.

Source code in fractal_tasks_core/tasks/_registration_utils.py

def add_zero_translation_columns(ad_table: ad.AnnData):
    """
    Add three zero-filled columns (`translation_{x,y,z}`) to an AnnData table.
    """
    columns = ["translation_z", "translation_y", "translation_x"]
    if ad_table.var.index.isin(columns).any().any():
        raise ValueError(
            "The roi table already contains translation columns. Did you "
            "enter a wrong reference acquisition?"
        )
    df = pd.DataFrame(np.zeros([len(ad_table), 3]), columns=columns)
    df.index = ad_table.obs.index
    ad_new = ad.concat([ad_table, ad.AnnData(df)], axis=1)
    return ad_new

apply_registration_to_single_ROI_table(roi_table, max_df, min_df)

Applies the registration to a ROI table

Calculates the new position as: p = position + max(shift, 0) - own_shift Calculates the new len as: l = len - max(shift, 0) + min(shift, 0)

PARAMETER	DESCRIPTION
roi_table	AnnData table which contains a Fractal ROI table. Rows are ROIs TYPE: AnnData
max_df	Max translation shift in z, y, x for each ROI. Rows are ROIs, columns are translation_z, translation_y, translation_x TYPE: DataFrame
min_df	Min translation shift in z, y, x for each ROI. Rows are ROIs, columns are translation_z, translation_y, translation_x TYPE: DataFrame

Returns: ROI table where all ROIs are registered to the smallest common area across all acquisitions.

Source code in fractal_tasks_core/tasks/_registration_utils.py

def apply_registration_to_single_ROI_table(
    roi_table: ad.AnnData,
    max_df: pd.DataFrame,
    min_df: pd.DataFrame,
) -> ad.AnnData:
    """
    Applies the registration to a ROI table

    Calculates the new position as: p = position + max(shift, 0) - own_shift
    Calculates the new len as: l = len - max(shift, 0) + min(shift, 0)

    Args:
        roi_table: AnnData table which contains a Fractal ROI table.
            Rows are ROIs
        max_df: Max translation shift in z, y, x for each ROI. Rows are ROIs,
            columns are translation_z, translation_y, translation_x
        min_df: Min translation shift in z, y, x for each ROI. Rows are ROIs,
            columns are translation_z, translation_y, translation_x
    Returns:
        ROI table where all ROIs are registered to the smallest common area
        across all acquisitions.
    """
    roi_table = copy.deepcopy(roi_table)
    rois = roi_table.obs.index
    if (rois != max_df.index).all() or (rois != min_df.index).all():
        raise ValueError(
            "ROI table and max & min translation need to contain the same "
            f"ROIS, but they were {rois=}, {max_df.index=}, {min_df.index=}"
        )

    for roi in rois:
        roi_table[[roi], ["z_micrometer"]] = (
            roi_table[[roi], ["z_micrometer"]].X
            + float(max_df.loc[roi, "translation_z"])
            - roi_table[[roi], ["translation_z"]].X
        )
        roi_table[[roi], ["y_micrometer"]] = (
            roi_table[[roi], ["y_micrometer"]].X
            + float(max_df.loc[roi, "translation_y"])
            - roi_table[[roi], ["translation_y"]].X
        )
        roi_table[[roi], ["x_micrometer"]] = (
            roi_table[[roi], ["x_micrometer"]].X
            + float(max_df.loc[roi, "translation_x"])
            - roi_table[[roi], ["translation_x"]].X
        )
        # This calculation only works if all ROIs are the same size initially!
        roi_table[[roi], ["len_z_micrometer"]] = (
            roi_table[[roi], ["len_z_micrometer"]].X
            - float(max_df.loc[roi, "translation_z"])
            + float(min_df.loc[roi, "translation_z"])
        )
        roi_table[[roi], ["len_y_micrometer"]] = (
            roi_table[[roi], ["len_y_micrometer"]].X
            - float(max_df.loc[roi, "translation_y"])
            + float(min_df.loc[roi, "translation_y"])
        )
        roi_table[[roi], ["len_x_micrometer"]] = (
            roi_table[[roi], ["len_x_micrometer"]].X
            - float(max_df.loc[roi, "translation_x"])
            + float(min_df.loc[roi, "translation_x"])
        )
    return roi_table

calculate_physical_shifts(shifts, level, coarsening_xy, full_res_pxl_sizes_zyx)

Calculates shifts in physical units based on pixel shifts

PARAMETER	DESCRIPTION
shifts	array of shifts, zyx or yx TYPE: array
level	resolution level TYPE: int
coarsening_xy	coarsening factor between levels TYPE: int
full_res_pxl_sizes_zyx	pixel sizes in physical units as zyx TYPE: list[float]

RETURNS	DESCRIPTION
list[float]	shifts in physical units as zyx

Source code in fractal_tasks_core/tasks/_registration_utils.py

def calculate_physical_shifts(
    shifts: np.array,
    level: int,
    coarsening_xy: int,
    full_res_pxl_sizes_zyx: list[float],
) -> list[float]:
    """
    Calculates shifts in physical units based on pixel shifts

    Args:
        shifts: array of shifts, zyx or yx
        level: resolution level
        coarsening_xy: coarsening factor between levels
        full_res_pxl_sizes_zyx: pixel sizes in physical units as zyx

    Returns:
        shifts in physical units as zyx
    """

    curr_pixel_size = np.array(full_res_pxl_sizes_zyx) * coarsening_xy**level
    if len(shifts) == 3:
        shifts_physical = shifts * curr_pixel_size
    elif len(shifts) == 2:
        shifts_physical = [
            0,
            shifts[0] * curr_pixel_size[1],
            shifts[1] * curr_pixel_size[2],
        ]
    else:
        raise ValueError(
            f"Wrong input for calculate_physical_shifts ({shifts=})"
        )
    return shifts_physical

chi2_shift_out(img_ref, img_cycle_x)

Helper function to get the output of chi2_shift into the same format as phase_cross_correlation. Calculates the shift between two images using the chi2_shift method.

PARAMETER	DESCRIPTION
img_ref	First image. TYPE: ndarray
img_cycle_x	Second image. TYPE: ndarray

RETURNS	DESCRIPTION
list[ndarray]	List containing numpy array of shift in y and x direction.

Source code in fractal_tasks_core/tasks/_registration_utils.py

def chi2_shift_out(img_ref, img_cycle_x) -> list[np.ndarray]:
    """
    Helper function to get the output of chi2_shift into the same format as
    phase_cross_correlation. Calculates the shift between two images using
    the chi2_shift method.

    Args:
        img_ref (np.ndarray): First image.
        img_cycle_x (np.ndarray): Second image.

    Returns:
        List containing numpy array of shift in y and x direction.
    """
    x, y, a, b = chi2_shift(np.squeeze(img_ref), np.squeeze(img_cycle_x))

    """
    Running into issues when using direct float output for fractal.
    When rounding to integer and using integer dtype, it typically works
    but for some reasons fails when run over a whole 384 well plate (but
    the well where it fails works fine when run alone). For now, rounding
    to integer, but still using float64 dtype (like the scikit-image
    phase cross correlation function) seems to be the safest option.
    """
    shifts = np.array([-np.round(y), -np.round(x)], dtype="float64")
    # return as a list to adhere to the phase_cross_correlation output format
    return [shifts]

get_ROI_table_with_translation(ROI_table, new_shifts)

Adds translation columns to a ROI table

PARAMETER	DESCRIPTION
ROI_table	Fractal ROI table TYPE: AnnData
new_shifts	zyx list of shifts TYPE: dict[str, list[float]]

RETURNS	DESCRIPTION
AnnData	Fractal ROI table with 3 additional columns for calculated translations

Source code in fractal_tasks_core/tasks/_registration_utils.py

def get_ROI_table_with_translation(
    ROI_table: ad.AnnData,
    new_shifts: dict[str, list[float]],
) -> ad.AnnData:
    """
    Adds translation columns to a ROI table

    Args:
        ROI_table: Fractal ROI table
        new_shifts: zyx list of shifts

    Returns:
        Fractal ROI table with 3 additional columns for calculated translations
    """

    shift_table = pd.DataFrame(new_shifts).T
    shift_table.columns = ["translation_z", "translation_y", "translation_x"]
    shift_table = shift_table.rename_axis("FieldIndex")
    new_roi_table = ROI_table.to_df().merge(
        shift_table, left_index=True, right_index=True
    )
    if len(new_roi_table) != len(ROI_table):
        raise ValueError(
            "New ROI table with registration info has a "
            f"different length ({len(new_roi_table)=}) "
            f"from the original ROI table ({len(ROI_table)=})"
        )

    adata = ad.AnnData(X=new_roi_table.astype(np.float32))
    adata.obs_names = new_roi_table.index
    adata.var_names = list(map(str, new_roi_table.columns))
    return adata

is_3D(dask_array)

Check if a dask array is 3D.

Treats singelton Z dimensions as 2D images. (1, 2000, 2000) => False (10, 2000, 2000) => True

PARAMETER	DESCRIPTION
dask_array	Input array to be checked TYPE: array

RETURNS	DESCRIPTION
bool	bool on whether the array is 3D

Source code in fractal_tasks_core/tasks/_registration_utils.py

def is_3D(dask_array: da.array) -> bool:
    """
    Check if a dask array is 3D.

    Treats singelton Z dimensions as 2D images.
    (1, 2000, 2000) => False
    (10, 2000, 2000) => True

    Args:
        dask_array: Input array to be checked

    Returns:
        bool on whether the array is 3D
    """
    if len(dask_array.shape) == 3 and dask_array.shape[0] > 1:
        return True
    else:
        return False