Merge branch 'feature/shed-data-reformatting' into 'develop'

Feature/shed data reformatting

See merge request !13

examples/shed.py

@@ -6,11 +6,12 @@ import sys
 from argparse import ArgumentParser
 from collections import namedtuple
 from multiprocessing import Pool
+from os import path
+from pprint import pprint
 
-from pydtn import Network, RandomTraffic, Node, EpidemicNode
+from pydtn import Network, RandomTraffic, Node, EpidemicNode, CSVTrace
 from pydtn.community import BubbleKCliqueNode, BubbleLouvainNode
 from pydtn.community import HCBFKCliqueNode, HCBFLouvainNode
-from pydtn.shed import ShedTrace
 
 
 Simulation = namedtuple('Simulation', ['trace', 'node_type', 'seed'])
@@ -20,7 +21,9 @@ def run_simulation(simulation):
     """Run a simulation."""
     seed = simulation.seed
 
-    trace = ShedTrace(simulation.trace)
+    csv = path.join(simulation.trace, 'contact.csv')
+    metadata = path.join(simulation.trace, 'metadata.json')
+    trace = CSVTrace(csv, metadata=metadata)
 
     epoch = 7*24*60*60  # 7 days
 
@@ -39,7 +42,7 @@ def run_simulation(simulation):
     traffic_options = {
         'seed': seed,
         'start': epoch,
-        'step': 1 * 60,  # 1 packet every 1 mins
+        'step': 60 * 60,  # 1 packet every 1 mins
     }
     traffic = RandomTraffic(nodes, **traffic_options)
 
@@ -47,7 +50,7 @@ def run_simulation(simulation):
     network.run()
 
     stats = {
-        'trace': trace.path,
+        'trace': simulation.trace,
         'node_type': node_type.__name__,
         'seed': seed,
     }
@@ -59,9 +62,11 @@ def run_simulation(simulation):
 
 def main(args):
     """Run simulation for each seed in args."""
-    trace = args['shed']
+    log = pprint if args['pretty'] else print
     pool = Pool()
     simulations = []
