1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
|
"""Run Leiden community detection (modularity or CPM)."""
import argparse
import sys
import os
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import igraph as ig
import leidenalg
from config import NETWORKS, RESULTS_DIR, SEED
from load_data import load_edge_list, build_igraph, save_communities
def run_leiden(network_name, method_name, quality, resolution=None):
net = NETWORKS[network_name]
edge_df = load_edge_list(net["edge_tsv"])
g, name_to_idx, idx_to_name = build_igraph(edge_df)
print(f" Graph: {g.vcount()} nodes, {g.ecount()} edges")
if quality == "modularity":
partition = leidenalg.find_partition(
g, leidenalg.ModularityVertexPartition, seed=SEED
)
elif quality == "cpm":
partition = leidenalg.find_partition(
g, leidenalg.CPMVertexPartition,
resolution_parameter=resolution, seed=SEED
)
else:
raise ValueError(f"Unknown quality function: {quality}")
print(f" Found {len(partition)} communities, modularity={partition.modularity:.4f}")
# Convert partition to node2com dict
node2com = {}
for comm_id, members in enumerate(partition):
for idx in members:
node2com[idx_to_name[idx]] = str(comm_id)
out_path = os.path.join(RESULTS_DIR, network_name, method_name, "com.tsv")
save_communities(node2com, out_path)
print(f" Saved to {out_path}")
return node2com
if __name__ == "__main__":
parser = argparse.ArgumentParser()
parser.add_argument("--network", required=True)
parser.add_argument("--method", required=True,
choices=["leiden_mod", "leiden_cpm_01", "leiden_cpm_001"])
args = parser.parse_args()
method_configs = {
"leiden_mod": {"quality": "modularity"},
"leiden_cpm_01": {"quality": "cpm", "resolution": 0.1},
"leiden_cpm_001": {"quality": "cpm", "resolution": 0.01},
}
cfg = method_configs[args.method]
print(f"Running {args.method} on {args.network}...")
run_leiden(args.network, args.method, **cfg)
|
