Tangential (Delaunay) Lifting (PointCloud 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/tangential_lifting.yaml

@@ -0,0 +1,3 @@
+transform_type: "lifting"
+transform_name: "TangentialLifting"
+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.tangential_lifting import (
+    TangentialLifting,
+)
 
 TRANSFORMS = {
     # Graph -> Hypergraph
@@ -23,6 +26,8 @@
     "SimplicialCliqueLifting": SimplicialCliqueLifting,
     # Graph -> Cell Complex
     "CellCycleLifting": CellCycleLifting,
+    # Point-cloud -> Simplicial Complex
+    "TangentialLifting": TangentialLifting,
     # Feature Liftings
     "ProjectionSum": ProjectionSum,
     # Data Manipulations

modules/transforms/liftings/pointcloud2simplicial/tangential_lifting.py

@@ -0,0 +1,42 @@
+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 TangentialLifting(PointCloud2SimplicialLifting):
+    # intrinsic dimension of the manifold set to 1 by default
+    def __init__(self, intrisic_dim=2, **kwargs):
+        super().__init__(**kwargs)
+        self.intrisic_dim = intrisic_dim
+
+    def _get_lifted_topology(self, simplicial_complex: SimplicialComplex) -> dict:
+        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, data: torch_geometric.data.Data, **kwargs) -> dict:
+
+        # initialize tangential complex object
+        tangential_complex = gd.TangentialComplex(self.intrisic_dim, data.pos)
+
+        # build the complex
+        tangential_complex.compute_tangential_complex()
+
+        simplicial_complex = SimplicialComplex().from_gudhi(tangential_complex.create_simplex_tree())
+
+        self.complex_dim = simplicial_complex.dim
+
+        node_features = {i: data.x[i, :] for i in range(data.x.shape[0])}
+        simplicial_complex.set_simplex_attributes(node_features, name="features")
+
+        return self._get_lifted_topology(simplicial_complex)

test/transforms/liftings/pointcloud2simplicial/test_tangential_lifting.py

@@ -0,0 +1,62 @@
+import torch
+
+from modules.data.utils.utils import load_point_cloud
+from modules.transforms.liftings.pointcloud2simplicial.tangential_lifting import (
+    TangentialLifting,
+)
+
+
+class TestTangentialLifting:
+    """Test the DelaunayLifting class."""
+
+    def setup_method(self):
+        # Load the point cloud
+        SEED = 42
+        self.data = load_point_cloud(num_points=5, seed=SEED)
+
+        # Initialise the TangentialLifting class
+        self.lifting_signed = TangentialLifting(signed=True)
+        self.lifting_unsigned = TangentialLifting(signed=False)
+
+    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., 1., 1., 0., 0., 0., 0.],
+                [1., 0., 0., 1., 0., 0., 0.],
+                [0., 0., 0., 0., 1., 1., 0.],
+                [0., 1., 0., 1., 1., 0., 1.],
+                [0., 0., 1., 0., 0., 1., 1.]
+            ]
+        )
+
+        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., 0.],
+                [1., 1., 0.],
+                [0., 1., 0.],
+                [1., 0., 0.],
+                [0., 0., 1.],
+                [0., 0., 1.],
+                [0., 1., 1.]
+            ]
+        )
+
+        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)."