final linting

pydtnsim/gui.py

@@ -9,7 +9,8 @@ import yaml
 
 from yaml.loader import Loader
 
-from pydtnsim import Node, EpidemicNode, ProphetNode, ProphetNodeSingle
+from pydtnsim import Node
+from pydtnsim.nodes import EpidemicNode, ProphetNode, ProphetNodeSingle
 from pydtnsim.community import BubbleKCliqueNode, BubbleLouvainNode
 from pydtnsim.community import HCBFKCliqueNode, HCBFLouvainNode
 from pydtnsim.community import HCBFAGKmeansNode, BubbleAGKmeansNode
@@ -396,11 +397,11 @@ class Gui(tk.Tk):
     def submit(self):
         """Save to config file."""
         param = {}
-        param["SimOptions"] = self.simOptions.get_options()
-        param["NodeOptions"] = self.nodeOptions.get_options()
-        param["NodeChoices"] = self.nodeChoices.get_choices()
-        param["TrafficOptions"] = self.trafficOptions.get_options()
-        param["GraphingOget_cptions"] = self.graphingOptions.get_options()
+        param["SimOptions"] = self.sim_options.get_options()
+        param["NodeOptions"] = self.node_options.get_options()
+        param["NodeChoices"] = self.node_choices.get_choices()
+        param["TrafficOptions"] = self.traffic_options.get_options()
+        param["GraphingOptions"] = self.graphing_options.get_options()
 
         self.param = param
 

pydtnsim/nodes.py

@@ -8,6 +8,12 @@ Implement non clustering routing algorithms here
 - PRoPHET single copy
 """
 
+__all__ = [
+    "EpidemicNode",
+    "FloodingNode",
+    "ProphetNode",
+    "ProphetNodeSingle",
+]
 from collections import defaultdict
 from pydtnsim import Node
 
@@ -78,7 +84,6 @@ class FloodingNode(Node):
 
         Do nothing, overrides default of removing from buffer.
         """
-        pass
 
 
 class ProphetNode(EpidemicNode):
@@ -121,6 +126,7 @@ class ProphetNode(EpidemicNode):
 
     def tick(self):
         """Apply aging constant every tick."""
+        # pylint:disable=duplicate-code
         tick = self.options["tick_rate"]
 
         while True:

pydtnsim/shed.py

-"""pydtnsim module for SHED dataset specific tools."""
-
-__all__ = [
-    "write_meta_file",
-    "read_meta_file",
-    "ShedTrace",
-]
-
-__author__ = "Jarrod Pas <j.pas@usask.ca>"
-
-
-import csv
-import json
-from collections import defaultdict
-from itertools import groupby, count
-from os import path
-
-from pydtnsim import Trace
-
-
-def write_meta_file(meta_path, csv_path, duty_cycle_length=300):
-    """Return metadata for a data set, from the dataset."""
-    nodes = set()
-    last = -1
-    with open(csv_path, "r", newline="", encoding="utf8") as csv_file:
-        csv_file = csv.reader(csv_file)
-        next(csv_file)
-        for row in csv_file:
-            _, source, _, target, _, slot = row
-            nodes.add(source)
-            nodes.add(target)
-            last = max(last, int(slot))
-
-    common = path.commonprefix(
-        [
-            path.dirname(meta_path),
-            path.dirname(csv_path),
-        ]
-    )
-    csv_path = path.relpath(csv_path, common)
-
-    meta_data = {
-        "data": csv_path,
-        "nodes": len(nodes),
-        "duration": last * duty_cycle_length,
-        "duty_cycle_length": duty_cycle_length,
-    }
-
-    with open(meta_path, "w", newline="", encoding="utf8") as meta_file:
-        json.dump(meta_data, meta_file, sort_keys=True, indent=2)
-        meta_file.write("\n")
-
-
-def read_meta_file(meta_path):
-    """Return metadata for a data set, from a metadata file."""
-    with open(meta_path, "r", newline="", encoding="utf8") as meta_file:
-        meta = json.load(meta_file)
-        meta["data"] = path.join(path.dirname(meta_path), meta["data"])
-        return meta
-
-    raise RuntimeError("Should not get here...")
-
-
-class ShedTrace(Trace):
-    """Generator for contact traces from duty cycle based SHED datasets."""
-
-    def __init__(self, meta_path):
-        """
-        Create SHED trace generator.
-
-        Arguments:
-        meta_path -- path to a metadata file created with `write_meta_file`
-        """
-        self.path = meta_path
-        self.meta = read_meta_file(meta_path)
-
-        super().__init__(self.meta["nodes"])
-
-        self._pairs = None
-        self._contacts = None
-
-    @property
-    def contact_pairs(self):
-        """Return all pairs and their duty cycles that they are in contact."""
-        if self._pairs is not None:
-            return self._pairs
-
-        pairs = defaultdict(set)
-        with open(
-            self.meta["data"], "r", newline="", encoding="utf8"
-        ) as csv_file:
-            csv_file = csv.reader(csv_file)
-            next(csv_file)
-            for row in csv_file:
-                _, source, _, target, _, slot = row
-                pair = min(source, target), max(source, target)
-                slot = int(slot)
-                pairs[pair].add(slot)
-
-        self._pairs = dict(pairs)
-        return self._pairs
-
-    @property
-    def contacts(self):
-        """
-        Calculate contacts in the SHED dataset.
-
-        Computes contacts by finding consecutive duty cycles that any give pair
-        of nodes saw each other. The end of each contact is at the start of
-        the duty cycle that they did not see each other.
-        """
-        if self._contacts is not None:
-            return self._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