Track: Track2; Team name: rayquaza; Model: HyperGCN

configs/model/hypergraph/hypergraph_convolution.yaml

@@ -0,0 +1,44 @@
+_target_: topobench.model.TBModel
+
+model_name: hypergraph_convolution
+model_domain: hypergraph
+
+feature_encoder:
+  _target_: topobench.nn.encoders.${model.feature_encoder.encoder_name}
+  encoder_name: AllCellFeatureEncoder
+  in_channels: ${infer_in_channels:${dataset},${oc.select:transforms,null}}
+  out_channels: 32
+  proj_dropout: 0.0
+  selected_dimensions:
+    - 0
+    - 1
+
+backbone:
+  _target_: topobench.nn.backbones.hypergraph.hypergraph_convolution.HyperGraphConvolution
+  a: ${model.feature_encoder.out_channels}
+  b: ${model.feature_encoder.out_channels}
+  # False -> propagate with the raw incidence matrix (works only because the khop
+  #          lifting produces a square [N, N] incidence).
+  # True  -> build the actual HyperGCN Laplacian from the hyperedges each forward.
+  reapproximate: True
+  cuda: 0
+
+backbone_wrapper:
+  _target_: topobench.nn.wrappers.HypergraphWrapper
+  _partial_: true
+  wrapper_name: HypergraphWrapper
+  out_channels: ${model.feature_encoder.out_channels}
+  num_cell_dimensions: 1
+
+readout:
+  _target_: topobench.nn.readouts.${model.readout.readout_name}
+  readout_name: PropagateSignalDown
+  # Must stay 1: PropagateSignalDown would otherwise look for model_out["x_1"]
+  # and batch["incidence_1"], neither of which exists for the hypergraph domain.
+  num_cell_dimensions: 1
+  hidden_dim: ${model.feature_encoder.out_channels}
+  out_channels: ${dataset.parameters.num_classes}
+  task_level: ${define_task_level:${dataset.parameters.task_level},${dataset.split_params.learning_setting}}
+  pooling_type: sum
+
+compile: false

configs/transforms/hypergraph_laplacian.yaml

@@ -0,0 +1,4 @@
+_target_: topobench.transforms.data_transform.DataTransform
+transform_name: "HypergraphLaplacian"
+transform_type: "liftings"
+m: true

pyproject.toml

@@ -120,12 +120,12 @@ explicit = true
 
 # Default find-links (will be overwritten by bash script)
 [tool.uv]
-find-links = ["https://data.pyg.org/whl/torch-2.3.0+cu121.html"]
+find-links = ["https://data.pyg.org/whl/torch-2.3.0+cpu.html"]
 
 [tool.uv.sources]
 torch = [
     { index = "pytorch-cpu", marker = "sys_platform == 'darwin' or sys_platform == 'win32'" },
-    { index = "pytorch-cu121", marker = "sys_platform == 'linux'" },
+    { index = "pytorch-cpu", marker = "sys_platform == 'linux'" },
 ]
 
 [tool.uv.extra-build-dependencies]

test/nn/backbones/hypergraph/test_hypergraph_convolution.py

