adding in default pipeline configuration

Author: pmalacho-mit

release/__init__.py

@@ -1,5 +1,115 @@
-from .skeletonize import Skeletonize
+from .skeletonize import Skeletonize, ImageSource
 from .segment import Segment
 from .graph import StrokeGraph
 from .vectorize.low_geometry import Vectorize as LowGeometryVectorize
 from .vectorize.high_geometry import Vectorize as HighGeometryVectorize
+
+import numpy as np
+
+
+def default_pipeline(source: ImageSource):
+    skeleton = Skeletonize(
+        source,
+        Skeletonize.Config.Binarize(threshold=0.5),
+        Skeletonize.Config.Skeletonize(method="zhang"),
+        Skeletonize.Config.Collapse(
+            skeletonize_method="lee",
+            max_hole_area=10,
+            max_thin_thickness=3.0,
+            reskeletonize=True,
+        ),
+        detect_config={
+            "local_tau_radius": 40,
+            "fat_ratio": 1.3,
+            "min_fat_area": 8,
+            "group_dilate": 15,
+            "skel_ring_dilate": 5,
+            "pairing_tangent_steps": 8,
+            "pairing_threshold": 1.2,
+            "min_chromosome_skel_length": 15,
+        },
+    )
+
+    junction_tol = 2.5
+    tangent_sample = 10
+
+    segment = Segment(
+        skeleton.uncrossed,
+        skeleton.binary,
+        Segment.Config.Segment(min_length=10.0),
+        Segment.Config.Fuse(
+            max_path_length=20,
+            lookback=10,
+            min_tangent_score=0.5,
+            gap_penalty=0.05,
+            curvature_penalty=3.0,
+        ),
+        Segment.Config.Repair(
+            junction_tol=junction_tol,
+            stable_skip=2,
+            stable_sample=6,
+            max_junction_region_length=20,
+            min_output_polyline_length=2,
+            min_tangent_spread_deg=15.0,
+            interp_max_spacing=1.0,
+            min_curvature_spike_ratio=2.0,
+            curvature_context_window=8,
+        ),
+        Segment.Config.PostRepairFuse(
+            junction_tol=junction_tol,
+            tangent_skip=2,
+            tangent_sample=tangent_sample,
+            min_tangent_score=0.6,
+            curvature_penalty=1.0,
+        ),
+    )
+
+    graph = StrokeGraph(
+        segment.fused_post_repair,
+        StrokeGraph.Config.Build(
+            junction_tol=junction_tol,  # match Repair.junction_tol
+            terminal_tangent_window=10,  # should match fuse.lookback
+            crossing_tangent_skip=2,  # baseline; dynamic walk handles arbitrary bridges
+            crossing_tangent_half_window=6,
+            cusp_angle_threshold_deg=50.0,  # raise to ~50 to handle bikelove's cusp-like junction
+            cluster_merge_centroid_distance=10.0,
+            cluster_merge_index_gap=10,
+        ),
+    )
+
+    start_pos = np.array([0.0, 0.0])
+    start_heading = 0.0
+    low_geometry = LowGeometryVectorize(
+        graph,
+        start_pos=start_pos,
+        start_heading=start_heading,
+    )
+    high_geometry = HighGeometryVectorize(
+        segment.fused_post_repair,
+        start_pos=start_pos,
+        start_heading=start_heading,
+        commands=HighGeometryVectorize.Config.ToCommands(
+            sigma=2.0,
+            corner_threshold=0.25,
+            max_fit_residual=5.0,
+        ),
+        consolidate=HighGeometryVectorize.Config.Consolidate(
+            center_tol_rel=0.25,
+            radius_tol_rel=0.25,
+            center_tol_abs=3.0,
+            radius_tol_abs=3.0,
+            max_endpoint_snap_rel=0.15,
+            max_endpoint_snap_abs=6.0,
+            proximity_min_radius_ratio=0.4,
+            line_angle_tol_deg=6.0,
+            line_offset_tol_abs=5.0,
+            min_line_length=5.0,
+            max_line_endpoint_snap_abs=5.0,
+            junction_epsilon=3.0,
+            merge_arcs=True,
+            merge_lines=True,
+            return_report=False,
+        ),
+    )
+
+    return skeleton, segment, graph, low_geometry, high_geometry

