Add ZarrTIFFWSIReader class.

Author: aacic

requirements/requirements.txt

@@ -1,5 +1,6 @@
 # torch installation
 --extra-index-url https://download.pytorch.org/whl/cu118; sys_platform != "darwin"
+aiohttp>=3.8.1
 albumentations>=1.3.0
 bokeh>=3.1.1, <3.6.0
 Click>=8.1.3

tests/zarrtiff/tiff_fsspec.py

@@ -0,0 +1,152 @@
+import json
+from datetime import datetime
+from numbers import Number
+from typing import Union
+
+from tifffile import TiffFile, TiffPages, tiff2fsspec
+
+# Path to the JSON file
+json_file_path = "../../samples/fsspec/73c69d24-6f9e-44e2-bfe5-a608d4cf5c27_fsspec.json"
+svs_file_path = "../../samples/slides/TCGA-22-1017-01Z-00-DX1.9562FE79-A261-42D3-B394-F3E0E2FF7DDA.svs"
+url_place_holder = "https://replace.me/"
+url_to_replace = (
+    url_place_holder
+    + "TCGA-22-1017-01Z-00-DX1.9562FE79-A261-42D3-B394-F3E0E2FF7DDA.svs"
+)
+final_url = "https://api.gdc.cancer.gov/data/73c69d24-6f9e-44e2-bfe5-a608d4cf5c27"
+
+
+def _parse_svs_metadata(pages: TiffPages) -> dict:
+    """Extract SVS specific metadata.
+
+    Returns:
+        dict:
+            Dictionary of kwargs for WSIMeta.
+
+    """
+    raw = {}
+    mpp = None
+    objective_power = None
+    vendor = "Aperio"
+
+    description = pages[0].description
+    raw["Description"] = description
+    parts = description.split("|")
+    description_headers, key_value_pairs = parts[0], parts[1:]
+    description_headers = description_headers.split(";")
+
+    software, photometric_info = description_headers[0].splitlines()
+    raw["Software"] = software
+    raw["Photometric Info"] = photometric_info
+
+    def parse_svs_tag(string: str) -> tuple[str, Union[Number, str]]:
+        """Parse SVS key-value string.
+
+        Infers type(s) of data by trial and error with a fallback to
+        the original string type.
+
+        Args:
+            string (str):
+                Key-value string in SVS format: "key=value".
+
+        Returns:
+            tuple:
+                Key-value pair.
+
+        """
+        pair = string.split("=")
+        if len(pair) != 2:
+            msg = "Invalid metadata. Expected string of the format 'key=value'."
+            raise ValueError(
+                msg,
+            )
+        key, value_string = pair
+        key = key.strip()
+        value_string = value_string.strip()
+
+        def us_date(string: str) -> datetime:
+            """Return datetime parsed according to US date format."""
+            return datetime.strptime(string, r"%m/%d/%y").astimezone()
+
+        def time(string: str) -> datetime:
+            """Return datetime parsed according to HMS format."""
+            return datetime.strptime(string, r"%H:%M:%S").astimezone()
+
+        casting_precedence = [us_date, time, int, float]
+        value = value_string
+        for cast in casting_precedence:
+            try:
+                value = cast(value_string)
+            except ValueError:  # noqa: PERF203
+                continue
+            else:
+                return key, value
+
+        return key, value
+
+    svs_tags = dict(parse_svs_tag(string) for string in key_value_pairs)
+    raw["SVS Tags"] = svs_tags
+    mpp = svs_tags.get("MPP")
+    if mpp is not None:
+        mpp = [mpp] * 2
+    objective_power = svs_tags.get("AppMag")
+
+    return {
+        "objective_power": objective_power,
+        "vendor": vendor,
+        "mpp": mpp,
+        "raw": raw,
+    }
+
+
+tifffilePages = TiffFile(svs_file_path).pages
+
+tiff2fsspec(svs_file_path, url="https://replace.me", out=json_file_path)
+
+
+metadata = _parse_svs_metadata(pages=tifffilePages)
+
+
+# Recursive function to convert datetime objects to JSON-compatible strings
+def convert_metadata(metadata):
+    if isinstance(metadata, dict):
+        return {key: convert_metadata(value) for key, value in metadata.items()}
+    if isinstance(metadata, list):
+        return [convert_metadata(item) for item in metadata]
+    if isinstance(metadata, datetime):
+        return metadata.isoformat()  # Convert datetime to ISO 8601 string
+    return metadata
+
+
+metadataSerializable = convert_metadata(metadata)
+
+# Read the JSON data from the file
+with open(json_file_path) as file:
+    json_data = json.load(file)
+
+
+# Decode `.zattrs` JSON string into a dictionary
+zattrs = json.loads(json_data[".zattrs"])
+
+# Update metadata into `.zattrs`
+if "multiscales" in zattrs and isinstance(zattrs["multiscales"], list):
+    zattrs["multiscales"][0]["metadata"] = metadataSerializable
+
+# Convert back to a JSON string
+json_data[".zattrs"] = json.dumps(zattrs)
+
+# Dump the updated JSON
+json_content = json.dumps(json_data, indent=2)
+
+
+def replace_url(data, output_path, old_url, new_url):
+    for key, value in data.items():
+        if isinstance(value, list) and value[0] == old_url:
+            value[0] = new_url
+
+    with open(output_path, "w") as json_file:
+        json.dump(data, json_file, indent=2)
+
+
+# Replace URLs in the JSON file
+replace_url(json_data, json_file_path, url_to_replace, final_url)

tests/zarrtiff/tileserver.py

@@ -0,0 +1,20 @@
+from flask_cors import CORS
+
+from tiatoolbox.visualization import TileServer
+from tiatoolbox.wsicore import WSIReader
+
+## Before running this script run: pip install aiohttp
+
+svs = "../../samples/fsspec/73c69d24-6f9e-44e2-bfe5-a608d4cf5c27_fsspec.json"
+
+reader = WSIReader.open(svs)
+
+# Initialize and run the TileServer
+tile_server = TileServer(
+    title="Tiatoolbox TileServer",
+    layers={"layer": reader},
+)
+CORS(tile_server, send_wildcard=True)
+
+
+tile_server.run(host="127.0.0.1", port=5000)