shorten some lines

Author: geraintpalmer

ciw/arrival_node.py

@@ -33,13 +33,15 @@ def __repr__(self):
 
     def decide_baulk(self, next_node, next_individual):
         """
-        Either makes an individual baulk, or sends the individual to the next node
+        Either makes an individual baulk, or sends the individual
+        to the next node
         """
         if next_node.baulking_functions[self.next_class] == None:
             self.send_individual(next_node, next_individual)
         else:
             rnd_num = random()
-            if rnd_num < next_node.baulking_functions[self.next_class](next_node.number_of_individuals):
+            if rnd_num < next_node.baulking_functions[self.next_class](
+                next_node.number_of_individuals):
                 self.record_baulk(next_node)
             else:
                 self.send_individual(next_node, next_individual)
@@ -65,7 +67,8 @@ def have_event(self):
         Send new arrival to relevent node.
         """
         self.number_of_individuals += 1
-        priority_class = self.simulation.network.priority_class_mapping[self.next_class]
+        priority_class = self.simulation.network.priority_class_mapping[
+            self.next_class]
         next_individual = Individual(self.number_of_individuals,
                                      self.next_class,
                                      priority_class)
@@ -99,7 +102,8 @@ def inter_arrival(self, nd, cls, current_time):
         """
         Samples the inter-arrival time for next class and node.
         """
-        if self.simulation.network.customer_classes[cls].arrival_distributions[nd-1][0] == "TimeDependent":
+        if self.simulation.network.customer_classes[
+            cls].arrival_distributions[nd-1][0] == "TimeDependent":
             return self.simulation.inter_arrival_times[nd][cls](current_time)
         return self.simulation.inter_arrival_times[nd][cls]()
 

ciw/import_params.py

@@ -6,7 +6,8 @@
 
 def create_network(params):
     """
-    Identifies the type of parameters that is input and calls the correct function
+    Identifies the type of parameters that is input and
+    calls the correct function
     """
     if isinstance(params, dict):
         return create_network_from_dictionary(params)

ciw/node.py

@@ -28,7 +28,8 @@ def __init__(self, id_, simulation):
             self.schedule = [list(pair) for pair in zip(boundaries, servers)]
             self.c = self.schedule[0][1]
             raw_schedule_boundaries = [row[0] for row in raw_schedule]
-            self.date_generator = self.date_from_schedule_generator(raw_schedule_boundaries)
+            self.date_generator = self.date_from_schedule_generator(
+                raw_schedule_boundaries)
             self.next_shift_change = next(self.date_generator)
         else:
             self.c = node.number_of_servers
@@ -130,8 +131,10 @@ def begin_interrupted_individuals_service(self, current_time, srvr):
         """
         ind = [i for i in self.interrupted_individuals][0]
         self.attach_server(srvr, ind)
-        ind.service_time = self.get_service_time(ind.customer_class, current_time)
-        ind.service_end_date = self.increment_time(self.get_now(current_time), ind.service_time)
+        ind.service_time = self.get_service_time(ind.customer_class,
+                                                 current_time)
+        ind.service_end_date = self.increment_time(self.get_now(current_time),
+                                                   ind.service_time)
         self.interrupted_individuals.remove(ind)
 
     def begin_service_if_possible_change_shift(self, current_time):
@@ -352,7 +355,8 @@ def get_service_time(self, cls, current_time):
         """
         Returns a service time for the given customer class
         """
-        if self.simulation.network.customer_classes[cls].service_distributions[self.id_number-1][0] == 'TimeDependent':
+        if self.simulation.network.customer_classes[cls].service_distributions[
+            self.id_number-1][0] == 'TimeDependent':
             return self.simulation.service_times[self.id_number][cls](current_time)
         return self.simulation.service_times[self.id_number][cls]()
 
@@ -431,11 +435,15 @@ def write_individual_record(self, individual):
         individual.destination = False
 
     def date_from_schedule_generator(self, boundaries):
