Commit 58089657 authored by Jarrod Pas's avatar Jarrod Pas
Browse files

Updates ShedTrace to inherit from Trace class

parent 74d01f92
...@@ -3,7 +3,7 @@ ...@@ -3,7 +3,7 @@
__all__ = [ __all__ = [
'write_meta_file', 'write_meta_file',
'read_meta_file', 'read_meta_file',
'shed_trace', 'ShedTrace',
] ]
__author__ = 'Jarrod Pas <j.pas@usask.ca>' __author__ = 'Jarrod Pas <j.pas@usask.ca>'
...@@ -15,7 +15,7 @@ from collections import defaultdict ...@@ -15,7 +15,7 @@ from collections import defaultdict
from itertools import groupby, count from itertools import groupby, count
from os import path from os import path
from pydtn import Contact from pydtn import Trace
def write_meta_file(meta_path, csv_path, duty_cycle_length=300): def write_meta_file(meta_path, csv_path, duty_cycle_length=300):
...@@ -56,53 +56,88 @@ def read_meta_file(meta_path): ...@@ -56,53 +56,88 @@ def read_meta_file(meta_path):
raise RuntimeError('Should not get here...') raise RuntimeError('Should not get here...')
def _get_contact_pairs(csv_path): class ShedTrace(Trace):
pairs = defaultdict(set) """Generator for contact traces from duty cycle based SHED datasets."""
with open(csv_path) as csv_file:
csv_file = csv.reader(csv_file) def __init__(self, meta_path):
next(csv_file) """
for row in csv_file: Create SHED trace generator.
_, source, _, target, _, slot = row
pair = min(source, target), max(source, target) Arguments:
slot = int(slot) meta_path -- path to a metadata file created with `write_meta_file`
pairs[pair].add(slot) """
return dict(pairs) self.path = meta_path
self.meta = read_meta_file(meta_path)
def shed_trace(meta_path): super().__init__(self.meta['nodes'])
"""
Generate contact trace for a duty cycle based SHED dataset. self._pairs = None
self._contacts = None
Keyword Arguments:
duty_cycle_length -- duration of each duty cycle (default 300) @property
""" def contact_pairs(self):
meta = read_meta_file(meta_path) """Return all pairs and their duty cycles that they are in contact."""
pairs = _get_contact_pairs(meta['data']) if self._pairs is not None:
node = count() return self._pairs
nodes = {}
contacts = [] pairs = defaultdict(set)
for (source, target), slots in pairs.items(): with open(self.meta['data']) as csv_file:
# get canonical node id for source csv_file = csv.reader(csv_file)
if source not in nodes: next(csv_file)
nodes[source] = next(node) for row in csv_file:
source = nodes[source] _, source, _, target, _, slot = row
pair = min(source, target), max(source, target)
# get canonical node id for source slot = int(slot)
if target not in nodes: pairs[pair].add(slot)
nodes[target] = next(node)
target = nodes[target] self._pairs = dict(pairs)
return self._pairs
slots = sorted(slots)
@property
# groups consecutive slots def contacts(self):
# if the lambda is mapped it will return: """
# [1, 2, 3, 6, 7, 9] -> [-1, -1, -1, -3, -3, -4] Calculate contacts in the SHED dataset.
for _, group in groupby(enumerate(slots), lambda p: p[0]-p[1]):
times = list(map(lambda g: g[1], group)) Computes contacts by finding consecutive duty cycles that any give pair
start = times[0] * meta['duty_cycle_length'] of nodes saw each other. The end of each contact is at the start of
end = (times[-1] + 1) * meta['duty_cycle_length'] the duty cycle that they did not see each other.
"""
contacts.append(Contact(start, source, target, True)) if self._contacts is not None:
contacts.append(Contact(end, source, target, False)) return self._contacts
yield from sorted(contacts) node = count()
nodes = {}
contacts = []
for (source, target), slots in self.contact_pairs.items():
# get canonical node id for source
if source not in nodes:
nodes[source] = next(node)
source = nodes[source]
# get canonical node id for source
if target not in nodes:
nodes[target] = next(node)
target = nodes[target]
slots = sorted(slots)
# groups consecutive slots
# if the lambda is mapped it will return:
# [1, 2, 3, 6, 7, 9] -> [-1, -1, -1, -3, -3, -4]
for _, group in groupby(enumerate(slots), lambda p: p[0]-p[1]):
times = list(map(lambda g: g[1], group))
start = times[0] * self.meta['duty_cycle_length']
end = (times[-1] + 1) * self.meta['duty_cycle_length']
contacts.extend([
self.create_contact(start, source, target, True),
self.create_contact(end, source, target, False),
])
contacts.sort()
self._contacts = contacts
return self._contacts
def __iter__(self):
"""Yield contacts in SHED dataset."""
yield from self.contacts
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment