Coverage for src/cell_abm_pipeline/flows/sample_image.py: 0%
56 statements
« prev ^ index » next coverage.py v7.1.0, created at 2024-06-05 19:14 +0000
« prev ^ index » next coverage.py v7.1.0, created at 2024-06-05 19:14 +0000
1"""
2Workflow for sampling cell ids from an image.
4Working location structure:
6.. code-block:: bash
8 (name)
9 ├── images
10 │ └── (name)_(key).(extension)
11 ├── plots
12 │ └── plots.SAMPLE
13 │ └── (name)_(key)_C(channel)_R(resolution).SAMPLE.png
14 └── samples
15 └── samples.RAW
16 └── (name)_(key)_C(channel)_R(resolution).RAW.csv
18Input images to be sampled are loaded from **images**. Resulting image sample(s)
19are placed into **samples.RAW** and corresponding contact sheet(s) are placed
20into **plots.SAMPLE**.
21"""
23from dataclasses import dataclass, field
24from typing import Optional
26from abm_initialization_collection.image import get_image_bounds, plot_contact_sheet
27from abm_initialization_collection.sample import (
28 get_image_samples,
29 get_sample_indices,
30 scale_sample_coordinates,
31)
32from io_collection.keys import make_key
33from io_collection.load import load_image
34from io_collection.save import save_dataframe, save_figure
35from prefect import flow
37# Default pixel resolution for images in x/y in um/pixel
38SCALE_MICRONS_XY: float = 0.108333
40# Default pixel resolution for images in z in um/pixel
41SCALE_MICRONS_Z: float = 0.29
44@dataclass
45class ParametersConfig:
46 """Parameter configuration for sample image flow."""
48 key: str
49 """Image key to sample."""
51 channels: list[int] = field(default_factory=lambda: [0])
52 """Image channels to sample."""
54 grid: str = "rect"
55 """Type of sampling grid (rect = rectangular, hex = hexagonal)."""
57 coordinate_type: Optional[str] = None
58 """Coordinate scaling type (None = unscaled, absolute = in um, step = by step size)."""
60 resolution: float = 1.0
61 """Distance between samples in um."""
63 scale_xy: float = SCALE_MICRONS_XY
64 """ Resolution scaling in x/y in um/pixel."""
66 scale_z: float = SCALE_MICRONS_Z
67 """Resolution scaling in z in um/pixel."""
69 extension: str = ".ome.tiff"
70 """Image extension."""
72 contact_sheet: bool = True
73 """True to save contact sheet of processed samples, False otherwise."""
76@dataclass
77class ContextConfig:
78 """Context configuration for sample image flow."""
80 working_location: str
81 """Location for input and output files (local path or S3 bucket)."""
84@dataclass
85class SeriesConfig:
86 """Series configuration for sample image flow."""
88 name: str
89 """Name of the simulation series."""
92@flow(name="sample-image")
93def run_flow(context: ContextConfig, series: SeriesConfig, parameters: ParametersConfig) -> None:
94 """Main sample image flow."""
96 image_key = make_key(
97 series.name, "images", f"{series.name}_{parameters.key}{parameters.extension}"
98 )
99 image = load_image(
100 context.working_location,
101 image_key,
102 dim_order="ZYX" if parameters.extension == ".tiff" else None,
103 )
104 image_bounds = get_image_bounds(image)
106 sample_indices = get_sample_indices(
107 parameters.grid,
108 image_bounds,
109 parameters.resolution,
110 parameters.scale_xy,
111 parameters.scale_z,
112 )
114 for channel in parameters.channels:
115 resolution_key = f"R{round(parameters.resolution * 10):03d}"
116 channel_key = f"C{channel:02d}"
117 item_key = f"{series.name}_{parameters.key}_{channel_key}_{resolution_key}"
119 samples = get_image_samples(image, sample_indices, channel)
120 samples = scale_sample_coordinates(
121 samples,
122 parameters.coordinate_type,
123 parameters.resolution,
124 parameters.scale_xy,
125 parameters.scale_z,
126 )
127 sample_key = make_key(series.name, "samples", "samples.RAW", f"{item_key}.RAW.csv")
128 save_dataframe(context.working_location, sample_key, samples, index=False)
130 if parameters.contact_sheet:
131 contact_sheet = plot_contact_sheet(samples)
132 plot_key = make_key(series.name, "plots", "plots.SAMPLE", f"{item_key}.SAMPLE.png")
133 save_figure(context.working_location, plot_key, contact_sheet)