diff --git a/Makefile b/Makefile index 5e7c20e6..0d4a2013 100644 --- a/Makefile +++ b/Makefile @@ -42,7 +42,7 @@ test: pytest -s tests --import-mode=importlib -W ignore::DeprecationWarning style: - pre-commit run --all-files + PATH="$(CURDIR)/.venv/bin:$$PATH" .venv/bin/pre-commit run --all-files check: style test diff --git a/notebooks/wsi/01-ingest-wsi.py b/notebooks/wsi/01-ingest-wsi.py new file mode 100644 index 00000000..32213510 --- /dev/null +++ b/notebooks/wsi/01-ingest-wsi.py @@ -0,0 +1,144 @@ +# Databricks notebook source +# /// script +# [tool.databricks.environment] +# environment_version = "5" +# /// +# DBTITLE 1,WSI Ingest — Overview +# MAGIC %md +# MAGIC # Whole Slide Image (WSI) Ingest Pipeline +# MAGIC +# MAGIC Unified ingest for **all OpenSlide-supported formats**: Aperio SVS, Hamamatsu NDPI, Leica SCN, MIRAX MRXS, Philips TIFF, Ventana BIF/TIF, Sakura SVSLIDE, generic TIFF. +# MAGIC +# MAGIC **Pipeline** +# MAGIC 1. Install dependencies (`openslide-python`, `openslide-bin`, `tifffile`) +# MAGIC 2. Add `src/` to path for editable development +# MAGIC 3. Configure source paths and target Delta table +# MAGIC 4. `WSICatalog.catalog()` — discover all WSI files (multi-extension) +# MAGIC 5. `WSIMetaExtractor.transform()` — extract metadata into `meta VARIANT` +# MAGIC 6. Save enriched catalog to Delta +# MAGIC 7. Verify and explore results + +# COMMAND ---------- + +# DBTITLE 1,Install dependencies +# MAGIC %pip install openslide-python openslide-bin tifffile imagecodecs Pillow -q + +# COMMAND ---------- + +# DBTITLE 1,Add src/ to sys.path +import sys + +SRC_PATH = "/Workspace/Users/douglas.moore@databricks.com/pixels-tiff/src" +if SRC_PATH not in sys.path: + sys.path.insert(0, SRC_PATH) + print(f"Added to sys.path: {SRC_PATH}") +else: + print(f"Already on sys.path: {SRC_PATH}") + +# COMMAND ---------- + +# DBTITLE 1,Configuration +# --- Source paths (all UC Volume locations with WSI files) --- +SOURCE_PATHS = [ + "/Volumes/hls_radiology_east/orthanc_demo/raw_images/Aperio", + "/Volumes/hls_radiology_east/orthanc_demo/raw_images/Generic-TIFF", + "/Volumes/hls_radiology_east/orthanc_demo/raw_images/Philips-TIFF", + "/Volumes/hls_radiology_east/osuwmc/sample", + "/Volumes/hls_radiology_east/jsl/samples", +] + +# --- Target Delta table and volume --- +TABLE = "dmoore.wsi.object_catalog" +VOLUME = "dmoore.wsi.wsi_volume" + +print(f"Source paths: {len(SOURCE_PATHS)}") +print(f"Target table: {TABLE}") +print(f"Volume: {VOLUME}") + +# COMMAND ---------- + +# DBTITLE 1,Initialize WSICatalog +from dbx.pixels.wsi import WSICatalog, WSIMetaExtractor, OPENSLIDE_PATTERNS + +catalog = WSICatalog(spark, table=TABLE, volume=VOLUME) +print(f"Catalog: {catalog}") +print(f"WSI patterns: {OPENSLIDE_PATTERNS}") + +# COMMAND ---------- + +# DBTITLE 1,Discover & catalog all WSI files +from functools import reduce +from pyspark.sql import DataFrame + +# Catalog each source path and union results +dfs = [] +for path in SOURCE_PATHS: + try: + df = catalog.catalog(path=path) # Uses all WSI patterns by default + dfs.append(df) + print(f" ✓ {path}") + except Exception as e: + print(f" ✗ {path}: {e}") + +if dfs: + catalog_df = reduce(DataFrame.union, dfs).dropDuplicates(["path"]).repartition(8) + print(f"\nTotal WSI files discovered: {catalog_df.count()}") + display(catalog_df.select("path", "modificationTime", "length", "extension")) +else: + raise ValueError("No WSI files found in any source path") + +# COMMAND ---------- + +# DBTITLE 1,Extract WSI metadata +# WSIMetaExtractor adds a `meta` VARIANT column with: +# - All OpenSlide properties (vendor-specific: aperio.*, hamamatsu.*, philips.*, etc.) +# - Derived fields: _wsi_vendor, _wsi_format, _wsi_width, _wsi_height, _wsi_level_count, +# _wsi_mpp_x, _wsi_mpp_y, _wsi_objective_power, _wsi_has_label, _wsi_has_macro +# - phi_tag_report: PHI classification of all properties + +extractor = WSIMetaExtractor(catalog, inputCol="local_path", outputCol="meta") +enriched_df = extractor.transform(catalog_df) + +print(f"Schema: {[f.name for f in enriched_df.schema.fields]}") +display(enriched_df.select("local_path", "extension", "meta")) + +# COMMAND ---------- + +# DBTITLE 1,Save to Delta table +# Save the enriched catalog to Delta +catalog.save(enriched_df, mode="overwrite") + +print(f"✓ Saved to {TABLE}") +print(f" Rows: {spark.table(TABLE).count()}") + +# COMMAND ---------- + +# DBTITLE 1,Explore metadata — vendor breakdown +# MAGIC %sql +# MAGIC -- Vendor breakdown from extracted metadata +# MAGIC SELECT +# MAGIC meta:_wsi_vendor::STRING AS vendor, +# MAGIC meta:_wsi_format::STRING AS format, +# MAGIC meta:_wsi_backend::STRING AS backend, +# MAGIC COUNT(*) AS file_count, +# MAGIC AVG(meta:_wsi_width::INT) AS avg_width, +# MAGIC AVG(meta:_wsi_height::INT) AS avg_height, +# MAGIC AVG(meta:_wsi_mpp_x::DOUBLE) AS avg_mpp_x +# MAGIC FROM dmoore.wsi.object_catalog +# MAGIC GROUP BY 1, 2, 3 +# MAGIC ORDER BY file_count DESC + +# COMMAND ---------- + +# DBTITLE 1,Explore metadata — PHI exposure +# MAGIC %sql +# MAGIC -- PHI tag exposure summary +# MAGIC SELECT +# MAGIC local_path, +# MAGIC meta:_wsi_vendor::STRING AS vendor, +# MAGIC meta:_wsi_has_label::BOOLEAN AS has_label, +# MAGIC meta:_wsi_has_macro::BOOLEAN AS has_macro, +# MAGIC SIZE(CAST(meta:phi_tag_report AS ARRAY)) AS phi_tag_count +# MAGIC FROM dmoore.wsi.object_catalog +# MAGIC WHERE SIZE(CAST(meta:phi_tag_report AS ARRAY)) > 0 +# MAGIC ORDER BY phi_tag_count DESC diff --git a/notebooks/wsi/02-phi-pixel-detection.py b/notebooks/wsi/02-phi-pixel-detection.py new file mode 100644 index 00000000..f8a5ab6b --- /dev/null +++ b/notebooks/wsi/02-phi-pixel-detection.py @@ -0,0 +1,112 @@ +# Databricks notebook source +# /// script +# [tool.databricks.environment] +# environment_version = "5" +# /// +# DBTITLE 1,PHI Pixel Detection — Overview +# MAGIC %md +# MAGIC # WSI Pixel-Level PHI Detection +# MAGIC +# MAGIC Uses `WSIVLMPhiDetector` (OpenSlide-backed) to detect Protected Health Information +# MAGIC in the **label** and **macro** associated images of Whole Slide Image files. +# MAGIC +# MAGIC OpenSlide extracts the sub-image (no full-resolution data loaded), then a Databricks +# MAGIC VLM serving endpoint identifies PHI entities (patient names, DOBs, accession numbers, barcodes). +# MAGIC +# MAGIC **Prerequisite**: Run `01-ingest-wsi` first to populate `dmoore.wsi.object_catalog`. + +# COMMAND ---------- + +# DBTITLE 1,Install dependencies +# MAGIC %pip install openslide-python openslide-bin tifffile imagecodecs Pillow mlflow openai -q + +# COMMAND ---------- + +# DBTITLE 1,Add src/ to sys.path +import sys + +SRC_PATH = "/Workspace/Users/douglas.moore@databricks.com/pixels-tiff/src" +if SRC_PATH not in sys.path: + sys.path.insert(0, SRC_PATH) + print(f"Added to sys.path: {SRC_PATH}") +else: + print(f"Already on sys.path: {SRC_PATH}") + +# COMMAND ---------- + +# DBTITLE 1,Configuration +# --- Source: enriched metadata from 01-ingest-wsi --- +TABLE = "dmoore.wsi.object_catalog" + +# --- VLM endpoint for PHI detection --- +VLM_ENDPOINT = "phi-detection" + +print(f"Source table: {TABLE}") +print(f"VLM endpoint: {VLM_ENDPOINT}") + +# COMMAND ---------- + +# DBTITLE 1,VLM PHI detection — label images +from pyspark.sql.functions import lit +from dbx.pixels.wsi import WSIVLMPhiDetector + +# Run VLM PHI detection on all three sub-image series: +# label — printed patient name, barcode, accession (primary PHI target) +# macro — overview image (may contain handwritten labels) +# tissue — smallest pyramid level (rarely contains PHI, but checks embedded text) +# +# wsi_to_image returns None (recorded as error, no VLM call) when a series is absent. + +base_df = spark.sql(f""" + SELECT local_path, extension, + meta:_wsi_vendor::STRING AS vendor, + meta:_wsi_format::STRING AS format + FROM {TABLE} +""").repartition(8) + +print(f"Total WSI files: {base_df.count()}") + +results = [] +for series in ["label", "macro", "tissue"]: + detector = WSIVLMPhiDetector( + endpoint=VLM_ENDPOINT, + inputCol="local_path", + outputCol="vlm_phi", + series=series, + max_width=768, + temperature=0.0, + ) + series_df = detector.transform(base_df).withColumn("series", lit(series)) + results.append(series_df) + +phi_df = results[0].unionByName(results[1]).unionByName(results[2]) + +print(f"Total results (files × series): {phi_df.count()}") +display(phi_df.select("local_path", "vendor", "format", "series", "vlm_phi.*")) + +# COMMAND ---------- + +# DBTITLE 1,PHI detection summary +from pyspark.sql.functions import col, size + +results = phi_df.select( + "local_path", + "vendor", + "series", + col("vlm_phi.content").alias("phi_entities"), + col("vlm_phi.error").alias("error"), + col("vlm_phi.total_tokens").alias("tokens"), +) + +total = results.count() +with_phi = results.filter(size("phi_entities") > 0).count() +with_error = results.filter(col("error").isNotNull()).count() + +print("=== VLM PHI Detection Summary ===") +print(f"Total scans (files × series): {total}") +print(f"Scans with PHI detected: {with_phi}") +print(f"Scans with errors: {with_error}") +print() + +# Show files/series where PHI was found +display(results.filter(size("phi_entities") > 0)) diff --git a/src/dbx/pixels/dicom/cache/bot_cache_builder.py b/src/dbx/pixels/dicom/cache/bot_cache_builder.py index 97b2f105..e4b1b89e 100644 --- a/src/dbx/pixels/dicom/cache/bot_cache_builder.py +++ b/src/dbx/pixels/dicom/cache/bot_cache_builder.py @@ -655,7 +655,8 @@ def get_preload_list( hl_used = self.half_lives["last_used"] * 3600.0 # hours → seconds hl_ins = self.half_lives["inserted"] * 3600.0 - query = sql.SQL(""" + query = sql.SQL( + """ SELECT filename, COUNT(*) AS frame_count, @@ -685,7 +686,8 @@ def get_preload_list( GROUP BY filename ORDER BY priority_score DESC LIMIT %s - """).format( + """ + ).format( table=sql.Identifier(self.lb.schema, "dicom_frames"), w_used=sql.Literal(w_used), w_ins=sql.Literal(w_ins), diff --git a/src/dbx/pixels/lakebase.py b/src/dbx/pixels/lakebase.py index a5769e32..43f3f509 100644 --- a/src/dbx/pixels/lakebase.py +++ b/src/dbx/pixels/lakebase.py @@ -799,7 +799,8 @@ def get_preload_priority_list( # avoids the json_agg / OID-114 deserialization issue where psycopg2 # may return the json type as Python None if the JSON adapter is not # registered in the connection context. - query = sql.SQL(""" + query = sql.SQL( + """ SELECT filename, COUNT(*) AS frame_count, @@ -829,7 +830,8 @@ def get_preload_priority_list( GROUP BY filename ORDER BY priority_score DESC LIMIT %s - """).format(table=sql.Identifier(self.schema, table)) + """ + ).format(table=sql.Identifier(self.schema, table)) rows = self.execute_and_fetch_query(query, (uc_table_name, limit)) results = [] diff --git a/src/dbx/pixels/wsi/__init__.py b/src/dbx/pixels/wsi/__init__.py new file mode 100644 index 00000000..6b67d252 --- /dev/null +++ b/src/dbx/pixels/wsi/__init__.py @@ -0,0 +1,64 @@ +"""dbx.pixels.wsi — Unified Whole Slide Image handler for databricks-pixels. + +Supports all OpenSlide-recognized WSI formats: +- Aperio SVS +- Hamamatsu NDPI / VMS / VMU +- Leica SCN +- MIRAX MRXS +- Philips TIFF +- Sakura SVSLIDE +- Trestle TIF +- Ventana BIF / TIF +- Generic TIFF + +Install alongside ``databricks-pixels`` and extend the ``dbx.pixels`` namespace:: + + import sys + SRC_PATH = "/Workspace/Users//pixels-tiff/src" + if SRC_PATH not in sys.path: + sys.path.insert(0, SRC_PATH) + + from dbx.pixels.wsi import WSICatalog, WSIMetaExtractor, WSIVLMPhiDetector, wsi_to_image +""" + +from dbx.pixels.wsi.catalog import WSICatalog +from dbx.pixels.wsi.wsi_meta_extractor import WSIMetaExtractor +from dbx.pixels.wsi.wsi_phi_tags import ( + LARGE_TAGS, + NOT_PHI_TAGS, + OPENSLIDE_PATTERNS, + PHI_TAGS, + QUESTIONABLE_TAGS, + SUPPORTED_EXTENSIONS, + classify_tag, + classify_tags, + scrub_image_description, +) +from dbx.pixels.wsi.wsi_utils import ( + wsi_detect_format, + wsi_get_properties, + wsi_to_image, +) +from dbx.pixels.wsi.wsi_vlm_phi_detector import WSIVLMPhiDetector + +__all__ = [ + # Classes + "WSICatalog", + "WSIMetaExtractor", + "WSIVLMPhiDetector", + # Utils + "wsi_to_image", + "wsi_detect_format", + "wsi_get_properties", + # PHI tags + "classify_tag", + "classify_tags", + "scrub_image_description", + "PHI_TAGS", + "QUESTIONABLE_TAGS", + "NOT_PHI_TAGS", + "LARGE_TAGS", + # Constants + "SUPPORTED_EXTENSIONS", + "OPENSLIDE_PATTERNS", +] diff --git a/src/dbx/pixels/wsi/catalog.py b/src/dbx/pixels/wsi/catalog.py new file mode 100644 index 00000000..acda13db --- /dev/null +++ b/src/dbx/pixels/wsi/catalog.py @@ -0,0 +1,95 @@ +"""WSICatalog — extends the base Catalog for all Whole Slide Image formats. + +Unified catalog handler that discovers and indexes WSI files across all +OpenSlide-supported formats (Aperio SVS, Hamamatsu NDPI/VMS/VMU, Leica SCN, +MIRAX MRXS, Philips TIFF, Sakura SVSLIDE, Trestle TIF, Ventana BIF/TIF, +generic TIFF). + +Overrides: + - catalog() uses a multi-extension pattern by default + - Supports per-format or all-format discovery + +Note on glob limitations: Spark's binaryFile reader ``pathGlobFilter`` does +not support brace expansion (``{*.svs,*.tiff}``). When multiple extensions +are needed, WSICatalog issues multiple catalog() calls and unions them. +""" + +from __future__ import annotations + +from pyspark.sql import DataFrame + +from dbx.pixels.catalog import Catalog +from dbx.pixels.wsi.wsi_phi_tags import OPENSLIDE_PATTERNS + +# Default patterns for common WSI formats +_DEFAULT_PATTERNS = OPENSLIDE_PATTERNS + + +class WSICatalog(Catalog): + """Object catalog for Whole Slide Images (all OpenSlide-supported formats). + + Extends :class:`dbx.pixels.Catalog` with WSI-specific defaults: + - ``catalog()`` discovers files matching all WSI extensions by default + - Supports single-pattern or multi-pattern modes + + Args: + spark: Active SparkSession. + table: Fully qualified UC table name (e.g. ``dmoore.wsi.object_catalog``). + volume: Fully qualified UC volume name (e.g. ``dmoore.wsi.wsi_volume``). + """ + + def __init__(self, spark, table: str, volume: str = None): + if volume is None: + # Derive volume from catalog.schema + catalog_name, schema_name, _ = table.split(".") + volume = f"{catalog_name}.{schema_name}.wsi_volume" + super().__init__(spark, table=table, volume=volume) + + def catalog( + self, + path: str, + pattern: str = None, + patterns: list[str] = None, + **kwargs, + ) -> DataFrame: + """Catalog WSI files at the given path. + + Args: + path: Root directory to scan. + pattern: Single glob pattern (e.g. ``'*.svs'``). If provided, + only this pattern is used. + patterns: List of glob patterns to scan. Defaults to all + OpenSlide-supported extensions. Each pattern triggers + a separate scan; results are unioned. + **kwargs: Additional arguments forwarded to + :meth:`Catalog.catalog` (recurse, streaming, etc.). + + Returns: + DataFrame with cataloged file metadata. + """ + if pattern is not None: + # Single pattern mode (backward-compatible) + return super().catalog(path, pattern=pattern, **kwargs) + + if patterns is None: + patterns = _DEFAULT_PATTERNS + + # Multi-pattern mode: scan each extension and union results + dfs = [] + for pat in patterns: + try: + df = super().catalog(path, pattern=pat, **kwargs) + dfs.append(df) + except Exception: + # Some patterns may match zero files — that's fine + pass + + if not dfs: + # Return empty DataFrame with correct schema + return super().catalog(path, pattern="*.NONEXISTENT_EXTENSION_PLACEHOLDER", **kwargs) + + result = dfs[0] + for df in dfs[1:]: + result = result.union(df) + + return result.dropDuplicates(["path"]) diff --git a/src/dbx/pixels/wsi/wsi_meta_extractor.py b/src/dbx/pixels/wsi/wsi_meta_extractor.py new file mode 100644 index 00000000..875e25e2 --- /dev/null +++ b/src/dbx/pixels/wsi/wsi_meta_extractor.py @@ -0,0 +1,311 @@ +"""WSIMetaExtractor — Spark ML Transformer that extracts metadata from any OpenSlide-supported WSI. + +Unified handler for all Whole Slide Image formats: +- Aperio SVS +- Hamamatsu NDPI / VMS / VMU +- Leica SCN +- MIRAX MRXS +- Philips TIFF +- Sakura SVSLIDE +- Trestle TIF +- Ventana BIF / TIF +- Generic TIFF + +Primary backend: ``openslide-python`` + ``openslide-bin``. +Fallback backend: ``tifffile`` (for non-WSI TIFFs that OpenSlide cannot open). + +Architecture mirrors ``SVSMetaExtractor`` and ``TiffMetaExtractor``: +- Extends ``pyspark.ml.pipeline.Transformer`` +- Implements ``_transform(df)`` +- Uses ``mapInPandas`` with ``ThreadPoolExecutor`` for concurrent I/O +- Outputs a single ``meta`` column as a parsed VARIANT + +All derived fields (vendor, dimensions, levels, associated images, mpp, +magnification, phi_tag_report) are merged into the properties dict before +JSON serialisation. +""" + +from __future__ import annotations + +import json +from concurrent.futures import ThreadPoolExecutor +from typing import Iterator + +import pandas as pd +import pyspark.sql.types as t +from pyspark.ml.pipeline import Transformer +from pyspark.sql.functions import expr + + +class WSIMetaExtractor(Transformer): + """Extract metadata from any WSI file into the ``meta VARIANT`` column. + + Uses ``openslide`` as the primary backend; falls back to ``tifffile`` + for files that OpenSlide cannot detect (plain TIFFs, OME-TIFF without + WSI structure). + + Args: + catalog: Catalog instance (provides path/anon context). + inputCol: Column with worker-accessible file paths (default ``local_path``). + outputCol: Output column name (default ``meta``). + maxWorkers: ``ThreadPoolExecutor`` concurrency per Spark task (default 32). + useVariant: Parse JSON string to VARIANT via ``parse_json()`` (default True). + filterLargeTags: Drop large binary tags before JSON serialisation (default True). + """ + + MAX_WORKERS = 32 + + def __init__( + self, + catalog, + inputCol: str = "local_path", + outputCol: str = "meta", + maxWorkers: int = None, + useVariant: bool = True, + filterLargeTags: bool = True, + ): + self.catalog = catalog + self.inputCol = inputCol + self.outputCol = outputCol + self.maxWorkers = maxWorkers or self.MAX_WORKERS + self.useVariant = useVariant + self.filterLargeTags = filterLargeTags + + # ------------------------------------------------------------------ + # Internal: OpenSlide backend (handles all WSI formats) + # ------------------------------------------------------------------ + + @staticmethod + def _process_openslide(path: str) -> str: + """Extract metadata using OpenSlide (primary backend).""" + import openslide + + from dbx.pixels.wsi.wsi_phi_tags import classify_tags + + try: + slide = openslide.OpenSlide(path) + props = dict(slide.properties) + associated = list(slide.associated_images.keys()) + + # Vendor detection + vendor = props.get("openslide.vendor", "unknown") + + # MPP (microns per pixel) — universal across vendors + mpp_x = props.get(openslide.PROPERTY_NAME_MPP_X) + mpp_y = props.get(openslide.PROPERTY_NAME_MPP_Y) + objective_power = props.get(openslide.PROPERTY_NAME_OBJECTIVE_POWER) + + meta = { + **props, + # --- Derived WSI fields --- + "_wsi_vendor": vendor, + "_wsi_format": _detect_format_name(path, vendor), + "_wsi_width": slide.dimensions[0], + "_wsi_height": slide.dimensions[1], + "_wsi_level_count": slide.level_count, + "_wsi_level_dimensions": [list(d) for d in slide.level_dimensions], + "_wsi_level_downsamples": list(slide.level_downsamples), + "_wsi_mpp_x": float(mpp_x) if mpp_x else None, + "_wsi_mpp_y": float(mpp_y) if mpp_y else None, + "_wsi_objective_power": float(objective_power) if objective_power else None, + "_wsi_has_label": "label" in associated, + "_wsi_has_macro": "macro" in associated, + "_wsi_has_thumbnail": "thumbnail" in associated, + "_wsi_associated_images": associated, + "_wsi_backend": "openslide", + "phi_tag_report": classify_tags(props), + } + slide.close() + return json.dumps(meta, default=str) + + except Exception as err: + return json.dumps( + {"error": str(err), "udf": "wsi_meta_extractor_openslide", "path": path} + ) + + # ------------------------------------------------------------------ + # Internal: tifffile fallback (non-WSI TIFFs) + # ------------------------------------------------------------------ + + @staticmethod + def _process_tifffile(path: str, filter_large: bool = True) -> str: + """Extract metadata with tifffile (fallback for non-WSI TIFFs).""" + import tifffile + + from dbx.pixels.wsi.wsi_phi_tags import LARGE_TAGS, classify_tags + + try: + with tifffile.TiffFile(path) as tif: + page = tif.pages[0] + + tags: dict[str, str] = { + tag.name: str(tag.value) + for tag in page.tags.values() + if not filter_large or tag.name not in LARGE_TAGS + } + + # Per-series summary + series_info = [] + for s in tif.series: + entry: dict = { + "shape": list(s.shape), + "axes": s.axes, + "dtype": str(s.dtype), + } + if hasattr(s, "levels"): + entry["level_count"] = len(s.levels) + series_info.append(entry) + + meta = { + **tags, + "_wsi_vendor": "generic-tiff", + "_wsi_format": _detect_tifffile_format(tif), + "_wsi_width": page.imagewidth, + "_wsi_height": page.imagelength, + "_wsi_level_count": ( + len(tif.series[0].levels) + if tif.series and hasattr(tif.series[0], "levels") + else len(tif.pages) + ), + "_wsi_level_dimensions": None, + "_wsi_level_downsamples": None, + "_wsi_mpp_x": None, + "_wsi_mpp_y": None, + "_wsi_objective_power": None, + "_wsi_has_label": _has_labeled_ifd(tif, "label"), + "_wsi_has_macro": _has_labeled_ifd(tif, "macro"), + "_wsi_has_thumbnail": False, + "_wsi_associated_images": [], + "_wsi_backend": "tifffile", + "_wsi_page_count": len(tif.pages), + "_wsi_series_count": len(tif.series), + "_wsi_is_bigtiff": tif.is_bigtiff, + "_wsi_is_ome": tif.is_ome, + "_wsi_is_svs": tif.is_svs, + "_wsi_is_ndpi": getattr(tif, "is_ndpi", False), + "_wsi_series": series_info, + "phi_tag_report": classify_tags(tags), + } + return json.dumps(meta, default=str) + + except Exception as err: + return json.dumps( + {"error": str(err), "udf": "wsi_meta_extractor_tifffile", "path": path} + ) + + # ------------------------------------------------------------------ + # Internal: Dispatcher + # ------------------------------------------------------------------ + + @staticmethod + def _process_file(path: str, filter_large: bool = True) -> str: + """Dispatch to OpenSlide (primary) or tifffile (fallback).""" + try: + import openslide + + # Check if OpenSlide can detect the format + try: + fmt = openslide.OpenSlide.detect_format(path) + except Exception: + fmt = None + + if fmt is not None: + return WSIMetaExtractor._process_openslide(path) + else: + # OpenSlide doesn't recognize it; try tifffile + return WSIMetaExtractor._process_tifffile(path, filter_large) + + except ImportError: + # openslide not installed; fall back to tifffile + return WSIMetaExtractor._process_tifffile(path, filter_large) + + # ------------------------------------------------------------------ + # Transformer entry point + # ------------------------------------------------------------------ + + def _transform(self, df): + """Apply WSI metadata extraction using mapInPandas with concurrent I/O.""" + input_col = self.inputCol + output_col = self.outputCol + max_workers = self.maxWorkers + filter_large = self.filterLargeTags + + out_schema = t.StructType( + list(df.schema.fields) + [t.StructField(output_col, t.StringType(), True)] + ) + + def _extract_meta_batch( + iterator: Iterator[pd.DataFrame], + ) -> Iterator[pd.DataFrame]: + for pdf in iterator: + paths = pdf[input_col].tolist() + with ThreadPoolExecutor(max_workers=max_workers) as executor: + results = list( + executor.map( + lambda p: WSIMetaExtractor._process_file(p, filter_large), + paths, + ) + ) + pdf[output_col] = results + yield pdf + + df = df.mapInPandas(_extract_meta_batch, schema=out_schema) + + if self.useVariant: + df = df.withColumn( + self.outputCol, + expr(f"parse_json(`{self.outputCol}`)"), + ) + + return df + + +# --------------------------------------------------------------------------- +# Module-level helpers +# --------------------------------------------------------------------------- + + +def _detect_format_name(path: str, vendor: str) -> str: + """Return a human-readable format name based on vendor + extension.""" + ext = path.rsplit(".", 1)[-1].lower() if "." in path else "" + _vendor_map = { + "aperio": "Aperio SVS", + "hamamatsu": "Hamamatsu NDPI" if ext == "ndpi" else "Hamamatsu VMS/VMU", + "leica": "Leica SCN", + "mirax": "MIRAX MRXS", + "philips": "Philips TIFF", + "sakura": "Sakura SVSLIDE", + "trestle": "Trestle TIF", + "ventana": "Ventana BIF" if ext == "bif" else "Ventana TIF", + "generic-tiff": "Generic TIFF", + "dicom": "DICOM WSI", + } + return _vendor_map.get(vendor, f"Unknown ({vendor})") + + +def _detect_tifffile_format(tif) -> str: + """Detect format from tifffile attributes.""" + if tif.is_svs: + return "Aperio SVS (tifffile)" + if getattr(tif, "is_ndpi", False): + return "Hamamatsu NDPI (tifffile)" + if tif.is_ome: + return "OME-TIFF" + if tif.is_bigtiff: + return "BigTIFF" + return "Standard TIFF" + + +def _has_labeled_ifd(tif, target: str) -> bool: + """Check if a tifffile TiffFile has an IFD with a matching ImageDescription.""" + target_lc = target.strip().lower() + for page in tif.pages: + try: + desc = page.description + if isinstance(desc, bytes): + desc = desc.decode("utf-8", errors="replace") + if desc.strip().strip("\x00").lower() == target_lc: + return True + except Exception: + continue + return False diff --git a/src/dbx/pixels/wsi/wsi_phi_tags.py b/src/dbx/pixels/wsi/wsi_phi_tags.py new file mode 100644 index 00000000..b00edb1b --- /dev/null +++ b/src/dbx/pixels/wsi/wsi_phi_tags.py @@ -0,0 +1,314 @@ +"""PHI tag classification for Whole Slide Images (all OpenSlide-supported formats). + +Unified classification covering vendor-specific metadata properties from: +- Aperio SVS (openslide.vendor = "aperio") +- Hamamatsu NDPI/VMS/VMU (openslide.vendor = "hamamatsu") +- Leica SCN (openslide.vendor = "leica") +- MIRAX MRXS (openslide.vendor = "mirax") +- Philips TIFF (openslide.vendor = "philips") +- Sakura SVSLIDE (openslide.vendor = "sakura") +- Trestle TIF (openslide.vendor = "trestle") +- Ventana BIF/TIF (openslide.vendor = "ventana") +- Generic TIFF (openslide.vendor = "generic-tiff") +- Standard TIFF tags (via tifffile fallback) + +Classification tiers +-------------------- +PHI — Directly identifies a patient or operator. +QUESTIONABLE — May contain PHI depending on site configuration. +NOT_PHI — Pure scanner / geometry / technical parameters. + +Public API +---------- + classify_tag(key) -> str + classify_tags(properties: dict) -> list[dict] + scrub_image_description(desc) -> str + SUPPORTED_EXTENSIONS -> set[str] + OPENSLIDE_PATTERNS -> list[str] +""" + +from __future__ import annotations + +# --------------------------------------------------------------------------- +# Supported WSI file extensions (OpenSlide + tifffile fallback) +# --------------------------------------------------------------------------- +SUPPORTED_EXTENSIONS: set[str] = { + ".svs", # Aperio + ".tif", # Generic TIFF, Trestle, Ventana + ".tiff", # Generic TIFF, Philips + ".bif", # Ventana + ".ndpi", # Hamamatsu + ".mrxs", # MIRAX (index file) + ".vms", # Hamamatsu (virtual microscope slide) + ".vmu", # Hamamatsu (virtual microscope uncompressed) + ".scn", # Leica + ".svslide", # Sakura + ".dcm", # DICOM WSI +} + +# Glob patterns for use with Spark binaryFile reader or Catalog.catalog() +OPENSLIDE_PATTERNS: list[str] = [ + "*.svs", + "*.tif", + "*.tiff", + "*.bif", + "*.ndpi", + "*.mrxs", + "*.vms", + "*.vmu", + "*.scn", + "*.svslide", +] + +# --------------------------------------------------------------------------- +# PHI: Directly identifies a person (any vendor) +# --------------------------------------------------------------------------- +PHI_TAGS: set[str] = { + # --- Aperio SVS --- + "aperio.Patient", + "aperio.PatientID", + "aperio.DOB", + "aperio.MRN", + "aperio.AccessionNumber", + "aperio.ClinicID", + "aperio.ClinicalTrialID", + "aperio.Procedure", + "aperio.Diagnosis", + "aperio.Id", + # Aperio short-key variants (from ImageDescription pipe section) + "Patient", + "PatientID", + "DOB", + "MRN", + "AccessionNumber", + "ClinicID", + "ClinicalTrialID", + "Procedure", + "Diagnosis", + "Id", + # --- Hamamatsu NDPI --- + "hamamatsu.SourceLens", # Operator-configured; may encode tech ID + "hamamatsu.Reference", # Patient/case reference + # --- Leica SCN --- + "leica.device-model", # Sometimes contains operator info + # --- Philips --- + "philips.PIM_DP_UFS_BARCODE", # Barcode text (patient label) + # --- Sakura --- + "sakura.NominalLensMagnification", # Technical, but barcode fields below + # --- Generic TIFF tags (from tifffile) --- + "Artist", # tag 315 — person who created the image + "HostComputer", # tag 316 — workstation/operator ID +} + +# --------------------------------------------------------------------------- +# QUESTIONABLE: May contain PHI depending on site configuration +# --------------------------------------------------------------------------- +QUESTIONABLE_TAGS: set[str] = { + # --- Aperio --- + "aperio.Date", + "aperio.Time", + "aperio.Clinic", + "aperio.Pathologist", + "aperio.Title", + "aperio.Filename", + "aperio.User", + "aperio.ImageID", + # Aperio short-key variants + "Date", + "Time", + "Clinic", + "Pathologist", + "Title", + "Filename", + "User", + "ImageID", + # --- Hamamatsu --- + "hamamatsu.Created", # Scan timestamp + "hamamatsu.Updated", # Modification timestamp + # --- Leica --- + "leica.creation-date", + "leica.device-version", + # --- Philips --- + "philips.DICOM_ACQUISITION_DATETIME", + "philips.DICOM_DATE_OF_LAST_CALIBRATION", + "philips.PIM_DP_SCANNER_OPERATOR_ID", + # --- Ventana --- + "ventana.ScanDate", + "ventana.ScanTime", + "ventana.Operator", + # --- MIRAX --- + "mirax.GENERAL.SLIDE_CREATIONDATETIME", + "mirax.GENERAL.SLIDE_NAME", + # --- Generic TIFF --- + "ImageDescription", # tag 270 — free-text; may embed patient info + "DateTime", # tag 306 — creation timestamp + "Copyright", # tag 33432 + "DateTimeOriginal", # EXIF 36867 + "DateTimeDigitized", # EXIF 36868 + # --- OpenSlide common --- + "openslide.comment", # May contain free-text with PHI + # --- Artist/Copyright short keys --- + "tiff.Artist", + "tiff.Copyright", +} + +# --------------------------------------------------------------------------- +# LARGE_TAGS: Binary/array tags to skip during metadata extraction +# (avoids bloating the JSON output) +# --------------------------------------------------------------------------- +LARGE_TAGS: set[str] = { + "JPEGTables", + "JPEGQTables", + "JPEGDCTables", + "JPEGACTables", + "TileOffsets", + "TileByteCounts", + "StripOffsets", + "StripByteCounts", + "ICCProfile", + "ColorMap", + "TransferFunction", + "ReferenceBlackWhite", + "XMP", + "IPTCNAA", + "Photoshop", + "ExifIFD", + "GeoKeyDirectoryTag", + "GeoDoubleParamsTag", + "GeoAsciiParamsTag", + "ImageDepth", + "SubIFDs", +} + +# --------------------------------------------------------------------------- +# NOT_PHI: Pure scanner / technical parameters (representative, not exhaustive) +# --------------------------------------------------------------------------- +NOT_PHI_TAGS: set[str] = { + # OpenSlide standard + "openslide.level-count", + "openslide.mpp-x", + "openslide.mpp-y", + "openslide.objective-power", + "openslide.vendor", + "openslide.quickhash-1", + # Aperio technical + "aperio.AppMag", + "aperio.MPP", + "aperio.ScanScope ID", + "aperio.StripeWidth", + "aperio.Parmset", + "aperio.Filtered", + "aperio.ICC Profile", + # Hamamatsu technical + "hamamatsu.XOffsetFromSlideCentre", + "hamamatsu.YOffsetFromSlideCentre", + "hamamatsu.SourceLens", + # TIFF baseline + "ImageWidth", + "ImageLength", + "BitsPerSample", + "Compression", + "PhotometricInterpretation", + "SamplesPerPixel", + "RowsPerStrip", + "XResolution", + "YResolution", + "PlanarConfiguration", + "ResolutionUnit", + "Software", + "Make", + "Model", + "TileWidth", + "TileLength", + "NewSubfileType", + "Orientation", + "SampleFormat", + # Aperio short keys + "AppMag", + "MPP", + "ScanScope ID", + "StripeWidth", + "Parmset", + "Filtered", + "ICC Profile", +} + + +# --------------------------------------------------------------------------- +# Public API +# --------------------------------------------------------------------------- + + +def classify_tag(key: str) -> str: + """Classify a single WSI metadata property key. + + Returns one of: ``'PHI'``, ``'QUESTIONABLE'``, ``'NOT_PHI'``. + Tags not in any lookup default to ``'NOT_PHI'``. + """ + if key in PHI_TAGS: + return "PHI" + if key in QUESTIONABLE_TAGS: + return "QUESTIONABLE" + return "NOT_PHI" + + +def classify_tags(properties: dict) -> list[dict]: + """Classify all WSI metadata properties and return the structured PHI report. + + Args: + properties: Dict mapping property key (str) -> value (str). + + Returns: + List of ``{"tag": key, "value": val, "classification": cls}`` dicts. + Only PHI and QUESTIONABLE entries are included (NOT_PHI omitted). + """ + report = [] + for key, value in properties.items(): + cls = classify_tag(key) + if cls in ("PHI", "QUESTIONABLE"): + report.append({"tag": key, "value": str(value), "classification": cls}) + return report + + +def scrub_image_description(image_desc: str) -> str: + """Scrub PHI/QUESTIONABLE values from Aperio-style ImageDescription strings. + + Aperio format:: + + {header_line}|key = val|key = val|... + + Header line is technical-only (preserved as-is). Each key=value pair + whose key matches PHI or QUESTIONABLE is replaced with ``key = REDACTED``. + + Also handles free-text ImageDescription fields from other vendors by + returning 'REDACTED' if the entire string is classified as PHI. + + Args: + image_desc: Raw ImageDescription string. + + Returns: + Scrubbed string with PHI values replaced. + """ + if not image_desc: + return image_desc + + # Aperio pipe-delimited format + if "|" in image_desc: + parts = image_desc.split("|") + header = parts[0] + rebuilt = [header] + + for kv in parts[1:]: + if " = " in kv: + k, _v = kv.split(" = ", 1) + key_stripped = k.strip() + if key_stripped in PHI_TAGS or key_stripped in QUESTIONABLE_TAGS: + rebuilt.append(f"{k} = REDACTED") + else: + rebuilt.append(kv) + else: + rebuilt.append(kv) + + return "|".join(rebuilt) + + return image_desc diff --git a/src/dbx/pixels/wsi/wsi_utils.py b/src/dbx/pixels/wsi/wsi_utils.py new file mode 100644 index 00000000..2cfbe089 --- /dev/null +++ b/src/dbx/pixels/wsi/wsi_utils.py @@ -0,0 +1,298 @@ +"""WSI utility functions — image conversion for all OpenSlide-supported formats. + +Provides ``wsi_to_image()``, the unified WSI equivalent of +``dbx.pixels.tiff.tiff_utils.tiff_to_image()``. + +Uses OpenSlide as the primary backend (handles all 13 WSI formats). +Falls back to tifffile for non-WSI TIFFs that OpenSlide cannot open. + +Key features: +- Thumbnail extraction at configurable max_width (uses OpenSlide's + get_thumbnail() which leverages the pyramid — never loads full-res data) +- Associated image extraction (label, macro, thumbnail) via OpenSlide +- Format-agnostic: works with SVS, NDPI, MRXS, SCN, BIF, Philips TIFF, etc. + +No dependency on ``dbx.pixels.dicom`` or ``pydicom``. +""" + +from __future__ import annotations + +import base64 +import io +from typing import Optional + +import numpy as np +from PIL import Image + +from dbx.pixels.logging import LoggerProvider + +logger = LoggerProvider() + + +# --------------------------------------------------------------------------- +# Internal helpers +# --------------------------------------------------------------------------- + + +def _normalize_to_uint8_rgb(arr: np.ndarray) -> np.ndarray: + """Coerce any numpy array to uint8 RGB suitable for JPEG encoding. + + Handles: + - 16-bit / float arrays → normalise to 0-255 + - RGBA (4-channel) → composite onto white background + - Grayscale (2-D) → replicate to 3 channels + - Single-channel 3-D → replicate to 3 channels + """ + # RGBA → RGB with white background composite + if arr.ndim == 3 and arr.shape[-1] == 4: + # Alpha composite onto white + alpha = arr[:, :, 3:4].astype(np.float32) / 255.0 + rgb = arr[:, :, :3].astype(np.float32) + white = np.full_like(rgb, 255.0) + arr = (rgb * alpha + white * (1.0 - alpha)).astype(np.uint8) + + # Convert to uint8 + if arr.dtype != np.uint8: + lo, hi = float(arr.min()), float(arr.max()) + if hi > lo: + arr = ((arr.astype(np.float32) - lo) / (hi - lo) * 255).astype(np.uint8) + else: + arr = np.zeros_like(arr, dtype=np.uint8) + + # Grayscale → RGB + if arr.ndim == 2: + arr = np.stack([arr, arr, arr], axis=-1) + elif arr.ndim == 3 and arr.shape[-1] == 1: + arr = np.concatenate([arr, arr, arr], axis=-1) + + return arr + + +def _pil_to_output( + img: Image.Image, + max_width: int, + output_path: Optional[str], + return_type: str, +) -> Optional[str | bytes]: + """Resize, encode to JPEG, and return in requested format.""" + # Resize if needed + if max_width > 0 and img.width > max_width: + ratio = max_width / img.width + new_size = (max_width, int(img.height * ratio)) + img = img.resize(new_size, Image.LANCZOS) + + # Ensure RGB (not RGBA or palette) + if img.mode != "RGB": + img = img.convert("RGB") + + # Encode to JPEG + buf = io.BytesIO() + img.save(buf, format="JPEG", quality=85) + jpeg_bytes = buf.getvalue() + + # Optionally save to disk + if output_path: + with open(output_path, "wb") as f: + f.write(jpeg_bytes) + + if return_type == "binary": + return jpeg_bytes + return base64.b64encode(jpeg_bytes).decode("utf-8") + + +# --------------------------------------------------------------------------- +# OpenSlide backend +# --------------------------------------------------------------------------- + + +def _wsi_openslide( + path: str, + series: str = "tissue", + max_width: int = 768, + output_path: Optional[str] = None, + return_type: str = "str", +) -> Optional[str | bytes]: + """Extract image from a WSI using OpenSlide.""" + import openslide + + try: + slide = openslide.OpenSlide(path) + + if series in ("label", "macro", "thumbnail"): + # Associated image extraction + if series in slide.associated_images: + img = slide.associated_images[series] + slide.close() + return _pil_to_output(img, max_width, output_path, return_type) + else: + logger.warning( + f"wsi_to_image: no '{series}' associated image in {path}; " + f"available: {list(slide.associated_images.keys())}. " + f"Falling back to tissue thumbnail." + ) + + # Tissue thumbnail (default) + # Use OpenSlide's get_thumbnail which respects the pyramid + thumb_size = (max_width, max_width) if max_width > 0 else slide.dimensions + img = slide.get_thumbnail(thumb_size) + slide.close() + return _pil_to_output(img, max_width=0, output_path=output_path, return_type=return_type) + + except Exception as e: + logger.exception(f"OpenSlide read failed for {path}: {e}") + return None + + +# --------------------------------------------------------------------------- +# tifffile fallback +# --------------------------------------------------------------------------- + + +def _wsi_tifffile( + path: str, + series: str = "tissue", + max_width: int = 768, + output_path: Optional[str] = None, + return_type: str = "str", +) -> Optional[str | bytes]: + """Extract image from a TIFF using tifffile (fallback).""" + import tifffile + + try: + with tifffile.TiffFile(path) as tif: + arr = None + + if series in ("label", "macro"): + # Search for labeled IFD + target_lc = series.strip().lower() + for page in tif.pages: + try: + desc = page.description + if isinstance(desc, bytes): + desc = desc.decode("utf-8", errors="replace") + if desc.strip().strip("\x00").lower() == target_lc: + arr = page.asarray() + break + except Exception: + continue + + if arr is None: + logger.warning( + f"wsi_to_image: no '{series}' IFD in {path}; falling back to tissue" + ) + + # Tissue (default) — smallest level of first series + if arr is None: + if tif.series: + s = tif.series[0] + if hasattr(s, "levels") and len(s.levels) > 1: + arr = s.levels[-1].asarray() + else: + arr = s.asarray() + else: + arr = tif.pages[0].asarray() + + if arr is None: + return None + + arr = _normalize_to_uint8_rgb(arr) + img = Image.fromarray(arr) + return _pil_to_output(img, max_width, output_path, return_type) + + except Exception as e: + logger.exception(f"tifffile read failed for {path}: {e}") + return None + + +# --------------------------------------------------------------------------- +# Public API +# --------------------------------------------------------------------------- + + +def wsi_to_image( + path: str, + max_width: int = 768, + output_path: Optional[str] = None, + return_type: str = "str", + series: str = "tissue", +) -> Optional[str | bytes]: + """Convert a WSI file to a JPEG thumbnail. + + Supports all OpenSlide formats (SVS, NDPI, MRXS, SCN, BIF, Philips TIFF, + generic TIFF, etc.) with tifffile fallback for non-WSI TIFFs. + + Args: + path: Local path to the WSI file. + max_width: Resize thumbnail to this width (default 768). + Set to 0 to skip resize. + output_path: If set, also save the JPEG to this path. + return_type: ``"str"`` — base64-encoded JPEG string (default). + ``"binary"`` — raw JPEG bytes. + series: Which sub-image to extract: + ``"tissue"`` — tissue thumbnail (default; pyramid-aware). + ``"label"`` — label associated image (PHI target). + ``"macro"`` — macro/overview associated image. + ``"thumbnail"`` — built-in thumbnail if available. + + Returns: + Base64 JPEG string, raw JPEG bytes, or ``None`` on failure. + """ + # Try OpenSlide first + try: + import openslide + + fmt = openslide.OpenSlide.detect_format(path) + if fmt is not None: + result = _wsi_openslide(path, series, max_width, output_path, return_type) + if result is not None: + return result + except ImportError: + logger.warn( + "wsi_to_image: preferred openslide not available. " + "Install openslide-python openslide-bin." + ) + except Exception: + pass + + # Fallback to tifffile + try: + return _wsi_tifffile(path, series, max_width, output_path, return_type) + except ImportError: + logger.error( + "wsi_to_image: neither openslide nor tifffile available. " + "Install openslide-python openslide-bin or tifffile." + ) + return None + + +def wsi_detect_format(path: str) -> Optional[str]: + """Detect the WSI format of a file without fully opening it. + + Returns the OpenSlide vendor string (e.g. 'aperio', 'hamamatsu', + 'philips', 'generic-tiff') or None if not a recognized WSI format. + """ + try: + import openslide + + return openslide.OpenSlide.detect_format(path) + except ImportError: + return None + except Exception: + return None + + +def wsi_get_properties(path: str) -> dict: + """Read all OpenSlide properties from a WSI file. + + Returns a dict of property key -> value strings. + Returns an empty dict if the file cannot be opened. + """ + try: + import openslide + + slide = openslide.OpenSlide(path) + props = dict(slide.properties) + slide.close() + return props + except Exception: + return {} diff --git a/src/dbx/pixels/wsi/wsi_vlm_phi_detector.py b/src/dbx/pixels/wsi/wsi_vlm_phi_detector.py new file mode 100644 index 00000000..7a33d666 --- /dev/null +++ b/src/dbx/pixels/wsi/wsi_vlm_phi_detector.py @@ -0,0 +1,258 @@ +"""WSIVLMPhiDetector — Spark ML Transformer for pixel-level PHI detection in WSI files. + +Fully self-contained: no dependency on ``dbx.pixels.dicom`` or ``pydicom``. +Uses OpenSlide (via ``wsi_to_image()``) as the primary backend for all WSI formats. + +Supported formats (via OpenSlide): +- Aperio SVS +- Hamamatsu NDPI / VMS / VMU +- Leica SCN +- MIRAX MRXS +- Philips TIFF +- Sakura SVSLIDE +- Trestle TIF +- Ventana BIF / TIF +- Generic TIFF + +Architecture: +- Extends ``pyspark.ml.base.Transformer`` +- Applies ``wsi_to_image()`` to extract the label/macro/tissue sub-image as a + JPEG thumbnail (uses OpenSlide's pyramid — never loads full-res data), then + calls a Databricks VLM serving endpoint via the OpenAI-compatible API. +- Output column schema mirrors ``VLMPhiDetector`` from the DICOM module + (content array, completion_tokens int, prompt_tokens int, + total_tokens int, error string). +""" + +from __future__ import annotations + +import base64 +from dataclasses import dataclass +from typing import Iterator, List, Optional + +import pandas as pd +from pyspark.ml.base import Transformer +from pyspark.sql.functions import col, pandas_udf + +from dbx.pixels.logging import LoggerProvider + +logger = LoggerProvider() + +__all__ = ["WSIVLMPhiDetector", "VlmResult"] + + +DEFAULT_SYSTEM_PROMPT = ( + "You are an expert in privacy and personal health information (PHI). " + "Per HIPAA rules there are 18 fields considered PHI. " + "Please identify all of the PHI fields found in the image and return a " + "list of pipe-separated named entities, e.g. 'John Smith'|'04-31-1954'|'123 Drury Lane' " + "and nothing else. " + "Don't be fooled by text fields especially acronyms that are not PHI. " + "If there's no PHI detected, return 'No PHI' and nothing else. " + "Answer concisely as requested without explanations." +) + + +@dataclass +class VlmResult: + """Per-image result returned by the VLM PHI detector.""" + + content: Optional[List[str]] + completion_tokens: int + prompt_tokens: int + total_tokens: int + error: Optional[str] + + +# --------------------------------------------------------------------------- +# Module-level UDF factory +# --------------------------------------------------------------------------- +# Defined outside the class so Spark Connect can serialise the closure +# without capturing ``self``. Only primitive types are closed over. + + +def _make_wsi_phi_detector_udf( + endpoint: str, + system_prompt: str, + temperature: float, + num_output_tokens: int, + input_type: str, + max_width: int, + series: str, +): + """Return a ``pandas_udf`` configured with the given inference parameters. + + Uses ``wsi_to_image()`` (OpenSlide primary, tifffile fallback) for image + extraction from any WSI format. + """ + + @pandas_udf( + "content array, completion_tokens int, " + "prompt_tokens int, total_tokens int, error string" + ) + def _extract_udf(paths: Iterator[pd.Series]) -> Iterator[pd.DataFrame]: + from dataclasses import replace as dc_replace + + from mlflow.utils.databricks_utils import get_databricks_host_creds + from openai import OpenAI + + from dbx.pixels.wsi.wsi_utils import wsi_to_image + + creds = get_databricks_host_creds("databricks") + client = OpenAI( + api_key=creds.token, + base_url=f"{creds.host}/serving-endpoints", + timeout=300, + max_retries=3, + ) + _null = VlmResult(None, 0, 0, 0, None) + + for batch in paths: + results = [] + for path in batch: + try: + # --- 1. Get image as base64 JPEG --- + if input_type == "wsi": + # OpenSlide-backed: handles SVS, NDPI, MRXS, SCN, BIF, + # Philips TIFF, generic TIFF, etc. + b64 = wsi_to_image( + path, + max_width=max_width, + return_type="str", + series=series, + ) + if b64 is None: + results.append( + dc_replace( + _null, + error=f"wsi_to_image returned None: {path}", + ) + ) + continue + elif input_type == "image": + # Raw image file (JPEG/PNG) — read and base64-encode + local = path[5:] if path.startswith("dbfs:") else path + with open(local, "rb") as fh: + b64 = base64.b64encode(fh.read()).decode("utf-8") + else: # "base64" — already encoded + b64 = path + + # --- 2. VLM inference --- + response = client.chat.completions.create( + model=endpoint, + messages=[ + {"role": "system", "content": system_prompt}, + { + "role": "user", + "content": [ + { + "type": "image_url", + "image_url": { + "url": f"data:image/jpeg;base64,{b64}", + "detail": "low", + }, + } + ], + }, + ], + temperature=float(temperature), + max_tokens=int(num_output_tokens), + ) + + # --- 3. Parse response --- + content = response.choices[0].message.content + if "|" in content: + phi_list = content.split("|") + elif content.strip().lower() == "no phi": + phi_list = [] + else: + phi_list = [content] + + results.append( + VlmResult( + phi_list, + response.usage.completion_tokens, + response.usage.prompt_tokens, + response.usage.total_tokens, + None, + ) + ) + + except Exception as exc: + logger.error(f"Error processing {path}: {exc}") + results.append(dc_replace(_null, error=str(exc))) + + yield pd.DataFrame(results) + + return _extract_udf + + +# --------------------------------------------------------------------------- +# Transformer +# --------------------------------------------------------------------------- + + +class WSIVLMPhiDetector(Transformer): + """Detect pixel-level PHI in Whole Slide Images using a Databricks VLM endpoint. + + Supports all OpenSlide-recognized formats (SVS, NDPI, MRXS, SCN, BIF, + Philips TIFF, generic TIFF, etc.) via ``wsi_to_image()`` which uses + OpenSlide as primary backend with tifffile fallback. + + No dependency on ``dbx.pixels.dicom`` or ``pydicom`` — fully self-contained. + + Args: + endpoint: Databricks serving endpoint name for the VLM. + system_prompt: Override the default HIPAA PHI detection prompt. + temperature: VLM sampling temperature (default 0.0). + num_output_tokens: Maximum tokens in the VLM response (default 200). + inputCol: Input column name (default ``local_path``). + outputCol: Output column name (default ``response``). + input_type: ``"wsi"`` — path to any WSI file (default). + ``"image"`` — path to a JPEG/PNG file. + ``"base64"`` — already base64-encoded image string. + max_width: Resize thumbnail width before VLM (default 768). + Set to 0 to disable. + series: Sub-image to extract from WSI files (default ``"label"``): + ``"label"`` — label associated image (rendered PHI text; recommended). + ``"macro"`` — macro/overview associated image. + ``"thumbnail"`` — built-in thumbnail if available. + ``"tissue"`` — smallest tissue pyramid level. + Falls back to tissue if the requested sub-image is absent. + """ + + def __init__( + self, + endpoint: str, + system_prompt: str = None, + temperature: float = 0.0, + num_output_tokens: int = 200, + inputCol: str = "local_path", + outputCol: str = "response", + input_type: str = "wsi", + max_width: int = 768, + series: str = "label", + ): + super().__init__() + self.endpoint = endpoint + self.system_prompt = system_prompt + self.temperature = temperature + self.num_output_tokens = num_output_tokens + self.inputCol = inputCol + self.outputCol = outputCol + self.input_type = input_type + self.max_width = max_width + self.series = series + + def _transform(self, df): + """Apply VLM PHI detection via ``pandas_udf``.""" + _udf = _make_wsi_phi_detector_udf( + endpoint=self.endpoint, + system_prompt=self.system_prompt or DEFAULT_SYSTEM_PROMPT, + temperature=self.temperature, + num_output_tokens=self.num_output_tokens, + input_type=self.input_type, + max_width=self.max_width, + series=self.series, + ) + return df.withColumn(self.outputCol, _udf(col(self.inputCol))) diff --git a/tests/dbx/test_wsi.py b/tests/dbx/test_wsi.py new file mode 100644 index 00000000..768d7d87 --- /dev/null +++ b/tests/dbx/test_wsi.py @@ -0,0 +1,338 @@ +"""Tests for dbx.pixels.wsi — Unified WSI handler. + +Runs against real SVS and TIFF files in UC Volumes: +- SVS: /Volumes/hls_radiology_east/orthanc_demo/raw_images/Aperio/ +- TIFF: /Volumes/hls_radiology_east/orthanc_demo/raw_images/Generic-TIFF/ + /Volumes/hls_radiology_east/orthanc_demo/raw_images/Philips-TIFF/ + /Volumes/hls_radiology_east/osuwmc/sample/ + +Requires: openslide-python, openslide-bin, tifffile + +Usage (from project root): + %pip install openslide-python openslide-bin tifffile imagecodecs -q + pytest tests/dbx/test_wsi.py -v +""" + +import json +import os +import sys +from pathlib import Path + +import pytest + +# Ensure src is on path +SRC_PATH = str(Path(__file__).resolve().parents[2] / "src") +if SRC_PATH not in sys.path: + sys.path.insert(0, SRC_PATH) + + +# --------------------------------------------------------------------------- +# Test fixtures: real file paths from UC Volumes +# --------------------------------------------------------------------------- + +SVS_DIR = "/Volumes/hls_radiology_east/orthanc_demo/raw_images/Aperio" +TIFF_DIRS = [ + "/Volumes/hls_radiology_east/orthanc_demo/raw_images/Generic-TIFF", + "/Volumes/hls_radiology_east/orthanc_demo/raw_images/Philips-TIFF", + "/Volumes/hls_radiology_east/osuwmc/sample", +] + + +def _discover_files(directory: str, extensions: tuple) -> list[str]: + """Discover test files by extension.""" + if not os.path.isdir(directory): + return [] + return sorted( + os.path.join(directory, f) + for f in os.listdir(directory) + if f.lower().endswith(extensions) and not f.startswith(".") + ) + + +@pytest.fixture(scope="module") +def svs_files(): + """All SVS test files.""" + files = _discover_files(SVS_DIR, (".svs",)) + if not files: + pytest.skip(f"No SVS files found in {SVS_DIR}") + return files + + +@pytest.fixture(scope="module") +def tiff_files(): + """All TIFF test files.""" + files = [] + for d in TIFF_DIRS: + files.extend(_discover_files(d, (".tiff", ".tif"))) + if not files: + pytest.skip(f"No TIFF files found in {TIFF_DIRS}") + return files + + +@pytest.fixture(scope="module") +def all_wsi_files(svs_files, tiff_files): + """All WSI files (SVS + TIFF).""" + return svs_files + tiff_files + + +# --------------------------------------------------------------------------- +# Test: phi_tags module +# --------------------------------------------------------------------------- + + +class TestPhiTags: + """Tests for wsi_phi_tags.py""" + + def test_classify_phi_tag(self): + from dbx.pixels.wsi.wsi_phi_tags import classify_tag + + assert classify_tag("aperio.Patient") == "PHI" + assert classify_tag("Patient") == "PHI" + assert classify_tag("Artist") == "PHI" + assert classify_tag("HostComputer") == "PHI" + + def test_classify_questionable_tag(self): + from dbx.pixels.wsi.wsi_phi_tags import classify_tag + + assert classify_tag("aperio.Date") == "QUESTIONABLE" + assert classify_tag("DateTime") == "QUESTIONABLE" + assert classify_tag("ImageDescription") == "QUESTIONABLE" + assert classify_tag("ventana.ScanDate") == "QUESTIONABLE" + + def test_classify_not_phi_tag(self): + from dbx.pixels.wsi.wsi_phi_tags import classify_tag + + assert classify_tag("openslide.mpp-x") == "NOT_PHI" + assert classify_tag("ImageWidth") == "NOT_PHI" + assert classify_tag("SomeUnknownTag") == "NOT_PHI" + + def test_classify_tags_returns_only_phi_and_questionable(self): + from dbx.pixels.wsi.wsi_phi_tags import classify_tags + + props = { + "aperio.Patient": "John Doe", + "aperio.AppMag": "40", + "openslide.mpp-x": "0.2525", + "DateTime": "2024-01-01", + } + report = classify_tags(props) + tags_in_report = {r["tag"] for r in report} + assert "aperio.Patient" in tags_in_report + assert "DateTime" in tags_in_report + assert "aperio.AppMag" not in tags_in_report + assert "openslide.mpp-x" not in tags_in_report + + def test_scrub_image_description_aperio_format(self): + from dbx.pixels.wsi.wsi_phi_tags import scrub_image_description + + desc = "Aperio Image Library v12.0.15|Patient = John Doe|AppMag = 40|Date = 2024-01-01" + scrubbed = scrub_image_description(desc) + assert "John Doe" not in scrubbed + assert "REDACTED" in scrubbed + assert "AppMag = 40" in scrubbed # NOT_PHI preserved + + def test_supported_extensions(self): + from dbx.pixels.wsi.wsi_phi_tags import SUPPORTED_EXTENSIONS + + assert ".svs" in SUPPORTED_EXTENSIONS + assert ".tiff" in SUPPORTED_EXTENSIONS + assert ".bif" in SUPPORTED_EXTENSIONS + assert ".ndpi" in SUPPORTED_EXTENSIONS + assert ".mrxs" in SUPPORTED_EXTENSIONS + + def test_openslide_patterns(self): + from dbx.pixels.wsi.wsi_phi_tags import OPENSLIDE_PATTERNS + + assert "*.svs" in OPENSLIDE_PATTERNS + assert "*.tiff" in OPENSLIDE_PATTERNS + + +# --------------------------------------------------------------------------- +# Test: wsi_utils module +# --------------------------------------------------------------------------- + + +class TestWSIUtils: + """Tests for wsi_utils.py — requires real files.""" + + def test_detect_format_svs(self, svs_files): + from dbx.pixels.wsi.wsi_utils import wsi_detect_format + + # All SVS files should be detected as 'aperio' + for f in svs_files: + fmt = wsi_detect_format(f) + assert fmt == "aperio", f"Expected 'aperio' for {f}, got {fmt}" + + def test_detect_format_tiff(self, tiff_files): + from dbx.pixels.wsi.wsi_utils import wsi_detect_format + + # TIFF files should be detected as some OpenSlide vendor or None + for f in tiff_files: + fmt = wsi_detect_format(f) + # Philips TIFFs should be detected; generic may return None + assert fmt is None or isinstance(fmt, str), f"Unexpected format for {f}: {fmt}" + + def test_get_properties_svs(self, svs_files): + from dbx.pixels.wsi.wsi_utils import wsi_get_properties + + props = wsi_get_properties(svs_files[0]) + assert len(props) > 0 + assert "openslide.vendor" in props + assert props["openslide.vendor"] == "aperio" + + def test_wsi_to_image_tissue_svs(self, svs_files): + from dbx.pixels.wsi.wsi_utils import wsi_to_image + + result = wsi_to_image(svs_files[0], max_width=256, series="tissue") + assert result is not None + assert isinstance(result, str) # base64 string + assert len(result) > 100 # Non-trivial content + + def test_wsi_to_image_label_svs(self, svs_files): + from dbx.pixels.wsi.wsi_utils import wsi_to_image + + # CMU-1.svs has a label image + cmu1 = next((f for f in svs_files if "CMU-1.svs" in f), svs_files[0]) + result = wsi_to_image(cmu1, max_width=256, series="label") + assert result is not None + assert isinstance(result, str) + + def test_wsi_to_image_macro_svs(self, svs_files): + from dbx.pixels.wsi.wsi_utils import wsi_to_image + + cmu1 = next((f for f in svs_files if "CMU-1.svs" in f), svs_files[0]) + result = wsi_to_image(cmu1, max_width=256, series="macro") + assert result is not None + assert isinstance(result, str) + + def test_wsi_to_image_binary_output(self, svs_files): + from dbx.pixels.wsi.wsi_utils import wsi_to_image + + result = wsi_to_image(svs_files[0], max_width=128, return_type="binary") + assert result is not None + assert isinstance(result, bytes) + assert result[:2] == b"\xff\xd8" # JPEG magic bytes + + def test_wsi_to_image_tiff(self, tiff_files): + from dbx.pixels.wsi.wsi_utils import wsi_to_image + + result = wsi_to_image(tiff_files[0], max_width=256, series="tissue") + assert result is not None + assert isinstance(result, str) + + +# --------------------------------------------------------------------------- +# Test: WSIMetaExtractor module (standalone, no Spark) +# --------------------------------------------------------------------------- + + +class TestWSIMetaExtractorStandalone: + """Tests for the static extraction methods (no Spark required).""" + + def test_process_openslide_svs(self, svs_files): + from dbx.pixels.wsi.wsi_meta_extractor import WSIMetaExtractor + + result = WSIMetaExtractor._process_openslide(svs_files[0]) + meta = json.loads(result) + assert "error" not in meta, f"Error: {meta.get('error')}" + assert meta["_wsi_vendor"] == "aperio" + assert meta["_wsi_backend"] == "openslide" + assert meta["_wsi_width"] > 0 + assert meta["_wsi_height"] > 0 + assert meta["_wsi_level_count"] >= 1 + assert isinstance(meta["_wsi_level_dimensions"], list) + assert isinstance(meta["_wsi_associated_images"], list) + assert isinstance(meta["phi_tag_report"], list) + + def test_process_openslide_has_mpp(self, svs_files): + from dbx.pixels.wsi.wsi_meta_extractor import WSIMetaExtractor + + result = WSIMetaExtractor._process_openslide(svs_files[0]) + meta = json.loads(result) + # CMU SVS files have MPP defined + assert meta["_wsi_mpp_x"] is not None or meta.get("openslide.mpp-x") is not None + + def test_process_openslide_svs_associated_images(self, svs_files): + from dbx.pixels.wsi.wsi_meta_extractor import WSIMetaExtractor + + # CMU-1.svs should have label and macro + cmu1 = next((f for f in svs_files if "CMU-1.svs" in f), svs_files[0]) + result = WSIMetaExtractor._process_openslide(cmu1) + meta = json.loads(result) + assert meta["_wsi_has_label"] is True + assert meta["_wsi_has_macro"] is True + + def test_process_file_dispatches_to_openslide_for_svs(self, svs_files): + from dbx.pixels.wsi.wsi_meta_extractor import WSIMetaExtractor + + result = WSIMetaExtractor._process_file(svs_files[0]) + meta = json.loads(result) + assert meta["_wsi_backend"] == "openslide" + + def test_process_tifffile_fallback(self, tiff_files): + from dbx.pixels.wsi.wsi_meta_extractor import WSIMetaExtractor + + # Force tifffile path + result = WSIMetaExtractor._process_tifffile(tiff_files[0]) + meta = json.loads(result) + assert "error" not in meta, f"Error: {meta.get('error')}" + assert meta["_wsi_backend"] == "tifffile" + assert meta["_wsi_width"] > 0 + assert meta["_wsi_height"] > 0 + + def test_process_file_tiff_uses_openslide_when_possible(self, tiff_files): + import openslide + + from dbx.pixels.wsi.wsi_meta_extractor import WSIMetaExtractor + + # Find a TIFF that OpenSlide recognizes + for f in tiff_files: + fmt = openslide.OpenSlide.detect_format(f) + if fmt is not None: + result = WSIMetaExtractor._process_file(f) + meta = json.loads(result) + assert meta["_wsi_backend"] == "openslide" + return + pytest.skip("No TIFF files recognized by OpenSlide") + + def test_all_svs_files_extract_without_error(self, svs_files): + from dbx.pixels.wsi.wsi_meta_extractor import WSIMetaExtractor + + for f in svs_files: + result = WSIMetaExtractor._process_file(f) + meta = json.loads(result) + assert "error" not in meta, f"Error for {f}: {meta.get('error')}" + + def test_all_tiff_files_extract_without_error(self, tiff_files): + from dbx.pixels.wsi.wsi_meta_extractor import WSIMetaExtractor + + for f in tiff_files: + result = WSIMetaExtractor._process_file(f) + meta = json.loads(result) + assert "error" not in meta, f"Error for {f}: {meta.get('error')}" + + +# --------------------------------------------------------------------------- +# Test: Format constants and patterns +# --------------------------------------------------------------------------- + + +class TestConstants: + """Verify module constants are consistent.""" + + def test_openslide_patterns_match_extensions(self): + from dbx.pixels.wsi.wsi_phi_tags import OPENSLIDE_PATTERNS, SUPPORTED_EXTENSIONS + + for pat in OPENSLIDE_PATTERNS: + ext = "." + pat.lstrip("*.") + assert ext in SUPPORTED_EXTENSIONS, f"Pattern {pat} not in SUPPORTED_EXTENSIONS" + + def test_imports(self): + """Verify the package imports cleanly.""" + from dbx.pixels.wsi import ( + WSICatalog, + WSIMetaExtractor, + ) + + assert WSICatalog is not None + assert WSIMetaExtractor is not None