adding the read_state functionality and recreating the demo

Author: MohamedHmini

.gitignore

@@ -1 +1,2 @@
 __pycache__/
+.vscode/

.ipynb_checkpoints/demo-checkpoint.ipynb

@@ -35,6 +35,16 @@ def get_vertices_schema(self):
         )
         return vertices_schema
 
+    def get_edges_schema(self):
+        edges_schema = T.StructType(
+            [
+                T.StructField(name = "src", dataType=T.IntegerType(), nullable = False),
+                T.StructField(name = "dst", dataType=T.IntegerType(), nullable = False),
+            ]
+        )
+        return edges_schema
+
+
     def toVertices(self, sdv):
         return sdv.rdd.toDF(['id', 'score'])
 
@@ -73,11 +83,12 @@ def interact(self):
         self.edges.show()
         self.vertices.show()
         edges = self.edges.rdd.map(lambda x: (x.src, x.dst)).collect()
-        vertices = self.vertices.rdd.map(lambda x: (x.id, x.score)).collect()
+        keys,values = tuple(zip(*self.vertices.rdd.map(lambda x: (x.id, x.score)).collect()))
+        print(keys, values, edges)
         G = nx.Graph()
-        G.add_nodes_from([key for key,val in vertices])
+        G.add_nodes_from(keys)
         G.add_edges_from(edges)
         plt.figure(figsize = (15,10))
-        nx.draw(G, pos = nx.spring_layout(G,scale=10), node_size = 800, cmap=plt.get_cmap('viridis'), node_color=[val for key,val in vertices], with_labels=True, font_color='white')
+        nx.draw(G, pos = nx.spring_layout(G,scale=10), node_size = 800, cmap=plt.get_cmap('brg'), node_color=values, with_labels=True, font_color='white')
         plt.show()
 

PandiNetwork.py

@@ -1,22 +1,25 @@
 
+from pyspark import StorageLevel
+
 import sys
 import os
 sys.path.insert(1, './utils')
 
 import PandiSimConfigInjection as config
+import SparkDependencyInjection as sdi
 
 
 
 
-class PandiSim(config.PandiSimConfigInjection):
+class PandiSim(sdi.SparkDependencyInjection, config.PandiSimConfigInjection):
 
-    def __init__(self, network, epi_model, scoring_model, edge_model, params = {'take_screenshots':False}):
+    def __init__(self, network, epi_model, scoring_model, edge_model, params = {'take_screenshots':False, 'destroy':False}):
         self.network = network
         self.epi_model = epi_model
         self.scoring_model = scoring_model
         self.edge_model = edge_model
         self.params = params
-        self.params['t_end'] = epi_model.params['t_end']
+        # self.params['t_end'] = epi_model.params['t_end']
 
     def move(self):
         sotw = self.epi_model.next_sotw()[1]
@@ -31,6 +34,8 @@ def run(self, perc = 0.1):
         stopAt = self._perc_to_steps(perc)
 
         for _ in range(stopAt):
+            if self.epi_model.step >= 2 and self.params['destroy']:
+                self.read_state()
             self.move()
             self.take_screenshot()
 
@@ -46,4 +51,15 @@ def take_screenshot(self):
                 .option('header', False).mode('overwrite').save(vertices_fil)
             self.network.edges\
                 .write.format("csv").option("delimiter", ',')\
-                .option('header', False).mode('overwrite').save(edges_fil)
+                .option('header', False).mode('overwrite').save(edges_fil)
+
+    def read_state(self):
+        hdfs = "hdfs://namenode:9000/"
+        edges_fil = os.path.join(hdfs, self.read_from, f"step_{self.epi_model.step}", "edges.csv")
+        vertices_fil = os.path.join(hdfs, self.read_from, f"step_{self.epi_model.step}", "vertices.csv")
+        self.network.vertices = self.spark.read.format("csv").option("delimiter", ',')\
+            .option('header', False).option('inferSchema', True).load(vertices_fil).toDF('id', 'score', 'health_status')\
+            .sort('id').cache()
+        self.network.edges = self.spark.read.format("csv").option("delimiter", ',')\
+            .option('header', False).option('inferSchema', True).load(edges_fil).toDF('src', 'dst')\
+            .cache()

PandiSim.py

@@ -25,8 +25,12 @@ def run(self):
         se.rdd = se.rdd.persist(StorageLevel.MEMORY_AND_DISK)
         D = sdm.SparseDistributedMatrix.diag(se) 
         M = A.dot(D)
-        C = A.dot(se).apply(lambda x: 1/x).outer(sdv.SparseDistributedVector.repeat(1, A.numRows()))
+        b = A.dot(se).apply(lambda x: 1/x)
+        print(b.rdd.collect())
+        C = b.outer(sdv.SparseDistributedVector.repeat(1, A.numRows()))
+        # print(C.entries.collect())
         P = M.multiply(C).transpose()