release/skeletonize/__init__.py

@@ -1,6 +1,7 @@
 import numpy as np
 from numpy.typing import NDArray
 from skimage import io
+from skimage.color import rgb2gray, rgba2rgb
 from skimage.morphology import skeletonize as _skeletonize
 
 from .cleanup import collapse_small_holes, CollapseConfig
@@ -11,17 +12,45 @@
     resolve_crossings,
 )
 
-from typing import Literal, TypedDict, NamedTuple
+from typing import Literal, TypedDict, NamedTuple, Union
 
 
 class BinarizeConfig(TypedDict):
     threshold: float  # 0.0 to 1.0
 
 
-def to_binary(path: str, config: BinarizeConfig) -> NDArray[np.bool_]:
-    img: np.ndarray = io.imread(path, as_gray=True)
-    if img.dtype != np.float64 and img.dtype != np.float32:
-        img = img / 255.0
+ImageSource = Union[str, np.ndarray]
+
+
+def _to_grayscale_float(img: np.ndarray) -> np.ndarray:
+    """Normalize an arbitrary image array to a float grayscale in [0, 1]."""
+    if img.ndim == 3:
+        if img.shape[-1] == 4:
+            img = rgba2rgb(img)  # also yields float in [0, 1]
+        if img.shape[-1] == 3:
+            return rgb2gray(img)  # float in [0, 1]
+        raise ValueError(
+            f"Unsupported channel count {img.shape[-1]} for 3D image input; "
+            "expected 3 (RGB) or 4 (RGBA)."
+        )
+    if img.ndim != 2:
+        raise ValueError(
+            f"Unsupported image ndim {img.ndim}; expected 2 (grayscale) or 3."
+        )
+    if img.dtype == np.bool_:
+        return img.astype(np.float64)
+    if np.issubdtype(img.dtype, np.integer):
+        return img / np.float64(np.iinfo(img.dtype).max)
+    return img.astype(np.float64, copy=False)
+
+
+def to_binary(source: ImageSource, config: BinarizeConfig) -> NDArray[np.bool_]:
+    if isinstance(source, np.ndarray):
+        img = _to_grayscale_float(source)
+    else:
+        img = io.imread(source, as_gray=True)
+        if img.dtype != np.float64 and img.dtype != np.float32:
+            img = img / 255.0
     return img < config["threshold"]
 
 
@@ -80,13 +109,13 @@ class Output(NamedTuple):
 
     def __init__(
         self,
-        path: str,
+        source: ImageSource,
         binarize_config: Config.Binarize,
         skeletonize_config: Config.Skeletonize,
         collapse_config: Config.Collapse,
         detect_config: Config.Detect,
     ):
-        self.binary = to_binary(path, binarize_config)
+        self.binary = to_binary(source, binarize_config)
         self.skeletonized = skeletonize(self.binary, skeletonize_config)
         self.collapsed = collapse_small_holes(self.skeletonized, collapse_config)
         # Detection uses the BINARY for its distance-transform analysis

tests/__init__.py

@@ -1,9 +1,9 @@
 import os
 
 from release import (
+    default_pipeline,
     Skeletonize,
     Segment,
-    StrokeGraph,
     LowGeometryVectorize,
     HighGeometryVectorize,
 )
@@ -141,123 +141,22 @@ def process():
     for example in examples:
         if only and len(only) > 0 and example not in only:
             continue
-        skeleton = Skeletonize(
-            f"examples/{example}.png",
-            Skeletonize.Config.Binarize(threshold=0.5),
-            Skeletonize.Config.Skeletonize(method="zhang"),
-            Skeletonize.Config.Collapse(
-                skeletonize_method="lee",
-                max_hole_area=10,
-                max_thin_thickness=3.0,
-                reskeletonize=True,
-            ),
-            detect_config={
-                "local_tau_radius": 40,
-                "fat_ratio": 1.3,
-                "min_fat_area": 8,
-                "group_dilate": 15,
-                "skel_ring_dilate": 5,
-                "pairing_tangent_steps": 8,
-                "pairing_threshold": 1.2,
-                "min_chromosome_skel_length": 15,
-            },
-        )
-        Visualize.skeleton(skeleton, example)
 