-        """A generator that yields the next time according to a given schedule"""
+        """
+        A generator that yields the next time according to a given schedule.
+        """
         boundaries_len = len(boundaries)
         index = 0
         date = 0
         while True:
-            date = self.increment_time(boundaries[index % boundaries_len], (index) // boundaries_len * boundaries[-1])
+            date = self.increment_time(
+                boundaries[index % boundaries_len],
+                (index) // boundaries_len * boundaries[-1])
             index += 1
             yield date

ciw/simulation.py

@@ -22,7 +22,11 @@ class Simulation(object):
     """
     Overall simulation class
     """
-    def __init__(self, network, exact=False, name='Simulation', tracker=False, deadlock_detector=False,
+    def __init__(self, network,
+                 exact=False,
+                 name='Simulation',
+                 tracker=False,
+                 deadlock_detector=False,
         node_class=None, arrival_node_class=None):
         """
         Initialise a queue instance.
@@ -62,13 +66,13 @@ def check_userdef_dist(self, func):
         """
         sample = func()
         if not isinstance(sample, float) or sample < 0:
-            raise ValueError("UserDefined function must return positive float.")
+            raise ValueError("UserDefined func must return positive float.")
         return sample
 
     def check_timedependent_dist(self, func, current_time):
         sample = func(current_time)
         if not isinstance(sample, float) or sample < 0:
-            raise ValueError("TimeDependent function must return positive float.")
+            raise ValueError("TimeDependent func must return positive float.")
         return sample
 
 

ciw/tests/test_sampling.py

@@ -79,8 +79,10 @@ def test_sampling_uniform_dist_hypothesis(self, u, rm):
         Q = ciw.Simulation(ciw.create_network(params))
         Nu = Q.transitive_nodes[0]
         for itr in range(10): # Because repition happens in the simulation
-            self.assertTrue(ul <= Nu.simulation.service_times[Nu.id_number][0]() <= uh)
-            self.assertTrue(ul <= Nu.simulation.inter_arrival_times[Nu.id_number][0]() <= uh)
+            self.assertTrue(
+                ul <= Nu.simulation.service_times[Nu.id_number][0]() <= uh)
+            self.assertTrue(
+                ul <= Nu.simulation.inter_arrival_times[Nu.id_number][0]() <= uh)
 
     def test_error_uniform_dist(self):
         Arrival_distributions = [['Uniform', 2.2, 3.3]]
@@ -156,8 +158,10 @@ def test_sampling_deterministic_dist_hypothesis(self, d, rm):
         Q = ciw.Simulation(ciw.create_network(params))
         Nd = Q.transitive_nodes[0]
         for itr in range(10): # Because repition happens in the simulation
-            self.assertEqual(Nd.simulation.service_times[Nd.id_number][0](), d)
-            self.assertEqual(Nd.simulation.inter_arrival_times[Nd.id_number][0](), d)
+            self.assertEqual(
+                Nd.simulation.service_times[Nd.id_number][0](), d)
+            self.assertEqual(
+                Nd.simulation.inter_arrival_times[Nd.id_number][0](), d)
 
     def test_error_deterministic_dist(self):
         Arrival_distributions = [['Deterministic', 2.2]]
@@ -225,8 +229,10 @@ def test_sampling_triangular_dist_hypothesis(self, t, rm):
         Q = ciw.Simulation(ciw.create_network(params))
         Nt = Q.transitive_nodes[0]
         for itr in range(10): # Because repition happens in the simulation
-            self.assertTrue(tl <= Nt.simulation.service_times[Nt.id_number][0]() <= th)
-            self.assertTrue(tl <= Nt.simulation.inter_arrival_times[Nt.id_number][0]() <= th)
+            self.assertTrue(
+                tl <= Nt.simulation.service_times[Nt.id_number][0]() <= th)
+            self.assertTrue(
+                tl <= Nt.simulation.inter_arrival_times[Nt.id_number][0]() <= th)
 
     def test_error_triangular_dist(self):
         Arrival_distributions = [['Triangular', 2.2, 3.3, 2.8]]
@@ -620,10 +626,12 @@ def test_error_custom_dist(self):
 
     def test_userdefined_function_dist(self):
         params = {
-            'Arrival_distributions': [['UserDefined', lambda : random()],
-                                      ['UserDefined', lambda : custom_function()]],
-            'Service_distributions': [['UserDefined', lambda : random()],
-                                      ['UserDefined', lambda : custom_function()]],
+            'Arrival_distributions': [
+                ['UserDefined', lambda : random()],
+                ['UserDefined', lambda : custom_function()]],
+            'Service_distributions': [
+                ['UserDefined', lambda : random()],
+                ['UserDefined', lambda : custom_function()]],
             'Number_of_servers': [1, 1],
             'Transition_matrices': [[0.1, 0.1],
                                     [0.1, 0.1]]
@@ -681,10 +689,12 @@ def test_userdefined_function_dist(self):
     def test_userdefined_function_dist_hypothesis(self, const, dist, rm):
         my_empirical_dist = [8.0, 8.0, 8.0, 8.8, 8.8, 12.3]
         params = {
-            'Arrival_distributions': [['UserDefined', lambda : choice(my_empirical_dist)],
-                                      ['UserDefined', lambda : const]],
-            'Service_distributions': [['UserDefined', lambda : random()],
-                                      ['UserDefined', lambda : custom_function()]],
+            'Arrival_distributions': [
+                ['UserDefined', lambda : choice(my_empirical_dist)],
+                ['UserDefined', lambda : const]],
+            'Service_distributions': [
+                ['UserDefined', lambda : random()],
+                ['UserDefined', lambda : custom_function()]],
             'Number_of_servers': [1, 1],
             'Transition_matrices': [[0.1, 0.1],
                                     [0.1, 0.1]]
@@ -694,11 +704,15 @@ def test_userdefined_function_dist_hypothesis(self, const, dist, rm):
         N2 = Q.transitive_nodes[1]
         ciw.seed(5)
         for itr in range(10): # Because repition happens in the simulation
-            self.assertTrue(N1.simulation.inter_arrival_times[N1.id_number][0]()
+            self.assertTrue(
+                N1.simulation.inter_arrival_times[N1.id_number][0]()
                 in set(my_empirical_dist))
-            self.assertTrue(N2.simulation.inter_arrival_times[N2.id_number][0]() == const)
-            self.assertTrue(0.0 <= N1.simulation.service_times[N1.id_number][0]() <= 1.0)
-            self.assertTrue(0.0 <= N2.simulation.service_times[N2.id_number][0]() <= 2.0)
+            self.assertTrue(
+                N2.simulation.inter_arrival_times[N2.id_number][0]() == const)
+            self.assertTrue(
+                0.0 <= N1.simulation.service_times[N1.id_number][0]() <= 1.0)
+            self.assertTrue(
+                0.0 <= N2.simulation.service_times[N2.id_number][0]() <= 2.0)
 
     def test_no_arrivals_dist(self):
         params = {
@@ -708,9 +722,10 @@ def test_no_arrivals_dist(self):
             'Transition_matrices': [[0.1]]
         }
         Q = ciw.Simulation(ciw.create_network(params))
-        Nna = Q.transitive_nodes[0]
+        Na = Q.transitive_nodes[0]
         ciw.seed(5)
-        self.assertEqual(Nna.simulation.inter_arrival_times[Nna.id_number][0](), float('Inf'))
+        self.assertEqual(
+            Na.simulation.inter_arrival_times[Na.id_number][0](), float('Inf'))
 
     def test_error_dist(self):
         params = {'Arrival_distributions': ['NoArrivals'],
@@ -732,18 +747,24 @@ def test_error_dist(self):
            rm=random_module())
     def test_check_userdef_dist(self, positive_float, negative_float, word, rm):
         assume(negative_float < 0)
-        Q = ciw.Simulation(ciw.create_network('ciw/tests/testing_parameters/params.yml'))
-        self.assertEqual(Q.check_userdef_dist(lambda : positive_float), positive_float)
-        self.assertRaises(ValueError, Q.check_userdef_dist, lambda : negative_float)
-        self.assertRaises(ValueError, Q.check_userdef_dist, lambda : word)
+        Q = ciw.Simulation(
+            ciw.create_network('ciw/tests/testing_parameters/params.yml'))
+        self.assertEqual(
+            Q.check_userdef_dist(lambda : positive_float), positive_float)
+        self.assertRaises(
+            ValueError, Q.check_userdef_dist, lambda : negative_float)
+        self.assertRaises(
+            ValueError, Q.check_userdef_dist, lambda : word)
 
 
     def test_timedependent_function_dist(self):
         params = {
-            'Arrival_distributions': [['TimeDependent', lambda t : time_dependent_function_1(t)],
-                                      ['TimeDependent', lambda t : time_dependent_function_2(t)]],
-            'Service_distributions': [['TimeDependent', lambda t : time_dependent_function_1(t)],
-                                      ['TimeDependent', lambda t : time_dependent_function_2(t)]],
+            'Arrival_distributions': [
+                ['TimeDependent', lambda t : time_dependent_function_1(t)],
+                ['TimeDependent', lambda t : time_dependent_function_2(t)]],
+            'Service_distributions': [
+                ['TimeDependent', lambda t : time_dependent_function_1(t)],
+                ['TimeDependent', lambda t : time_dependent_function_2(t)]],
             'Number_of_servers': [1, 1],
             'Transition_matrices': [[0.1, 0.1],
                                     [0.1, 0.1]]
@@ -752,16 +773,26 @@ def test_timedependent_function_dist(self):
         N1 = Q.transitive_nodes[0]
         N2 = Q.transitive_nodes[1]
         ciw.seed(5)
-        self.assertEqual(N1.simulation.service_times[N1.id_number][0](3.0), 3.0)
-        self.assertEqual(N1.simulation.service_times[N1.id_number][0](9.0), 3.0)
-        self.assertEqual(N1.simulation.service_times[N1.id_number][0](9.0), 3.0)
-        self.assertEqual(N1.simulation.service_times[N1.id_number][0](11.0), 5.0)
-        self.assertEqual(N1.simulation.service_times[N1.id_number][0](11.0), 5.0)
-        self.assertEqual(N2.simulation.service_times[N2.id_number][0](4.0), 2.0)
-        self.assertEqual(N2.simulation.service_times[N2.id_number][0](4.0), 2.0)
-        self.assertEqual(N2.simulation.service_times[N2.id_number][0](17.0), 8.5)
-        self.assertEqual(N2.simulation.service_times[N2.id_number][0](22.0), 8.0)
-        self.assertEqual(N2.simulation.service_times[N2.id_number][0](22.0), 8.0)
+        self.assertEqual(
+            N1.simulation.service_times[N1.id_number][0](3.0), 3.0)
+        self.assertEqual(
+            N1.simulation.service_times[N1.id_number][0](9.0), 3.0)
+        self.assertEqual(
+            N1.simulation.service_times[N1.id_number][0](9.0), 3.0)
+        self.assertEqual(
+            N1.simulation.service_times[N1.id_number][0](11.0), 5.0)
+        self.assertEqual(
+            N1.simulation.service_times[N1.id_number][0](11.0), 5.0)
+        self.assertEqual(
+            N2.simulation.service_times[N2.id_number][0](4.0), 2.0)
+        self.assertEqual(
+            N2.simulation.service_times[N2.id_number][0](4.0), 2.0)
+        self.assertEqual(
+            N2.simulation.service_times[N2.id_number][0](17.0), 8.5)
+        self.assertEqual(
+            N2.simulation.service_times[N2.id_number][0](22.0), 8.0)
+        self.assertEqual(
+            N2.simulation.service_times[N2.id_number][0](22.0), 8.0)
 
         self.assertEqual(N1.get_service_time(0, 3.0), 3.0)
         self.assertEqual(N1.get_service_time(0, 9.0), 3.0)
@@ -774,21 +805,33 @@ def test_timedependent_function_dist(self):
         self.assertEqual(N2.get_service_time(0, 22.0), 8.0)
         self.assertEqual(N2.get_service_time(0, 22.0), 8.0)
 
-        self.assertEqual(N1.simulation.inter_arrival_times[N1.id_number][0](3.0), 3.0)
-        self.assertEqual(N1.simulation.inter_arrival_times[N1.id_number][0](9.0), 3.0)
-        self.assertEqual(N1.simulation.inter_arrival_times[N1.id_number][0](9.0), 3.0)
-        self.assertEqual(N1.simulation.inter_arrival_times[N1.id_number][0](11.0), 5.0)
-        self.assertEqual(N1.simulation.inter_arrival_times[N1.id_number][0](11.0), 5.0)
-        self.assertEqual(N2.simulation.inter_arrival_times[N2.id_number][0](4.0), 2.0)
-        self.assertEqual(N2.simulation.inter_arrival_times[N2.id_number][0](4.0), 2.0)
-        self.assertEqual(N2.simulation.inter_arrival_times[N2.id_number][0](17.0), 8.5)
-        self.assertEqual(N2.simulation.inter_arrival_times[N2.id_number][0](22.0), 8.0)
-        self.assertEqual(N2.simulation.inter_arrival_times[N2.id_number][0](22.0), 8.0)
+        self.assertEqual(
+            N1.simulation.inter_arrival_times[N1.id_number][0](3.0), 3.0)
+        self.assertEqual(
+            N1.simulation.inter_arrival_times[N1.id_number][0](9.0), 3.0)
+        self.assertEqual(
+            N1.simulation.inter_arrival_times[N1.id_number][0](9.0), 3.0)
+        self.assertEqual(
+            N1.simulation.inter_arrival_times[N1.id_number][0](11.0), 5.0)
+        self.assertEqual(
+            N1.simulation.inter_arrival_times[N1.id_number][0](11.0), 5.0)
+        self.assertEqual(
+            N2.simulation.inter_arrival_times[N2.id_number][0](4.0), 2.0)
+        self.assertEqual(
+            N2.simulation.inter_arrival_times[N2.id_number][0](4.0), 2.0)
+        self.assertEqual(
+            N2.simulation.inter_arrival_times[N2.id_number][0](17.0), 8.5)
+        self.assertEqual(
+            N2.simulation.inter_arrival_times[N2.id_number][0](22.0), 8.0)
+        self.assertEqual(
+            N2.simulation.inter_arrival_times[N2.id_number][0](22.0), 8.0)
 
     def test_broken_timedependent_function_dist(self):
         params = {
-            'Arrival_distributions': [['TimeDependent', lambda t : time_dependent_function_1(t)]],
-            'Service_distributions': [['TimeDependent', lambda t : broken_td_func(t)]],
+            'Arrival_distributions': [
+                ['TimeDependent', lambda t : time_dependent_function_1(t)]],
+            'Service_distributions': [
+                ['TimeDependent', lambda t : broken_td_func(t)]],
             'Number_of_servers': [1],
             'Transition_matrices': [[0.1]]
         }
@@ -800,10 +843,12 @@ def test_broken_timedependent_function_dist(self):
 
     def test_timedependent_exact(self):
         params = {
-            'Arrival_distributions': [['TimeDependent', lambda t : time_dependent_function_1(t)],
-                                      ['TimeDependent', lambda t : time_dependent_function_2(t)]],
-            'Service_distributions': [['TimeDependent', lambda t : time_dependent_function_1(t)],
-                                      ['TimeDependent', lambda t : time_dependent_function_2(t)]],
+            'Arrival_distributions': [
+                ['TimeDependent', lambda t : time_dependent_function_1(t)],
+                ['TimeDependent', lambda t : time_dependent_function_2(t)]],
+            'Service_distributions': [
+                ['TimeDependent', lambda t : time_dependent_function_1(t)],
+                ['TimeDependent', lambda t : time_dependent_function_2(t)]],
             'Number_of_servers': [1, 1],
             'Transition_matrices': [[0.1, 0.1],
                                     [0.1, 0.1]]