+        # print(P.entries.collect())
         P.entries = P.entries.persist(StorageLevel.MEMORY_AND_DISK)
 
         # running the walker:

demo.ipynb

@@ -44,7 +44,7 @@
 logger.LogManager.getLogger("akka").setLevel( logger.Level.ERROR )
 
 sdi.SparkDependencyInjection.set_spark(spark).set_spark_context(sc)
-pci.PandiSimConfigInjection.set_write_to("d_pandisim")
+pci.PandiSimConfigInjection.set_write_to("d_pandisim").set_read_from("d_pandisim")
 
 # a = SparseDistributedMatrix(sc, sc.parallelize([MatrixEntry(0, 0, 1),MatrixEntry(2, 0, 3),MatrixEntry(4, 0, 1)]), 4, 1).transpose()
 # o = SparseDistributedMatrix.ones(sc, 4).transpose()
@@ -74,7 +74,7 @@
 
 # print(u.dot(a).rdd.collect())
 # print(u.dot(a).rdd.collect())
-# print(a.dot(a.dot(a.dot(u))).rdd.collect())
+# print(a.dot(u).rdd.collect())
 # print(v.dot(u))
 # print(v.outer(u).entries.collect())
 # print(u.op(v).rdd.collect())
@@ -110,60 +110,76 @@
 # print(v.rdd.collect())
 # print(v.op(ns, 'add').rdd.collect())
 
-sir = ssir.Simple_SIR(
-    inits = {'S':0.9, 'I':0.1, 'R':0}, 
-    params = {'beta':0.35, 'gamma':0.07, 'N':6, 't_end':20, 'step_size':1}
-)
-sir.run()
-dr = sir.current_sotw()[1]
-
-init = Initializer101(
-    nbr_vertices = 6, 
-    nbr_edges = 2, 
-    nbr_infected = int(dr[0]), 
-    nbr_recovered = int(dr[1])
-)
+# sir = ssir.Simple_SIR(
+#     inits = {'S':0.9, 'I':0.1, 'R':0}, 
+#     params = {'beta':0.35, 'gamma':0.07, 'N':6, 't_end':20, 'step_size':1}
+# )
+# sir.run()
+# dr = sir.current_sotw()[1]
+
+# init = Initializer101(
+#     nbr_vertices = 6, 
+#     nbr_edges = 2, 
+#     nbr_infected = int(dr[0]), 
+#     nbr_recovered = int(dr[1])
+# )
 # init = Initializer101(
 #     nbr_vertices = 20, 
 #     nbr_edges = 4, 
 #     nbr_infected = 3, 
 #     nbr_recovered = 2
 # )
-init.initialize_vertices()
-init.initialize_edges(init.vertices)
+# init.initialize_vertices()
+# init.initialize_edges(init.vertices)
 
 # network = pn.PandiNetwork(init.vertices, init.edges, init.nbr_vertices)
-network = init.toPandiNetwork()
+# network = init.toPandiNetwork()
 
-walker = sw.ScoringWalker(
-    network, 
-    params = {'alpha-scaler':-2, 'walker-steps':3}
-)
+# walker = sw.ScoringWalker(
+#     network, 
+#     params = {'alpha-scaler':-2, 'walker-steps':3}
+# )
 
 # walker.run()
 # walker.annotate((2,1))
 # network.vertices.show()
 # network.edges.show()
 
-edge_est = see.StochasticEdgeEstimator(
-    network,
-    params = {'SDF': 100, 'alpha': 80, 'beta': 100}
-)
+# edge_est = see.StochasticEdgeEstimator(
+#     network,
+#     params = {'SDF': 100, 'alpha': 80, 'beta': 100}
+# )
 
 # edge_est.run()
 # network.vertices.show()
 # network.edges.show(50, False)
 
+# pandisim = ps.PandiSim(
+#     network = network, 
+#     epi_model = sir, 
+#     scoring_model = walker, 
+#     edge_model = edge_est, 
+#     params = {'take_screenshots':False}
+# )
+
+# pandisim.move()
+# pandisim.take_screenshot()
+
+# network.vertices.show()
+# network.edges.show()
+
+
+
+network = pn.PandiNetwork(None,None,6)
+
 pandisim = ps.PandiSim(
     network = network, 
-    epi_model = sir, 
-    scoring_model = walker, 
-    edge_model = edge_est, 
+    epi_model = None, 
+    scoring_model = None, 
+    edge_model = None, 
     params = {'take_screenshots':False}
 )
 
-pandisim.move()
-pandisim.take_screenshot()
-
+pandisim.read_state(1)
 network.vertices.show()
 network.edges.show()