🆕Multichannel Image Reading

tests/test_app_bokeh.py

@@ -138,6 +138,7 @@ def run_app() -> None:
         title="Tiatoolbox TileServer",
         layers={},
     )
+    app.json.sort_keys = False
     CORS(app, send_wildcard=True)
     app.run(host="127.0.0.1", threaded=True)
 

tests/test_wsireader.py

@@ -1482,6 +1482,7 @@ def test_wsireader_open(
     sample_ome_tiff: Path,
     sample_ventana_tif: Path,
     sample_regular_tif: Path,
+    sample_qptiff: Path,
     source_image: Path,
     tmp_path: pytest.TempPathFactory,
 ) -> None:
@@ -1513,6 +1514,9 @@ def test_wsireader_open(
     wsi = WSIReader.open(Path(source_image))
     assert isinstance(wsi, wsireader.VirtualWSIReader)
 
+    wsi = WSIReader.open(sample_qptiff)
+    assert isinstance(wsi, wsireader.TIFFWSIReader)
+
     img = utils.misc.imread(str(Path(source_image)))
     wsi = WSIReader.open(input_img=img)
     assert isinstance(wsi, wsireader.VirtualWSIReader)
@@ -2581,6 +2585,11 @@ def test_jp2_no_header(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
             "sample_key": "jp2-omnyx-small",
             "kwargs": {},
         },
+        {
+            "reader_class": TIFFWSIReader,
+            "sample_key": "qptiff_sample",
+            "kwargs": {},
+        },
     ],
     ids=[
         "AnnotationReaderOverlaid",
@@ -2591,6 +2600,7 @@ def test_jp2_no_header(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
         "NGFFWSIReader",
         "OpenSlideWSIReader (Small SVS)",
         "OmnyxJP2WSIReader",
+        "TIFFReader_Multichannel",
     ],
 )
 def wsi(request: requests.request, remote_sample: Callable) -> WSIReader:
@@ -2812,3 +2822,15 @@ def test_read_multi_channel(source_image: Path) -> None:
     assert region.shape == (100, 50, (new_img_array.shape[-1]))
     assert np.abs(np.median(region.astype(int) - target.astype(int))) == 0
     assert np.abs(np.mean(region.astype(int) - target.astype(int))) < 0.2
+
+
+def test_visualise_multi_channel(sample_qptiff: Path) -> None:
+    """Test visualising a multi-channel qptiff multiplex image."""
+    wsi = wsireader.TIFFWSIReader(sample_qptiff, post_proc="auto")
+    wsi2 = wsireader.TIFFWSIReader(sample_qptiff, post_proc=None)
+
+    region = wsi.read_rect(location=(0, 0), size=(50, 100))
+    region2 = wsi2.read_rect(location=(0, 0), size=(50, 100))
+
+    assert region.shape == (100, 50, 3)
+    assert region2.shape == (100, 50, 7)

tiatoolbox/cli/visualize.py

@@ -25,6 +25,7 @@ def run_app() -> None:
             title="Tiatoolbox TileServer",
             layers={},
         )
+        app.json.sort_keys = False
         CORS(app, send_wildcard=True)
         app.run(host="127.0.0.1", threaded=True)
 

tiatoolbox/data/remote_samples.yaml

@@ -145,3 +145,7 @@ files:
     url: [ *testdata, "annotation/test2_config.json"]
   nuclick-output:
     url: [*modelroot, "predictions/nuclei_mask/nuclick-output.npy"]
+  qptiff_sample:
+    url: [*wsis, "multiplexed_example.qptiff"]
+  qptiff_sample_small:
+    url: [ *wsis, "multiplexed_example_small.qptiff" ]

tiatoolbox/models/dataset/classification.py

