Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 9 additions & 1 deletion src/MuxVizPy/versatility.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,14 @@
from MuxVizPy.utils.katz_utils import _katz_neumann, _katz_krylov, _VALID_SOLVERS


def get_largest_eigenvalue(
adj: sps.spmatrix,
logger: logging.Logger | None = None,
) -> tuple[float, np.ndarray]:
"""Return the largest-magnitude eigenpair using the legacy function name."""
return get_largest_magnitude_eigenvalue(adj, logger=logger)


# ---------------------------------------------------------------------------
# Block accumulation and aggregation helpers (integrated from hornet/node_based)
# ---------------------------------------------------------------------------
Expand Down Expand Up @@ -888,4 +896,4 @@ def get_multi_Kcore_centrality(supra: sps.spmatrix, layers: int, nodes: int):
kcore_table[:,l] = gt.topology.kcore_decomposition(g_tmp).get_array()

centrality_vector = np.min(kcore_table, axis=1)
return centrality_vector
return centrality_vector
106 changes: 89 additions & 17 deletions tests/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,8 @@
- Numerical comparison helpers
"""

from collections.abc import Callable

import pytest
import subprocess
import tempfile
Expand Down Expand Up @@ -373,15 +375,11 @@ def generate_script(


def _ensure_edges_csv(config_name: str) -> Path:
"""Return the path to an edges CSV for a config, writing it if needed."""
config = NETWORK_CONFIGS[config_name]
"""Write and return the deterministic edges CSV for a config."""
data_dir = TESTS_DATA_DIR / config_name
data_dir.mkdir(parents=True, exist_ok=True)
edges_path = data_dir / "edges.csv"
if edges_path.exists():
return edges_path
# For configs without a data_dir (e.g. toy), write from TOY_EDGES
save_network_for_muxviz(TOY_EDGES, edges_path)
save_network_for_muxviz(_network_edges(config_name), edges_path)
return edges_path


Expand Down Expand Up @@ -561,26 +559,103 @@ def toy_interaction(toy_network):
# Parametrized network configs for cross-network testing
# ---------------------------------------------------------------------------

NETWORK_CONFIGS = {
Edge = tuple[int, int, int, int, float]


def _add_ordered_interlayer_edges(
edges: list[Edge],
n_nodes: int,
n_layers: int,
) -> None:
"""Add bidirectional coupling between adjacent replicas of each node."""
for layer in range(n_layers - 1):
for node in range(n_nodes):
edges.append((node, layer, node, layer + 1, 1.0))
edges.append((node, layer + 1, node, layer, 1.0))


def _generate_random_large_edges() -> list[Edge]:
"""Generate a deterministic sparse random multilayer network."""
n_nodes = 1000
n_layers = 4
rng = np.random.default_rng(1729)
edges: list[Edge] = []

for layer in range(n_layers):
for source in range(n_nodes):
raw_targets = rng.choice(n_nodes - 1, size=2, replace=False)
targets = raw_targets + (raw_targets >= source)
for target in targets:
weight = float(rng.integers(1, 4))
edges.append((source, layer, int(target), layer, weight))

_add_ordered_interlayer_edges(edges, n_nodes, n_layers)
return edges


def _generate_scalefree_small_edges() -> list[Edge]:
"""Generate a deterministic preferential-attachment multilayer network."""
n_nodes = 10
n_layers = 2
rng = np.random.default_rng(2718)
edges: list[Edge] = []

for layer in range(n_layers):
degree = np.ones(n_nodes, dtype=float)
for source in range(1, n_nodes):
n_targets = min(2, source)
probabilities = degree[:source] / degree[:source].sum()
targets = rng.choice(
source,
size=n_targets,
replace=False,
p=probabilities,
)
for target in targets:
target = int(target)
edges.append((source, layer, target, layer, 1.0))
edges.append((target, layer, source, layer, 1.0))
degree[source] += 1
degree[target] += 1

_add_ordered_interlayer_edges(edges, n_nodes, n_layers)
return edges


NETWORK_CONFIGS: dict[str, dict[str, int]] = {
"toy": {
"n_nodes": TOY_N_NODES,
"n_layers": TOY_N_LAYERS,
},
"random_large": {
"n_nodes": 1000,
"n_layers": 4,
"data_dir": "random_large",
},
"scalefree_small": {
"n_nodes": 10,
"n_layers": 2,
"data_dir": "scalefree_small",
},
}

NETWORK_EDGE_FACTORIES: dict[str, Callable[[], list[Edge]]] = {
"random_large": _generate_random_large_edges,
"scalefree_small": _generate_scalefree_small_edges,
}

TESTS_DATA_DIR = Path(__file__).parent / "data"


def _network_edges(config_name: str) -> list[Edge]:
"""Return a fresh deterministic edge list for a test configuration."""
if config_name not in NETWORK_CONFIGS:
raise KeyError(f"Unknown network configuration: {config_name}")

edge_factory = NETWORK_EDGE_FACTORIES.get(config_name)
if edge_factory is None:
return list(TOY_EDGES)
return edge_factory()


@pytest.fixture(scope="session", params=list(NETWORK_CONFIGS.keys()))
def network_config(request):
"""Parametrized network configuration name."""
Expand All @@ -594,14 +669,11 @@ def net_info(network_config):
n_nodes = config["n_nodes"]
n_layers = config["n_layers"]

if "data_dir" in config:
df = pl.read_csv(str(TESTS_DATA_DIR / config["data_dir"] / "edges.csv"))
else:
df = pl.DataFrame(
TOY_EDGES,
schema=["node.from", "layer.from", "node.to", "layer.to", "weight"],
orient="row",
)
df = pl.DataFrame(
_network_edges(network_config),
schema=["node.from", "layer.from", "node.to", "layer.to", "weight"],
orient="row",
)

tensor = parsing.build_tensor_from_dataframe(df)
adj = parsing.build_supra_adjacency_matrix_from_tensor(tensor)
Expand Down
35 changes: 35 additions & 0 deletions tests/test_fixture_generation.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,35 @@
"""Tests for deterministic parametrized network fixture generation."""

import polars as pl
import pytest
from conftest import NETWORK_CONFIGS, _network_edges

from MuxVizPy.utils import parsing


@pytest.mark.parametrize("config_name", ["random_large", "scalefree_small"])
def test_generated_network_is_deterministic(config_name):
assert _network_edges(config_name) == _network_edges(config_name)


@pytest.mark.parametrize("config_name", ["random_large", "scalefree_small"])
def test_generated_network_has_declared_dimensions(config_name):
config = NETWORK_CONFIGS[config_name]
n_nodes = config["n_nodes"]
n_layers = config["n_layers"]
edges = _network_edges(config_name)

node_ids = {edge[0] for edge in edges} | {edge[2] for edge in edges}
layer_ids = {edge[1] for edge in edges} | {edge[3] for edge in edges}

assert node_ids == set(range(n_nodes))
assert layer_ids == set(range(n_layers))

frame = pl.DataFrame(
edges,
schema=["node.from", "layer.from", "node.to", "layer.to", "weight"],
orient="row",
)
tensor = parsing.build_tensor_from_dataframe(frame)

assert tensor.shape == (n_nodes, n_layers, n_nodes, n_layers)
9 changes: 8 additions & 1 deletion tests/test_versatility.py
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,13 @@ def test_get_largest_eigenvalue_positive(self, net_adjacency):
lam, _ = versatility.get_largest_eigenvalue(net_adjacency)
assert lam > 0

def test_get_largest_eigenvalue_preserves_largest_magnitude_behavior(self):
adj = sp.diags([-5.0, 2.0, 1.0], format="csr")

lam, _ = versatility.get_largest_eigenvalue(adj)

assert lam == pytest.approx(-5.0)

def test_approximate_largest_eigenvalue_returns_tuple(self, net_adjacency, net_nl):
lam, vec = versatility.approximate_largest_eigenvalue(net_adjacency)
assert isinstance(lam, float)
Expand Down Expand Up @@ -409,4 +416,4 @@ def test_multi_degree_hornet_vs_muxviz(self, net_adjacency, net_n, net_l, net_mu
"""get_multi_degree hornet backend matches R GetMultiDegree."""
expected = np.array(net_muxviz_results["indegree"]) + np.array(net_muxviz_results["outdegree"])
computed = versatility.get_multi_degree(net_adjacency, net_l, net_n, backend="hornet")
compare_metrics(computed, expected, "MultiDegree hornet (vs muxViz R derived)")
compare_metrics(computed, expected, "MultiDegree hornet (vs muxViz R derived)")