♻️ Refactor WSIReader.open

tests/test_tiffreader.py

@@ -11,6 +11,9 @@
 from defusedxml import ElementTree
 from PIL import Image
 
+from tiatoolbox.wsicore.wsireader import base
+from tiatoolbox.wsicore.wsireader.factory import try_tiff
+
 if TYPE_CHECKING:
     from collections.abc import Callable
     from pathlib import Path
@@ -144,25 +147,36 @@ def __init__(
             error_msg = "Unsupported TIFF WSI format"
             raise ValueError(error_msg)
 
-    monkeypatch.setattr(wsireader, "TIFFWSIReader", DummyTIFFWSIReader)
+    def raise_unsupported(*args, **kwargs) -> None:  # noqa: ARG001, ANN002, ANN003
+        """Help raise unsupported TIFF formats error."""
+        msg = "Unsupported TIFF WSI format"
+        raise ValueError(msg)
+
+    monkeypatch.setattr(
+        wsireader.base,
+        "_handle_tiff_wsi",
+        raise_unsupported,
+    )
+
+    monkeypatch.setattr(wsireader.base, "TIFFWSIReader", DummyTIFFWSIReader)
 
     dummy_file = track_tmp_path / "dummy.tiff"
     dummy_img = np.zeros((10, 10, 3), dtype=np.uint8)
     cv2.imwrite(str(dummy_file), dummy_img)
 
-    reader = wsireader.WSIReader.try_tiff(dummy_file, ".tiff", None, None, None)
+    reader = try_tiff(dummy_file, ".tiff", None, None, None)
     assert isinstance(reader, wsireader.VirtualWSIReader)
 
 
-def test_try_tiff_raises_other_valueerror(
+def test_try_tiff_raises_other_value_error(
     monkeypatch: pytest.MonkeyPatch, track_tmp_path: Path
 ) -> None:
     """Test try_tiff raises ValueError if not an unsupported TIFF format."""
     tiff_path = track_tmp_path / "test.tiff"
     Image.new("RGB", (10, 10), color="white").save(tiff_path)
 
     # Patch TIFFWSIReader to raise a different ValueError
-    def raise_other_valueerror(*args: object, **kwargs: object) -> None:
+    def raise_other_value_error(*args: object, **kwargs: object) -> None:
         """Raise a ValueError emulating a non-unsupported TIFF error.
 
         This helper is patched in tests to ensure that `WSIReader.try_tiff`
@@ -174,10 +188,14 @@ def raise_other_valueerror(*args: object, **kwargs: object) -> None:
         msg = "Some other TIFF error"
         raise ValueError(msg)
 
-    monkeypatch.setattr(wsireader.base, "TIFFWSIReader", raise_other_valueerror)
+    monkeypatch.setattr(
+        base,
+        "_handle_tiff_wsi",
+        raise_other_value_error,
+    )
 
     with pytest.raises(ValueError, match="Some other TIFF error"):
-        wsireader.WSIReader.try_tiff(
+        try_tiff(
             input_path=tiff_path,
             last_suffix=".tiff",
             mpp=(0.5, 0.5),

tests/test_wsireader.py

@@ -53,12 +53,26 @@
     VirtualWSIReader,
     _handle_tiff_wsi,
     _handle_virtual_wsi,
+    detection,
     is_dicom,
     is_ngff,
     is_tiled_tiff,
     is_url,
     is_zarr,
 )
+from tiatoolbox.wsicore.wsireader.detection import is_valid_zarr_fsspec
+from tiatoolbox.wsicore.wsireader.factory import (
+    _handle_special_cases,
+    _validate_input,
+    try_annotation_store,
+    try_dicom,
+    try_fsspec,
+    try_ngff,
+    try_ome_tiff,
+    try_openslide,
+    try_tiff,
+    verify_supported_wsi,
+)
 
 if TYPE_CHECKING:  # pragma: no cover
     from collections.abc import Callable, Iterable
@@ -780,7 +794,7 @@ def test_read_rect_tiffreader_ome_tiff_baseline(sample_ome_tiff: Path) -> None:
 def test_is_tiled_tiff(source_image: Path) -> None:
     """Test if source_image is a tiled tiff."""
     source_image.replace(source_image.with_suffix(".tiff"))
-    assert wsireader.is_tiled_tiff(source_image.with_suffix(".tiff")) is False
+    assert detection.is_tiled_tiff(source_image.with_suffix(".tiff")) is False
     source_image.with_suffix(".tiff").replace(source_image)
 
 
@@ -792,7 +806,7 @@ def test_is_not_tiled_tiff(tmp_samples_path: Path) -> None:
     with tifffile.TiffWriter(temp_tiff_path) as tif:
         for image in images:
             tif.write(image, compression=None, tile=None)
-    assert wsireader.is_tiled_tiff(temp_tiff_path) is False
+    assert detection.is_tiled_tiff(temp_tiff_path) is False
 
 
 def test_read_rect_openslide_levels(sample_ndpi: Path) -> None:
@@ -2433,7 +2447,7 @@ def test_ngff_missing_multiscales_returns_false(
     del zattrs["multiscales"]
     with Path.open(sample_copy / ".zattrs", "w") as fh:
         json.dump(zattrs, fh, indent=2)
-    assert not wsireader.is_ngff(sample_copy)
+    assert not detection.is_ngff(sample_copy)
 
 
 def test_ngff_wrong_format_metadata(
@@ -2453,7 +2467,7 @@ def test_ngff_wrong_format_metadata(
     with Path.open(sample_copy / ".zattrs", "w") as fh:
         json.dump(zattrs, fh, indent=2)
     with caplog.at_level(logging.WARNING):
-        assert not wsireader.is_ngff(sample_copy)
+        assert not detection.is_ngff(sample_copy)
     assert "must be present and of the correct type" in caplog.text
 
 
@@ -3122,7 +3136,7 @@ def test_fsspec_json_wsi_reader_instantiation() -> None:
 
     with (
         patch(
-            "tiatoolbox.wsicore.wsireader.base.FsspecJsonWSIReader.is_valid_zarr_fsspec",
+            "tiatoolbox.wsicore.wsireader.factory.is_valid_zarr_fsspec",
             return_value=True,
         ),
         patch("tiatoolbox.wsicore.wsireader.base.FsspecJsonWSIReader") as mock_reader,
@@ -3134,7 +3148,7 @@ def test_fsspec_json_wsi_reader_instantiation() -> None:
 def test_generate_fsspec_json_file_and_validate(
     sample_svs: Path, track_tmp_path: Path
 ) -> None:
-    """Test generate fsspec json file and validate it."""
+    """Test generate fsspec JSON file and validate it."""
     file_types = ("*.svs",)
 
     files_all = utils.misc.grab_files_from_dir(
@@ -3150,9 +3164,7 @@ def test_generate_fsspec_json_file_and_validate(
 
     assert Path(json_file_path).exists(), "Output JSON file was not created."
 
-    assert FsspecJsonWSIReader.is_valid_zarr_fsspec(json_file_path), (
-        "FSSPEC JSON file is invalid."
-    )
+    assert is_valid_zarr_fsspec(json_file_path), "FSSPEC JSON file is invalid."
 
 
 def test_fsspec_wsireader_info_read(sample_svs: Path, track_tmp_path: Path) -> None:
@@ -3215,17 +3227,17 @@ def test_fsspec_reader_open_invalid_json_file(track_tmp_path: Path) -> None:
     json_path = track_tmp_path / "invalid.json"
     json_path.write_text("{invalid json}")  # Corrupt JSON
 
-    assert not FsspecJsonWSIReader.is_valid_zarr_fsspec(str(json_path))
+    assert not is_valid_zarr_fsspec(str(json_path))
 
 
 def test_fsspec_reader_open_oserror_handling() -> None:
     """Ensure OSError is handled properly.
 
-    Pass non existent JSON to  FsspecJsonWSIReader.is_valid_zarr_fsspec.
+    Pass non-existent JSON to  FsspecJsonWSIReader.is_valid_zarr_fsspec.
 
     """
     with patch("builtins.open", side_effect=OSError("File not found")):
-        result = FsspecJsonWSIReader.is_valid_zarr_fsspec("non_existent.json")
+        result = is_valid_zarr_fsspec("non_existent.json")
 
     assert result is False, "Function should return False for OSError"
 
@@ -3241,7 +3253,7 @@ def test_fsspec_reader_open_pass_empty_json(track_tmp_path: Path) -> None:
     json_path = track_tmp_path / "empty.json"
     json_path.write_text("{}")
 
-    assert not FsspecJsonWSIReader.is_valid_zarr_fsspec(str(json_path))
+    assert not is_valid_zarr_fsspec(str(json_path))
 
 
 def test_fsspec_reader_group_branch(monkeypatch: pytest.MonkeyPatch) -> None:
@@ -3445,28 +3457,28 @@ def test_read_bounds_transformedreader_baseline(
 def test_wsireader_validate_input_edge_cases() -> None:
     """Test WSIReader._validate_input with various edge cases."""
     # Test with valid inputs
-    WSIReader._validate_input("test.svs")
-    WSIReader._validate_input(Path("test.svs"))
-    WSIReader._validate_input(np.array([1, 2, 3]))
+    _validate_input("test.svs")
+    _validate_input(Path("test.svs"))
+    _validate_input(np.array([1, 2, 3]))
 
     # Test with invalid inputs
     with pytest.raises(TypeError, match="Invalid input"):
-        WSIReader._validate_input(123)
+        _validate_input(123)
 
     with pytest.raises(TypeError, match="Invalid input"):
-        WSIReader._validate_input({"invalid": "dict"})
+        _validate_input({"invalid": "dict"})
 
 
 def test_wsireader_verify_supported_wsi_edge_cases(track_tmp_path: Path) -> None:
     """Test WSIReader.verify_supported_wsi with edge cases."""
     # Test with unsupported extension
     unsupported_file = track_tmp_path / "test.xyz"
     with pytest.raises(FileNotSupportedError, match="not a supported file format"):
-        WSIReader.verify_supported_wsi(unsupported_file)
+        verify_supported_wsi(unsupported_file)
 
     # Test with no extension
     no_ext_file = track_tmp_path / "test"
-    WSIReader.verify_supported_wsi(no_ext_file)
+    verify_supported_wsi(no_ext_file)
 
 
 def test_wsireader_handle_virtual_wsi_edge_cases(track_tmp_path: Path) -> None:
@@ -3502,7 +3514,7 @@ def test_wsireader_special_cases_coverage(track_tmp_path: Path) -> None:
         ValueError,
         match="No metadata found in store",
     ):
-        WSIReader._handle_special_cases(db_file, db_file, None, None, None, info=None)
+        _handle_special_cases(db_file, db_file, None, None, None, info=None)
 
 
 def test_wsireader_get_post_proc_edge_cases() -> None:
@@ -4038,29 +4050,27 @@ def test_wsireader_find_read_bounds_params_edge_cases(sample_svs: Path) -> None:
 def test_wsireader_try_methods_comprehensive() -> None:
     """Test WSIReader.try_* methods comprehensively."""
     # Test try_dicom with non-DICOM path
-    result = WSIReader.try_dicom(Path("test.txt"), None, None, None)
+    result = try_dicom(Path("test.txt"), None, None, None)
     assert result is None
 
     # Test try_fsspec with invalid input
-    result = WSIReader.try_fsspec("invalid.txt", None, None)
+    result = try_fsspec("invalid.txt", None, None)
     assert result is None
 
     # Test try_annotation_store with non-.db file
-    result = WSIReader.try_annotation_store(Path("test.txt"), ".txt", None, {})
+    result = try_annotation_store(Path("test.txt"), ".txt", None, {})
     assert result is None
 
     # Test try_ngff with non-.zarr file
-    result = WSIReader.try_ngff(Path("test.txt"), ".txt", None, None)
+    result = try_ngff(Path("test.txt"), ".txt", None, None)
     assert result is None
 
     # Test try_ome_tiff with non-OME file
-    result = WSIReader.try_ome_tiff(
-        Path("test.txt"), [".txt"], ".txt", None, None, None
-    )
+    result = try_ome_tiff(Path("test.txt"), [".txt"], ".txt", None, None, None)
     assert result is None
 
     # Test try_tiff with non-TIFF file
-    result = WSIReader.try_tiff(Path("test.txt"), ".txt", None, None, None)
+    result = try_tiff(Path("test.txt"), ".txt", None, None, None)
     assert result is None
 
 
@@ -4269,7 +4279,7 @@ def test_tiff_suffix_success(self, mock_reader: MagicMock) -> None:
         mock_instance = MagicMock()
         mock_reader.return_value = mock_instance
 
-        result: OpenSlideWSIReader | None = WSIReader.try_openslide(
+        result: OpenSlideWSIReader | None = try_openslide(
             input_path=Path("sample.tif"),
             last_suffix=".tif",
             mpp=(0.5, 0.5),
@@ -4288,7 +4298,7 @@ def test_tiff_suffix_raises_openslide_error(self, mock_reader: MagicMock) -> Non
         """Test that OpenSlide errors are caught and the function returns None."""
         mock_reader.side_effect = openslide.OpenSlideError("bad file")
 
-        result: OpenSlideWSIReader | None = WSIReader.try_openslide(
+        result: OpenSlideWSIReader | None = try_openslide(
             input_path=Path("bad.tiff"),
             last_suffix=".tiff",
             mpp=None,
@@ -4351,3 +4361,49 @@ def test_handle_tiff_wsi_returns_none_when_no_handlers_match(
 def test_is_url(input_path: str | Path, *, expected: bool) -> None:
     """Verify that is_url correctly identifies URLs and ignores local paths."""
     assert is_url(input_path) is expected
+
+
+def test_is_ngff_keyerror_returns_false(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Test is_ngff returns False when attrs access raises KeyError."""
+
+    class BrokenAttrs(dict):
+        """Dummy BrokenAttrs."""
+
+        def get(self, key, default=None) -> None:  # noqa: ANN001, ARG002
+            """Dummy get function raises error."""
+            raise KeyError(key)
+
+    class MockAttrs:
+        """Dummy MockAttrs."""
+
+        @staticmethod
+        def asdict() -> BrokenAttrs:
+            """Dummy asdict, returns BrokenAttrs."""
+            return BrokenAttrs()
+
+    class MockGroup:
+        """Mock Group."""
+
+        attrs = MockAttrs()
+
+    monkeypatch.setattr(
+        zarr,
+        "open",
+        lambda *args, **kwargs: MockGroup(),  # noqa: ARG005
+    )
+
+    # Make isinstance(zarr_group, zarr.Group) pass
+    monkeypatch.setattr(
+        zarr,
+        "Group",
+        MockGroup,
+    )
+
+    assert is_ngff("dummy.zarr") is False
+
+
+def test_is_valid_zarr_not_str_path() -> None:
+    """Tests is_valid_zarr_fsspec if not str or path."""
+    assert not is_valid_zarr_fsspec(np.zeros(0))

tiatoolbox/wsicore/__init__.py

@@ -1,8 +1,5 @@
 """Package to read whole slide images."""
 
-from numbers import Number
-from typing import TypedDict
-
 from tiatoolbox.wsicore import metadata, wsimeta, wsireader
 
 from .wsimeta import WSIMeta
@@ -15,11 +12,4 @@
     "WSIReaderParams",
 ]
 
-
-class WSIReaderParams(TypedDict, total=False):
-    """Parameters for reading whole slide images."""
-
-    meta: WSIMeta | None
-    mpp: tuple[Number, Number] | Number
-    power: Number
-    storage_options: dict  # For FsspecStore
+from .wsireader.types import WSIReaderParams

tiatoolbox/wsicore/wsireader/__init__.py

@@ -20,13 +20,8 @@
     WSIReader,
     _handle_tiff_wsi,
     _handle_virtual_wsi,
-    fix_mangled_url_by_pathlib,
-    is_dicom,
-    is_ngff,
-    is_tiled_tiff,
-    is_url,
-    is_zarr,
 )
+from .detection import is_dicom, is_ngff, is_tiled_tiff, is_url, is_zarr
 
 __all__ = [
     "AnnotationStoreReader",
@@ -44,7 +39,6 @@
     "WSIReader",
     "_handle_tiff_wsi",
     "_handle_virtual_wsi",
-    "fix_mangled_url_by_pathlib",
     "is_dicom",
     "is_ngff",
     "is_tiled_tiff",