Witness Lifting (Point Cloud to Simplicial)

configs/datasets/toy_point_cloud.yaml

@@ -0,0 +1,12 @@
+data_domain: point_cloud
+data_type: toy_dataset
+data_name: toy_point_cloud
+data_dir: datasets/${data_domain}/${data_type}
+
+# Dataset parameters
+num_points: 8
+num_classes: 2
+
+num_features: 1
+task: classification
+loss_type: cross_entropy

configs/transforms/liftings/pointcloud2simplicial/witness_lifting.yaml

@@ -0,0 +1,3 @@
+transform_type: "lifting"
+transform_name: "WitnessLifting"
+feature_lifting: ProjectionSum

modules/data/load/loaders.py

@@ -12,6 +12,7 @@
     load_cell_complex_dataset,
     load_hypergraph_pickle_dataset,
     load_manual_graph,
+    load_point_cloud,
     load_simplicial_dataset,
 )
 
@@ -204,3 +205,34 @@ def load(
             torch_geometric.data.Dataset object containing the loaded data.
         """
         return load_hypergraph_pickle_dataset(self.parameters)
+
+
+class PointCloudLoader(AbstractLoader):
+    r"""Loader for point-cloud dataset.
+    Parameters
+    ----------
+    parameters: DictConfig
+        Configuration parameters
+    """
+
+    def __init__(self, parameters: DictConfig):
+        super().__init__(parameters)
+        self.parameters = parameters
+        self.data_dir = self.parameters["data_dir"]
+        if "num_classes" not in self.cfg:
+            self.cfg["num_classes"] = 2
+
+    def load(self) -> torch_geometric.data.Dataset:
+        r"""Load point-cloud dataset.
+        Parameters
+        ----------
+        None
+        Returns
+        -------
+        torch_geometric.data.Dataset
+            torch_geometric.data.Dataset object containing the loaded data.
+        """
+        data = load_point_cloud(
+            num_classes=self.cfg["num_classes"], num_points=self.cfg["num_points"]
+        )
+        return CustomDataset([data], self.cfg["data_dir"])

modules/data/utils/utils.py

@@ -50,16 +50,16 @@ def get_complex_connectivity(complex, max_rank, signed=False):
                 )
             except ValueError:  # noqa: PERF203
                 if connectivity_info == "incidence":
-                    connectivity[f"{connectivity_info}_{rank_idx}"] = (
-                        generate_zero_sparse_connectivity(
-                            m=practical_shape[rank_idx - 1], n=practical_shape[rank_idx]
-                        )
+                    connectivity[
+                        f"{connectivity_info}_{rank_idx}"
+                    ] = generate_zero_sparse_connectivity(
+                        m=practical_shape[rank_idx - 1], n=practical_shape[rank_idx]
                     )
                 else:
-                    connectivity[f"{connectivity_info}_{rank_idx}"] = (
-                        generate_zero_sparse_connectivity(
-                            m=practical_shape[rank_idx], n=practical_shape[rank_idx]
-                        )
+                    connectivity[
+                        f"{connectivity_info}_{rank_idx}"
+                    ] = generate_zero_sparse_connectivity(
+                        m=practical_shape[rank_idx], n=practical_shape[rank_idx]
                     )
     connectivity["shape"] = practical_shape
     return connectivity
@@ -283,6 +283,17 @@ def load_hypergraph_pickle_dataset(cfg):
     return data
 
 
+def load_point_cloud(num_classes: int = 2, num_points: int = 18, seed: int = 42):
+    """Create a toy point cloud dataset"""
+    rng = np.random.default_rng(seed)
+
+    points = torch.tensor(rng.random((num_points, 2)), dtype=torch.float)
+    classes = torch.tensor(rng.integers(num_classes, size=num_points), dtype=torch.long)
+    features = torch.tensor(rng.integers(3, size=(num_points, 1)), dtype=torch.float)
+
+    return torch_geometric.data.Data(x=features, y=classes, pos=points)
+
+
 def load_manual_graph():
     """Create a manual graph for testing purposes."""
     # Define the vertices (just 8 vertices)

modules/transforms/data_transform.py

@@ -15,6 +15,9 @@
 from modules.transforms.liftings.graph2simplicial.clique_lifting import (
     SimplicialCliqueLifting,
 )
+from modules.transforms.liftings.pointcloud2simplicial.witness_lifting import (
+    WitnessLifting,
+)
 
 TRANSFORMS = {
     # Graph -> Hypergraph
@@ -23,6 +26,8 @@
     "SimplicialCliqueLifting": SimplicialCliqueLifting,
     # Graph -> Cell Complex
     "CellCycleLifting": CellCycleLifting,
+    # Point-cloud -> Simplicial Complex
+    "WitnessLifting": WitnessLifting,
     # Feature Liftings
     "ProjectionSum": ProjectionSum,
     # Data Manipulations

modules/transforms/liftings/pointcloud2simplicial/witness_lifting.py

@@ -0,0 +1,74 @@
+import gudhi as gd
+import torch
+import torch_geometric
+from toponetx.classes import SimplicialComplex
+
+from modules.data.utils.utils import get_complex_connectivity
+from modules.transforms.liftings.pointcloud2simplicial.base import (
+    PointCloud2SimplicialLifting,
+)
+
+
+class WitnessLifting(PointCloud2SimplicialLifting):
+    def __init__(
+        self,
+        is_weak=True,
+        is_euclidian=True,
+        landmark_proportion: int = 0.8,
+        max_alpha_square=0.15,
+        complex_dim=2,
+        seed=42,
+        **kwargs,
+    ):
+        super().__init__(**kwargs)
+        self.is_weak = is_weak
+        self.is_euclidian = is_euclidian
+        self.landmark_proportion = landmark_proportion
+        self.max_alpha_square = max_alpha_square
+        self.complex_dim = complex_dim
+        self.seed = seed
+        torch.manual_seed(seed)
+
+    def _get_lifted_topology(self, simplicial_complex: SimplicialComplex) -> dict:
+        r"""Returns the lifted topology.
+        Parameters
+        ----------
+        simplicial_complex : SimplicialComplex
+            The simplicial complex.
+        Returns
+        ---------
+        dict
+            The lifted topology.
+        """
+        lifted_topology = get_complex_connectivity(simplicial_complex, self.complex_dim)
+        lifted_topology["x_0"] = torch.stack(
+            list(simplicial_complex.get_simplex_attributes("features", 0).values())
+        )
+
+        return lifted_topology
+
+    def lift_topology(
+        self,
+        witnesses: torch_geometric.data.Data,
+    ) -> dict:
+        n = len(witnesses.pos)
+
+        perm = torch.randperm(n)
+        idx = perm[: round(n * self.landmark_proportion)]
+        landmarks_position = witnesses.pos[idx]
+
+        if self.is_euclidian:
+            if self.is_weak:
+                complex = gd.EuclideanWitnessComplex(witnesses.pos, landmarks_position)
+                simplex_tree = complex.create_simplex_tree(
+                    self.max_alpha_square, self.complex_dim
+                )
+
+                simplicial_complex = SimplicialComplex.from_gudhi(simplex_tree)
+            else:
+                pass
+
+        node_features = {i: witnesses.x[i, :] for i in range(witnesses.x.shape[0])}
+        simplicial_complex.set_simplex_attributes(node_features, name="features")
+
+        return self._get_lifted_topology(simplicial_complex)

test/transforms/liftings/pointcloud2simplicial/test_witness_lifting.py

@@ -0,0 +1,52 @@
+import torch
+
+from modules.data.utils.utils import load_point_cloud
+from modules.transforms.liftings.pointcloud2simplicial.witness_lifting import (
+    WitnessLifting,
+)
+
+
+class TestWitnessLifting:
+    """Test the WitnessLifting class."""
+
+    def setup_method(self):
+        # Load the point cloud
+        SEED = 42
+        self.data = load_point_cloud(num_points=5, seed=SEED)
+
+        # Initialise the WitnessLifting class
+        self.lifting_signed = WitnessLifting(signed=True, seed=SEED)
+        self.lifting_unsigned = WitnessLifting(signed=False, seed=SEED)
+
+    def test_lift_topology(self):
+        """Test the lift_topology method."""
+
+        # Test the lift_topology method
+        lifted_data_signed = self.lifting_signed.forward(self.data.clone())
+        lifted_data_unsigned = self.lifting_unsigned.forward(self.data.clone())
+
+        expected_incidence_1 = torch.tensor(
+            [
+                [1.0, 1.0, 1.0, 0.0, 0.0],
+                [1.0, 0.0, 0.0, 1.0, 0.0],
+                [0.0, 0.0, 0.0, 0.0, 1.0],
+                [0.0, 1.0, 0.0, 1.0, 0.0],
+                [0.0, 0.0, 1.0, 0.0, 1.0],
+            ]
+        )
+
+        assert (
+            abs(expected_incidence_1) == lifted_data_unsigned.incidence_1.to_dense()
+        ).all(), "Something is wrong with unsigned incidence_1 (nodes to edges)."
+        assert (
+            expected_incidence_1 == lifted_data_signed.incidence_1.to_dense()
+        ).all(), "Something is wrong with signed incidence_1 (nodes to edges)."
+
+        expected_incidence_2 = torch.tensor([[1.0], [1.0], [0.0], [1.0], [0.0]])
+
+        assert (
+            abs(expected_incidence_2) == lifted_data_unsigned.incidence_2.to_dense()
+        ).all(), "Something is wrong with unsigned incidence_2 (edges to triangles)."
+        assert (
+            expected_incidence_2 == lifted_data_signed.incidence_2.to_dense()
+        ).all(), "Something is wrong with signed incidence_2 (edges to triangles)."