dexined configurable

backend/py/app/gradio_demo/ui.py

@@ -35,6 +35,11 @@ def _gradio_runtime(
     hough_max_gap: int,
     ellipse_min_area: int,
     max_ellipses: int,
+    dexined_use_marching_squares: bool,
+    dexined_threshold_mode: str,
+    dexined_threshold_sigma: float,
+    dexined_threshold_value: float,
+    dexined_threshold_offset: float,
 ) -> Tuple[Optional[np.ndarray], Optional[np.ndarray], Dict[str, Any]]:
     if image is None:
         return None, None, {"info": "No image provided."}
@@ -50,6 +55,11 @@ def _gradio_runtime(
         "hough_max_gap": int(hough_max_gap),
         "ellipse_min_area": int(ellipse_min_area),
         "max_ellipses": int(max_ellipses),
+        "dexined_use_marching_squares": bool(dexined_use_marching_squares),
+        "dexined_threshold_mode": str(dexined_threshold_mode).lower(),
+        "dexined_threshold_sigma": float(dexined_threshold_sigma),
+        "dexined_threshold_value": float(dexined_threshold_value),
+        "dexined_threshold_offset": float(dexined_threshold_offset),
     }
 
     params["line_detector"] = LINE_METHOD_LABELS.get(line_method, "hough")
@@ -119,6 +129,41 @@ def build_demo() -> gr.Blocks:
                     ellipse_min_area = gr.Slider(10, 50000, value=defaults["ellipse_min_area"], step=10, label="Ellipse min area (px^2)")
                     max_ellipses = gr.Slider(1, 20, value=defaults["max_ellipses"], step=1, label="Max ellipses")
 
+                with gr.Accordion("DexiNed Options", open=False):
+                    dexined_use_marching_squares = gr.Checkbox(
+                        value=defaults.get("dexined_use_marching_squares", False),
+                        label="Use marching squares iso-contours",
+                        info="Enable subpixel iso-contours (marching squares) for DexiNed outputs.",
+                    )
+                    dexined_threshold_mode = gr.Radio(
+                        choices=["adaptive", "fixed"],
+                        value=defaults.get("dexined_threshold_mode", "adaptive"),
+                        label="Threshold mode",
+                        info="Adaptive uses mean + σ·std; fixed uses a constant probability threshold.",
+                    )
+                    dexined_threshold_sigma = gr.Slider(
+                        0.0,
+                        5.0,
+                        value=defaults.get("dexined_threshold_sigma", 1.0),
+                        step=0.05,
+                        label="Adaptive σ multiplier",
+                    )
+                    dexined_threshold_value = gr.Slider(
+                        0.0,
+                        1.0,
+                        value=defaults.get("dexined_threshold_value", 0.3),
+                        step=0.01,
+                        label="Fixed probability threshold",
+                    )
+                    dexined_threshold_offset = gr.Slider(
+                        -0.5,
+                        0.5,
+                        value=defaults.get("dexined_threshold_offset", 0.0),
+                        step=0.01,
+                        label="Threshold offset",
+                        info="Applied after the selected mode (use small adjustments).",
+                    )
+
                 with gr.Accordion("DL Models (how to enable)", open=False):
                     dl_lines = "\n".join(
                         f"- **{det}**: {', '.join(models) if models else 'n/a'}" for det, models in DL_MODELS.items()
@@ -160,6 +205,11 @@ def build_demo() -> gr.Blocks:
                 hough_max_gap,
                 ellipse_min_area,
                 max_ellipses,
+                dexined_use_marching_squares,
+                dexined_threshold_mode,
+                dexined_threshold_sigma,
+                dexined_threshold_value,
+                dexined_threshold_offset,
             ],
             outputs=[out_img_classical, out_img_dl, meta_json],
         )
@@ -183,6 +233,11 @@ def build_demo() -> gr.Blocks:
                 hough_max_gap,
                 ellipse_min_area,
                 max_ellipses,
+                dexined_use_marching_squares,
+                dexined_threshold_mode,
+                dexined_threshold_sigma,
+                dexined_threshold_value,
+                dexined_threshold_offset,
             ],
             outputs=[out_img_classical, out_img_dl, meta_json],
         )

backend/py/app/inference/dl.py

@@ -44,10 +44,27 @@ def _load_session(path: str):
         raise RuntimeError(f"Failed to load ONNX model '{path}': {e}")
 
 
+def _extract_adapter_options(model_path: str, params: Optional[Dict[str, Any]]) -> Dict[str, Any]:
+    if not params:
+        return {}
+    name = os.path.basename(model_path).lower()
+    if "dexined" in name:
+        keys = [
+            "dexined_threshold_mode",
+            "dexined_threshold_sigma",
+            "dexined_threshold_offset",
+            "dexined_threshold_value",
+            "dexined_use_marching_squares",
+        ]
+        return {k: params.get(k) for k in keys if k in params}
+    return {}
+
+
 def detect_dl(
     image: np.ndarray,
     detector: str,
     model_choice: Optional[str],
+    params: Optional[Dict[str, Any]] = None,
 ) -> Tuple[np.ndarray, Dict[str, Any]]:
     bgr = to_bgr(image)
     rgb = to_rgb(bgr)
@@ -77,6 +94,10 @@ def detect_dl(
         meta["error"] = f"Preprocess failed: {e}"
         return rgb, meta
 
+    adapter_options = _extract_adapter_options(model_path, params)
+    if adapter_options and isinstance(ctx.extra, dict):
+        ctx.extra.update(adapter_options)
+
     try:
         outputs = sess.run(None, feed)
     except Exception as e:
@@ -86,6 +107,8 @@ def detect_dl(
     try:
         overlay, post_meta = adapter.postprocess(outputs, rgb, ctx, detector)
         meta.update(post_meta)
+        if adapter_options:
+            meta["adapter_options"] = adapter_options
     except Exception as e:
         meta["error"] = f"Postprocess failed: {e}"
         return rgb, meta

backend/py/app/inference/dl_adapters.py

@@ -4,6 +4,8 @@ import os
 from dataclasses import dataclass
 from typing import Any, Dict, List, Optional, Tuple
 
+import math
+
 import cv2
 import numpy as np
 
@@ -114,9 +116,35 @@ class EdgesAdapter(DLAdapter):
 
 
 class DexiNedAdapter(DLAdapter):
-    """DexiNed-specific adapter replicating the reference OpenCV DNN pipeline."""
+    """DexiNed-specific adapter replicating the reference OpenCV DNN pipeline with optional refinements."""
 
     _MEAN_BGR = (103.5, 116.2, 123.6)
+    _MARCHING_CASES: Dict[int, Tuple[Tuple[int, int], ...]] = {
+        0: (),
+        1: ((3, 0),),
+        2: ((0, 1),),
+        3: ((3, 1),),
+        4: ((1, 2),),
+        5: ((3, 2), (0, 1)),
+        6: ((0, 2),),
+        7: ((3, 2),),
+        8: ((2, 3),),
+        9: ((2, 0),),
+        10: ((1, 3), (0, 2)),
+        11: ((2, 1),),
+        12: ((1, 3),),
+        13: ((1, 0),),
+        14: ((3, 2),),
+        15: (),
+    }
+    _EDGE_CORNERS: Tuple[Tuple[int, int], ...] = (
+        (0, 1),  # top
+        (1, 2),  # right
+        (2, 3),  # bottom
+        (3, 0),  # left
+    )
+    _SUBPIXEL_MAX_SAMPLES = 48
+    _MS_MAX_SAMPLES = 48
 
     def preprocess(self, rgb: np.ndarray, sess) -> Tuple[Dict[str, np.ndarray], AdapterContext]:
         input_name, in_wh = _first_input(sess)
@@ -157,14 +185,150 @@ class DexiNedAdapter(DLAdapter):
         ave_u8 = np.mean(np.stack(maps_u8, axis=0), axis=0).astype(np.uint8)
         return fuse_prob, ave_prob, fuse_u8, ave_u8
 
+    @staticmethod
+    def _clamp_threshold(threshold: float) -> float:
+        return float(np.clip(threshold, 1e-6, 1.0))
+
+    def _resolve_threshold(self, fuse_prob: np.ndarray, options: Dict[str, Any]) -> Tuple[float, str]:
+        mode_raw = options.get("dexined_threshold_mode", "adaptive")
+        mode = mode_raw.lower() if isinstance(mode_raw, str) else "adaptive"
+        offset = float(options.get("dexined_threshold_offset", 0.0) or 0.0)
+        mean = float(np.mean(fuse_prob))
+        std = float(np.std(fuse_prob))
+        if mode == "fixed":
+            value = float(options.get("dexined_threshold_value", 0.3) or 0.0)
+            threshold = self._clamp_threshold(value + offset)
+            return threshold, "fixed"
+        sigma = float(options.get("dexined_threshold_sigma", 1.0) or 0.0)
+        threshold = self._clamp_threshold(mean + sigma * std + offset)
+        return threshold, "adaptive"
+
+    @staticmethod
+    def _interpolate(
+        p0: Tuple[float, float], p1: Tuple[float, float], v0: float, v1: float, level: float
+    ) -> Tuple[float, float]:
+        denom = v1 - v0
+        if abs(denom) < 1e-6:
+            t = 0.5
+        else:
+            t = (level - v0) / denom
+        t = float(np.clip(t, 0.0, 1.0))
+        return p0[0] + t * (p1[0] - p0[0]), p0[1] + t * (p1[1] - p0[1])
+
+    def _marching_squares(
+        self, prob: np.ndarray, level: float
+    ) -> List[Tuple[Tuple[float, float], Tuple[float, float]]]:
+        H, W = prob.shape
+        segments: List[Tuple[Tuple[float, float], Tuple[float, float]]] = []
+        for y in range(H - 1):
+            for x in range(W - 1):
+                v0 = prob[y, x]
+                v1 = prob[y, x + 1]
+                v2 = prob[y + 1, x + 1]
+                v3 = prob[y + 1, x]
+                mask = 0
+                if v0 >= level:
+                    mask |= 1
+                if v1 >= level:
+                    mask |= 2
+                if v2 >= level:
+                    mask |= 4
+                if v3 >= level:
+                    mask |= 8
+                edges = self._MARCHING_CASES[mask]
+                if not edges:
+                    continue
+                corners = (
+                    (float(x), float(y)),
+                    (float(x + 1), float(y)),
+                    (float(x + 1), float(y + 1)),
+                    (float(x), float(y + 1)),
+                )
+                values = (v0, v1, v2, v3)
+                for edge_a, edge_b in edges:
+                    a0, a1 = self._EDGE_CORNERS[edge_a]
+                    b0, b1 = self._EDGE_CORNERS[edge_b]
+                    pa = self._interpolate(corners[a0], corners[a1], values[a0], values[a1], level)
+                    pb = self._interpolate(corners[b0], corners[b1], values[b0], values[b1], level)
+                    segments.append((pa, pb))
+        return segments
+
+    @staticmethod
+    def _skeletonize(mask: np.ndarray) -> np.ndarray:
+        skeleton = np.zeros_like(mask)
+        element = cv2.getStructuringElement(cv2.MORPH_CROSS, (3, 3))
+        working = mask.copy()
+        while True:
+            eroded = cv2.erode(working, element)
+            opened = cv2.dilate(eroded, element)
+            temp = cv2.subtract(working, opened)
+            skeleton = cv2.bitwise_or(skeleton, temp)
+            working = eroded
+            if cv2.countNonZero(working) == 0:
+                break
+        return skeleton
+
+    @staticmethod
+    def _sample_bilinear(prob: np.ndarray, x: float, y: float, width: int, height: int) -> float:
+        x = float(np.clip(x, 0.0, width - 1.0))
+        y = float(np.clip(y, 0.0, height - 1.0))
+        x0 = int(math.floor(x))
+        y0 = int(math.floor(y))
+        x1 = min(x0 + 1, width - 1)
+        y1 = min(y0 + 1, height - 1)
+        dx = x - x0
+        dy = y - y0
+        return float(
+            (1 - dx) * (1 - dy) * prob[y0, x0]
+            + dx * (1 - dy) * prob[y0, x1]
+            + (1 - dx) * dy * prob[y1, x0]
+            + dx * dy * prob[y1, x1]
+        )
+
+    def _refine_subpixel(self, prob: np.ndarray, skeleton: np.ndarray) -> Tuple[int, List[Tuple[float, float]]]:
+        if not np.any(skeleton):
+            return 0, []
+        smooth = cv2.GaussianBlur(prob, (0, 0), sigmaX=1.0, sigmaY=1.0)
+        gx = cv2.Sobel(smooth, cv2.CV_32F, 1, 0, ksize=3)
+        gy = cv2.Sobel(smooth, cv2.CV_32F, 0, 1, ksize=3)
+        ys, xs = np.where(skeleton > 0)
+        height, width = prob.shape
+        count = 0
+        samples: List[Tuple[float, float]] = []
+        for x, y in zip(xs.tolist(), ys.tolist()):
+            gx_val = float(gx[y, x])
+            gy_val = float(gy[y, x])
+            mag = math.hypot(gx_val, gy_val)
+            if mag < 1e-6:
+                continue
+            nx = gx_val / mag
+            ny = gy_val / mag
+            p0 = self._sample_bilinear(smooth, x, y, width, height)
+            pm = self._sample_bilinear(smooth, x - nx, y - ny, width, height)
+            pp = self._sample_bilinear(smooth, x + nx, y + ny, width, height)
+            denom = pm - 2 * p0 + pp
+            if abs(denom) < 1e-8:
+                t = 0.0
+            else:
+                t = 0.5 * (pm - pp) / denom
+                t = float(np.clip(t, -1.0, 1.0))
+            x_sub = float(np.clip(x + t * nx, 0.0, width - 1.0))
+            y_sub = float(np.clip(y + t * ny, 0.0, height - 1.0))
+            count += 1
+            if len(samples) < self._SUBPIXEL_MAX_SAMPLES:
+                samples.append((x_sub, y_sub))
+        return count, samples
+
     def postprocess(
         self, outputs: List[np.ndarray], rgb: np.ndarray, ctx: AdapterContext, detector: str
     ) -> Tuple[np.ndarray, Dict[str, Any]]:
         fuse_prob, ave_prob, fuse_u8, ave_u8 = self._post_process_maps(outputs, ctx.orig_size)
-        threshold = float(fuse_prob.mean() + fuse_prob.std())
-        mask = fuse_prob > threshold
+        options = ctx.extra if isinstance(ctx.extra, dict) else {}
+        threshold, threshold_mode = self._resolve_threshold(fuse_prob, options)
+        use_marching_squares = bool(options.get("dexined_use_marching_squares", False))
+
         overlay = rgb.copy()
-        overlay[mask] = (0, 255, 0)
+        height, width = fuse_prob.shape
         meta: Dict[str, Any] = {
             "edge_prob_mean": float(np.mean(fuse_prob)),
             "edge_prob_std": float(np.std(fuse_prob)),
@@ -174,9 +338,41 @@ class DexiNedAdapter(DLAdapter):
             "ave_mean": float(np.mean(ave_u8) / 255.0),
             "fuse_min": float(np.min(fuse_prob)),
             "fuse_max": float(np.max(fuse_prob)),
+            "dexined_threshold": threshold,
+            "dexined_threshold_mode": threshold_mode,
+            "dexined_use_marching_squares": use_marching_squares,
             "adapter": "dexined",
             "resize": {"h": ctx.in_size[0], "w": ctx.in_size[1]},
         }
+
+        if use_marching_squares:
+            segments = self._marching_squares(fuse_prob, threshold)
+            for p0, p1 in segments:
+                x0 = int(np.clip(round(p0[0]), 0, width - 1))
+                y0 = int(np.clip(round(p0[1]), 0, height - 1))
+                x1 = int(np.clip(round(p1[0]), 0, width - 1))
+                y1 = int(np.clip(round(p1[1]), 0, height - 1))
+                cv2.line(overlay, (x0, y0), (x1, y1), (0, 255, 0), 1, lineType=cv2.LINE_AA)
+            meta["marching_squares_segments"] = int(len(segments))
+            if segments:
+                sample_segments = [
+                    [round(p0[0], 3), round(p0[1], 3), round(p1[0], 3), round(p1[1], 3)]
+                    for p0, p1 in segments[: self._MS_MAX_SAMPLES]
+                ]
+                meta["marching_squares_samples"] = sample_segments
+        else:
+            binary = (fuse_prob >= threshold).astype(np.uint8)
+            mask = (binary * 255).astype(np.uint8)
+            skeleton = self._skeletonize(mask)
+            overlay[skeleton > 0] = (0, 255, 0)
+            subpixel_count, subpixel_samples = self._refine_subpixel(fuse_prob, skeleton)
+            meta["skeleton_pixels"] = int(np.count_nonzero(skeleton))
+            meta["subpixel_points"] = int(subpixel_count)
+            if subpixel_samples:
+                meta["subpixel_samples"] = [
+                    [round(x, 3), round(y, 3)] for x, y in subpixel_samples[: self._SUBPIXEL_MAX_SAMPLES]
+                ]
+
         return overlay, meta
 
 

backend/py/app/services/runtime_adapter.py

@@ -22,8 +22,21 @@ DEFAULT_PARAMS: Dict[str, Any] = {
     "ellipse_min_area": 300,
     "max_ellipses": 5,
     "line_detector": "hough",
+    "dexined_threshold_mode": "adaptive",
+    "dexined_threshold_sigma": 1.0,
+    "dexined_threshold_offset": 0.0,
+    "dexined_threshold_value": 0.3,
+    "dexined_use_marching_squares": False,
 }
 
+def _to_bool(value: Any) -> bool:
+    if isinstance(value, bool):
+        return value
+    if isinstance(value, str):
+        return value.strip().lower() in {"1", "true", "yes", "on"}
+    return bool(value)
+
+
 PARAM_TYPES: Dict[str, Any] = {
     "canny_low": int,
     "canny_high": int,
@@ -36,6 +49,11 @@ PARAM_TYPES: Dict[str, Any] = {
     "ellipse_min_area": int,
     "max_ellipses": int,
     "line_detector": lambda x: str(x).lower(),
+    "dexined_threshold_mode": lambda x: str(x).lower(),
+    "dexined_threshold_sigma": float,
+    "dexined_threshold_offset": float,
+    "dexined_threshold_value": float,
+    "dexined_use_marching_squares": _to_bool,
 }
 
 CLASSICAL_MODEL_INFO = {"name": "opencv-classical", "version": cv2.__version__}
@@ -118,7 +136,7 @@ def run_detection(
 
     if execute_dl:
         t0 = time.perf_counter()
-        dl_img, dl_meta = detect_dl(image, detector, dl_choice)
+        dl_img, dl_meta = detect_dl(image, detector, dl_choice, params=merged)
         t_ms = (time.perf_counter() - t0) * 1000.0
         overlays["dl"] = dl_img
         features["dl"] = dl_meta

tests/test_edge_detection.py

@@ -2,6 +2,7 @@ import sys
 from pathlib import Path
 
 import numpy as np
+import pytest
 
 # Ensure project root is on the import path when tests run directly.
 ROOT = Path(__file__).resolve().parents[1]
@@ -46,9 +47,9 @@ def test_detect_classical_edges_highlights_pixels():
     assert np.any(np.all(overlay == np.array([0, 255, 0], dtype=np.uint8), axis=-1))
 
 
-def test_detect_dl_dexined_preprocess_postprocess_matches_expectations():
+def test_detect_dl_dexined_morphological_thinning_default():
     img = _synthetic_edge_image()
-    overlay, meta = detect_dl(img, detector="Edges (Canny)", model_choice="dexined.onnx")
+    overlay, meta = detect_dl(img, detector="Edges (Canny)", model_choice="dexined.onnx", params={})
 
     assert overlay.shape == img.shape
     assert overlay.dtype == np.uint8
@@ -59,4 +60,31 @@ def test_detect_dl_dexined_preprocess_postprocess_matches_expectations():
     assert 0.0 <= meta["edge_map_mean"] <= 1.0
     assert 0.0 <= meta["edge_map_std"] <= 1.0
     assert meta["model_path"].endswith("dexined.onnx")
+    assert meta["dexined_use_marching_squares"] is False
+    assert meta["dexined_threshold_mode"] == "adaptive"
+    assert 0.0 <= meta["dexined_threshold"] <= 1.0
+    assert meta.get("skeleton_pixels", 0) > 0
+    assert meta.get("subpixel_points", 0) >= 0
     assert np.any(np.all(overlay == np.array([0, 255, 0], dtype=np.uint8), axis=-1))
+
+
+def test_detect_dl_dexined_marching_squares_with_fixed_threshold():
+    img = _synthetic_edge_image()
+    params = {
+        "dexined_use_marching_squares": True,
+        "dexined_threshold_mode": "fixed",
+        "dexined_threshold_value": 0.05,
+    }
+    overlay, meta = detect_dl(img, detector="Edges (Canny)", model_choice="dexined.onnx", params=params)
+
+    assert overlay.shape == img.shape
+    assert overlay.dtype == np.uint8
+    assert meta.get("error") is None
+    assert meta["dexined_use_marching_squares"] is True
+    assert meta["dexined_threshold_mode"] == "fixed"
+    assert meta["dexined_threshold"] == pytest.approx(0.05, abs=1e-4)
+    assert meta.get("marching_squares_segments", 0) >= 0
+    if meta.get("marching_squares_segments", 0):
+        assert "marching_squares_samples" in meta
+        greenish = (overlay[..., 1] > overlay[..., 0] + 10) & (overlay[..., 1] > overlay[..., 2] + 10)
+        assert np.count_nonzero(greenish) > 0