@@ -0,0 +1,195 @@
+"""Unit tests for HyperGraphConvolution."""
+
+import pytest
+import torch
+import torch_geometric
+
+from topobench.nn.backbones.hypergraph.hypergraph_convolution import (
+    HyperGraphConvolution,
+    SparseMM,
+    incidence_to_hyperedges,
+)
+from topobench.nn.wrappers import HypergraphWrapper
+
+
+def _square_incidence(num_nodes, seed=0):
+    """Build a square sparse incidence matrix with no empty hyperedge.
+
+    Parameters
+    ----------
+    num_nodes : int
+        Number of nodes, also used as the number of hyperedges.
+    seed : int, optional
+        Seed for reproducibility, by default 0.
+
+    Returns
+    -------
+    torch.Tensor
+        Sparse incidence matrix of shape ``[num_nodes, num_nodes]``.
+    """
+    generator = torch.Generator().manual_seed(seed)
+    incidence = (
+        torch.rand(num_nodes, num_nodes, generator=generator) > 0.4
+    ).float()
+    # Guarantee every hyperedge has at least two members.
+    incidence[0, :] = 1.0
+    incidence[1, :] = 1.0
+    return incidence.to_sparse_coo()
+
+
+def test_incidence_to_hyperedges():
+    """Unit test for incidence_to_hyperedges."""
+    incidence = torch.tensor(
+        [
+            [1.0, 0.0, 1.0],
+            [1.0, 0.0, 0.0],
+            [0.0, 1.0, 0.0],
+            [1.0, 0.0, 0.0],
+        ]
+    ).to_sparse_coo()
+
+    # Hyperedges 1 and 2 are singletons and must be dropped.
+    hyperedges = incidence_to_hyperedges(incidence)
+    assert set(hyperedges) == {0}
+    assert sorted(hyperedges[0]) == [0, 1, 3]
+
+    # Lowering min_size keeps them.
+    hyperedges = incidence_to_hyperedges(incidence, min_size=1)
+    assert set(hyperedges) == {0, 1, 2}
+    assert sorted(hyperedges[1]) == [2]
+    assert sorted(hyperedges[2]) == [0]
+
+    # Every hyperedge is a singleton -> empty dict.
+    assert incidence_to_hyperedges(torch.eye(4).to_sparse_coo()) == {}
+
+
+def test_forward_without_reapproximation():
+    """Unit test for the forward pass reusing the incidence matrix."""
+    num_nodes, in_channels, out_channels = 6, 5, 3
+    x_0 = torch.randn(num_nodes, in_channels)
+    incidence = _square_incidence(num_nodes)
+
+    model = HyperGraphConvolution(
+        in_channels, out_channels, reapproximate=False
+    )
+    x_0_out, x_1_out = model(x_0, incidence)
+
+    assert x_0_out.shape == (num_nodes, out_channels)
+    assert x_1_out.shape == (num_nodes, out_channels)
+    assert torch.isfinite(x_0_out).all()
+
+
+@pytest.mark.parametrize("mediators", [True, False])
+def test_forward_with_reapproximation(mediators):
+    """Unit test for the forward pass rebuilding the Laplacian.
+
+    Parameters
+    ----------
+    mediators : bool
+        Whether the Laplacian approximation uses mediators.
+    """
+    num_nodes, num_hyperedges, in_channels, out_channels = 8, 5, 4, 3
+    x_0 = torch.randn(num_nodes, in_channels)
+
+    incidence = torch.zeros(num_nodes, num_hyperedges)
+    for edge in range(num_hyperedges):
+        members = torch.arange(edge, min(edge + 3, num_nodes))
+        incidence[members, edge] = 1.0
+    incidence = incidence.to_sparse_coo()
+
+    model = HyperGraphConvolution(
+        in_channels, out_channels, reapproximate=True
+    )
+    x_0_out, x_1_out = model(x_0, incidence, m=mediators)
+
+    assert x_0_out.shape == (num_nodes, out_channels)
+    assert x_1_out.shape == (num_hyperedges, out_channels)
+    assert torch.isfinite(x_0_out).all()
+
+
+def test_forward_with_only_singleton_hyperedges():
+    """Unit test for the identity fallback when no hyperedge survives."""
+    num_nodes, in_channels, out_channels = 4, 3, 2
+    x_0 = torch.randn(num_nodes, in_channels)
+    incidence = torch.eye(num_nodes).to_sparse_coo()
+
+    model = HyperGraphConvolution(
+        in_channels, out_channels, reapproximate=True
+    )
+    x_0_out, _ = model(x_0, incidence)
+
+    expected = x_0 @ model.W + model.bias
+    assert torch.allclose(x_0_out, expected, atol=1e-5)
+
+
+def test_backward():
+    """Unit test that gradients reach the layer parameters."""
+    num_nodes, in_channels, out_channels = 6, 5, 3
+    x_0 = torch.randn(num_nodes, in_channels)
+    incidence = _square_incidence(num_nodes)
+
+    model = HyperGraphConvolution(
+        in_channels, out_channels, reapproximate=False
+    )
+    x_0_out, _ = model(x_0, incidence)
+    x_0_out.sum().backward()
+
+    assert model.W.grad is not None
+    assert model.W.grad.shape == model.W.shape
+    assert model.bias.grad is not None
+
+
+def test_sparse_mm():
+    """Unit test for SparseMM covering both backward branches."""
+    m1 = torch.randn(3, 4, requires_grad=True)
+    m2 = torch.randn(4, 2, requires_grad=True)
+
+    out = SparseMM.apply(m1, m2)
+    assert torch.allclose(out, m1 @ m2, atol=1e-6)
+
+    out.sum().backward()
+    assert m1.grad.shape == m1.shape
+    assert m2.grad.shape == m2.shape
+
+
+def test_reset_parameters():
+    """Unit test that parameters are reinitialised in range."""
+    model = HyperGraphConvolution(4, 16)
+    model.reset_parameters()
+
+    bound = 1.0 / (16**0.5)
+    assert model.W.abs().max().item() <= bound
+    assert model.bias.abs().max().item() <= bound
+
+
+def test_repr():
+    """Unit test for the string representation."""
+    model = HyperGraphConvolution(4, 3)
+    assert repr(model) == "HyperGraphConvolution (4 -> 3)"
+
+
+def test_hypergraph_wrapper():
+    """Unit test for HyperGraphConvolution behind its wrapper."""
+    num_nodes, channels = 6, 4
+    x_0 = torch.randn(num_nodes, channels)
+    incidence = _square_incidence(num_nodes)
+
+    batch = torch_geometric.data.Data(
+        x_0=x_0,
+        y=torch.randint(0, 2, (num_nodes,)),
+        incidence_hyperedges=incidence,
+        batch_0=torch.zeros(num_nodes, dtype=torch.long),
+    )
+
+    backbone = HyperGraphConvolution(channels, channels, reapproximate=False)
+    wrapper = HypergraphWrapper(
+        backbone, **{"out_channels": channels, "num_cell_dimensions": 1}
+    )
+
+    _ = wrapper.__repr__()
+    model_out = wrapper(batch)
+
+    assert model_out["x_0"].shape == x_0.shape
+    assert model_out["hyperedge"].shape == (num_nodes, channels)
+    assert "labels" in model_out
+    assert "batch_0" in model_out

