diff --git a/tda_ml/metrics.py b/tda_ml/metrics.py index 5ef49eb..e6335f6 100644 --- a/tda_ml/metrics.py +++ b/tda_ml/metrics.py @@ -32,15 +32,16 @@ def compute_recall_specificity_gmean_mcc_wdist( *, points: np.ndarray | None = None, gt_inliers: np.ndarray | None = None, - empty_pred_wdist: float = 9.99, ) -> tuple[float, float, float, float, float]: """ Return the four classification metrics plus W-Dist. - W-Dist is computed only when ``points`` and ``gt_inliers`` are provided. - If every predicted label is outlier (no predicted inliers), returns - ``empty_pred_wdist`` (default ``9.99``), distinct from - :func:`tda_ml.persistence.compute_w_distance` returning ``999.0`` for an empty point cloud. + W-Dist is the H1 1-Wasserstein distance between persistence diagrams of + predicted inliers and ``gt_inliers``. It is computed only when ``points`` + and ``gt_inliers`` are provided; otherwise ``w_dist`` is ``0.0``. + An empty predicted-inlier set uses the standard OT distance + ``W(empty_diagram, PD(gt_inliers))`` — see + :func:`tda_ml.persistence.compute_w_distance`. """ recall, specificity, gmean, mcc = compute_recall_specificity_gmean_mcc( labels_gt, labels_pred @@ -48,8 +49,5 @@ def compute_recall_specificity_gmean_mcc_wdist( w_dist = 0.0 if points is not None and gt_inliers is not None: pred_inliers = points[np.asarray(labels_pred) == 0] - if len(pred_inliers) > 0: - w_dist = float(compute_w_distance(pred_inliers, gt_inliers)) - else: - w_dist = float(empty_pred_wdist) + w_dist = float(compute_w_distance(pred_inliers, gt_inliers)) return recall, specificity, gmean, mcc, w_dist diff --git a/tda_ml/persistence.py b/tda_ml/persistence.py index d9f21c7..cd75be0 100644 --- a/tda_ml/persistence.py +++ b/tda_ml/persistence.py @@ -27,26 +27,23 @@ def wasserstein_h1( def compute_w_distance(points_pred, points_gt): """ - Computes Wasserstein distance between persistence diagrams (dim 1) of two point clouds. + Wasserstein distance between H1 persistence diagrams of two point clouds. Args: points_pred: (N, 2) array or tensor points_gt: (M, 2) array or tensor Returns: - distance: float. Returns ``999.0`` when ``points_pred`` is empty (no diagram to match), - and ``0.0`` when ``points_gt`` is empty. + 1-Wasserstein distance (``order=1``, ``internal_p=2``) via Gudhi. + Empty point clouds yield empty H1 diagrams; the distance is the + standard optimal-transport cost (including matching to the diagonal), + not a sentinel penalty. """ if isinstance(points_pred, torch.Tensor): points_pred = points_pred.detach().cpu().numpy() if isinstance(points_gt, torch.Tensor): points_gt = points_gt.detach().cpu().numpy() - if len(points_pred) == 0: - return 999.0 - if len(points_gt) == 0: - return 0.0 - def get_pd(pts): if len(pts) < 3: return [] @@ -59,33 +56,24 @@ def get_pd(pts): pd_pred = get_pd(points_pred) pd_gt = get_pd(points_gt) - if len(pd_pred) == 0 and len(pd_gt) == 0: - return 0.0 - if len(pd_pred) == 0: - pd_pred = np.empty((0, 2)) - if len(pd_gt) == 0: - pd_gt = np.empty((0, 2)) - return wasserstein_h1(pd_pred, pd_gt, order=1, internal_p=2) def compute_bottleneck_distance(points_pred, points_gt): """ - Computes Bottleneck distance between persistence diagrams (dim 1). + Bottleneck distance between H1 persistence diagrams of two point clouds. Hold note (B-3 / GitHub #24): currently unused in active tda_ml/scripts/tests, but referenced in legacy code under `trash/`; keep until removal is fully validated. + + Empty point clouds yield empty H1 diagrams; distance is computed via Gudhi + without sentinel penalties (same policy as :func:`compute_w_distance`). """ if isinstance(points_pred, torch.Tensor): points_pred = points_pred.detach().cpu().numpy() if isinstance(points_gt, torch.Tensor): points_gt = points_gt.detach().cpu().numpy() - if len(points_pred) == 0: - return 99.0 - if len(points_gt) == 0: - return 0.0 - def get_pd(pts): if len(pts) < 3: return [] @@ -98,15 +86,12 @@ def get_pd(pts): pd_pred = get_pd(points_pred) pd_gt = get_pd(points_gt) - if len(pd_pred) == 0 and len(pd_gt) == 0: - return 0.0 if len(pd_pred) == 0: pd_pred = np.empty((0, 2)) if len(pd_gt) == 0: pd_gt = np.empty((0, 2)) - dist = gudhi.bottleneck_distance(pd_pred, pd_gt) - return dist + return float(gudhi.bottleneck_distance(pd_pred, pd_gt)) def compute_betti_error(points_pred, points_gt): diff --git a/tests/test_metrics.py b/tests/test_metrics.py index 1d89687..dcbb0a3 100644 --- a/tests/test_metrics.py +++ b/tests/test_metrics.py @@ -6,6 +6,7 @@ compute_recall_specificity_gmean_mcc, compute_recall_specificity_gmean_mcc_wdist, ) +from tda_ml.persistence import compute_w_distance class TestMetrics(unittest.TestCase): @@ -47,12 +48,19 @@ def test_compute_metrics_with_wdist_empty_pred_inliers(self): y_true = np.array([0, 0, 1, 1], dtype=int) y_pred = np.array([1, 1, 1, 1], dtype=int) points = np.array([[0.0, 0.0], [0.5, 0.2], [1.0, 1.0], [-1.0, -1.0]]) - gt_inliers = np.array([[0.0, 0.0], [0.5, 0.2]]) + gt_inliers = self._circle_gt_inliers() _, _, _, _, wdist = compute_recall_specificity_gmean_mcc_wdist( - y_true, y_pred, points=points, gt_inliers=gt_inliers, empty_pred_wdist=9.99 + y_true, y_pred, points=points, gt_inliers=gt_inliers ) - self.assertAlmostEqual(wdist, 9.99, places=6) + expected = compute_w_distance(np.empty((0, 2)), gt_inliers) + self.assertAlmostEqual(wdist, expected, places=6) + self.assertTrue(np.isfinite(wdist)) + + @staticmethod + def _circle_gt_inliers(n: int = 12) -> np.ndarray: + theta = np.linspace(0, 2 * np.pi, n, endpoint=False) + return np.stack([np.cos(theta), np.sin(theta)], axis=1) def test_compute_metrics_wdist_extreme_scale_inputs(self): """極小/極大スケールの点群でも指標計算が有限値で返る。""" diff --git a/tests/test_persistence.py b/tests/test_persistence.py new file mode 100644 index 0000000..4238ebd --- /dev/null +++ b/tests/test_persistence.py @@ -0,0 +1,36 @@ +import unittest + +import numpy as np + +from tda_ml.persistence import compute_bottleneck_distance, compute_w_distance + + +class TestPersistenceWasserstein(unittest.TestCase): + def _circle_points(self, n: int = 20) -> np.ndarray: + theta = np.linspace(0, 2 * np.pi, n, endpoint=False) + return np.stack([np.cos(theta), np.sin(theta)], axis=1) + + def test_empty_pred_returns_finite_wasserstein_not_sentinel(self): + circle = self._circle_points() + w = compute_w_distance([], circle) + self.assertTrue(np.isfinite(w)) + self.assertLess(w, 10.0) + + def test_empty_gt_symmetric_with_empty_pred(self): + circle = self._circle_points() + w_pred_empty = compute_w_distance([], circle) + w_gt_empty = compute_w_distance(circle, []) + self.assertAlmostEqual(w_pred_empty, w_gt_empty, places=6) + + def test_both_empty_point_clouds(self): + self.assertEqual(compute_w_distance([], []), 0.0) + + def test_bottleneck_empty_pred_not_sentinel(self): + circle = self._circle_points() + b = compute_bottleneck_distance([], circle) + self.assertTrue(np.isfinite(b)) + self.assertLess(b, 10.0) + + +if __name__ == "__main__": + unittest.main()