Added multilayer hypergraph

dhg/__init__.py

@@ -10,7 +10,7 @@
 
 from .structure import load_structure
 from .structure import BaseGraph, Graph, DiGraph, BiGraph
-from .structure import BaseHypergraph, Hypergraph
+from .structure import BaseHypergraph, Hypergraph, MultilayerHypergraph
 
 from ._global import AUTHOR_EMAIL, CACHE_ROOT, DATASETS_ROOT, REMOTE_DATASETS_ROOT
 
@@ -37,4 +37,5 @@
     "BiGraph",
     "BaseHypergraph",
     "Hypergraph",
-}
+    "MultilayerHypergraph",
+}

dhg/structure/__init__.py

@@ -1,7 +1,7 @@
 from .base import load_structure, BaseGraph, BaseHypergraph
 from .graphs import Graph, DiGraph, BiGraph
 from .hypergraphs import Hypergraph
-
+from .hypergraphs import MultilayerHypergraph
 
 __all__ = [
     "load_structure",
@@ -11,4 +11,5 @@
     "DiGraph",
     "BiGraph",
     "Hypergraph",
+    "MultilayerHypergraph"
 ]

dhg/structure/hypergraphs/__init__.py

@@ -1 +1,2 @@
 from .hypergraph import Hypergraph
+from .multilayer_hypergraph import MultilayerHypergraph

dhg/structure/hypergraphs/multilayer_hypergraph.py

@@ -0,0 +1,153 @@
+import itertools
+import random
+import re
+from typing import Iterable, List, Optional, Union
+
+import torch
+
+from .hypergraph import Hypergraph
+
+
+class MultilayerHypergraph(Hypergraph):
+    """A hypergraph composed of multiple hypergraph layers."""
+
+    def __init__(
+        self,
+        num_v: int,
+        num_layers: int,
+        layers_list: List[Hypergraph],
+        e_list=None,
+        e_weight=None,
+        v_weight=None,
+        prob_inner_layer_connect: float = 0,
+        device: torch.device = torch.device("cpu"),
+    ) -> None:
+        super().__init__(num_v, e_list, e_weight, v_weight, device=device)
+        if num_v <= 1:
+            raise ValueError("num_v must be greater than 1")
+        if num_layers <= 1:
+            raise ValueError("num_layers must be greater than 1")
+        if layers_list is None or num_layers != len(layers_list):
+            raise ValueError("num_layers must equal the length of layers_list")
+        if not 0 <= prob_inner_layer_connect <= 1:
+            raise ValueError("prob_inner_layer_connect must be between 0 and 1")
+
+        self.num_layers = num_layers
+        self.layers_list = list(layers_list)
+        self.prob_inner_layer_connect = prob_inner_layer_connect
+        self.node_layer: List[List[int]] = []
+
+    def _compute_offsets(self) -> List[int]:
+        """Return each layer's starting vertex index in the combined structure."""
+        offsets = []
+        total = 0
+        for layer in self.layers_list:
+            offsets.append(total)
+            total += layer.num_v
+        return offsets
+
+    def construct_multi_layer_hypergraph(self) -> Hypergraph:
+        """Build and return the flattened hypergraph for all layers."""
+        offsets = self._compute_offsets()
+        layer_nodes = [
+            self.select_nodes(range(layer.num_v), self.prob_inner_layer_connect)
+            for layer in self.layers_list
+        ]
+        self.node_layer = [
+            [node + offset for node in nodes]
+            for nodes, offset in zip(layer_nodes, offsets)
+        ]
+
+        inter_layer_hyperedges = list(itertools.product(*self.node_layer))
+        layer_hyperedges = self._inter_layer_connect()
+        all_hyperedges = inter_layer_hyperedges + [
+            edge for edges in layer_hyperedges for edge in edges
+        ]
+        return Hypergraph(
+            self.num_v_multi(), all_hyperedges, device=self.device
+        )
+
+    def num_v_multi(self) -> int:
+        """Return the total number of vertices across all layers."""
+        return sum(layer.num_v for layer in self.layers_list)
+
+    def _inter_layer_connect(self) -> List[list]:
+        """Return each layer's hyperedges with its global vertex offset applied."""
+        offsets = self._compute_offsets()
+        return [
+            self.add_numbers_list(layer.e[0], offset)
+            for layer, offset in zip(self.layers_list, offsets)
+        ]
+
+    def load_interlayer_mapping(
+        self,
+        path: Optional[Union[str, List[str]]],
+        save: bool = False,
+        save_path=None,
+    ) -> List[tuple]:
+        """Load gene mappings and return hyperedges for genes shared across layers.
+
+        ``path`` may be one file containing ``gene [codes]`` records, or a list
+        of tab-separated ``gene<TAB>code`` files, one per layer.
+        """
+        if isinstance(path, str):
+            gene_encoding = {}
+            with open(path, "r", encoding="utf-8") as file:
+                for line in file:
+                    match = re.findall(r"([^\[]+)\s+\[([^\]]+)\]", line)
+                    if not match:
+                        continue
+                    gene, codes = match[0]
+                    gene_encoding[gene.strip()] = [
+                        int(code.strip()) for code in codes.split(",")
+                    ]
+        elif isinstance(path, list):
+            gene_encoding = {}
+            for file_path in path:
+                with open(file_path, "r", encoding="utf-8") as file:
+                    for line in file:
+                        gene, code = line.rstrip("\n").split("\t", 1)
+                        gene_encoding.setdefault(gene, []).append(int(code))
+        else:
+            raise TypeError("path must be a file path or a list of file paths")
+
+        if save:
+            if save_path is None:
+                raise ValueError("save_path is required when save=True")
+            with open(save_path, "w", encoding="utf-8") as file:
+                for gene, codes in gene_encoding.items():
+                    file.write(f"{gene} {codes}\n")
+
+        return [tuple(codes) for codes in gene_encoding.values() if len(codes) > 1]
+
+    def node_layer_list(self) -> List[List[int]]:
+        """Return the vertices in each layer using flattened global indices."""
+        offsets = self._compute_offsets()
+        return [
+            [node + offset for node in layer.v]
+            for layer, offset in zip(self.layers_list, offsets)
+        ]
+
+    @staticmethod
+    def select_nodes(num_v: Iterable[int], prob: float) -> list:
+        """Select each vertex independently with probability ``prob``."""
+        return [node for node in num_v if random.random() < prob]
+
+    @staticmethod
+    def add_numbers_list(lst: Iterable[Iterable[int]], num: int) -> list:
+        """Add ``num`` to every vertex in each hyperedge."""
+        return [tuple(vertex + num for vertex in edge) for edge in lst]
+
+    def __str__(self):
+        return (
+            "MultilayerHypergraph(num_v={}, num_e={}, num_layers={}, "
+            "layers_list={}, prob_inner_layer_connect={})"
+        ).format(
+            self.num_v,
+            self.num_e,
+            self.num_layers,
+            self.layers_list,
+            self.prob_inner_layer_connect,
+        )
+
+    __repr__ = __str__