test/pipeline/test_pipeline.py

@@ -7,7 +7,7 @@
 
 
 DATASET = "graph/MUTAG"  # ADD YOUR DATASET HERE
-MODELS = ["graph/gcn", "cell/topotune", "simplicial/topotune"]  # ADD ONE OR SEVERAL MODELS
+MODELS = ["graph/gcn", "cell/topotune", "simplicial/topotune","hypergraph/hypergraph_convolution"]  # ADD ONE OR SEVERAL MODELS
 
 
 class TestPipeline:

test/transforms/liftings/graph2hypergraph/test_hypergraph_laplacian.py

@@ -0,0 +1,100 @@
+"""Unit tests for the hypergraph Laplacian approximation."""
+
+import numpy as np
+import pytest
+import scipy.sparse as sp
+import torch
+
+from topobench.transforms.liftings.graph2hypergraph.hypergraph_laplacian import (
+    Laplacian,
+    adjacency,
+    normalise,
+    ssm2tst,
+    symnormalise,
+    update,
+)
+
+
+@pytest.mark.parametrize("mediators", [True, False])
+def test_laplacian(mediators):
+    """Unit test for Laplacian.
+
+    Parameters
+    ----------
+    mediators : bool
+        Whether the approximation uses mediators.
+    """
+    num_nodes = 6
+    hyperedges = {0: [0, 1, 2], 1: [2, 3], 2: [3, 4, 5]}
+    features = np.random.default_rng(0).normal(size=(num_nodes, 4))
+
+    A = Laplacian(num_nodes, hyperedges, features, mediators)
+
+    assert A.shape == (num_nodes, num_nodes)
+    assert A.is_sparse
+    dense = A.to_dense()
+    assert torch.isfinite(dense).all()
+    # Self loops are added before normalisation, so the diagonal is non-zero.
+    assert (dense.diagonal() > 0).all()
+
+
+def test_update():
+    """Unit test for update."""
+    weights = update(0, 1, 2, {}, c=3.0)
+
+    assert set(weights) == {(0, 2), (1, 2), (2, 0), (2, 1)}
+    for value in weights.values():
+        assert value == pytest.approx(1 / 3)
+
+    # Calling again accumulates on the existing keys.
+    weights = update(0, 1, 2, weights, c=3.0)
+    for value in weights.values():
+        assert value == pytest.approx(2 / 3)
+
+
+def test_adjacency():
+    """Unit test for adjacency."""
+    edges = [[0, 1], [1, 0], [0, 1]]  # duplicated pair is deduplicated
+    weights = {(0, 1): 0.5, (1, 0): 0.5}
+
+    A = adjacency(edges, weights, n=3)
+    dense = A.to_dense()
+
+    assert dense.shape == (3, 3)
+    assert torch.allclose(dense, dense.t(), atol=1e-6)
+    # Isolated node 2 only has its self loop, normalised to one.
+    assert dense[2, 2].item() == pytest.approx(1.0, abs=1e-6)
+
+
+def test_symnormalise():
+    """Unit test for symnormalise."""
+    M = sp.csr_matrix(np.array([[2.0, 0.0], [0.0, 4.0]], dtype=np.float32))
+    out = np.asarray(symnormalise(M).todense())
+
+    assert np.allclose(out, np.eye(2), atol=1e-6)
+
+    # A zero row yields a zero scaling factor rather than an infinity.
+    M = sp.csr_matrix(np.array([[0.0, 0.0], [0.0, 4.0]], dtype=np.float32))
+    out = np.asarray(symnormalise(M).todense())
+    assert np.isfinite(out).all()
+
+
+def test_normalise():
+    """Unit test for normalise."""
+    M = sp.csr_matrix(np.array([[1.0, 3.0], [0.0, 0.0]], dtype=np.float32))
+    out = np.asarray(normalise(M).todense())
+
+    assert out[0].sum() == pytest.approx(1.0)
+    assert np.isfinite(out).all()
+
+
+def test_ssm2tst():
+    """Unit test for ssm2tst."""
+    M = sp.coo_matrix(np.array([[1.0, 0.0], [0.0, 2.0]], dtype=np.float32))
+    A = ssm2tst(M)
+
+    assert A.is_sparse
+    assert A.shape == (2, 2)
+    assert torch.allclose(
+        A.to_dense(), torch.tensor([[1.0, 0.0], [0.0, 2.0]]), atol=1e-6
+    )