+
+    trace = args['shed']
     node_types = [
         Node,
         EpidemicNode,
@@ -77,7 +82,7 @@ def main(args):
             simulations.append(sim)
 
     for stats in pool.imap_unordered(run_simulation, simulations):
-        print(stats)
+        log(stats)
 
 
 def parse_args(args):
@@ -85,6 +90,7 @@ def parse_args(args):
     parser = ArgumentParser()
 
     parser.add_argument('shed')
+    parser.add_argument('--pretty', action='store_true')
     parser.add_argument('--seeds', '-s',
                         metavar='SEED', type=int, nargs='+', default=[None])
 

pydtn/__init__.py

@@ -35,6 +35,7 @@ __author__ = 'Jarrod Pas <j.pas@usask.ca>'
 
 from collections import ChainMap, defaultdict, namedtuple, OrderedDict
 import csv
+import json
 from itertools import count
 from random import Random
 from time import time
@@ -176,6 +177,10 @@ class Network:
         stats['broadcasts'] = sum(nodes['broadcasts'])
         stats['delivery-cost'] = stats['broadcasts'] / stats['recieved']
 
+        for stat, values in packets.items():
+            if stat.startswith('hops'):
+                stats[stat] = sum(values)
+
         stats.update(self._stats)
         return stats
 
@@ -546,6 +551,8 @@ class Trace:
 
     def __init__(self, nodes):
         """Create a contact trace."""
+        if nodes < 2:
+            raise ValueError('A trace requires at least 2 nodes')
         self._nodes = nodes
 
     @property
@@ -569,36 +576,41 @@ class CSVTrace(Trace):
 
     The first row is used as headers.
 
-    The second row is the following information about the trace:
-    [ int(duration), int(node_count), ignored, ignored ]
-
     The csv uses the following columns:
     [ int(time), int(node_a), int(node_b), int(join) ]
     time -- simulation time that the contact even happens
     node_{a,b} -- nodes involved in contact
     join -- whether the contact is going up (1) or going down (0)
+
+    Optionally accepts a metadata file so it does not have to read the trace
+    twice.
     """
 
-    def __init__(self, path):
+    def __init__(self, path, nodes=0, metadata=None):
         """Create a csv trace generator."""
-        with open(path) as csv_file:
-            csv_file = csv.reader(csv_file)
-            # skip header
-            next(csv_file)
-            # get node count from first row of csv. it is additional stats
-            # about the trace
-            # duration, placeholder, placehold, node_count
-            nodes = int(next(csv_file)[1])
+        self.path = path
+        self._contacts = None
+
+        if metadata is not None:
+            with open(metadata) as metadata_file:
+                nodes = json.load(metadata_file)['nodes']
+        else:
+            with open(self.path) as csv_file:
+                csv_file = csv.reader(csv_file)
+                next(csv_file)
+                nodes = set()
+                for _, node_a, node_b, _ in csv_file:
+                    nodes.add(node_a)
+                    nodes.add(node_b)
+                nodes = len(nodes)
 
         super().__init__(nodes)
-        self.path = path
 
     def __iter__(self):
         """Yield contacts from csv file."""
         with open(self.path) as csv_file:
             csv_file = csv.reader(csv_file)
-            # skip header and first row
-            next(csv_file)
+            # skip header
             next(csv_file)
             for row in csv_file:
                 yield self.create_contact(*map(int, row))

util/shed_process

+#!/usr/bin/env python
+
+import csv
+import json
+import sys
+from argparse import ArgumentParser
+from collections import defaultdict, namedtuple
+from itertools import groupby
+from os import path
+
+import networkx as nx
+
+def main(args):
+    seers = set()
+    observations = set()
+    with open(args['csv_path']) as csv_file:
+        csv_file = csv.DictReader(csv_file)
+        for row in csv_file:
+            time = int(row[args['time_column']])
+            seer = row[args['seer_column']]
+            seen = row[args['seen_column']]
+            if time >= 0:
+                seers.add(seer)
+                observations.add((time, seer, seen))
+
+    graph = nx.Graph()
+    for time, seer, seen in observations:
+        if seen not in seers:
+            continue
+
+        if not graph.has_edge(seer, seen):
+            graph.add_edge(seer, seen, {'times': set()})
+        graph[seer][seen]['times'].add(time)
+
+    nodes = max(nx.connected_components(graph), key=len)
+    nodes = {node: index for index, node in enumerate(nodes)}
+
+    contacts = []
+
+    for node_a, node_b, times in graph.edges(nbunch=nodes, data='times'):
+        times = sorted(times)
+        node_a, node_b = nodes[node_a], nodes[node_b]
+
+        for _, group in groupby(enumerate(times), lambda p: p[0]-p[1]):
+            contact = list(map(lambda g: g[1], group))
+            contacts.append((contact[0], node_a, node_b, True))
+            contacts.append((contact[-1] + 1, node_a, node_b, False))
+
+    contacts.sort(key=lambda c: c[0])
+
+    start = contacts[0][0]
+    duration = contacts[-1][0] - start
+
+    metadata = {
+        'nodes': len(nodes),
+        'contacts': len(contacts)/2,  # account for both up and down
+        'duration': duration,
+    }
+
+    meta_path = path.join(args['outdir'], 'metadata.json')
+    with open(meta_path, 'w') as meta_file:
+        json.dump(metadata, meta_file)
+        print('wrote: %s' % meta_path)
+
+    contact_path = path.join(args['outdir'], 'contact.csv')
+    with open(contact_path, 'w') as contact_file:
+        writer = csv.writer(contact_file)
+        writer.writerow(['time','a','b','join'])
+        for contact in contacts:
+            contact = list(map(int, contact))
+            contact[0] = (contact[0] - start) * args['duty_cycle_length']
+            print(contact)
+            writer.writerow(contact)
+        print('wrote: %s' % contact_path)
+
+
+def parse_args(args):
+    parser = ArgumentParser()
+
+    parser.add_argument('csv_path',
+                        help='path to csv to process')
+
+    parser.add_argument('outdir',
+                        help='directory to output metadata and contacts')
+
+    parser.add_argument('time_column',
+                        help='column used for time slots')
+
+    parser.add_argument('seer_column',
+                        help='column used for devices that scan for others.')
+
+    parser.add_argument('seen_column',
+                        help='column used for devices are seen during a scan.')
+
+    parser.add_argument('--duty_cycle_length', default=300, type=int)
+
+    return vars(parser.parse_args(args))
+
+if __name__ == '__main__':
+    exit(main(parse_args(sys.argv[1:])))