test_project.py 15 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407
  1. import pytest
  2. import networkx as nx
  3. from networkx.algorithms import bipartite
  4. from networkx.utils import edges_equal, nodes_equal
  5. class TestBipartiteProject:
  6. def test_path_projected_graph(self):
  7. G = nx.path_graph(4)
  8. P = bipartite.projected_graph(G, [1, 3])
  9. assert nodes_equal(list(P), [1, 3])
  10. assert edges_equal(list(P.edges()), [(1, 3)])
  11. P = bipartite.projected_graph(G, [0, 2])
  12. assert nodes_equal(list(P), [0, 2])
  13. assert edges_equal(list(P.edges()), [(0, 2)])
  14. G = nx.MultiGraph([(0, 1)])
  15. with pytest.raises(nx.NetworkXError, match="not defined for multigraphs"):
  16. bipartite.projected_graph(G, [0])
  17. def test_path_projected_properties_graph(self):
  18. G = nx.path_graph(4)
  19. G.add_node(1, name="one")
  20. G.add_node(2, name="two")
  21. P = bipartite.projected_graph(G, [1, 3])
  22. assert nodes_equal(list(P), [1, 3])
  23. assert edges_equal(list(P.edges()), [(1, 3)])
  24. assert P.nodes[1]["name"] == G.nodes[1]["name"]
  25. P = bipartite.projected_graph(G, [0, 2])
  26. assert nodes_equal(list(P), [0, 2])
  27. assert edges_equal(list(P.edges()), [(0, 2)])
  28. assert P.nodes[2]["name"] == G.nodes[2]["name"]
  29. def test_path_collaboration_projected_graph(self):
  30. G = nx.path_graph(4)
  31. P = bipartite.collaboration_weighted_projected_graph(G, [1, 3])
  32. assert nodes_equal(list(P), [1, 3])
  33. assert edges_equal(list(P.edges()), [(1, 3)])
  34. P[1][3]["weight"] = 1
  35. P = bipartite.collaboration_weighted_projected_graph(G, [0, 2])
  36. assert nodes_equal(list(P), [0, 2])
  37. assert edges_equal(list(P.edges()), [(0, 2)])
  38. P[0][2]["weight"] = 1
  39. def test_directed_path_collaboration_projected_graph(self):
  40. G = nx.DiGraph()
  41. nx.add_path(G, range(4))
  42. P = bipartite.collaboration_weighted_projected_graph(G, [1, 3])
  43. assert nodes_equal(list(P), [1, 3])
  44. assert edges_equal(list(P.edges()), [(1, 3)])
  45. P[1][3]["weight"] = 1
  46. P = bipartite.collaboration_weighted_projected_graph(G, [0, 2])
  47. assert nodes_equal(list(P), [0, 2])
  48. assert edges_equal(list(P.edges()), [(0, 2)])
  49. P[0][2]["weight"] = 1
  50. def test_path_weighted_projected_graph(self):
  51. G = nx.path_graph(4)
  52. with pytest.raises(nx.NetworkXAlgorithmError):
  53. bipartite.weighted_projected_graph(G, [1, 2, 3, 3])
  54. P = bipartite.weighted_projected_graph(G, [1, 3])
  55. assert nodes_equal(list(P), [1, 3])
  56. assert edges_equal(list(P.edges()), [(1, 3)])
  57. P[1][3]["weight"] = 1
  58. P = bipartite.weighted_projected_graph(G, [0, 2])
  59. assert nodes_equal(list(P), [0, 2])
  60. assert edges_equal(list(P.edges()), [(0, 2)])
  61. P[0][2]["weight"] = 1
  62. def test_digraph_weighted_projection(self):
  63. G = nx.DiGraph([(0, 1), (1, 2), (2, 3), (3, 4)])
  64. P = bipartite.overlap_weighted_projected_graph(G, [1, 3])
  65. assert nx.get_edge_attributes(P, "weight") == {(1, 3): 1.0}
  66. assert len(P) == 2
  67. def test_path_weighted_projected_directed_graph(self):
  68. G = nx.DiGraph()
  69. nx.add_path(G, range(4))
  70. P = bipartite.weighted_projected_graph(G, [1, 3])
  71. assert nodes_equal(list(P), [1, 3])
  72. assert edges_equal(list(P.edges()), [(1, 3)])
  73. P[1][3]["weight"] = 1
  74. P = bipartite.weighted_projected_graph(G, [0, 2])
  75. assert nodes_equal(list(P), [0, 2])
  76. assert edges_equal(list(P.edges()), [(0, 2)])
  77. P[0][2]["weight"] = 1
  78. def test_star_projected_graph(self):
  79. G = nx.star_graph(3)
  80. P = bipartite.projected_graph(G, [1, 2, 3])
  81. assert nodes_equal(list(P), [1, 2, 3])
  82. assert edges_equal(list(P.edges()), [(1, 2), (1, 3), (2, 3)])
  83. P = bipartite.weighted_projected_graph(G, [1, 2, 3])
  84. assert nodes_equal(list(P), [1, 2, 3])
  85. assert edges_equal(list(P.edges()), [(1, 2), (1, 3), (2, 3)])
  86. P = bipartite.projected_graph(G, [0])
  87. assert nodes_equal(list(P), [0])
  88. assert edges_equal(list(P.edges()), [])
  89. def test_project_multigraph(self):
  90. G = nx.Graph()
  91. G.add_edge("a", 1)
  92. G.add_edge("b", 1)
  93. G.add_edge("a", 2)
  94. G.add_edge("b", 2)
  95. P = bipartite.projected_graph(G, "ab")
  96. assert edges_equal(list(P.edges()), [("a", "b")])
  97. P = bipartite.weighted_projected_graph(G, "ab")
  98. assert edges_equal(list(P.edges()), [("a", "b")])
  99. P = bipartite.projected_graph(G, "ab", multigraph=True)
  100. assert edges_equal(list(P.edges()), [("a", "b"), ("a", "b")])
  101. def test_project_collaboration(self):
  102. G = nx.Graph()
  103. G.add_edge("a", 1)
  104. G.add_edge("b", 1)
  105. G.add_edge("b", 2)
  106. G.add_edge("c", 2)
  107. G.add_edge("c", 3)
  108. G.add_edge("c", 4)
  109. G.add_edge("b", 4)
  110. P = bipartite.collaboration_weighted_projected_graph(G, "abc")
  111. assert P["a"]["b"]["weight"] == 1
  112. assert P["b"]["c"]["weight"] == 2
  113. def test_directed_projection(self):
  114. G = nx.DiGraph()
  115. G.add_edge("A", 1)
  116. G.add_edge(1, "B")
  117. G.add_edge("A", 2)
  118. G.add_edge("B", 2)
  119. P = bipartite.projected_graph(G, "AB")
  120. assert edges_equal(list(P.edges()), [("A", "B")])
  121. P = bipartite.weighted_projected_graph(G, "AB")
  122. assert edges_equal(list(P.edges()), [("A", "B")])
  123. assert P["A"]["B"]["weight"] == 1
  124. P = bipartite.projected_graph(G, "AB", multigraph=True)
  125. assert edges_equal(list(P.edges()), [("A", "B")])
  126. G = nx.DiGraph()
  127. G.add_edge("A", 1)
  128. G.add_edge(1, "B")
  129. G.add_edge("A", 2)
  130. G.add_edge(2, "B")
  131. P = bipartite.projected_graph(G, "AB")
  132. assert edges_equal(list(P.edges()), [("A", "B")])
  133. P = bipartite.weighted_projected_graph(G, "AB")
  134. assert edges_equal(list(P.edges()), [("A", "B")])
  135. assert P["A"]["B"]["weight"] == 2
  136. P = bipartite.projected_graph(G, "AB", multigraph=True)
  137. assert edges_equal(list(P.edges()), [("A", "B"), ("A", "B")])
  138. class TestBipartiteWeightedProjection:
  139. @classmethod
  140. def setup_class(cls):
  141. # Tore Opsahl's example
  142. # http://toreopsahl.com/2009/05/01/projecting-two-mode-networks-onto-weighted-one-mode-networks/
  143. cls.G = nx.Graph()
  144. cls.G.add_edge("A", 1)
  145. cls.G.add_edge("A", 2)
  146. cls.G.add_edge("B", 1)
  147. cls.G.add_edge("B", 2)
  148. cls.G.add_edge("B", 3)
  149. cls.G.add_edge("B", 4)
  150. cls.G.add_edge("B", 5)
  151. cls.G.add_edge("C", 1)
  152. cls.G.add_edge("D", 3)
  153. cls.G.add_edge("E", 4)
  154. cls.G.add_edge("E", 5)
  155. cls.G.add_edge("E", 6)
  156. cls.G.add_edge("F", 6)
  157. # Graph based on figure 6 from Newman (2001)
  158. cls.N = nx.Graph()
  159. cls.N.add_edge("A", 1)
  160. cls.N.add_edge("A", 2)
  161. cls.N.add_edge("A", 3)
  162. cls.N.add_edge("B", 1)
  163. cls.N.add_edge("B", 2)
  164. cls.N.add_edge("B", 3)
  165. cls.N.add_edge("C", 1)
  166. cls.N.add_edge("D", 1)
  167. cls.N.add_edge("E", 3)
  168. def test_project_weighted_shared(self):
  169. edges = [
  170. ("A", "B", 2),
  171. ("A", "C", 1),
  172. ("B", "C", 1),
  173. ("B", "D", 1),
  174. ("B", "E", 2),
  175. ("E", "F", 1),
  176. ]
  177. Panswer = nx.Graph()
  178. Panswer.add_weighted_edges_from(edges)
  179. P = bipartite.weighted_projected_graph(self.G, "ABCDEF")
  180. assert edges_equal(list(P.edges()), Panswer.edges())
  181. for u, v in list(P.edges()):
  182. assert P[u][v]["weight"] == Panswer[u][v]["weight"]
  183. edges = [
  184. ("A", "B", 3),
  185. ("A", "E", 1),
  186. ("A", "C", 1),
  187. ("A", "D", 1),
  188. ("B", "E", 1),
  189. ("B", "C", 1),
  190. ("B", "D", 1),
  191. ("C", "D", 1),
  192. ]
  193. Panswer = nx.Graph()
  194. Panswer.add_weighted_edges_from(edges)
  195. P = bipartite.weighted_projected_graph(self.N, "ABCDE")
  196. assert edges_equal(list(P.edges()), Panswer.edges())
  197. for u, v in list(P.edges()):
  198. assert P[u][v]["weight"] == Panswer[u][v]["weight"]
  199. def test_project_weighted_newman(self):
  200. edges = [
  201. ("A", "B", 1.5),
  202. ("A", "C", 0.5),
  203. ("B", "C", 0.5),
  204. ("B", "D", 1),
  205. ("B", "E", 2),
  206. ("E", "F", 1),
  207. ]
  208. Panswer = nx.Graph()
  209. Panswer.add_weighted_edges_from(edges)
  210. P = bipartite.collaboration_weighted_projected_graph(self.G, "ABCDEF")
  211. assert edges_equal(list(P.edges()), Panswer.edges())
  212. for u, v in list(P.edges()):
  213. assert P[u][v]["weight"] == Panswer[u][v]["weight"]
  214. edges = [
  215. ("A", "B", 11 / 6.0),
  216. ("A", "E", 1 / 2.0),
  217. ("A", "C", 1 / 3.0),
  218. ("A", "D", 1 / 3.0),
  219. ("B", "E", 1 / 2.0),
  220. ("B", "C", 1 / 3.0),
  221. ("B", "D", 1 / 3.0),
  222. ("C", "D", 1 / 3.0),
  223. ]
  224. Panswer = nx.Graph()
  225. Panswer.add_weighted_edges_from(edges)
  226. P = bipartite.collaboration_weighted_projected_graph(self.N, "ABCDE")
  227. assert edges_equal(list(P.edges()), Panswer.edges())
  228. for u, v in list(P.edges()):
  229. assert P[u][v]["weight"] == Panswer[u][v]["weight"]
  230. def test_project_weighted_ratio(self):
  231. edges = [
  232. ("A", "B", 2 / 6.0),
  233. ("A", "C", 1 / 6.0),
  234. ("B", "C", 1 / 6.0),
  235. ("B", "D", 1 / 6.0),
  236. ("B", "E", 2 / 6.0),
  237. ("E", "F", 1 / 6.0),
  238. ]
  239. Panswer = nx.Graph()
  240. Panswer.add_weighted_edges_from(edges)
  241. P = bipartite.weighted_projected_graph(self.G, "ABCDEF", ratio=True)
  242. assert edges_equal(list(P.edges()), Panswer.edges())
  243. for u, v in list(P.edges()):
  244. assert P[u][v]["weight"] == Panswer[u][v]["weight"]
  245. edges = [
  246. ("A", "B", 3 / 3.0),
  247. ("A", "E", 1 / 3.0),
  248. ("A", "C", 1 / 3.0),
  249. ("A", "D", 1 / 3.0),
  250. ("B", "E", 1 / 3.0),
  251. ("B", "C", 1 / 3.0),
  252. ("B", "D", 1 / 3.0),
  253. ("C", "D", 1 / 3.0),
  254. ]
  255. Panswer = nx.Graph()
  256. Panswer.add_weighted_edges_from(edges)
  257. P = bipartite.weighted_projected_graph(self.N, "ABCDE", ratio=True)
  258. assert edges_equal(list(P.edges()), Panswer.edges())
  259. for u, v in list(P.edges()):
  260. assert P[u][v]["weight"] == Panswer[u][v]["weight"]
  261. def test_project_weighted_overlap(self):
  262. edges = [
  263. ("A", "B", 2 / 2.0),
  264. ("A", "C", 1 / 1.0),
  265. ("B", "C", 1 / 1.0),
  266. ("B", "D", 1 / 1.0),
  267. ("B", "E", 2 / 3.0),
  268. ("E", "F", 1 / 1.0),
  269. ]
  270. Panswer = nx.Graph()
  271. Panswer.add_weighted_edges_from(edges)
  272. P = bipartite.overlap_weighted_projected_graph(self.G, "ABCDEF", jaccard=False)
  273. assert edges_equal(list(P.edges()), Panswer.edges())
  274. for u, v in list(P.edges()):
  275. assert P[u][v]["weight"] == Panswer[u][v]["weight"]
  276. edges = [
  277. ("A", "B", 3 / 3.0),
  278. ("A", "E", 1 / 1.0),
  279. ("A", "C", 1 / 1.0),
  280. ("A", "D", 1 / 1.0),
  281. ("B", "E", 1 / 1.0),
  282. ("B", "C", 1 / 1.0),
  283. ("B", "D", 1 / 1.0),
  284. ("C", "D", 1 / 1.0),
  285. ]
  286. Panswer = nx.Graph()
  287. Panswer.add_weighted_edges_from(edges)
  288. P = bipartite.overlap_weighted_projected_graph(self.N, "ABCDE", jaccard=False)
  289. assert edges_equal(list(P.edges()), Panswer.edges())
  290. for u, v in list(P.edges()):
  291. assert P[u][v]["weight"] == Panswer[u][v]["weight"]
  292. def test_project_weighted_jaccard(self):
  293. edges = [
  294. ("A", "B", 2 / 5.0),
  295. ("A", "C", 1 / 2.0),
  296. ("B", "C", 1 / 5.0),
  297. ("B", "D", 1 / 5.0),
  298. ("B", "E", 2 / 6.0),
  299. ("E", "F", 1 / 3.0),
  300. ]
  301. Panswer = nx.Graph()
  302. Panswer.add_weighted_edges_from(edges)
  303. P = bipartite.overlap_weighted_projected_graph(self.G, "ABCDEF")
  304. assert edges_equal(list(P.edges()), Panswer.edges())
  305. for u, v in list(P.edges()):
  306. assert P[u][v]["weight"] == Panswer[u][v]["weight"]
  307. edges = [
  308. ("A", "B", 3 / 3.0),
  309. ("A", "E", 1 / 3.0),
  310. ("A", "C", 1 / 3.0),
  311. ("A", "D", 1 / 3.0),
  312. ("B", "E", 1 / 3.0),
  313. ("B", "C", 1 / 3.0),
  314. ("B", "D", 1 / 3.0),
  315. ("C", "D", 1 / 1.0),
  316. ]
  317. Panswer = nx.Graph()
  318. Panswer.add_weighted_edges_from(edges)
  319. P = bipartite.overlap_weighted_projected_graph(self.N, "ABCDE")
  320. assert edges_equal(list(P.edges()), Panswer.edges())
  321. for u, v in P.edges():
  322. assert P[u][v]["weight"] == Panswer[u][v]["weight"]
  323. def test_generic_weighted_projected_graph_simple(self):
  324. def shared(G, u, v):
  325. return len(set(G[u]) & set(G[v]))
  326. B = nx.path_graph(5)
  327. G = bipartite.generic_weighted_projected_graph(
  328. B, [0, 2, 4], weight_function=shared
  329. )
  330. assert nodes_equal(list(G), [0, 2, 4])
  331. assert edges_equal(
  332. list(G.edges(data=True)),
  333. [(0, 2, {"weight": 1}), (2, 4, {"weight": 1})],
  334. )
  335. G = bipartite.generic_weighted_projected_graph(B, [0, 2, 4])
  336. assert nodes_equal(list(G), [0, 2, 4])
  337. assert edges_equal(
  338. list(G.edges(data=True)),
  339. [(0, 2, {"weight": 1}), (2, 4, {"weight": 1})],
  340. )
  341. B = nx.DiGraph()
  342. nx.add_path(B, range(5))
  343. G = bipartite.generic_weighted_projected_graph(B, [0, 2, 4])
  344. assert nodes_equal(list(G), [0, 2, 4])
  345. assert edges_equal(
  346. list(G.edges(data=True)), [(0, 2, {"weight": 1}), (2, 4, {"weight": 1})]
  347. )
  348. def test_generic_weighted_projected_graph_custom(self):
  349. def jaccard(G, u, v):
  350. unbrs = set(G[u])
  351. vnbrs = set(G[v])
  352. return len(unbrs & vnbrs) / len(unbrs | vnbrs)
  353. def my_weight(G, u, v, weight="weight"):
  354. w = 0
  355. for nbr in set(G[u]) & set(G[v]):
  356. w += G.edges[u, nbr].get(weight, 1) + G.edges[v, nbr].get(weight, 1)
  357. return w
  358. B = nx.bipartite.complete_bipartite_graph(2, 2)
  359. for i, (u, v) in enumerate(B.edges()):
  360. B.edges[u, v]["weight"] = i + 1
  361. G = bipartite.generic_weighted_projected_graph(
  362. B, [0, 1], weight_function=jaccard
  363. )
  364. assert edges_equal(list(G.edges(data=True)), [(0, 1, {"weight": 1.0})])
  365. G = bipartite.generic_weighted_projected_graph(
  366. B, [0, 1], weight_function=my_weight
  367. )
  368. assert edges_equal(list(G.edges(data=True)), [(0, 1, {"weight": 10})])
  369. G = bipartite.generic_weighted_projected_graph(B, [0, 1])
  370. assert edges_equal(list(G.edges(data=True)), [(0, 1, {"weight": 2})])