🆕Multichannel Image Reading

pyproject.toml

@@ -105,7 +105,7 @@ lint.select = [
   "ASYNC", # flake8-async
 ]
 # Ignore rules which conflict with ruff formatter.
-lint.ignore = ["COM812", "ISC001",]
+lint.ignore = ["COM812", "ISC001"]
 # Allow Ruff to discover `*.ipynb` files.
 include = ["*.py", "*.pyi", "**/pyproject.toml", "*.ipynb"]
 

tests/conftest.py

@@ -124,6 +124,16 @@ def sample_svs(remote_sample: Callable) -> Path:
     return remote_sample("svs-1-small")
 
 
+@pytest.fixture(scope="session")
+def sample_qptiff(remote_sample: Callable) -> Path:
+    """Sample pytest fixture for qptiff images.
+
+    Download qptiff image for pytest.
+
+    """
+    return remote_sample("qptiff_sample")
+
+
 @pytest.fixture(scope="session")
 def sample_ome_tiff(remote_sample: Callable) -> Path:
     """Sample pytest fixture for ome-tiff (brightfield pyramid) images.

tests/models/test_patch_predictor.py

@@ -429,6 +429,23 @@ def __getitem__(self: Proto, idx: int) -> object:
     assert roi1.shape[1] == roi2.shape[1]
     assert np.min(correlation) > 0.9, correlation
 
+    # check automask works if only mpp in metadata
+    reader = WSIReader.open(mini_wsi_svs)
+    meta = reader.info
+    meta.objective_power = None
+    reader._m_info = meta
+    ds = WSIPatchDataset(
+        img_path=mini_wsi_svs,
+        mode="wsi",
+        patch_input_shape=patch_size,
+        stride_shape=stride_size,
+        auto_get_mask=True,
+        resolution=0.5,
+        units="mpp",
+    )
+    mask_reader = ds._setup_mask_reader(None, reader, auto_get_mask=True)
+    assert mask_reader is not None
+
 
 def test_patch_dataset_abc() -> None:
     """Test for ABC methods.

tests/test_app_bokeh.py

@@ -123,6 +123,24 @@ def get_renderer_prop(prop: str) -> json:
     return resp.json()
 
 
