Merge branch 'martin-carrasco-feat_rsc'

Author: gbg141

configs/transforms/liftings/pointcloud2simplicial/random_flag_complex_lifting.yaml

@@ -0,0 +1,7 @@
+transform_type: 'lifting'
+transform_name: "RandomFlagComplexLifting"
+complex_dim: 3
+alpha: 1.5
+steps: 10
+signed: True
+feature_lifting: ProjectionSum

modules/transforms/data_transform.py

@@ -80,6 +80,9 @@
 from modules.transforms.liftings.pointcloud2simplicial.delaunay_lifting import (
     DelaunayLifting,
 )
+from modules.transforms.liftings.pointcloud2simplicial.random_flag_complex import (
+    RandomFlagComplexLifting,
+)
 
 TRANSFORMS = {
     # Graph -> Hypergraph
@@ -107,6 +110,7 @@
     "AlphaComplexLifting": AlphaComplexLifting,
     # Point-cloud -> Simplicial Complex
     "DelaunayLifting": DelaunayLifting,
+    "RandomFlagComplexLifting": RandomFlagComplexLifting,
     # Pointcloud -> Hypergraph
     "VoronoiLifting": VoronoiLifting,
     "MoGMSTLifting": MoGMSTLifting,

modules/transforms/liftings/pointcloud2simplicial/random_flag_complex.py

@@ -0,0 +1,109 @@
+from itertools import combinations
+
+import gudhi
+import gudhi.simplex_tree
+import networkx as nx
+import numpy as np
+import torch
+from toponetx.classes import SimplicialComplex
+from torch_geometric.data import Data
+
+from modules.data.utils.utils import get_complex_connectivity
+from modules.transforms.liftings.pointcloud2simplicial.base import (
+    PointCloud2SimplicialLifting,
+)
+
+
+class RandomFlagComplexLifting(PointCloud2SimplicialLifting):
+    """  Lifting of pointclouds to simplicial complexes using the Random Flag Complex construction.
+    """
+    def __init__(self, steps, alpha: float | None = None, p: float | None = None, **kwargs):
+        self.alpha = alpha
+        self.steps = steps
+        self.p = p
+        super().__init__(**kwargs)
+
+    def lift_topology(self, data: Data) -> dict:
+        # Get the number of points and generate an empty graph
+        n = data["x"].size(0)
+        if self.p is None:
+            self.p = np.power(n, -self.alpha)
+        self.p = float(self.p)
+
+        adj_mat = np.zeros((n, n))
+        indices = np.tril_indices(n)
+
+        st = gudhi.SimplexTree()
+
+        generator = np.random.default_rng()
+        # For each step, sample from random binomial distribution
+        # for each edge appearign
+        for _ in range(self.steps):
+            number_of_edges = n*(n+1)//2
+            prob = generator.binomial(1, self.p, size=number_of_edges)
+            print(prob)
+            tmp_mat = np.zeros((n, n))
+            tmp_mat[indices] = prob
+            np.logical_or(adj_mat, tmp_mat, out=adj_mat)
+        np.fill_diagonal(adj_mat, 0)
+
+        # Insert all vertices
+        for i in range(n):
+            st.insert([i])
+
+        graph: nx.Graph = nx.from_numpy_matrix(adj_mat).to_undirected()
+
+        # Insert all edges
+        for v, u in graph.edges:
+            st.insert([v, u])
+
+        simplicial_complex = SimplicialComplex(graph)
+
+        # Add features to the vertices
+        feats = {
+            i: f
+            for i, f in enumerate(data["x"])
+        }
+
+        simplicial_complex.set_simplex_attributes(feats, name="features")
+
+        # Find the cliques up to the maximum dimension specified
+        cliques = nx.find_cliques(graph)
+        simplices = [set() for _ in range(2, self.complex_dim + 1)]
+
+        for clique in cliques:
+            for i in range(2, self.complex_dim + 1):
+                for c in combinations(clique, i + 1):
+                    simplices[i - 2].add(tuple(c))
+
+
+        # Add the k-tuples as simplices
+        for set_k_simplices in simplices:
+            for k_simplex in set_k_simplices:
+                st.insert(k_simplex)
+            simplicial_complex.add_simplices_from(list(set_k_simplices))
+
+        return self._get_lifted_topology(simplicial_complex, st)
+
+    def _get_lifted_topology(self, simplicial_complex: SimplicialComplex, st: gudhi.SimplexTree) -> dict:
+
+        # Connectivity of the complex
+        lifted_topology = get_complex_connectivity(
+            simplicial_complex, self.complex_dim, signed=False
+        )
+        # Computing the persitence to obtain the Betti numbers
+        st.compute_persistence(persistence_dim_max=True)
+
+        # Save the Betti numbers in the Data object
+        lifted_topology["betti"] = torch.tensor(st.betti_numbers())
+
+
+        lifted_topology["x_0"] = torch.stack(
+            list(simplicial_complex.get_simplex_attributes("features", 0).values())
+        )
+
+        # Add the indices of the simplices
+        for r in range(simplicial_complex.dim):
+            lifted_topology[f"x_idx_{r}"] = torch.tensor(simplicial_complex.skeleton(r))
+
+        return lifted_topology

test/transforms/liftings/pointcloud2simplicial/test_random_flag_complex_lifting.py

@@ -0,0 +1,38 @@
+"""Test the message passing module."""
+
+
+from modules.data.utils.utils import load_manual_graph
+from modules.transforms.liftings.pointcloud2simplicial.random_flag_complex import (
+    RandomFlagComplexLifting,
+)
+
+
+class TestRandomFlagComplexLifting:
+    """Test the SimplicialCliqueLifting class."""
+
+    def setup_method(self):
+        # Load the graph
+        self.data = load_manual_graph()
+        del self.data["edge_attr"]
+        del self.data["edge_index"]
+
+        self.lifting_p_0 = RandomFlagComplexLifting(steps=10, p=0)
+        self.lifting_p_1 = RandomFlagComplexLifting(steps=1, p=1)
+        self.lifting_hp = RandomFlagComplexLifting(steps=100, alpha=0.01)
+
+
+    def test_empty(self):
+        lifted_data = self.lifting_p_0.forward(self.data.clone())
+        assert(lifted_data.x_1.size(0) == 0)
+
+    def test_not_empty(self):
+        lifted_data = self.lifting_hp.forward(self.data.clone())
+        assert(lifted_data.x_1.size(0) > 0)
+
+    def test_full_graph(self):
+        lifted_data = self.lifting_p_1.forward(self.data.clone())
+        possible_edges = lifted_data.num_nodes * (self.data.num_nodes - 1) / 2
+        assert(lifted_data.x_1.size(0) == possible_edges)
+
+
+        assert(lifted_data)