Skip to content

roi

Subpackage for ROI-related functions.

apply_shift_in_one_direction(tmp_df_well, line_1, line_2, mu, tol=1e-10)

TBD

PARAMETER DESCRIPTION
tmp_df_well

TBD

TYPE: DataFrame

line_1

TBD

TYPE: Sequence[float]

line_2

TBD

TYPE: Sequence[float]

mu

TBD

TYPE: str

tol

TBD

TYPE: float DEFAULT: 1e-10

Source code in fractal_tasks_core/roi/v1_overlaps.py
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
def apply_shift_in_one_direction(
    tmp_df_well: pd.DataFrame,
    line_1: Sequence[float],
    line_2: Sequence[float],
    mu: str,
    tol: float = 1e-10,
):
    """
    TBD

    Args:
        tmp_df_well: TBD
        line_1: TBD
        line_2: TBD
        mu: TBD
        tol: TBD
    """
    min_1, max_1 = line_1[:]
    min_2, max_2 = line_2[:]
    min_max = min(max_1, max_2)
    max_min = max(min_1, min_2)
    shift = min_max - max_min
    logging.debug(f"{mu}-shifting by {shift=}")
    ind = tmp_df_well.loc[:, f"{mu}min"] >= max_min - tol
    if not (shift > 0.0 and ind.to_numpy().max() > 0):
        raise ValueError(
            "Something wrong in apply_shift_in_one_direction\n"
            f"{mu=}\n{shift=}\n{ind.to_numpy()=}"
        )
    tmp_df_well.loc[ind, f"{mu}min"] += shift
    tmp_df_well.loc[ind, f"{mu}max"] += shift
    tmp_df_well.loc[ind, f"{mu}_micrometer"] += shift
    return tmp_df_well

are_ROI_table_columns_valid(*, table)

Verify some validity assumptions on a ROI table.

This function reflects our current working assumptions (e.g. the presence of some specific columns); this may change in future versions.

PARAMETER DESCRIPTION
table

AnnData table to be checked

TYPE: AnnData

Source code in fractal_tasks_core/roi/v1_checks.py
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
def are_ROI_table_columns_valid(*, table: ad.AnnData) -> None:
    """
    Verify some validity assumptions on a ROI table.

    This function reflects our current working assumptions (e.g. the presence
    of some specific columns); this may change in future versions.

    Args:
        table: AnnData table to be checked
    """

    # Hard constraint: table columns must include some expected ones
    columns = [
        "x_micrometer",
        "y_micrometer",
        "z_micrometer",
        "len_x_micrometer",
        "len_y_micrometer",
        "len_z_micrometer",
    ]
    for column in columns:
        if column not in table.var_names:
            raise ValueError(f"Column {column} is not present in ROI table")

array_to_bounding_box_table(mask_array, pxl_sizes_zyx, origin_zyx=(0, 0, 0))

Construct bounding-box ROI table for a mask array.

PARAMETER DESCRIPTION
mask_array

Original array to construct bounding boxes.

TYPE: ndarray

pxl_sizes_zyx

Physical-unit pixel ZYX sizes.

TYPE: list[float]

origin_zyx

Shift ROI origin by this amount of ZYX pixels.

TYPE: tuple[int, int, int] DEFAULT: (0, 0, 0)

RETURNS DESCRIPTION
DataFrame

DataFrame with each line representing the bounding-box ROI that corresponds to a unique value of mask_array. ROI properties are expressed in physical units (with columns defined as elsewhere this module - see e.g. prepare_well_ROI_table), and positions are optionally shifted (if origin_zyx is set). An additional column label keeps track of the mask_array value corresponding to each ROI.

Source code in fractal_tasks_core/roi/v1.py
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
def array_to_bounding_box_table(
    mask_array: np.ndarray,
    pxl_sizes_zyx: list[float],
    origin_zyx: tuple[int, int, int] = (0, 0, 0),
) -> pd.DataFrame:
    """
    Construct bounding-box ROI table for a mask array.

    Args:
        mask_array: Original array to construct bounding boxes.
        pxl_sizes_zyx: Physical-unit pixel ZYX sizes.
        origin_zyx: Shift ROI origin by this amount of ZYX pixels.

    Returns:
        DataFrame with each line representing the bounding-box ROI that
            corresponds to a unique value of `mask_array`. ROI properties are
            expressed in physical units (with columns defined as elsewhere this
            module - see e.g. `prepare_well_ROI_table`), and positions are
            optionally shifted (if `origin_zyx` is set). An additional column
            `label` keeps track of the `mask_array` value corresponding to each
            ROI.
    """

    pxl_sizes_zyx_array = np.array(pxl_sizes_zyx)
    z_origin, y_origin, x_origin = origin_zyx[:]

    labels = np.unique(mask_array)
    labels = labels[labels > 0]
    elem_list = []
    for label in labels:
        # Compute bounding box
        label_match = np.where(mask_array == label)
        zmin, ymin, xmin = np.min(label_match, axis=1) * pxl_sizes_zyx_array
        zmax, ymax, xmax = (
            np.max(label_match, axis=1) + 1
        ) * pxl_sizes_zyx_array

        # Compute bounding-box edges
        length_x = xmax - xmin
        length_y = ymax - ymin
        length_z = zmax - zmin

        # Shift origin
        zmin += z_origin * pxl_sizes_zyx[0]
        ymin += y_origin * pxl_sizes_zyx[1]
        xmin += x_origin * pxl_sizes_zyx[2]

        elem_list.append((xmin, ymin, zmin, length_x, length_y, length_z))

    df_columns = [
        "x_micrometer",
        "y_micrometer",
        "z_micrometer",
        "len_x_micrometer",
        "len_y_micrometer",
        "len_z_micrometer",
    ]

    if len(elem_list) == 0:
        df = pd.DataFrame(columns=[x for x in df_columns] + ["label"])
    else:
        df = pd.DataFrame(np.array(elem_list), columns=df_columns)
        df["label"] = labels

    return df

check_valid_ROI_indices(list_indices, ROI_table_name)

Check that list of indices has zero origin on each axis.

See fractal-tasks-core issues #530 and #554.

This helper function is meant to provide informative error messages when ROI tables created with fractal-tasks-core up to v0.11 are used in v0.12. This function will be deprecated and removed as soon as the v0.11/v0.12 transition advances.

Note that only FOV_ROI_table and well_ROI_table have to fulfill this constraint, while ROI tables obtained through segmentation may have arbitrary (non-negative) indices.

PARAMETER DESCRIPTION
list_indices

Output of convert_ROI_table_to_indices; each item is like [start_z, end_z, start_y, end_y, start_x, end_x].

TYPE: list[list[int]]

ROI_table_name

Name of the ROI table.

TYPE: str

RAISES DESCRIPTION
ValueError

If the table name is FOV_ROI_table or well_ROI_table and the minimum value of start_x, start_y and start_z are not all zero.

