elf.htm.visualization

  1import string
  2from typing import Dict, List, Optional, Tuple, Union
  3
  4import numpy as np
  5
  6try:
  7    import napari
  8    from napari.experimental import link_layers
  9except ImportError:
 10    napari = None
 11    link_layers = None
 12
 13
 14def parse_wells(well_names, zero_based):
 15    """@private
 16    """
 17    well_x, well_y = [], []
 18    for name in well_names:
 19        letter, y = name[0], int(name[1:])
 20        if not zero_based:
 21            y -= 1
 22        assert y >= 0
 23        x = string.ascii_uppercase.index(letter)
 24        well_x.append(x)
 25        well_y.append(y)
 26    assert len(well_x) == len(well_y) == len(well_names)
 27    well_positions = {name: (x, y) for name, x, y in zip(well_names, well_x, well_y)}
 28    well_start = (min(well_x), min(well_y))
 29    well_stop = (max(well_x) + 1, max(well_y) + 1)
 30    return well_positions, well_start, well_stop
 31
 32
 33def get_world_position(well_x, well_y, pos, well_shape, well_spacing, site_spacing, shape):
 34    """@private
 35    """
 36    unraveled = np.unravel_index([pos], well_shape)
 37    pos_x, pos_y = unraveled[0][0], unraveled[1][0]
 38    i = well_x * well_shape[0] + pos_x
 39    j = well_y * well_shape[1] + pos_y
 40    # print(well_x, well_y, pos, ":", i, j)
 41
 42    sx = shape[-2]
 43    sy = shape[-1]
 44    x = sx * i + i * site_spacing + well_x * well_spacing
 45    y = sy * j + j * site_spacing + well_y * well_spacing
 46
 47    ndim_non_spatial = len(shape) - 2
 48    world_pos = ndim_non_spatial * [0] + [x, y]
 49    return world_pos
 50
 51
 52def add_grid_sources(
 53    grid_sources, well_positions, well_shape, well_spacing, site_spacing, add_source, source_settings=None
 54):
 55    """@private
 56    """
 57    if source_settings is None:
 58        source_settings = {}
 59    for channel_name, well_sources in grid_sources.items():
 60        shape = None
 61        settings = source_settings.get(channel_name, {})
 62        channel_layers = []
 63        for well_name, sources in well_sources.items():
 64            well_x, well_y = well_positions[well_name]
 65            for pos, source in enumerate(sources):
 66                layer = add_source(source, name=f"{channel_name}_{well_name}_{pos}", **settings)
 67                if shape is None:
 68                    shape = source.shape
 69                    if "scale" in settings:
 70                        scale = settings["scale"]
 71                        assert len(scale) == len(shape)
 72                        shape = tuple(sc * sh for sc, sh in zip(shape, scale))
 73                else:
 74                    assert source.shape == shape, f"{source.shape}, {shape}"
 75                world_pos = get_world_position(well_x, well_y, pos, well_shape, well_spacing, site_spacing, shape)
 76                layer.translate = world_pos
 77                channel_layers.append(layer)
 78        link_layers(channel_layers)
 79    return shape
 80
 81
 82def _add_layout(viewer, name, box_names, boxes, edge_width, measurements=None, color="coral"):
 83    properties = {"names": box_names}
 84    text = "{names}"
 85    if measurements is not None:
 86        text += " - "
 87        for measure_name, measure_values in measurements.items():
 88            this_measurements = [measure_values[box_name] for box_name in box_names]
 89            properties[measure_name] = this_measurements
 90            if isinstance(this_measurements[0], float):
 91                text += f"{measure_name}: {{{measure_name}:0.2f}},"
 92            else:
 93                text += f"{measure_name}: {{{measure_name}}},"
 94        # get rid of the last comma
 95        text = text[:-1]
 96    text_properties = {
 97        "text": text,
 98        "anchor": "upper_left",
 99        "translation": [-5, 0],
100        "size": 32,
101        "color": "coral"
102    }
103    viewer.add_shapes(boxes,
104                      name=name,
105                      properties=properties,
106                      text=text_properties,
107                      shape_type="rectangle",
108                      edge_width=edge_width,
109                      edge_color="coral",
110                      face_color="transparent")
111
112
113def add_plate_layout(
114    viewer, well_names, well_positions, well_shape, well_spacing, site_spacing, shape,
115    measurements=None
116):
117    """@private
118    """
119    well_boxes = []
120    for well_name in well_names:
121        well_x, well_y = well_positions[well_name]
122        xmin, ymin = get_world_position(
123            well_x, well_y, 0, well_shape, well_spacing, site_spacing, shape
124        )[-2:]
125        xmin -= well_spacing // 2
126        ymin -= well_spacing // 2
127
128        xmax, ymax = get_world_position(
129            well_x + 1, well_y + 1, 0, well_shape, well_spacing, site_spacing, shape
130        )[-2:]
131        xmax += well_spacing // 2
132        ymax += well_spacing // 2
133
134        well_boxes.append(np.array([[xmin, ymin], [xmax, ymax]]))
135    _add_layout(viewer, "wells", well_names, well_boxes, well_spacing // 2,
136                measurements=measurements)
137
138
139def set_camera(viewer, well_start, well_stop, well_shape, well_spacing, site_spacing, shape):
140    """@private
141    """
142    pix_start = get_world_position(
143        well_start[0], well_start[1], 0, well_shape, well_spacing, site_spacing, shape
144    )[-2:]
145    pix_stop = get_world_position(
146        well_stop[0], well_stop[1], 0, well_shape, well_spacing, site_spacing, shape
147    )[-2:]
148    camera_center = tuple((start + stop) // 2 for start, stop in zip(pix_start, pix_stop))
149    viewer.camera.center = camera_center
150
151    # zoom out so that we see all wells
152    plate_extent = [wstop - wstart for wstart, wstop in zip(well_start, well_stop)]
153    max_extent = max(plate_extent)
154    max_len = well_shape[np.argmax(plate_extent)]
155    viewer.camera.zoom /= (max_len * max_extent)
156
157
158def view_plate(
159    image_data: Dict[str, Dict[str, List[np.ndarray]]],
160    label_data: Optional[Dict[str, Dict[str, List[np.ndarray]]]] = None,
161    image_settings: Optional[Dict[str, Dict]] = None,
162    label_settings: Optional[Dict[str, Dict]] = None,
163    well_measurements: Optional[Dict[str, Dict[str, Union[float, int, str]]]] = None,
164    well_shape: Optional[Tuple[int, int]] = None,
165    zero_based: bool = True,
166    well_spacing: int = 16,
167    site_spacing: int = 4,
168    show: bool = True,
169) -> "napari.Viewer":
170    """Visualize data from a multi-well plate using napari.
171
172    Args:
173        image_data: Dict of image sources, each dict must map the channel names to
174            a dict which maps the well names (e.g. A1, B3) to
175            the image data for this well (one array per well position).
176        label_data: Dict of label sources, each dict must map the label name to
177            a dict which maps the well names (e.g. A1, B3) to
178            the label data for this well (one array per well position).
179        image_settings: Image settings for the channels.
180        label_settings: Settings for the label layers.
181        well_measurements: Measurements associated with the wells.
182        well_shape: the 2D shape of a well in terms of images, if not given will be derived.
183            Well shape can only be derived for square wells and must be passed otherwise.
184        zero_based: Whether the well indexing is zero-based.
185        well_spacing: Spacing between wells, in pixels.
186        site_spacing: Spacing between sites, in pixels.
187        show: Whether to show the viewer.
188
189    Returns:
190        The napari viewer.
191    """
192    assert napari is not None and link_layers is not None, "Requires napari"
193
194    # find the number of positions per well
195    first_channel_sources = next(iter(image_data.values()))
196    pos_per_well = len(next(iter(first_channel_sources.values())))
197
198    # find the well shape
199    if well_shape is None:  # well shape can only be derived for square wells
200        assert pos_per_well in (1, 4, 9, 25, 36, 49), f"well is not square: {pos_per_well}"
201        well_len = int(np.sqrt(pos_per_well))
202        well_shape = (well_len, well_len)
203    else:
204        assert len(well_shape) == 2
205        pos_per_well_exp = np.prod(list(well_shape))
206        assert pos_per_well_exp == pos_per_well, f"{pos_per_well_exp} != {pos_per_well}"
207
208    def process_sources(sources, well_names):
209        for well_sources in sources.values():
210            # make sure all wells have the same number of label
211            n_pos_well = [len(sources) for sources in well_sources.values()]
212            assert all(n_pos == pos_per_well for n_pos in n_pos_well), f"{pos_per_well} != {n_pos_well}"
213            well_names.extend(list(well_sources.keys()))
214        return well_names
215
216    # find the well names for all sources
217    well_names = process_sources(image_data, [])
218    if label_data is not None:
219        well_names = process_sources(label_data, well_names)
220    well_names = list(set(well_names))
221    well_names.sort()
222
223    # compute the well extent and well positions
224    well_positions, well_start, well_stop = parse_wells(well_names, zero_based)
225    assert len(well_positions) == len(well_names)
226
227    # start the veiwer and add all sources
228    viewer = napari.Viewer()
229    shape = add_grid_sources(
230        image_data, well_positions, well_shape, well_spacing, site_spacing, viewer.add_image, image_settings
231    )
232    if label_data is not None:
233        add_grid_sources(
234            label_data, well_positions, well_shape, well_spacing, site_spacing, viewer.add_labels, label_settings
235        )
236
237    # add shape layer corresponding to the well positions
238    add_plate_layout(
239        viewer, well_names, well_positions, well_shape, well_spacing, site_spacing, shape,
240        measurements=well_measurements
241    )
242
243    # set the camera so that the initial view is centered around the existing wells
244    # and zoom out so that the central well is fully visible
245    set_camera(viewer, well_start, well_stop, well_shape, well_spacing, site_spacing, shape)
246
247    if show:
248        napari.run()
249    return viewer
250
251
252def add_positional_sources(positional_sources, positions, add_source, source_settings=None):
253    """@private
254    """
255    if source_settings is None:
256        source_settings = {}
257    for channel_name, sources in positional_sources.items():
258        settings = source_settings.get(channel_name, {})
259        channel_layers = []
260        for sample, source in sources.items():
261            layer = add_source(source, name=f"{channel_name}_{sample}", **settings)
262            position = positions[sample]
263            if len(source.shape) > len(position):
264                ndim_non_spatial = len(source.shape) - len(position)
265                position = ndim_non_spatial * [0] + list(position)
266            layer.translate = list(position)
267            channel_layers.append(layer)
268            shape = source.shape
269        link_layers(channel_layers)
270    return shape
271
272
273def set_camera_positional(viewer, positions, shape):
274    """@private
275    """
276    coords = list(positions.values())
277    y = [coord[0] for coord in coords]
278    x = [coord[1] for coord in coords]
279    ymin, ymax = min(y), max(y)
280    xmin, xmax = min(x), max(x)
281
282    camera_center = [(ymax - ymin) // 2, (xmax - xmin) // 2]
283    viewer.camera.center = camera_center
284
285    extent = (ymax - ymin, xmax - xmin)
286    max_extent = max(extent)
287    zoom_out = max_extent / shape[np.argmax(extent)]
288    viewer.camera.zoom /= zoom_out
289
290
291def add_positional_layout(viewer, positions, shape, measurements=None, spacing=16):
292    """@private
293    """
294    boxes = []
295    sample_names = []
296    for sample, position in positions.items():
297        ymin, xmin = position
298        ymax, xmax = ymin + shape[0], xmin + shape[1]
299
300        xmin -= spacing
301        ymin -= spacing
302        xmax += spacing
303        ymax += spacing
304
305        boxes.append(np.array([[ymin, xmin], [ymax, xmax]]))
306        sample_names.append(sample)
307    _add_layout(viewer, "samples", sample_names, boxes, spacing // 2, measurements=measurements)
308
309
310def view_positional_images(
311    image_data: Dict[str, Dict[str, np.ndarray]],
312    positions: Dict[str, Tuple[int, int]],
313    label_data: Optional[Dict[str, Dict[str, np.ndarray]]] = None,
314    image_settings: Optional[Dict[str, Dict]] = None,
315    label_settings: Optional[Dict[str, Dict]] = None,
316    sample_measurements: Optional[Dict[str, Dict[str, Union[float, int, str]]]] = None,
317    show: bool = True,
318) -> "napari.Viewer":
319    """Similar to `view_plate`, but using position data to place the images.
320
321    Args:
322        image_data: The image data (outer dict is channels, inner is samples).
323        positions: The position for each sample.
324        label_data: The label data (outer dict is channels, inner is sample).
325        image_settings: Image settings for the channels.
326        label_settings: Settings for the label data.
327        sample_measurements: Measurements associated with the samples.
328        show: Whether to show the viewer.
329
330    Returns:
331        The napari viewer.
332    """
333    assert napari is not None and link_layers is not None, "Requires napari"
334
335    all_samples = []
336    for sources in image_data.values():
337        all_samples.extend(list(sources.keys()))
338    if label_data is not None:
339        for sources in label_data.values():
340            all_samples.extend(list(sources.keys()))
341
342    # make sure we have positional data for all the samples
343    assert all(sample in positions for sample in all_samples)
344
345    viewer = napari.Viewer()
346
347    shape = add_positional_sources(image_data, positions, viewer.add_image, image_settings)
348    if label_data is not None:
349        add_positional_sources(label_data, positions, viewer.add_labels, label_settings)
350
351    add_positional_layout(viewer, positions, shape, measurements=sample_measurements)
352
353    set_camera_positional(viewer, positions, shape)
354
355    if show:
356        napari.run()
357    return viewer