[pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

Author: pre-commit-ci[bot]

tiatoolbox/models/architecture/mapde.py

@@ -8,21 +8,20 @@
 
 from __future__ import annotations
 
+import dask.array as da
 import numpy as np
+import pandas as pd
 import torch
 import torch.nn.functional as F  # noqa: N812
-from skimage.feature import peak_local_max
-import dask.array as da
-from tiatoolbox.annotation.storage import SQLiteStore
-import pandas as pd
 
 from tiatoolbox.models.architecture.micronet import MicroNet
 from tiatoolbox.models.engine.nucleus_detector import (
-    peak_detection_mapoverlap,
     centroids_map_to_dask_dataframe,
     nucleus_detection_nms,
+    peak_detection_mapoverlap,
 )
 
+
 class MapDe(MicroNet):
     """Initialize MapDe [1].
 
@@ -238,13 +237,11 @@ def forward(self: MapDe, input_tensor: torch.Tensor) -> torch.Tensor:
         logits, _, _, _ = super().forward(input_tensor)
         out = F.conv2d(logits, self.dist_filter, padding="same")
         return F.relu(out)
-    
-
-    
-
 
     #  skipcq: PYL-W0221  # noqa: ERA001
-    def postproc(self: MapDe, prediction_map: da.Array, prediction_shape: tuple, dtype: np.dtype) -> pd.DataFrame:
+    def postproc(
+        self: MapDe, prediction_map: da.Array, prediction_shape: tuple, dtype: np.dtype
+    ) -> pd.DataFrame:
         """Post-processing script for MapDe.
 
         Performs peak detection and extracts coordinates in x, y format.
@@ -289,7 +286,6 @@ def postproc(self: MapDe, prediction_map: da.Array, prediction_shape: tuple, dty
 
         return nms_df
 
-
     @staticmethod
     def infer_batch(
         model: torch.nn.Module,

tiatoolbox/models/engine/nucleus_detector.py

@@ -2,53 +2,53 @@
 
 from __future__ import annotations
 
-import os
-import sys
 from pathlib import Path
 from typing import TYPE_CHECKING, Unpack
 
 import dask.array as da
 import dask.dataframe as dd
 import numpy as np
 import pandas as pd
-from shapely.geometry import Point
 from skimage.feature import peak_local_max
 from skimage.measure import label, regionprops
 
-from tiatoolbox.models.engine.io_config import IOSegmentorConfig
+from tiatoolbox import logger
+from tiatoolbox.annotation import AnnotationStore
 from tiatoolbox.models.engine.semantic_segmentor import (
     SemanticSegmentor,
     SemanticSegmentorRunParams,
 )
 from tiatoolbox.models.models_abc import ModelABC
-from tiatoolbox.annotation import Annotation, SQLiteStore, AnnotationStore
 from tiatoolbox.utils.misc import df_to_store_nucleus_detector
-from tiatoolbox import logger
 
 if TYPE_CHECKING:  # pragma: no cover
-    import os
-    from tiatoolbox.models.engine.io_config import IOSegmentorConfig
     from tiatoolbox.models.models_abc import ModelABC
-    from tiatoolbox.wsicore import WSIReader
 
 
 def probability_to_peak_map(
-    img2d: np.ndarray, min_distance: int, threshold_abs: float, threshold_rel: float = 0.0
+    img2d: np.ndarray,
+    min_distance: int,
+    threshold_abs: float,
+    threshold_rel: float = 0.0,
 ) -> np.ndarray:
     """Build a boolean mask (H, W) of objects from a 2D probability map using peak_local_max.
-    
+
     Args:
         img2d (np.ndarray): 2D probability map.
         min_distance (int): Minimum distance between peaks.
         threshold_abs (float): Absolute threshold for peak detection.
         threshold_rel (float, optional): Relative threshold for peak detection. Defaults to 0.0.
+
     Returns:
         mask (np.ndarray): Boolean mask (H, W) with True at peak locations.
     """
     H, W = img2d.shape
     mask = np.zeros((H, W), dtype=bool)
     coords = peak_local_max(
-        img2d, min_distance=min_distance, threshold_abs=threshold_abs, threshold_rel=threshold_rel
+        img2d,
+        min_distance=min_distance,
+        threshold_abs=threshold_abs,
+        threshold_rel=threshold_rel,
     )
     if coords.size:
         r, c = coords[:, 0], coords[:, 1]
@@ -67,7 +67,7 @@ def peak_detection_mapoverlap(
 ) -> np.ndarray:
     """Runs inside Dask.da.map_overlap on a padded NumPy block: (h_pad, w_pad, C).
     Builds a processed mask per channel, runs peak_local_max then
-    label+regionprops, and writes probability (mean_intensity) at centroid pixels. 
+    label+regionprops, and writes probability (mean_intensity) at centroid pixels.
     Keeps only centroids whose (row,col) lie in the interior window:
         rows [depth_h : depth_h + core_h), cols [depth_w : depth_w + core_w)
     Returns same spatial shape as input block: (h_pad, w_pad, C), float32.
@@ -81,6 +81,7 @@ def peak_detection_mapoverlap(
         depth_w: Halo size in pixels for width (cols).
         calculate_probabilities: If True, write mean_intensity at centroids;
             else write 1.0 at centroids.
+
     Returns:
         out: NumPy array (H, W, C) with probabilities at centroids, 0 elsewhere.
     """
@@ -120,7 +121,9 @@ def peak_detection_mapoverlap(
     return out
 
 
-def detection_with_map_overlap(probs: da.Array, min_distance: int, threshold_abs: float, depth_pixels: int) -> da.Array:
+def detection_with_map_overlap(
+    probs: da.Array, min_distance: int, threshold_abs: float, depth_pixels: int
+) -> da.Array:
     """probs: Dask array (H, W, C), float.
     depth_pixels: halo in pixels for H/W (use >= min_distance and >= any morphology radius).
 
@@ -143,18 +146,21 @@ def detection_with_map_overlap(probs: da.Array, min_distance: int, threshold_abs
     return scores
 
 
-def centroids_map_to_dask_dataframe(scores: da.Array, x_offset: int = 0, y_offset: int = 0) -> dd.DataFrame:
+def centroids_map_to_dask_dataframe(
+    scores: da.Array, x_offset: int = 0, y_offset: int = 0
+) -> dd.DataFrame:
     """Convert centroid map (H, W, C) into a Dask DataFrame with columns: x, y, type, prob.
 
     Args:
         scores: Dask array (H, W, C) with probabilities at centroids, 0 elsewhere.
         x_offset: global x offset to add to all x coordinates.
         y_offset: global y offset to add to all y coordinates.
+
     Returns:
         ddf: Dask DataFrame with columns: x, y, type, prob.
     """
     # 1) Build a boolean mask of detections
-    
+
     mask = scores > 0
     # 2) Get coordinates and class of detections (lazy 1D Dask arrays)
 
@@ -172,7 +178,7 @@ def centroids_map_to_dask_dataframe(scores: da.Array, x_offset: int = 0, y_offse
             dd.from_dask_array(ss.astype("float32"), columns="prob"),
         ],
         axis=1,
-        ignore_unknown_divisions=True
+        ignore_unknown_divisions=True,
     )
 
     # 5) Apply global offsets (if needed)
@@ -184,7 +190,9 @@ def centroids_map_to_dask_dataframe(scores: da.Array, x_offset: int = 0, y_offse
     return ddf
 
 
-def nucleus_detection_nms(df: pd.DataFrame, radius: int, overlap_threshold:float = 0.5) -> pd.DataFrame:
+def nucleus_detection_nms(
+    df: pd.DataFrame, radius: int, overlap_threshold: float = 0.5
+) -> pd.DataFrame:
     """Greedy NMS across ALL detections.
 
     Keeps the highest-prob detection, removes any other point within 'radius' pixels > overlap_threshold.
@@ -215,7 +223,7 @@ def nucleus_detection_nms(df: pd.DataFrame, radius: int, overlap_threshold:float
     coords = sub[["x", "y"]].to_numpy(dtype=np.float64)
     r = float(radius)
     two_r = 2.0 * r
-    two_r2 = (two_r * two_r)  # distance^2 cutoff for any overlap
+    two_r2 = two_r * two_r  # distance^2 cutoff for any overlap
 
     suppressed = np.zeros(len(sub), dtype=bool)
     keep_idx = []
@@ -232,18 +240,19 @@ def nucleus_detection_nms(df: pd.DataFrame, radius: int, overlap_threshold:float
         d2 = dx * dx + dy * dy
 
         # Only points with d < 2r can have nonzero overlap
-        cand = (d2 <= two_r2)
+        cand = d2 <= two_r2
         cand[i] = False  # don't suppress the kept point itself
         if not np.any(cand):
             continue
 
         d = np.sqrt(d2[cand])
 
-
         # Safe cosine argument = (distance ÷ diameter), Clamp for numerical stability
         u = np.clip(d / (2.0 * r), -1.0, 1.0)
         # Exact intersection area of two equal-radius circles.
-        inter = 2.0 * (r * r) * np.arccos(u) - 0.5 * d * np.sqrt(np.clip(4.0 * r * r - d * d, 0.0, None))
+        inter = 2.0 * (r * r) * np.arccos(u) - 0.5 * d * np.sqrt(
+            np.clip(4.0 * r * r - d * d, 0.0, None)
+        )
 
         union = 2.0 * np.pi * (r * r) - inter
         iou = inter / union
@@ -252,7 +261,7 @@ def nucleus_detection_nms(df: pd.DataFrame, radius: int, overlap_threshold:float
         idx_cand = np.where(cand)[0]
         to_suppress = idx_cand[iou >= overlap_threshold]
         suppressed[to_suppress] = True
-    
+
     kept = sub.iloc[keep_idx].copy()
     return kept
 
@@ -363,6 +372,7 @@ def post_process_patches(
             raw_predictions (da.Array): The raw predictions from the model.
             prediction_shape (tuple[int, ...]): The shape of the predictions.
             prediction_dtype (type): The data type of the predictions.
+
         Returns:
             A list of DataFrames containing the post-processed predictions for each patch.
 
@@ -376,7 +386,6 @@ def post_process_patches(
             batch_predictions.append(self.model.postproc_func(raw_predictions[i]))
         return batch_predictions
 
-
     def post_process_wsi(
         self: NucleusDetector,
         raw_predictions: da.Array,
@@ -396,8 +405,9 @@ def post_process_wsi(
         logger.info(f"Raw probabilities dtype: {prediction_dtype}")
         logger.info(f"Chunk size: {raw_predictions.chunks}")
 
-        detection_df = self.model.postproc(raw_predictions, prediction_shape, prediction_dtype)
-
+        detection_df = self.model.postproc(
+            raw_predictions, prediction_shape, prediction_dtype
+        )
 
         return detection_df
 
@@ -441,11 +451,9 @@ def save_predictions(
 
                 save_paths.append(out_file)
             return save_paths
-        else:
-            return df_to_store_nucleus_detector(
-                processed_predictions['predictions'],
-                scale_factor=scale_factor,
-                save_path=save_path,
-                class_dict=class_dict,
-            )
-            
+        return df_to_store_nucleus_detector(
+            processed_predictions["predictions"],
+            scale_factor=scale_factor,
+            save_path=save_path,
+            class_dict=class_dict,
+        )

tiatoolbox/utils/misc.py

@@ -21,7 +21,7 @@
 import zarr
 from filelock import FileLock
 from shapely.affinity import translate
-from shapely.geometry import Polygon, Point
+from shapely.geometry import Point, Polygon
 from shapely.geometry import shape as feature2geometry
 from skimage import exposure
 from tqdm import notebook as tqdm_notebook
@@ -1345,18 +1345,17 @@ def df_to_store_nucleus_detector(
     scale_factor: tuple[float, float],
     save_path: Path | None = None,
     class_dict: dict | None = None,
-    batch_size: int = 50_000
+    batch_size: int = 50_000,
 ) -> SQLiteStore | Path:
-    """
-    Convert a pandas DataFrame with columns ['x','y','type','prob']
+    """Convert a pandas DataFrame with columns ['x','y','type','prob']
     into an Annotation SQLiteStore efficiently using append_many().
 
     Args:
         df (pd.DataFrame):
             A pandas DataFrame with columns ['x','y','type','prob'].
         save_path (Path, optional):
             Optional Output directory to save the Annotation
-            Store results. 
+            Store results.
         scale_factor (tuple[float, float]):
                 The scale factor to use when saving the
                 annotations. All coordinates will be multiplied by this factor to allow
@@ -1373,7 +1372,6 @@ def df_to_store_nucleus_detector(
             or Path to file storing SQLiteStore containing Annotations
             for each nucleus.
     """
-
     # 1) Select & coerce dtypes once (compact + avoids per-row casts)
     x = df["x"].to_numpy(dtype=np.int64, copy=False)
     y = df["y"].to_numpy(dtype=np.int64, copy=False)
@@ -1387,7 +1385,7 @@ def df_to_store_nucleus_detector(
 
     def make_points(xb, yb):
         return [Point(int(xx), int(yy)) for xx, yy in zip(xb, yb)]
-    
+
     if class_dict is None:
         # identity over the actually present types (robust if types aren't 0..K)
         unique_types = np.unique(t)
@@ -1400,9 +1398,16 @@ def make_points(xb, yb):
 
         pts = make_points(xb, yb)  # array/list of Points
 
-        anns = [Annotation(geometry=pt,
-                            properties={"type": class_dict.get(int(tt), int(tt)), "probability": float(pp)})
-                for pt, tt, pp in zip(pts, tb, pb)]
+        anns = [
+            Annotation(
+                geometry=pt,
+                properties={
+                    "type": class_dict.get(int(tt), int(tt)),
+                    "probability": float(pp),
+                },
+            )
+            for pt, tt, pp in zip(pts, tb, pb)
+        ]
 
         store.append_many(anns)