+def get_channel_ui_elements() -> tuple[
+    bkmodels.DataTable,
+    bkmodels.DataTable,
+    bkmodels.ColorPicker,
+    bkmodels.Button,
+    bkmodels.Slider,
+]:
+    """Return channel selection UI widgets."""
+    channel_select = main.UI["channel_select"]
+    inner_column = channel_select.children[1]
+    table_row = inner_column.children[0]
+    channel_table, color_table = table_row.children
+    picker_row = inner_column.children[1]
+    color_picker, apply_button = picker_row.children
+    enhance_slider = inner_column.children[2]
+    return channel_table, color_table, color_picker, apply_button, enhance_slider
+
+
 @pytest.fixture(scope="module", autouse=True)
 def annotation_path(data_path: dict[str, Path]) -> dict[str, object]:
     """Download some testing slides and overlays.
@@ -143,6 +161,10 @@ def annotation_path(data_path: dict[str, Path]) -> dict[str, object]:
         "patch-extraction-vf",
         data_path["base_path"] / "slides",
     )
+    data_path["qptiff"] = fetch_sample_to_dir(
+        "qptiff_sample",
+        data_path["base_path"] / "slides",
+    )
     data_path["annotations"] = fetch_sample_to_dir(
         "annotation_store_svs_1",
         data_path["base_path"] / "overlays",
@@ -185,6 +207,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", port=int(main.port), threaded=True)
 
@@ -195,7 +218,21 @@ def doc(data_path: dict[str, object]) -> Generator[Document, object, None]:
     # start tile server
     p = multiprocessing.Process(target=run_app, daemon=True)
     p.start()
-    time.sleep(2)  # allow time for server to start
+    # wait until server is ready
+    start = time.time()
+    url = f"http://127.0.0.1:{main.port}/tileserver/session_id"
+    while True:
+        try:
+            resp = requests.get(url, timeout=1)
+            if resp.status_code == 200:
+                break
+        except requests.RequestException:
+            pass
+        if time.time() - start > 10:
+            p.terminate()
+            msg = f"Tileserver failed to start within 10s: {url}"
+            raise RuntimeError(msg)
+        time.sleep(0.2)
 
     main.doc_config.set_sys_args(argv=["dummy_str", str(data_path["base_path"])])
     handler = FunctionHandler(main.doc_config.setup_doc)
@@ -249,8 +286,8 @@ def test_config_loaded(data_path: pytest.TempPathFactory) -> None:
 def test_slide_select(doc: Document, data_path: pytest.TempPathFactory) -> None:
     """Test slide selection."""
     slide_select = doc.get_model_by_name("slide_select0")
-    # check there are three available slides
-    assert len(slide_select.options) == 3
+    # check there are four available slides
+    assert len(slide_select.options) == 4
     assert slide_select.options[0][0] == data_path["slide1"].name
 
     # select a slide and check it is loaded
@@ -272,7 +309,7 @@ def test_dual_window(doc: Document, data_path: pytest.TempPathFactory) -> None:
     doc.get_model_by_name("slide_windows")
     control_tabs.active = 1
     slide_select = doc.get_model_by_name("slide_select1")
-    assert len(slide_select.options) == 3
+    assert len(slide_select.options) == 4
     assert slide_select.options[0][0] == data_path["slide1"].name
 
     control_tabs.active = 0
@@ -309,13 +346,32 @@ def test_add_slide_layer(doc: Document, data_path: pytest.TempPathFactory) -> No
 
 def test_transform_overlay(doc: Document, data_path: pytest.TempPathFactory) -> None:
     """Test adding a transform overlay."""
-    layer_drop = doc.get_model_by_name("layer_drop0")
-    affine_layer_path = str(data_path["affine_trans"])  # sample .npy file
 
-    click = MenuItemClick(layer_drop, affine_layer_path)
-    layer_drop._trigger_event(click)
+    class DummyResponse:
+        """Dummy response for mocking requests.put()."""
 
-    assert len(layer_drop.menu) == 6
+        text = json.dumps("dummy.npy")
+        status_code = 200
+
+    def dummy_put(*_: object, **__: object) -> DummyResponse:
+        """Dummy put method to replace requests.Session.put()."""
+        return DummyResponse()
+
+    # Patch the method on the Session class
+    old_put = requests.sessions.Session.put
+    requests.sessions.Session.put = dummy_put
+
+    try:
+        layer_drop = doc.get_model_by_name("layer_drop0")
+        affine_layer_path = str(data_path["affine_trans"])
+
+        click = MenuItemClick(layer_drop, affine_layer_path)
+        layer_drop._trigger_event(click)
+
+        assert len(layer_drop.menu) == 6
+
+    finally:
+        requests.sessions.Session.put = old_put
 
 
 def test_add_annotation_layer(doc: Document, data_path: pytest.TempPathFactory) -> None:
@@ -855,7 +911,7 @@ def test_option_buttons() -> None:
 def test_populate_slide_list(doc: Document, data_path: pytest.TempPathFactory) -> None:
     """Test populating the slide list."""
     slide_select = doc.get_model_by_name("slide_select0")
-    assert len(slide_select.options) == 3
+    assert len(slide_select.options) == 4
     main.populate_slide_list(
         data_path["base_path"] / "slides",
         search_txt="TCGA-HE-7130-01Z-00-DX1",
@@ -864,7 +920,77 @@ def test_populate_slide_list(doc: Document, data_path: pytest.TempPathFactory) -
     main.populate_slide_list(
         data_path["base_path"] / "slides",
     )
-    assert len(slide_select.options) == 3
+    assert len(slide_select.options) == 4
+
+
+def test_channel_color_ui_callbacks(
+    doc: Document,
+    data_path: pytest.TempPathFactory,
+) -> None:
+    """Test channel color selection and apply changes callbacks on qptiff."""
+    slide_select = doc.get_model_by_name("slide_select0")
+    slide_select.value = [data_path["qptiff"].name]
+    assert main.UI["vstate"].slide_path == data_path["qptiff"]
+
+    channel_table, color_table, color_picker, apply_button, _ = (
+        get_channel_ui_elements()
+    )
+    # check we see 5 channels
+    assert len(channel_table.source.data["channels"]) == 5
+
+    # if no channels selected, check apply button does nothing
+    old_colors = color_table.source.data["colors"].copy()
+    color_picker.color = "#ffff00"
+    click = ButtonClick(apply_button)
+    apply_button._trigger_event(click)
+    assert color_table.source.data["colors"] == old_colors
+
+    # select the first channel and set it to red
+    channel_index = 0
+    color_table.source.selected.indices = [channel_index]
+    color_picker.color = "#ff0000"
+    channel_table.source.selected.indices = [channel_index]
+    click = ButtonClick(apply_button)
+    apply_button._trigger_event(click)
+    assert color_table.source.data["colors"] != old_colors
+
+    # check that getting a tile now red
+    tile = get_tile("slide", 0, 0, 0, show=False).astype(np.float32)
+    sum_r = tile[:, :, 0].sum()
+    sum_gb = tile[:, :, 1:].sum()
+    assert sum_r > 0
+    # may be tiny non-zero g and b values due to webp compression
+    # but should be almost pure red
+    assert (sum_gb) / (sum_r + 1e-5) < 0.1
+
+
+def test_enhance_slider_callback(
+    doc: Document,
+    data_path: pytest.TempPathFactory,
+) -> None:
+    """Test enhance slider callback on qptiff."""
+    slide_select = doc.get_model_by_name("slide_select0")
+    slide_select.value = [data_path["qptiff"].name]
+    assert main.UI["vstate"].slide_path == data_path["qptiff"]
+
+    channel_table, color_table, color_picker, apply_button, enhance_slider = (
+        get_channel_ui_elements()
+    )
+    assert len(channel_table.source.data["channels"]) > 0
+
+    channel_index = 0
+    color_table.source.selected.indices = [channel_index]
+    color_picker.color = "#ff0000"
+    channel_table.source.selected.indices = [channel_index]
+    click = ButtonClick(apply_button)
+    apply_button._trigger_event(click)
+
+    before = get_tile("slide", 0, 0, 0, show=False).astype(np.float32)
+    enhance_slider.value = 2.0
+    after = get_tile("slide", 0, 0, 0, show=False).astype(np.float32)
+    # enhance should have made it brighter
+    diff = after - before
+    assert np.max(diff) > 0
 
 
 def test_clearing_doc(doc: Document) -> None:

tests/test_bokeh_main_urls.py

@@ -0,0 +1,80 @@
+"""Tests for channel information retrieval and update in the Bokeh app."""
+
+import json
+from types import SimpleNamespace
+from typing import Any
+
+from tiatoolbox.visualization.bokeh_app import main
+
+
+class DummySession:
+    """A minimal session mock that records GET and PUT requests."""
+
+    def __init__(self) -> None:
+        """Initialize tracking attributes for request inspection."""
+        self.last_get: str | None = None
+        self.last_put: str | None = None
+        self.last_put_data: Any = None
+
+    def get(self, url: str) -> SimpleNamespace:
+        """Record GET requests and return a dummy successful response."""
+        self.last_get = url
+        return SimpleNamespace(
+            text=json.dumps({"channels": {"c0": [1, 0, 0]}, "active": []}),
+            status_code=200,
+        )
+
+    def put(self, url: str, data: str | None = None) -> SimpleNamespace:
+        """Record PUT requests and return a dummy successful response."""
+        self.last_put = url
+        self.last_put_data = data
+        return SimpleNamespace(status_code=200)
+
+
+def test_get_channel_info_uses_configured_port() -> None:
+    """Ensure get_channel_info uses the configured host and port."""
+    dummy = DummySession()
+    # replace UI with a minimal mapping exposing 's'
+    old_ui = getattr(main, "UI", None)
+    main.UI = {"s": dummy}
+    old_host2 = getattr(main, "host2", "127.0.0.1")
+    old_port = getattr(main, "port", "5000")
+    try:
+        main.host2 = "127.0.0.1"
+        main.port = "12345"
+        channels, _active = main.get_channel_info()
+        expected = f"http://{main.host2}:{main.port}/tileserver/channels"
+        assert dummy.last_get == expected
+        assert "c0" in channels
+    finally:
+        # restore
+        if old_ui is None:
+            delattr(main, "UI")
+        else:
+            main.UI = old_ui
+        main.host2 = old_host2
+        main.port = old_port
+
+
+def test_set_channel_info_uses_configured_port() -> None:
+    """Ensure set_channel_info uses the configured host and port."""
+    dummy = DummySession()
+    old_ui = getattr(main, "UI", None)
+    main.UI = {"s": dummy}
+    old_host2 = getattr(main, "host2", "127.0.0.1")
+    old_port = getattr(main, "port", "5000")
+
+    main.host2 = "127.0.0.1"
+    main.port = "54321"
+    expected = f"http://{main.host2}:{main.port}/tileserver/channels"
+    try:
+        colors = {"c0": "#ff0000"}
+        main.set_channel_info(colors, [0])
+    finally:
+        assert dummy.last_put == expected
+        if old_ui is None:
+            delattr(main, "UI")
+        else:
+            main.UI = old_ui
+        main.host2 = old_host2
+        main.port = old_port