Source code in fractal_tasks_core/roi/v1_checks.py
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
def check_valid_ROI_indices(
    list_indices: list[list[int]],
    ROI_table_name: str,
) -> None:
    """
    Check that list of indices has zero origin on each axis.

    See fractal-tasks-core issues #530 and #554.

    This helper function is meant to provide informative error messages when
    ROI tables created with fractal-tasks-core up to v0.11 are used in v0.12.
    This function will be deprecated and removed as soon as the v0.11/v0.12
    transition advances.

    Note that only `FOV_ROI_table` and `well_ROI_table` have to fulfill this
    constraint, while ROI tables obtained through segmentation may have
    arbitrary (non-negative) indices.

    Args:
        list_indices:
            Output of `convert_ROI_table_to_indices`; each item is like
            `[start_z, end_z, start_y, end_y, start_x, end_x]`.
        ROI_table_name: Name of the ROI table.

    Raises:
        ValueError:
            If the table name is `FOV_ROI_table` or `well_ROI_table` and the
                minimum value of `start_x`, `start_y` and `start_z` are not all
                zero.
    """
    if ROI_table_name not in ["FOV_ROI_table", "well_ROI_table"]:
        # This validation function only applies to the FOV/well ROI tables
        # generated with fractal-tasks-core
        return

    # Find minimum index along ZYX
    min_start_z = min(item[0] for item in list_indices)
    min_start_y = min(item[2] for item in list_indices)
    min_start_x = min(item[4] for item in list_indices)

    # Check that minimum indices are all zero
    for ind, min_index in enumerate((min_start_z, min_start_y, min_start_x)):
        if min_index != 0:
            axis = ["Z", "Y", "X"][ind]
            raise ValueError(
                f"{axis} component of ROI indices for table `{ROI_table_name}`"
                f" do not start with 0, but with {min_index}.\n"
                "Hint: As of fractal-tasks-core v0.12, FOV/well ROI "
                "tables with non-zero origins (e.g. the ones created with "
                "v0.11) are not supported."
            )

check_well_for_FOV_overlap(site_metadata, selected_well, plotting_function, tol=1e-10)

This function is currently only used in tests and examples.

The plotting_function parameter is exposed so that other tools (see examples in this repository) may use it to show the FOV ROIs.

PARAMETER DESCRIPTION
site_metadata

TBD

TYPE: DataFrame

selected_well

TBD

TYPE: str

plotting_function

TBD

TYPE: Callable

tol

TBD

TYPE: float DEFAULT: 1e-10

Source code in fractal_tasks_core/roi/v1_overlaps.py
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
def check_well_for_FOV_overlap(
    site_metadata: pd.DataFrame,
    selected_well: str,
    plotting_function: Callable,
    tol: float = 1e-10,
):
    """
    This function is currently only used in tests and examples.

    The `plotting_function` parameter is exposed so that other tools (see
    examples in this repository) may use it to show the FOV ROIs.

    Args:
        site_metadata: TBD
        selected_well: TBD
        plotting_function: TBD
        tol: TBD
    """

    df = site_metadata.loc[selected_well].copy()
    df["xmin"] = df["x_micrometer"]
    df["ymin"] = df["y_micrometer"]
    df["xmax"] = df["x_micrometer"] + df["pixel_size_x"] * df["x_pixel"]
    df["ymax"] = df["y_micrometer"] + df["pixel_size_y"] * df["y_pixel"]

    xmin = list(df.loc[:, "xmin"])
    ymin = list(df.loc[:, "ymin"])
    xmax = list(df.loc[:, "xmax"])
    ymax = list(df.loc[:, "ymax"])
    num_lines = len(xmin)

    list_overlapping_FOVs = []
    for line_1 in range(num_lines):
        min_x_1, max_x_1 = [a[line_1] for a in [xmin, xmax]]
        min_y_1, max_y_1 = [a[line_1] for a in [ymin, ymax]]
        for line_2 in range(line_1):
            min_x_2, max_x_2 = [a[line_2] for a in [xmin, xmax]]
            min_y_2, max_y_2 = [a[line_2] for a in [ymin, ymax]]
            overlap = is_overlapping_2D(
                (min_x_1, min_y_1, max_x_1, max_y_1),
                (min_x_2, min_y_2, max_x_2, max_y_2),
                tol=tol,
            )
            if overlap:
                list_overlapping_FOVs.append(line_1)
                list_overlapping_FOVs.append(line_2)

    # Call plotting_function
    plotting_function(
        xmin, xmax, ymin, ymax, list_overlapping_FOVs, selected_well
    )

    if len(list_overlapping_FOVs) > 0:
        # Increase values by one to switch from index to the label plotted
        return {selected_well: [x + 1 for x in list_overlapping_FOVs]}

convert_ROI_table_to_indices(ROI, full_res_pxl_sizes_zyx, level=0, coarsening_xy=2, cols_xyz_pos=['x_micrometer', 'y_micrometer', 'z_micrometer'], cols_xyz_len=['len_x_micrometer', 'len_y_micrometer', 'len_z_micrometer'])

Convert a ROI AnnData table into integer array indices.

PARAMETER DESCRIPTION
ROI

AnnData table with list of ROIs.

TYPE: AnnData

full_res_pxl_sizes_zyx

Physical-unit pixel ZYX sizes at the full-resolution pyramid level.

TYPE: Sequence[float]

level

Pyramid level.

TYPE: int DEFAULT: 0

coarsening_xy

Linear coarsening factor in the YX plane.

TYPE: int DEFAULT: 2

cols_xyz_pos

Column names for XYZ ROI positions.

TYPE: Sequence[str] DEFAULT: ['x_micrometer', 'y_micrometer', 'z_micrometer']

cols_xyz_len

Column names for XYZ ROI edges.

TYPE: Sequence[str] DEFAULT: ['len_x_micrometer', 'len_y_micrometer', 'len_z_micrometer']

RAISES DESCRIPTION
ValueError

If any of the array indices is negative.

RETURNS DESCRIPTION
list[list[int]]

Nested list of indices. The main list has one item per ROI. Each ROI item is a list of six integers as in [start_z, end_z, start_y, end_y, start_x, end_x]. The array-index interval for a given ROI is start_x:end_x along X, and so on for Y and Z.