-        junction_tol = 2.5
-        tangent_sample = 10
-
-        segment = Segment(
-            skeleton.uncrossed,
-            skeleton.binary,
-            Segment.Config.Segment(min_length=10.0),
-            Segment.Config.Fuse(
-                max_path_length=20,
-                lookback=10,
-                min_tangent_score=0.5,
-                gap_penalty=0.05,
-                curvature_penalty=3.0,
-            ),
-            Segment.Config.Repair(
-                junction_tol=junction_tol,
-                stable_skip=2,
-                stable_sample=6,
-                max_junction_region_length=20,
-                min_output_polyline_length=2,
-                min_tangent_spread_deg=15.0,
-                interp_max_spacing=1.0,
-                min_curvature_spike_ratio=2.0,
-                curvature_context_window=8,
-            ),
-            Segment.Config.PostRepairFuse(
-                junction_tol=junction_tol,
-                tangent_skip=2,
-                tangent_sample=tangent_sample,
-                min_tangent_score=0.6,
-                curvature_penalty=1.0,
-            ),
+        skeleton, segment, graph, low_geometry, high_geometry = default_pipeline(
+            f"examples/{example}.png"
         )
-        Visualize.segments(skeleton, segment, example)
 
-        graph = StrokeGraph(
-            segment.fused_post_repair,
-            StrokeGraph.Config.Build(
-                junction_tol=junction_tol,  # match Repair.junction_tol
-                terminal_tangent_window=10,  # should match fuse.lookback
-                crossing_tangent_skip=2,  # baseline; dynamic walk handles arbitrary bridges
-                crossing_tangent_half_window=6,
-                cusp_angle_threshold_deg=50.0,  # raise to ~50 to handle bikelove's cusp-like junction
-                cluster_merge_centroid_distance=10.0,
-                cluster_merge_index_gap=10,
-            ),
-        )
+        Visualize.skeleton(skeleton, example)
+        Visualize.segments(skeleton, segment, example)
         visualize_graph(
             skeleton.binary, graph, scale=1, output_path=f"examples/{example}.graph.png"
         )
-
-        start_pos = np.array([0.0, 0.0])
-        start_heading = 0.0
-        vectorized = {
-            "low_geometry": LowGeometryVectorize(
-                graph,
-                start_pos=start_pos,
-                start_heading=start_heading,
-            ),
-            "high_geometry": HighGeometryVectorize(
-                segment.fused_post_repair,
-                start_pos=start_pos,
-                start_heading=start_heading,
-                commands=HighGeometryVectorize.Config.ToCommands(
-                    sigma=2.0,
-                    corner_threshold=0.25,
-                    max_fit_residual=5.0,
-                ),
-                consolidate=HighGeometryVectorize.Config.Consolidate(
-                    center_tol_rel=0.25,
-                    radius_tol_rel=0.25,
-                    center_tol_abs=3.0,
-                    radius_tol_abs=3.0,
-                    max_endpoint_snap_rel=0.15,
-                    max_endpoint_snap_abs=6.0,
-                    proximity_min_radius_ratio=0.4,
-                    line_angle_tol_deg=6.0,
-                    line_offset_tol_abs=5.0,
-                    min_line_length=5.0,
-                    max_line_endpoint_snap_abs=5.0,
-                    junction_epsilon=3.0,
-                    merge_arcs=True,
-                    merge_lines=True,
-                    return_report=False,
-                ),
-            ),
-        }
-
         Visualize.vectorized(
-            vectorized["low_geometry"],
-            vectorized["high_geometry"],
+            low_geometry,
+            high_geometry,
             example,
-            start_pos=start_pos,
-            start_heading=start_heading,
+            start_pos=np.array([0.0, 0.0]),
+            start_heading=0.0,
         )