BlobRunner.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "from SIRStochasticDynamicsDisconnect import *\n",
    "from SIRStochasticDynamicsRewire import *\n",
    "from SIRStochasticDynamicsRewireDegree import *\n",
    "from SIRStochasticDynamicsRewireNeighbour import *\n",
    "\n",
    "from IPython.parallel import Client\n",
    "import time\n",
    "import csv"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "def make_simulation(N, M, desc, model_type, repetitions = 1, offset = 0 ):\n",
    "    '''Return a function to populate and run the simulation on a graph\n",
    "    with dynamics. If there is more than one repetition, return the\n",
    "    average of the outbreak parameters.\n",
    "    \n",
    "    N: the network size\n",
    "    p: model function for degree distribution\n",
    "    desc: description of distrbution, used for filename generation\n",
    "    repetitions: (optional) number of repetitions (defaults to 1)\n",
    "    offset: (optional) start repetition (defaults to 0)'''\n",
    "    \n",
    "    def run_simulation( g ):\n",
    "        \n",
    "        # Write the results to a csv file\n",
    "        filename = model_type + '_results.csv'\n",
    "\n",
    "        #with open(filename, 'wb') as csvfile:\n",
    "        #        writer = csv.writer(csvfile, delimiter=',',quotechar='|', quoting=csv.QUOTE_MINIMAL)\n",
    "        #        writer.writerow(['p_infect'] + ['p_rewire'] + ['p_recover'] + ['repetitions'] + \n",
    "        #                        ['start_time'] + ['end_time'] + ['avg_time_data'] + ['avg_node_data'])\n",
    "        \n",
    "        start = time.clock()\n",
    "        time_results = []\n",
    "        node_results = []\n",
    "        r_infinities = []\n",
    "        for rep in xrange(repetitions):\n",
    "            start_rep = time.clock()\n",
    "            # build the network topology using the given degree distribution\n",
    "            g.reset()\n",
    "            \n",
    "            g.rebuild_barabasi_albert(N,M)\n",
    "            \n",
    "            # run the simulation dynamics\n",
    "            steps = g.dynamics()\n",
    "\n",
    "            # compute the (partial) results\n",
    "            time_results.append(steps['peak_infection'][0])\n",
    "            node_results.append(steps['peak_infection'][1])\n",
    "            r_infinities.append(steps['r_infinity'])\n",
    "              \n",
    "        end = time.clock()\n",
    "        \n",
    "        with open(filename, 'ab') as csvfile:\n",
    "            spamwriter = csv.writer(csvfile, delimiter=',',quotechar='|', quoting=csv.QUOTE_MINIMAL)\n",
    "            spamwriter.writerow([N] + [g.p_infect] + [g.p_rewire] + [g.p_recover] + [repetitions] + \n",
    "                                [start] + [end] + [numpy.mean(time_results)] + [numpy.mean(node_results)] +\n",
    "                                [numpy.mean(r_infinities)])\n",
    "        \n",
    "        # construct metadata to wrap-up repetition results\n",
    "        r = dict()\n",
    "        r['nodes'] = N\n",
    "        r['p_infect'] = g.p_infect\n",
    "        r['p_rewire'] = g.p_rewire\n",
    "        r['p_recover'] = g.p_recover\n",
    "        r['repetitions'] = repetitions\n",
    "        r['start_time'] = start\n",
    "        r['end_time'] = end\n",
    "        r['avg_time_data'] = numpy.mean(time_results)\n",
    "        r['avg_node_data'] = numpy.mean(node_results)\n",
    "        r['r_infinity'] = numpy.mean(r_infinities)\n",
    "        \n",
    "        return r\n",
    "    \n",
    "    return run_simulation"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "def blob_runner( reps = 1, timelim = 10000, numnodes = 5000, seed = 3, startinfected = 0.01, pinf_min = 0.00, pinf_max = 0.02, \n",
    "                pinf_num = 10, prew_min = 0.00, prew_max = 0.00, prew_num = 10, prec_min = 0.00, prec_max = 0.00, \n",
    "                prec_num = 10, set_alpha = 2, model_type = 'BASE'):\n",
    "    \n",
    "    # IPython profile for our remote cluster\n",
    "    cluster_name = \"blob\"\n",
    "\n",
    "    # connect to cluster\n",
    "    cluster = Client(profile = cluster_name)\n",
    "    #cluster = Client()\n",
    "    print(\"Cluster has {n} engines available\".format(n = len(cluster[:])))\n",
    "\n",
    "    # Add the location of the files on the blob server to python path\n",
    "    #def parallel(x):\n",
    "    #    import sys\n",
    "    #    sys.path.append('//home//mjp22//08_Blob')\n",
    "    #    return sys.path\n",
    "        \n",
    "    #k = cluster[:].map_sync(parallel,range(1))\n",
    "    \n",
    "    #print k\n",
    "    \n",
    "    d = cluster[:]\n",
    "    \n",
    "    # set up imports on cluster machines\n",
    "    with d.sync_imports():\n",
    "        import time\n",
    "        import math\n",
    "        import numpy\n",
    "        import mpmath\n",
    "        import networkx\n",
    "        import dill\n",
    "        import collections\n",
    "        import operator\n",
    "        import os.path\n",
    "        import csv\n",
    "    \n",
    "    # use Dill as pickler\n",
    "    d.use_dill()\n",
    "\n",
    "    # load-balance work across the available compute nodes\n",
    "    view = cluster.load_balanced_view()\n",
    "    \n",
    "    def parallel(x):\n",
    "    \n",
    "        filename = model_type + '_results.csv'\n",
    "        with open(filename, 'wb') as csvfile:\n",
    "            writer = csv.writer(csvfile, delimiter=',',quotechar='|', quoting=csv.QUOTE_MINIMAL)\n",
    "            writer.writerow(['nodes'] + ['p_infect'] + ['p_rewire'] + ['p_recover'] + ['repetitions'] + \n",
    "                            ['start_time'] + ['end_time'] + ['avg_time_data'] + ['avg_node_data'] + ['r_infinity'])\n",
    "        \n",
    "        return 'done'\n",
    "    \n",
    "    print 'CSV file written: ', view.map_sync(parallel, range(1))\n",
    "    \n",
    "    # set up simulation parameters\n",
    "    repetitions = reps\n",
    "    time_limit = timelim\n",
    "\n",
    "    # network parameters\n",
    "    N = numnodes\n",
    "    M = seed\n",
    "    alpha = set_alpha\n",
    "    p_infected = startinfected\n",
    "    \n",
    "    # Parameter spaces\n",
    "    p_infects = numpy.linspace(pinf_min, pinf_max, endpoint = True, num = pinf_num)\n",
    "    p_rewires = numpy.linspace(prew_min, prew_max, endpoint = True, num = prew_num)\n",
    "    p_recovers = numpy.linspace(prec_min, prec_max, endpoint = True, num = prec_num)\n",
    "    \n",
    "    print 'p_infects: ', p_infects\n",
    "    print 'p_rewires: ', p_rewires\n",
    "    print 'p_recovers: ', p_recovers\n",
    "\n",
    "    simulations = []\n",
    "    \n",
    "    for pi in p_infects :\n",
    "        for prew in p_rewires:\n",
    "            for prec in p_recovers:\n",
    "                \n",
    "                if(model_type == 'REWIRE'):\n",
    "                    simulations.append(SIRStochasticDynamicsRewire(time_limit = time_limit, p_infected = p_infected, \n",
    "                                         p_infect = pi, p_recover = prec, p_rewire = prew))\n",
    "                elif(model_type == 'DISCONNECT'):\n",
    "                    simulations.append(SIRStochasticDynamicsDisconnect(time_limit = time_limit, p_infected = p_infected, \n",
    "                                         p_infect = pi, p_recover = prec, p_rewire = prew))\n",
    "                elif(model_type == 'NEIGHBOUR'):\n",
    "                    simulations.append(SIRStochasticDynamicsRewireNeighbour(time_limit = time_limit, p_infected = p_infected, \n",
    "                                         p_infect = pi, p_recover = prec, p_rewire = prew))\n",
    "                elif(model_type == 'DEGREE'):\n",
    "                    simulations.append(SIRStochasticDynamicsRewireDegree(time_limit = time_limit, p_infected = p_infected, \n",
    "                                         p_infect = pi, p_recover = prec, p_rewire = prew))\n",
    "                    \n",
    "    \n",
    "   \n",
    "    sim = make_simulation(N, M, desc = '', repetitions = repetitions, model_type = model_type)\n",
    "    \n",
    "    print 'Beginning simulations...'\n",
    "    \n",
    "    rc = view.map_async(sim, simulations)\n",
    "    \n",
    "    # wait for simulations to complete\n",
    "    #rc.wait()\n",
    "    \n",
    "    elapsed = 0\n",
    "    while True:\n",
    "        msgset = set(rc.msg_ids)\n",
    "        completed = len(msgset.difference(cluster.outstanding))\n",
    "        pending = len(msgset.intersection(cluster.outstanding))\n",
    "        print 'After %d secs: %d complete, %d outstanding' % (elapsed, completed, pending)\n",
    "        if rc.ready():\n",
    "            break\n",
    "        time.sleep(60)\n",
    "        elapsed += 60\n",
    "    \n",
    "    # Write the results to a csv file locally and on the server\n",
    "    filename = '.\\\\output\\\\' + model_type + '_results.csv'\n",
    "\n",
    "    with open(filename, 'wb') as csvfile:\n",
    "        awriter = csv.writer(csvfile, delimiter=',',quotechar='|', quoting=csv.QUOTE_MINIMAL)\n",
    "        awriter.writerow(rc.result[0].keys())\n",
    "        for i in xrange(0,len(rc.result)):\n",
    "            awriter.writerow(rc.result[i].values())\n",
    "    \n",
    "    print 'Simulations complete'\n",
    "    "
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 2",
   "language": "python",
   "name": "python2"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}