from __future__ import division

import networkx as nx
import pandas as pd
import random
from mrjob.job import MRJob
from mrjob.protocol import JSONValueProtocol
from mrjob.step import MRStep

import cascade


class MRJobNetworkXSimulations(MRJob):
    OUTPUT_PROTOCOL = JSONValueProtocol
    INPUT_PROTOCOL = JSONValueProtocol

    def configure_options(self):
        super(MRJobNetworkXSimulations, self).configure_options()
        self.add_file_option('--network')
        self.add_passthrough_option('--modle', type='int', default=0, help='...')
        self.add_passthrough_option('--sampelFraction', type='int', default=10, help='...')
        self.add_passthrough_option('--resampeling', type='int', default=10, help='...')

    def runCascade(self, C):
        cas = C
        idx = []
        values = []
        while True:
            try:
                cas.next()
                values.append(cas.getInfectedNode())
                idx.append(cas.getStep())
            except StopIteration as err:
                break
            except Exception as e:
                print e
        return idx, values

    def csize(self, _, line):
        client = hdfs.client.Client("http://" + urlparse(line).netloc)

        if line[-1] != "#":
            with client.read(urlparse(line).path) as r:
                # with open(urlparse(line).path) as r:
                buf = BytesIO(r.read())

                # If the data is in a GZipped file.
                if ".gz" in line:
                    gzip_f = gzip.GzipFile(fileobj=buf)
                    content = gzip_f.read()
                    buf = StringIO.StringIO(content)

                dtf = pd.read_csv(buf, index_col=False, header=None, sep="\t", engine="python",
                                  compression=None).drop_duplicates(subset=[2], keep='last')
                yield "apple", len(dft.index)

    def mapper_init(self):

        self.G = nx.read_gpickle(self.options.network)
        nx.set_node_attributes(self.G, 'activated', {node: 0 for node in self.G.nodes()})
        seed = random.choice([n for n, attrdict in self.G.node.items() if attrdict['activated'] == 0])
        nx.set_node_attributes(self.G, 'activated', {seed: 1})

        self.r_u_l = None
        self.r_a_l = None

    def mapper(self, _, line):

        iteration = int(line) * 10
        sampelFraction = self.options.sampelFraction

        if self.options.modle == 0:
            idx, values = self.runCascade(cascade.randomActive(self.G, itterations=iteration))
        elif self.options.modle == 1:
            idx, values = self.runCascade(cascade.CascadeNabours(self.G, itterations=iteration))
        elif self.options.modle == 2:
            idx, values = self.runCascade(cascade.NodeWithHighestActiveNabours(self.G, itterations=iteration))
        elif self.options.modle == 3:
            idx, values = self.runCascade(cascade.NodeInSameCommunity(self.G, itterations=iteration))
        elif self.options.modle == 4:
            idx, values = self.runCascade(cascade.CascadeNaboursWeight(self.G, itterations=iteration))

        df = pd.DataFrame({"ids": values}, index=idx)

        for i in range(1, self.options.resampeling):
            yield "tmp", df.sample(frac=(float(sampelFraction) / float(100))).to_json()

    def steps(self):
        return [
            MRStep(
                mapper=self.csize,
                mapper_init=self.mapper_init,
                mapper=self.mapper
            )
        ]


if __name__ == '__main__':
    MRJobNetworkXSimulations.run()

@@ -725,16 +725,108 @@
 p='...')
+%0A        self.add_passthrough_option('--numberofloops', type='int', default=100, help='...')
 %0A%0A    de
@@ -2482,16 +2482,75 @@
 action%0A%0A
+        for x in range(0, self.options.numberofloops):%0A    
         
@@ -2577,16 +2577,20 @@
 e == 0:%0A
+    
         
@@ -2676,32 +2676,36 @@
 ation))%0A        
+    
 elif self.option
@@ -2722,32 +2722,36 @@
  1:%0A            
+ 
+   
 idx, values = se
@@ -2811,32 +2811,36 @@
 ons=iteration))%0A
+    
         elif sel
@@ -2865,32 +2865,36 @@
  2:%0A            
+    
 idx, values = se
@@ -2976,32 +2976,36 @@
 ation))%0A        
+  
+  
 elif self.option
@@ -3018,16 +3018,20 @@
 e == 3:%0A
+    
         
@@ -3124,32 +3124,36 @@
 ation))%0A        
+    
 elif self.option
@@ -3170,32 +3170,36 @@
  4:%0A            
+   
+ 
 idx, values = se
@@ -3282,16 +3282,20 @@
         
+    
 df = pd.
@@ -3337,24 +3337,28 @@
 x)%0A%0A        
+  
+  
 for i in ran
@@ -3390,16 +3390,20 @@
 eling):%0A
+    
         

Load from the pre processed data