tests/structure/test_multilayer_hypergraph.py

@@ -0,0 +1,64 @@
+import torch
+import pytest
+
+from dhg import Hypergraph, MultilayerHypergraph
+
+
+@pytest.fixture
+def g1():
+    return Hypergraph(6, [(0, 1, 2, 5), (0, 1), (2, 3, 4), (3, 2, 4)])
+
+
+@pytest.fixture
+def g2():
+    return Hypergraph(5, [(1, 2, 3), (0, 1, 3), (0, 1), (2, 4, 3), (2, 3)])
+
+
+def test_construct_multilayer_hypergraph(g1, g2):
+    mhg = MultilayerHypergraph(11, 2, [g1, g2])
+
+    assert mhg.num_layers == 2
+    assert mhg.layers_list == [g1, g2]
+    assert mhg.num_v == 11
+    assert mhg.prob_inner_layer_connect == 0
+    assert mhg.device == torch.device("cpu")
+
+    combined = mhg.construct_multi_layer_hypergraph()
+    assert type(combined) is Hypergraph
+    assert combined.num_v == 11
+    assert combined.num_e == g1.num_e + g2.num_e
+
+
+def test_offsets_are_local_and_repeatable(g1, g2):
+    mhg = MultilayerHypergraph(11, 2, [g1, g2])
+
+    assert mhg.node_layer_list() == [list(g1.v), [v + 6 for v in g2.v]]
+    first = mhg._inter_layer_connect()
+    second = mhg._inter_layer_connect()
+    assert first == second
+    assert mhg.construct_multi_layer_hypergraph().num_e == g1.num_e + g2.num_e
+
+
+def test_select_nodes_accepts_range(monkeypatch):
+    monkeypatch.setattr(
+        "dhg.structure.hypergraphs.multilayer_hypergraph.random.random",
+        lambda: 0,
+    )
+    assert MultilayerHypergraph.select_nodes(range(3), 1.0) == [0, 1, 2]
+
+
+def test_mapping_save_writes_once(g1, g2, tmp_path):
+    source_a = tmp_path / "a.txt"
+    source_b = tmp_path / "b.txt"
+    source_a.write_text("geneA\t1\ngeneB\t2\n", encoding="utf-8")
+    source_b.write_text("geneA\t7\ngeneB\t8\n", encoding="utf-8")
+    output = tmp_path / "merged.txt"
+    mhg = MultilayerHypergraph(11, 2, [g1, g2])
+
+    assert mhg.load_interlayer_mapping(
+        [str(source_a), str(source_b)], True, output
+    ) == [(1, 7), (2, 8)]
+    assert output.read_text(encoding="utf-8").splitlines() == [
+        "geneA [1, 7]",
+        "geneB [2, 8]",
+    ]