@@ -320,7 +320,11 @@ def __init__(  # skipcq: PY-R1000  # noqa: PLR0915
         elif auto_get_mask and mode == "wsi" and mask_path is None:
             # if no mask provided and `wsi` mode, generate basic tissue
             # mask on the fly
-            mask_reader = self.reader.tissue_mask(resolution=1.25, units="power")
+            try:
+                mask_reader = self.reader.tissue_mask(resolution=1.25, units="power")
+            except:
+                # if power is None, try with mpp
+                mask_reader = self.reader.tissue_mask(resolution=6.0, units="mpp")
             # ? will this mess up  ?
             mask_reader.info = self.reader.info
 

tiatoolbox/utils/postproc_defs.py

@@ -0,0 +1,121 @@
+"""Module to provide postprocessing classes."""
+
+from __future__ import annotations
+
+import colorsys
+import warnings
+
+import numpy as np
+
+
+class MultichannelToRGB:
+    """Class to convert multi-channel images to RGB images."""
+
+    def __init__(
+        self: MultichannelToRGB,
+        color_dict: list[tuple[float, float, float]] | None = None,
+    ) -> None:
+        """Initialize the MultichannelToRGB converter.
+
+        Args:
+            color_dict: Dict of channel names with RGB colors for each channel. If not
+            provided, a set of distinct colors will be auto-generated.
+
+        """
+        self.colors = None
+        self.color_dict = color_dict
+        self.is_validated = False
+        self.channels = None
+        self.enhance = 1.0
+
+    def validate(self: MultichannelToRGB, n: int) -> None:
+        """Validate the input color_dict on first read from image.
+
+        Checks that n is either equal to the number of colors provided, or is
+        one less. In the latter case it is assumed that the last channel is background
+        autofluorescence and is not in the tiff and we will drop it from
+        the color_dict with a warning.
+
+        Args:
+            n (int): Number of channels
+
+        """
+        n_colors = len(self.colors)
+        if n_colors == n:
+            self.is_validated = True
+            return
+
+        if n_colors - 1 == n:
+            self.colors = self.colors[:n]
+            self.channels = [c for c in self.channels if c < n]
+            self.is_validated = True
+            warnings.warn(
+                """Number of channels in image is one less than number of channels in
+                dict. Assuming last channel is background autofluorescence and ignoring
+                it. If this is not the case please provide a manual color_dict.""",
+                stacklevel=2,
+            )
+            return
+
+        msg = f"Number of colors: {n_colors} does not match channels in image: {n}."
+        raise ValueError(msg)
+
+    def generate_colors(self: MultichannelToRGB, n_channels: int) -> np.ndarray:
+        """Generate a set of visually distinct colors.
+
+        Args:
+            n_channels (int): Number of channels/colors to generate
+
+        Returns:
+            np.ndarray: Array of RGB colors
+
+        """
+        self.color_dict = {
+            f"channel_{i}": colorsys.hsv_to_rgb(i / n_channels, 1, 1)
+            for i in range(n_channels)
+        }
+
+    def __call__(self: MultichannelToRGB, image: np.ndarray) -> np.ndarray:
+        """Convert a multi-channel image to an RGB image.
+
+        Args:
+            image (np.ndarray): Input image of shape (H, W, N)
+
+        Returns:
+            np.ndarray: RGB image of shape (H, W, 3)
+
+        """
+        n = image.shape[2]
+
+        if n < 5:  # noqa: PLR2004
+            # assume already rgb(a) so just return image
+            return image
+
+        if self.colors is None:
+            self.generate_colors(n)
+
+        if not self.is_validated:
+            self.validate(n)
+
+        # Convert to RGB image
+        rgb_image = (
+            np.einsum(
+                "hwn,nc->hwc",
+                image[:, :, self.channels],
+                self.colors[self.channels, :],
+                optimize=True,
+            )
+            * self.enhance
+        )
+
+        # Clip  to ensure in valid range and return
+        return np.clip(rgb_image, 0, 255).astype(np.uint8)
+
+    def __setattr__(self: MultichannelToRGB, name: str, value: np.Any) -> None:
+        """Ensure that colors is updated if color_dict is updated."""
+        if name == "color_dict" and value is not None:
+            self.colors = np.array(list(value.values()), dtype=np.float32)
+            if self.channels is None:
+                self.channels = list(range(len(value)))
+
+        super().__setattr__(name, value)