Source code in fractal_tasks_core/roi/v1.py
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
def convert_ROI_table_to_indices(
    ROI: ad.AnnData,
    full_res_pxl_sizes_zyx: Sequence[float],
    level: int = 0,
    coarsening_xy: int = 2,
    cols_xyz_pos: Sequence[str] = [
        "x_micrometer",
        "y_micrometer",
        "z_micrometer",
    ],
    cols_xyz_len: Sequence[str] = [
        "len_x_micrometer",
        "len_y_micrometer",
        "len_z_micrometer",
    ],
) -> list[list[int]]:
    """
    Convert a ROI AnnData table into integer array indices.

    Args:
        ROI: AnnData table with list of ROIs.
        full_res_pxl_sizes_zyx:
            Physical-unit pixel ZYX sizes at the full-resolution pyramid level.
        level: Pyramid level.
        coarsening_xy: Linear coarsening factor in the YX plane.
        cols_xyz_pos: Column names for XYZ ROI positions.
        cols_xyz_len: Column names for XYZ ROI edges.

    Raises:
        ValueError:
            If any of the array indices is negative.

    Returns:
        Nested list of indices. The main list has one item per ROI. Each ROI
            item is a list of six integers as in `[start_z, end_z, start_y,
            end_y, start_x, end_x]`. The array-index interval for a given ROI
            is `start_x:end_x` along X, and so on for Y and Z.
    """
    # Handle empty ROI table
    if len(ROI) == 0:
        return []

    # Set pyramid-level pixel sizes
    pxl_size_z, pxl_size_y, pxl_size_x = full_res_pxl_sizes_zyx
    prefactor = coarsening_xy**level
    pxl_size_x *= prefactor
    pxl_size_y *= prefactor

    x_pos, y_pos, z_pos = cols_xyz_pos[:]
    x_len, y_len, z_len = cols_xyz_len[:]

    list_indices = []
    for ROI_name in ROI.obs_names:
        # Extract data from anndata table
        x_micrometer = ROI[ROI_name, x_pos].X[0, 0]
        y_micrometer = ROI[ROI_name, y_pos].X[0, 0]
        z_micrometer = ROI[ROI_name, z_pos].X[0, 0]
        len_x_micrometer = ROI[ROI_name, x_len].X[0, 0]
        len_y_micrometer = ROI[ROI_name, y_len].X[0, 0]
        len_z_micrometer = ROI[ROI_name, z_len].X[0, 0]

        # Identify indices along the three dimensions
        start_x = x_micrometer / pxl_size_x
        end_x = (x_micrometer + len_x_micrometer) / pxl_size_x
        start_y = y_micrometer / pxl_size_y
        end_y = (y_micrometer + len_y_micrometer) / pxl_size_y
        start_z = z_micrometer / pxl_size_z
        end_z = (z_micrometer + len_z_micrometer) / pxl_size_z
        indices = [start_z, end_z, start_y, end_y, start_x, end_x]

        # Round indices to lower integer
        indices = list(map(round, indices))

        # Fail for negative indices
        if min(indices) < 0:
            raise ValueError(
                f"ROI {ROI_name} converted into negative array indices.\n"
                f"ZYX position: {z_micrometer}, {y_micrometer}, "
                f"{x_micrometer}\n"
                f"ZYX pixel sizes: {pxl_size_z}, {pxl_size_y}, "
                f"{pxl_size_x} ({level=})\n"
                "Hint: As of fractal-tasks-core v0.12, FOV/well ROI "
                "tables with non-zero origins (e.g. the ones created with "
                "v0.11) are not supported."
            )

        # Append ROI indices to to list
        list_indices.append(indices[:])

    return list_indices

convert_ROIs_from_3D_to_2D(adata, pixel_size_z)

TBD

Note that this function is only relevant when the ROIs in adata span the whole extent of the Z axis. TODO: check this explicitly.

PARAMETER DESCRIPTION
adata

TBD

TYPE: AnnData

pixel_size_z

TBD

TYPE: float

Source code in fractal_tasks_core/roi/v1.py
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
def convert_ROIs_from_3D_to_2D(
    adata: ad.AnnData,
    pixel_size_z: float,
) -> ad.AnnData:
    """
    TBD

    Note that this function is only relevant when the ROIs in adata span the
    whole extent of the Z axis.
    TODO: check this explicitly.

    Args:
        adata: TBD
        pixel_size_z: TBD
    """

    # Compress a 3D stack of images to a single Z plane,
    # with thickness equal to pixel_size_z
    df = adata.to_df()
    df["len_z_micrometer"] = pixel_size_z

    # Assign dtype explicitly, to avoid
    # >> UserWarning: X converted to numpy array with dtype float64
    # when creating AnnData object
    df = df.astype(np.float32)

    # Create an AnnData object directly from the DataFrame
    new_adata = ad.AnnData(X=df)

    # Rename rows and columns
    new_adata.obs_names = adata.obs_names
    new_adata.var_names = list(map(str, df.columns))

    return new_adata

convert_indices_to_regions(index)

Converts index tuples to region tuple

PARAMETER DESCRIPTION
index

Tuple containing 6 entries of (z_start, z_end, y_start, y_end, x_start, x_end).

TYPE: list[int]

RETURNS DESCRIPTION
region

tuple of three slices (ZYX)

TYPE: tuple[slice, slice, slice]

Source code in fractal_tasks_core/roi/v1.py
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
def convert_indices_to_regions(
    index: list[int],
) -> tuple[slice, slice, slice]:
    """
    Converts index tuples to region tuple

    Args:
        index: Tuple containing 6 entries of (z_start, z_end, y_start,
            y_end, x_start, x_end).

    Returns:
        region: tuple of three slices (ZYX)
    """
    return (
        slice(index[0], index[1]),
        slice(index[2], index[3]),
        slice(index[4], index[5]),
    )

create_roi_table_from_df_list(bbox_dataframe_list)

Creates an AnnData ROI table from a list of bounding-box tables

This function handles empty bbox lists, ensures that it has unique entries per label (and to address #810, it handles duplicate labels by only keeping the first entry for each label) & converts it to an AnnData table with a label column in obs.

PARAMETER DESCRIPTION
bbox_dataframe_list

List of bounding box dataframes. All dataframes are expected to have the same columns and they usually are: x_micrometer, y_micrometer, z_micrometer, len_x_micrometer, len_y_micrometer, len_z_micrometer, label. The label column is required.

TYPE: list[DataFrame]

RETURNS DESCRIPTION
AnnData

An AnnData table with all the ROIs.

Source code in fractal_tasks_core/roi/v1.py
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
def create_roi_table_from_df_list(
    bbox_dataframe_list: list[pd.DataFrame],
) -> ad.AnnData:
    """
    Creates an AnnData ROI table from a list of bounding-box tables

    This function handles empty bbox lists, ensures that it has unique entries
    per label (and to address #810, it handles duplicate labels by only
    keeping the first entry for each label) & converts it to an AnnData table
    with a label column in obs.

    Args:
        bbox_dataframe_list: List of bounding box dataframes. All dataframes
            are expected to have the same columns and they usually are:
            x_micrometer, y_micrometer, z_micrometer, len_x_micrometer,
            len_y_micrometer, len_z_micrometer, label. The label column is
            required.

    Returns:
        An `AnnData` table with all the ROIs.
    """
    # Handle the case where `bbox_dataframe_list` is empty (typically
    # because list_indices is also empty)
    if len(bbox_dataframe_list) == 0:
        bbox_dataframe_list = [empty_bounding_box_table()]
    # Concatenate all ROI dataframes
    df_well = pd.concat(bbox_dataframe_list, axis=0, ignore_index=True)

    # Drop duplicates based on the 'label' column, keeping only the first
    # occurrence (see #810 for details)
    df_well = df_well.drop_duplicates(subset="label", keep="first")

    # Extract labels and drop them from df_well
    labels = pd.DataFrame(df_well["label"].astype(str)).reset_index(drop=True)

    # Check that there are only unique labels. Should be ensured by check above
    if len(labels["label"]) != len(labels["label"].unique()):
        raise ValueError(
            "The output ROI table contains duplicate entries for labels: "
            f"It contains {len(labels['label'])} entries, but only "
            f"{len(labels['label'].unique())} unique labels"
        )

    df_well.index = labels["label"]
    df_well.drop(labels=["label"], axis=1, inplace=True)
    # Convert all to float (warning: some would be int, in principle)
    bbox_dtype = np.float32
    df_well = df_well.astype(bbox_dtype)
    # Convert to anndata
    bbox_table = ad.AnnData(df_well)
    bbox_table.obs = labels
    return bbox_table

empty_bounding_box_table()

Construct an empty bounding-box ROI table of given shape.

This function mirrors the functionality of array_to_bounding_box_table, for the specific case where the array includes no label. The advantages of this function are that:

  1. It does not require computing a whole array of zeros;
  2. We avoid hardcoding column names in the task functions.
RETURNS DESCRIPTION
DataFrame

DataFrame with no rows, and with columns corresponding to the output of array_to_bounding_box_table.

Source code in fractal_tasks_core/roi/v1.py
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
def empty_bounding_box_table() -> pd.DataFrame:
    """
    Construct an empty bounding-box ROI table of given shape.

    This function mirrors the functionality of `array_to_bounding_box_table`,
    for the specific case where the array includes no label. The advantages of
    this function are that:

    1. It does not require computing a whole array of zeros;
    2. We avoid hardcoding column names in the task functions.

    Returns:
        DataFrame with no rows, and with columns corresponding to the output of
            `array_to_bounding_box_table`.
    """

    df_columns = [
        "x_micrometer",
        "y_micrometer",
        "z_micrometer",
        "len_x_micrometer",
        "len_y_micrometer",
        "len_z_micrometer",
    ]
    df = pd.DataFrame(columns=[x for x in df_columns] + ["label"])
    return df

find_overlaps_in_ROI_indices(list_indices)

Given a list of integer ROI indices, find whether there are overlaps.

PARAMETER DESCRIPTION
list_indices

List of ROI indices, where each element in the list should look like [start_z, end_z, start_y, end_y, start_x, end_x].

TYPE: list[list[int]]

RETURNS DESCRIPTION
Optional[tuple[int, int]]

None if no overlap was detected, otherwise a tuple with the positional indices of a pair of overlapping ROIs.

Source code in fractal_tasks_core/roi/v1_overlaps.py
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
def find_overlaps_in_ROI_indices(
    list_indices: list[list[int]],
) -> Optional[tuple[int, int]]:
    """
    Given a list of integer ROI indices, find whether there are overlaps.

    Args:
        list_indices: List of ROI indices, where each element in the list
            should look like
            `[start_z, end_z, start_y, end_y, start_x, end_x]`.

    Returns:
        `None` if no overlap was detected, otherwise a tuple with the
            positional indices of a pair of overlapping ROIs.
    """

    for ind_1, ROI_1 in enumerate(list_indices):
        s_z, e_z, s_y, e_y, s_x, e_x = ROI_1[:]
        box_1 = [s_x, s_y, s_z, e_x, e_y, e_z]
        for ind_2 in range(ind_1):
            ROI_2 = list_indices[ind_2]
            s_z, e_z, s_y, e_y, s_x, e_x = ROI_2[:]
            box_2 = [s_x, s_y, s_z, e_x, e_y, e_z]
            if _is_overlapping_3D_int(box_1, box_2):
                return (ind_1, ind_2)
    return None

get_image_grid_ROIs(array_shape, pixels_ZYX, grid_YX_shape)

Produce a table with ROIS placed on a rectangular grid.

The main goal of this ROI grid is to allow processing of smaller subset of the whole array.

In a specific case (that is, if the image array was obtained by stitching together a set of FOVs placed on a regular grid), the ROIs correspond to the original FOVs.

TODO: make this flexible with respect to the presence/absence of Z.

PARAMETER DESCRIPTION
array_shape

ZYX shape of the image array.

TYPE: tuple[int, int, int]

pixels_ZYX

ZYX pixel sizes in micrometers.

TYPE: list[float]

grid_YX_shape

TYPE: tuple[int, int]

RETURNS DESCRIPTION
AnnData

An AnnData table with a single ROI.

Source code in fractal_tasks_core/roi/v1.py
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
def get_image_grid_ROIs(
    array_shape: tuple[int, int, int],
    pixels_ZYX: list[float],
    grid_YX_shape: tuple[int, int],
) -> ad.AnnData:
    """
    Produce a table with ROIS placed on a rectangular grid.

    The main goal of this ROI grid is to allow processing of smaller subset of
    the whole array.

    In a specific case (that is, if the image array was obtained by stitching
    together a set of FOVs placed on a regular grid), the ROIs correspond to
    the original FOVs.

    TODO: make this flexible with respect to the presence/absence of Z.

    Args:
        array_shape: ZYX shape of the image array.
        pixels_ZYX: ZYX pixel sizes in micrometers.
        grid_YX_shape:

    Returns:
        An `AnnData` table with a single ROI.
    """
    shape_z, shape_y, shape_x = array_shape[-3:]
    grid_size_y, grid_size_x = grid_YX_shape[:]
    X = []
    obs_names = []
    counter = 0
    start_z = 0
    len_z = shape_z

    # Find minimal len_y that covers [0,shape_y] with grid_size_y intervals
    len_y = math.ceil(shape_y / grid_size_y)
    len_x = math.ceil(shape_x / grid_size_x)
    for ind_y in range(grid_size_y):
        start_y = ind_y * len_y
        tmp_len_y = min(shape_y, start_y + len_y) - start_y
        for ind_x in range(grid_size_x):
            start_x = ind_x * len_x
            tmp_len_x = min(shape_x, start_x + len_x) - start_x
            X.append(
                [
                    start_x * pixels_ZYX[2],
                    start_y * pixels_ZYX[1],
                    start_z * pixels_ZYX[0],
                    tmp_len_x * pixels_ZYX[2],
                    tmp_len_y * pixels_ZYX[1],
                    len_z * pixels_ZYX[0],
                ]
            )
            counter += 1
            obs_names.append(f"ROI_{counter}")
    ROI_table = ad.AnnData(X=np.array(X, dtype=np.float32))
    ROI_table.obs_names = obs_names
    ROI_table.var_names = [
        "x_micrometer",
        "y_micrometer",
        "z_micrometer",
        "len_x_micrometer",
        "len_y_micrometer",
        "len_z_micrometer",
    ]
    return ROI_table

get_overlapping_pair(tmp_df, tol=1e-10)

Finds the indices for the next overlapping FOVs pair.

Note: the returned indices are positional indices, starting from 0.

PARAMETER DESCRIPTION
tmp_df

Dataframe with columns ["xmin", "ymin", "xmax", "ymax"].

TYPE: DataFrame

tol

Finite tolerance for floating-point comparisons.

TYPE: float DEFAULT: 1e-10

Source code in fractal_tasks_core/roi/v1_overlaps.py
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
def get_overlapping_pair(
    tmp_df: pd.DataFrame, tol: float = 1e-10
) -> Union[tuple[int, int], bool]:
    """
    Finds the indices for the next overlapping FOVs pair.

    Note: the returned indices are positional indices, starting from 0.

    Args:
        tmp_df: Dataframe with columns `["xmin", "ymin", "xmax", "ymax"]`.
        tol: Finite tolerance for floating-point comparisons.
    """

    num_lines = len(tmp_df.index)
    for pos_ind_1 in range(num_lines):
        for pos_ind_2 in range(pos_ind_1):
            if is_overlapping_2D(
                tmp_df.iloc[pos_ind_1], tmp_df.iloc[pos_ind_2], tol=tol
            ):
                return (pos_ind_1, pos_ind_2)
    return False

get_overlapping_pairs_3D(tmp_df, full_res_pxl_sizes_zyx)

Finds the indices for the all overlapping FOVs pair, in three dimensions.

Note: the returned indices are positional indices, starting from 0.

PARAMETER DESCRIPTION
tmp_df

Dataframe with columns {x,y,z}_micrometer and len_{x,y,z}_micrometer.

TYPE: DataFrame

full_res_pxl_sizes_zyx

TBD

TYPE: Sequence[float]

Source code in fractal_tasks_core/roi/v1_overlaps.py
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def get_overlapping_pairs_3D(
    tmp_df: pd.DataFrame,
    full_res_pxl_sizes_zyx: Sequence[float],
):
    """
    Finds the indices for the all overlapping FOVs pair, in three dimensions.

    Note: the returned indices are positional indices, starting from 0.

    Args:
        tmp_df: Dataframe with columns `{x,y,z}_micrometer` and
            `len_{x,y,z}_micrometer`.
        full_res_pxl_sizes_zyx: TBD
    """

    tol = 1e-10
    if tol > min(full_res_pxl_sizes_zyx) / 1e3:
        raise ValueError(f"{tol=} but {full_res_pxl_sizes_zyx=}")

    new_tmp_df = tmp_df.copy()

    new_tmp_df["x_micrometer_max"] = (
        new_tmp_df["x_micrometer"] + new_tmp_df["len_x_micrometer"]
    )
    new_tmp_df["y_micrometer_max"] = (
        new_tmp_df["y_micrometer"] + new_tmp_df["len_y_micrometer"]
    )
    new_tmp_df["z_micrometer_max"] = (
        new_tmp_df["z_micrometer"] + new_tmp_df["len_z_micrometer"]
    )
    # Remove columns which are not necessary for overlap checks
    list_columns = [
        "len_x_micrometer",
        "len_y_micrometer",
        "len_z_micrometer",
        "label",
    ]
    new_tmp_df.drop(labels=list_columns, axis=1, inplace=True)

    # Loop over all pairs, and construct list of overlapping ones
    num_lines = len(new_tmp_df.index)
    overlapping_list = []
    for pos_ind_1 in range(num_lines):
        for pos_ind_2 in range(pos_ind_1):
            overlap = is_overlapping_3D(
                new_tmp_df.iloc[pos_ind_1], new_tmp_df.iloc[pos_ind_2], tol=tol
            )
            if overlap:
                overlapping_list.append((pos_ind_1, pos_ind_2))
    return overlapping_list

get_single_image_ROI(array_shape, pixels_ZYX)

Produce a table with a single ROI that covers the whole array

TODO: make this flexible with respect to the presence/absence of Z.

PARAMETER DESCRIPTION
array_shape

ZYX shape of the image array.

TYPE: tuple[int, int, int]

pixels_ZYX

ZYX pixel sizes in micrometers.

TYPE: list[float]

RETURNS DESCRIPTION
AnnData

An AnnData table with a single ROI.

Source code in fractal_tasks_core/roi/v1.py
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
def get_single_image_ROI(
    array_shape: tuple[int, int, int],
    pixels_ZYX: list[float],
) -> ad.AnnData:
    """
    Produce a table with a single ROI that covers the whole array

    TODO: make this flexible with respect to the presence/absence of Z.

    Args:
        array_shape: ZYX shape of the image array.
        pixels_ZYX: ZYX pixel sizes in micrometers.

    Returns:
        An `AnnData` table with a single ROI.
    """
    shape_z, shape_y, shape_x = array_shape[-3:]
    ROI_table = ad.AnnData(
        X=np.array(
            [
                [
                    0.0,
                    0.0,
                    0.0,
                    shape_x * pixels_ZYX[2],
                    shape_y * pixels_ZYX[1],
                    shape_z * pixels_ZYX[0],
                ],
            ],
            dtype=np.float32,
        )
    )
    ROI_table.obs_names = ["image_1"]
    ROI_table.var_names = [
        "x_micrometer",
        "y_micrometer",
        "z_micrometer",
        "len_x_micrometer",
        "len_y_micrometer",
        "len_z_micrometer",
    ]
    return ROI_table

is_ROI_table_valid(*, table_path, use_masks)

Verify some validity assumptions on a ROI table.

This function reflects our current working assumptions (e.g. the presence of some specific columns); this may change in future versions.

If use_masks=True, we verify that the table is a valid masking_roi_table as of table specifications V1; if this check fails, use_masks should be set to False upstream in the parent function.

PARAMETER DESCRIPTION
table_path

Path of the AnnData ROI table to be checked.

TYPE: str

use_masks

If True, perform some additional checks related to masked loading.

TYPE: bool

RETURNS DESCRIPTION
Optional[bool]

Always None if use_masks=False, otherwise return whether the table is valid for masked loading.

Source code in fractal_tasks_core/roi/v1_checks.py
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def is_ROI_table_valid(*, table_path: str, use_masks: bool) -> Optional[bool]:
    """
    Verify some validity assumptions on a ROI table.

    This function reflects our current working assumptions (e.g. the presence
    of some specific columns); this may change in future versions.

    If `use_masks=True`, we verify that the table is a valid
    `masking_roi_table` as of table specifications V1; if this check fails,
    `use_masks` should be set to `False` upstream in the parent function.

    Args:
        table_path: Path of the AnnData ROI table to be checked.
        use_masks: If `True`, perform some additional checks related to
            masked loading.

    Returns:
        Always `None` if `use_masks=False`, otherwise return whether the table
            is valid for masked loading.
    """

    table = ad.read_zarr(table_path)
    are_ROI_table_columns_valid(table=table)
    if not use_masks:
        return None

    # Check whether the table can be used for masked loading
    attrs = zarr.group(table_path).attrs.asdict()
    logger.info(f"ROI table at {table_path} has attrs: {attrs}")
    try:
        MaskingROITableAttrs(**attrs)
        logging.info("ROI table can be used for masked loading")
        return True
    except ValidationError:
        logging.info("ROI table cannot be used for masked loading")
        return False

is_overlapping_1D(line1, line2, tol=1e-10)

Given two intervals, finds whether they overlap.

This is based on https://stackoverflow.com/a/70023212/19085332, and we additionally use a finite tolerance for floating-point comparisons.

PARAMETER DESCRIPTION
line1

The boundaries of the first interval, written as [x_min, x_max].

TYPE: Sequence[float]

line2

The boundaries of the second interval, written as [x_min, x_max].

TYPE: Sequence[float]

tol

Finite tolerance for floating-point comparisons.

TYPE: float DEFAULT: 1e-10

Source code in fractal_tasks_core/roi/_overlaps_common.py
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
def is_overlapping_1D(
    line1: Sequence[float], line2: Sequence[float], tol: float = 1e-10
) -> bool:
    """
    Given two intervals, finds whether they overlap.

    This is based on https://stackoverflow.com/a/70023212/19085332, and we
    additionally use a finite tolerance for floating-point comparisons.

    Args:
        line1: The boundaries of the first interval, written as
            `[x_min, x_max]`.
        line2: The boundaries of the second interval, written as
            `[x_min, x_max]`.
        tol: Finite tolerance for floating-point comparisons.
    """
    return line1[0] <= line2[1] - tol and line2[0] <= line1[1] - tol

is_overlapping_2D(box1, box2, tol=1e-10)

Given two rectangular boxes, finds whether they overlap.

This is based on https://stackoverflow.com/a/70023212/19085332, and we additionally use a finite tolerance for floating-point comparisons.

PARAMETER DESCRIPTION
box1

The boundaries of the first rectangle, written as [x_min, y_min, x_max, y_max].

TYPE: Sequence[float]

box2

The boundaries of the second rectangle, written as [x_min, y_min, x_max, y_max].

TYPE: Sequence[float]

tol

Finite tolerance for floating-point comparisons.

TYPE: float DEFAULT: 1e-10

Source code in fractal_tasks_core/roi/_overlaps_common.py
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
def is_overlapping_2D(
    box1: Sequence[float], box2: Sequence[float], tol: float = 1e-10
) -> bool:
    """
    Given two rectangular boxes, finds whether they overlap.

    This is based on https://stackoverflow.com/a/70023212/19085332, and we
    additionally use a finite tolerance for floating-point comparisons.

    Args:
        box1: The boundaries of the first rectangle, written as
            `[x_min, y_min, x_max, y_max]`.
        box2: The boundaries of the second rectangle, written as
            `[x_min, y_min, x_max, y_max]`.
        tol: Finite tolerance for floating-point comparisons.
    """
    overlap_x = is_overlapping_1D(
        [box1[0], box1[2]], [box2[0], box2[2]], tol=tol
    )
    overlap_y = is_overlapping_1D(
        [box1[1], box1[3]], [box2[1], box2[3]], tol=tol
    )
    return overlap_x and overlap_y

is_overlapping_3D(box1, box2, tol=1e-10)

Given two three-dimensional boxes, finds whether they overlap.

This is based on https://stackoverflow.com/a/70023212/19085332, and we additionally use a finite tolerance for floating-point comparisons.

PARAMETER DESCRIPTION
box1

The boundaries of the first box, written as [x_min, y_min, z_min, x_max, y_max, z_max].

TYPE: Sequence[float]

box2

The boundaries of the second box, written as [x_min, y_min, z_min, x_max, y_max, z_max].

TYPE: Sequence[float]

tol

Finite tolerance for floating-point comparisons.

TYPE: float DEFAULT: 1e-10

Source code in fractal_tasks_core/roi/_overlaps_common.py
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
def is_overlapping_3D(
    box1: Sequence[float], box2: Sequence[float], tol: float = 1e-10
) -> bool:
    """
    Given two three-dimensional boxes, finds whether they overlap.

    This is based on https://stackoverflow.com/a/70023212/19085332, and we
    additionally use a finite tolerance for floating-point comparisons.

    Args:
        box1: The boundaries of the first box, written as
            `[x_min, y_min, z_min, x_max, y_max, z_max]`.
        box2: The boundaries of the second box, written as
            `[x_min, y_min, z_min, x_max, y_max, z_max]`.
        tol: Finite tolerance for floating-point comparisons.
    """

    overlap_x = is_overlapping_1D(
        [box1[0], box1[3]], [box2[0], box2[3]], tol=tol
    )
    overlap_y = is_overlapping_1D(
        [box1[1], box1[4]], [box2[1], box2[4]], tol=tol
    )
    overlap_z = is_overlapping_1D(
        [box1[2], box1[5]], [box2[2], box2[5]], tol=tol
    )
    return overlap_x and overlap_y and overlap_z

is_standard_roi_table(table)

True if the name of the table contains one of the standard Fractal tables

If a table name is well_ROI_table, FOV_ROI_table or contains either of the two (e.g. registered_FOV_ROI_table), this function returns True.

PARAMETER DESCRIPTION
table

table name

TYPE: str

RETURNS DESCRIPTION
bool

bool of whether it's a standard ROI table

Source code in fractal_tasks_core/roi/v1.py
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
def is_standard_roi_table(table: str) -> bool:
    """
    True if the name of the table contains one of the standard Fractal tables

    If a table name is well_ROI_table, FOV_ROI_table or contains either of the
    two (e.g. registered_FOV_ROI_table), this function returns True.

    Args:
        table: table name

    Returns:
        bool of whether it's a standard ROI table

    """
    if "well_ROI_table" in table:
        return True
    elif "FOV_ROI_table" in table:
        return True
    else:
        return False

prepare_FOV_ROI_table(df, metadata=('time'))

Prepare an AnnData table for fields-of-view ROIs.

PARAMETER DESCRIPTION
df

Input dataframe, possibly prepared through parse_yokogawa_metadata.

TYPE: DataFrame

metadata

Columns of df to be stored (if present) into AnnData table obs.

TYPE: tuple[str, ...] DEFAULT: ('time')

Source code in fractal_tasks_core/roi/v1.py
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
def prepare_FOV_ROI_table(
    df: pd.DataFrame, metadata: tuple[str, ...] = ("time",)
) -> ad.AnnData:
    """
    Prepare an AnnData table for fields-of-view ROIs.

    Args:
        df:
            Input dataframe, possibly prepared through
            `parse_yokogawa_metadata`.
        metadata:
            Columns of `df` to be stored (if present) into AnnData table `obs`.
    """

    # Make a local copy of the dataframe, to avoid SettingWithCopyWarning
    df = df.copy()

    # Convert DataFrame index to str, to avoid
    # >> ImplicitModificationWarning: Transforming to str index
    # when creating AnnData object.
    # Do this in the beginning to allow concatenation with e.g. time
    df.index = df.index.astype(str)

    # Obtain box size in physical units
    df = df.assign(len_x_micrometer=df.x_pixel * df.pixel_size_x)
    df = df.assign(len_y_micrometer=df.y_pixel * df.pixel_size_y)
    df = df.assign(len_z_micrometer=df.z_pixel * df.pixel_size_z)

    # Select only the numeric positional columns needed to define ROIs
    # (to avoid) casting things like the data column to float32
    # or to use unnecessary columns like bit_depth
    positional_columns = [
        "x_micrometer",
        "y_micrometer",
        "z_micrometer",
        "len_x_micrometer",
        "len_y_micrometer",
        "len_z_micrometer",
        "x_micrometer_original",
        "y_micrometer_original",
    ]

    # Assign dtype explicitly, to avoid
    # >> UserWarning: X converted to numpy array with dtype float64
    # when creating AnnData object
    df_roi = df.loc[:, positional_columns].astype(np.float32)

    # Create an AnnData object directly from the DataFrame
    adata = ad.AnnData(X=df_roi)

    # Reset origin of the FOV ROI table, so that it matches with the well
    # origin
    adata = reset_origin(adata)

    # Save any metadata that is specified to the obs df
    for col in metadata:
        if col in df:
            # Cast all metadata to str.
            # Reason: AnnData Zarr writers don't support all pandas types.
            # e.g. pandas.core.arrays.datetimes.DatetimeArray can't be written
            adata.obs[col] = df[col].astype(str)

    # Rename rows and columns: Maintain FOV indices from the dataframe
    # (they are already enforced to be unique by Pandas and may contain
    # information for the user, as they are based on the filenames)
    adata.obs_names = "FOV_" + adata.obs.index
    adata.var_names = list(map(str, df_roi.columns))

    return adata

prepare_well_ROI_table(df, metadata=('time'))

Prepare an AnnData table with a single well ROI.

PARAMETER DESCRIPTION
df

Input dataframe, possibly prepared through parse_yokogawa_metadata.

TYPE: DataFrame

metadata

Columns of df to be stored (if present) into AnnData table obs.

TYPE: tuple[str, ...] DEFAULT: ('time')

Source code in fractal_tasks_core/roi/v1.py
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
def prepare_well_ROI_table(
    df: pd.DataFrame, metadata: tuple[str, ...] = ("time",)
) -> ad.AnnData:
    """
    Prepare an AnnData table with a single well ROI.

    Args:
        df:
            Input dataframe, possibly prepared through
            `parse_yokogawa_metadata`.
        metadata:
            Columns of `df` to be stored (if present) into AnnData table `obs`.
    """

    # Make a local copy of the dataframe, to avoid SettingWithCopyWarning
    df = df.copy()

    # Convert DataFrame index to str, to avoid
    # >> ImplicitModificationWarning: Transforming to str index
    # when creating AnnData object.
    # Do this in the beginning to allow concatenation with e.g. time
    df.index = df.index.astype(str)

    # Calculate bounding box extents in physical units
    for mu in ["x", "y", "z"]:
        # Obtain per-FOV properties in physical units.
        # NOTE: a FOV ROI is defined here as the interval [min_micrometer,
        # max_micrometer], with max_micrometer=min_micrometer+len_micrometer
        min_micrometer = df[f"{mu}_micrometer"]
        len_micrometer = df[f"{mu}_pixel"] * df[f"pixel_size_{mu}"]
        max_micrometer = min_micrometer + len_micrometer
        # Obtain well bounding box, in physical units
        min_min_micrometer = min_micrometer.min()
        max_max_micrometer = max_micrometer.max()
        df[f"{mu}_micrometer"] = min_min_micrometer
        df[f"len_{mu}_micrometer"] = max_max_micrometer - min_min_micrometer

    # Select only the numeric positional columns needed to define ROIs
    # (to avoid) casting things like the data column to float32
    # or to use unnecessary columns like bit_depth
    positional_columns = [
        "x_micrometer",
        "y_micrometer",
        "z_micrometer",
        "len_x_micrometer",
        "len_y_micrometer",
        "len_z_micrometer",
    ]

    # Assign dtype explicitly, to avoid
    # >> UserWarning: X converted to numpy array with dtype float64
    # when creating AnnData object
    df_roi = df.iloc[0:1, :].loc[:, positional_columns].astype(np.float32)

    # Create an AnnData object directly from the DataFrame
    adata = ad.AnnData(X=df_roi)

    # Reset origin of the single-entry well ROI table
    adata = reset_origin(adata)

    # Save any metadata that is specified to the obs df
    for col in metadata:
        if col in df:
            # Cast all metadata to str.
            # Reason: AnnData Zarr writers don't support all pandas types.
            # e.g. pandas.core.arrays.datetimes.DatetimeArray can't be written
            adata.obs[col] = df[col].astype(str)

    # Rename rows and columns: Maintain FOV indices from the dataframe
    # (they are already enforced to be unique by Pandas and may contain
    # information for the user, as they are based on the filenames)
    adata.obs_names = "well_" + adata.obs.index
    adata.var_names = list(map(str, df_roi.columns))

    return adata

remove_FOV_overlaps(df)

Given a metadata dataframe, shift its columns to remove FOV overlaps.

PARAMETER DESCRIPTION
df

Metadata dataframe.

TYPE: DataFrame

Source code in fractal_tasks_core/roi/v1_overlaps.py
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
def remove_FOV_overlaps(df: pd.DataFrame):
    """
    Given a metadata dataframe, shift its columns to remove FOV overlaps.

    Args:
        df: Metadata dataframe.
    """

    # Set tolerance (this should be much smaller than pixel size or expected
    # round-offs), and maximum number of iterations in constraint solver
    tol = 1e-10
    max_iterations = 200

    # Create a local copy of the dataframe
    df = df.copy()

    # Create temporary columns (to streamline overlap removals), which are
    # then removed at the end of the remove_FOV_overlaps function
    df["xmin"] = df["x_micrometer"]
    df["ymin"] = df["y_micrometer"]
    df["xmax"] = df["x_micrometer"] + df["pixel_size_x"] * df["x_pixel"]
    df["ymax"] = df["y_micrometer"] + df["pixel_size_y"] * df["y_pixel"]
    list_columns = ["xmin", "ymin", "xmax", "ymax"]

    # Create columns with the original positions (not to be removed)
    df["x_micrometer_original"] = df["x_micrometer"]
    df["y_micrometer_original"] = df["y_micrometer"]

    # Check that tolerance is much smaller than pixel sizes
    min_pixel_size = df[["pixel_size_x", "pixel_size_y"]].min().min()
    if tol > min_pixel_size / 1e3:
        raise ValueError(
            f"In remove_FOV_overlaps, {tol=} but {min_pixel_size=}"
        )

    # Loop over wells
    wells = sorted(list(set([ind[0] for ind in df.index])))
    for well in wells:

        logger.info(f"removing FOV overlaps for {well=}")
        df_well = df.loc[well].copy()

        # NOTE: these are positional indices (i.e. starting from 0)
        pair_pos_indices = get_overlapping_pair(df_well[list_columns], tol=tol)

        # Keep going until there are no overlaps, or until iteration reaches
        # max_iterations
        iteration = 0
        while pair_pos_indices:
            iteration += 1

            # Identify overlapping FOVs
            pos_ind_1, pos_ind_2 = pair_pos_indices
            fov_id_1 = df_well.index[pos_ind_1]
            fov_id_2 = df_well.index[pos_ind_2]
            xmin_1, ymin_1, xmax_1, ymax_1 = df_well[list_columns].iloc[
                pos_ind_1
            ]
            xmin_2, ymin_2, xmax_2, ymax_2 = df_well[list_columns].iloc[
                pos_ind_2
            ]
            logger.debug(
                f"{well=}, {iteration=}, removing overlap between"
                f" {fov_id_1=} and {fov_id_2=}"
            )

            # Check what kind of overlap is there (X, Y, or XY)
            is_x_equal = abs(xmin_1 - xmin_2) < tol and (xmax_1 - xmax_2) < tol
            is_y_equal = abs(ymin_1 - ymin_2) < tol and (ymax_1 - ymax_2) < tol
            is_x_overlap = is_overlapping_1D(
                [xmin_1, xmax_1], [xmin_2, xmax_2], tol=tol
            )
            is_y_overlap = is_overlapping_1D(
                [ymin_1, ymax_1], [ymin_2, ymax_2], tol=tol
            )

            if is_x_equal and is_y_overlap:
                # Y overlap
                df_well = apply_shift_in_one_direction(
                    df_well,
                    [ymin_1, ymax_1],
                    [ymin_2, ymax_2],
                    mu="y",
                    tol=tol,
                )
            elif is_y_equal and is_x_overlap:
                # X overlap
                df_well = apply_shift_in_one_direction(
                    df_well,
                    [xmin_1, xmax_1],
                    [xmin_2, xmax_2],
                    mu="x",
                    tol=tol,
                )
            elif not (is_x_equal or is_y_equal) and (
                is_x_overlap and is_y_overlap
            ):
                # XY overlap
                df_well = apply_shift_in_one_direction(
                    df_well,
                    [xmin_1, xmax_1],
                    [xmin_2, xmax_2],
                    mu="x",
                    tol=tol,
                )
                df_well = apply_shift_in_one_direction(
                    df_well,
                    [ymin_1, ymax_1],
                    [ymin_2, ymax_2],
                    mu="y",
                    tol=tol,
                )
            else:
                raise ValueError(
                    "Trying to remove overlap which is not there."
                )

            # Look for next overlapping FOV pair
            pair_pos_indices = get_overlapping_pair(
                df_well[list_columns], tol=tol
            )

            # Enforce maximum number of iterations
            if iteration >= max_iterations:
                raise ValueError(f"Reached {max_iterations=} for {well=}")

        # Note: using df.loc[well] = df_well leads to a NaN dataframe, see
        # for instance https://stackoverflow.com/a/28432733/19085332
        df.loc[well, :] = df_well.values

    # Remove temporary columns that were added only as part of this function
    df.drop(list_columns, axis=1, inplace=True)

    return df

reset_origin(ROI_table, x_pos='x_micrometer', y_pos='y_micrometer', z_pos='z_micrometer')

Return a copy of a ROI table, with shifted-to-zero origin for some columns.

PARAMETER DESCRIPTION
ROI_table

Original ROI table.

TYPE: AnnData

x_pos

Name of the column with X position of ROIs.

TYPE: str DEFAULT: 'x_micrometer'

y_pos

Name of the column with Y position of ROIs.

TYPE: str DEFAULT: 'y_micrometer'

z_pos

Name of the column with Z position of ROIs.

TYPE: str DEFAULT: 'z_micrometer'

RETURNS DESCRIPTION
AnnData

A copy of the ROI_table AnnData table, where values of x_pos, y_pos and z_pos columns have been shifted by their minimum values.

Source code in fractal_tasks_core/roi/v1.py
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
def reset_origin(
    ROI_table: ad.AnnData,
    x_pos: str = "x_micrometer",
    y_pos: str = "y_micrometer",
    z_pos: str = "z_micrometer",
) -> ad.AnnData:
    """
    Return a copy of a ROI table, with shifted-to-zero origin for some columns.

    Args:
        ROI_table: Original ROI table.
        x_pos: Name of the column with X position of ROIs.
        y_pos: Name of the column with Y position of ROIs.
        z_pos: Name of the column with Z position of ROIs.

    Returns:
        A copy of the `ROI_table` AnnData table, where values of `x_pos`,
            `y_pos` and `z_pos` columns have been shifted by their minimum
            values.
    """
    new_table = ROI_table.copy()

    origin_x = min(new_table[:, x_pos].X[:, 0])
    origin_y = min(new_table[:, y_pos].X[:, 0])
    origin_z = min(new_table[:, z_pos].X[:, 0])

    for FOV in new_table.obs_names:
        new_table[FOV, x_pos] = new_table[FOV, x_pos].X[0, 0] - origin_x
        new_table[FOV, y_pos] = new_table[FOV, y_pos].X[0, 0] - origin_y
        new_table[FOV, z_pos] = new_table[FOV, z_pos].X[0, 0] - origin_z

    return new_table

run_overlap_check(site_metadata, tol=1e-10, plotting_function=None)

Run an overlap check over all wells and optionally plots overlaps.

This function is currently only used in tests and examples.

The plotting_function parameter is exposed so that other tools (see examples in this repository) may use it to show the FOV ROIs. Its arguments are: [xmin, xmax, ymin, ymax, list_overlapping_FOVs, selected_well].

PARAMETER DESCRIPTION
site_metadata

TBD

TYPE: DataFrame

tol

TBD

TYPE: float DEFAULT: 1e-10

plotting_function

TBD

TYPE: Optional[Callable] DEFAULT: None

Source code in fractal_tasks_core/roi/v1_overlaps.py
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
def run_overlap_check(
    site_metadata: pd.DataFrame,
    tol: float = 1e-10,
    plotting_function: Optional[Callable] = None,
):
    """
    Run an overlap check over all wells and optionally plots overlaps.

    This function is currently only used in tests and examples.

    The `plotting_function` parameter is exposed so that other tools (see
    examples in this repository) may use it to show the FOV ROIs. Its arguments
    are: `[xmin, xmax, ymin, ymax, list_overlapping_FOVs, selected_well]`.

    Args:
        site_metadata: TBD
        tol: TBD
        plotting_function: TBD
    """

    if plotting_function is None:

        def plotting_function(
            xmin, xmax, ymin, ymax, list_overlapping_FOVs, selected_well
        ):
            pass

    wells = site_metadata.index.unique(level="well_id")
    overlapping_FOVs = []
    for selected_well in wells:
        overlap_curr_well = check_well_for_FOV_overlap(
            site_metadata,
            selected_well=selected_well,
            tol=tol,
            plotting_function=plotting_function,
        )
        if overlap_curr_well:
            print(selected_well)
            overlapping_FOVs.append(overlap_curr_well)

    return overlapping_FOVs