Merge branch 'dev' into feature/read_the_docs_final

simba/optimization.py

@@ -35,6 +35,7 @@ def service_optimization(schedule, scenario, args):
 
     negative_sets = {}
     for rot_key in negative_rotations:
+        logging.debug(f"Looking at negative rotation {rot_key}")
         # remove negative rotation from initial schedule
         rotation = schedule.rotations.pop(rot_key)
         if rotation.charging_type != "oppb":
@@ -48,19 +49,26 @@ def service_optimization(schedule, scenario, args):
         vid = rotation.vehicle_id
         last_neg_soc_time = scenario.negative_soc_tracker[vid][-1]
         last_neg_soc_time = datetime.datetime.fromisoformat(last_neg_soc_time)
-        dependent_station = {r: t for r, t in common_stations[rot_key].items()
-                             if t <= last_neg_soc_time}
-        while dependent_station:
-            r, t = dependent_station.popitem()
+        dependent_rotations = {r: t for r, t in common_stations[rot_key].items()
+                               if t <= last_neg_soc_time}
+        logging.debug(f"Dependent stations: {dependent_rotations}")
+        while dependent_rotations:
+            # get next rotation ID and last common time
+            r, t = dependent_rotations.popitem()
+            logging.debug(f"\tRotation {r} @ {t}:")
             if r not in negative_rotations:
+                # r not negative: add to set of dependent rotations
                 s.add(r)
-                # add dependencies of r
-                dependent_station.update({r2: t2 for r2, t2
-                                          in common_stations[r].items() if t2 <= t})
-            elif r.charging_type != "obbp":
+                # add dependencies of r, avoid circular dependency
+                dependencies = {r2: t2 for r2, t2 in common_stations[r].items()
+                                if t2 <= t and r2 not in s}
+                dependent_rotations.update(dependencies)
+                logging.debug(f"\t{dependencies}")
+            elif original.rotations[r].charging_type != "obbp":
                 logging.warning(f"Rotation {rot_key} depends on negative non-oppb rotation")
 
         negative_sets[rot_key] = s
+        logging.debug(f"Negative sets: {negative_sets}")
 
     # run scenario with non-negative rotations only
     if schedule.rotations:
@@ -96,11 +104,13 @@ def service_optimization(schedule, scenario, args):
         combined = negative_sets[r1].union(negative_sets[r2])
         schedule.rotations = {r: original[0].rotations[r] for r in combined}
         scenario = schedule.run(args)
+        logging.debug(f"{r1} + {r2}: {not scenario.negative_soc_tracker}")
         if not scenario.negative_soc_tracker:
             # compatible (don't interfere): keep union, remove r2
             negative_sets[r1] = combined
             # simplified. What about triangle? (r1+r2, r1+r3, r2!+r3)?
             possible = [t for t in possible if r2 not in t]
+            logging.debug(possible)
 
             # save scenario with highest electrification rate
             if optimal is None or len(scenario[0].rotations) > len(optimal[0].rotations):

simba/rotation.py

@@ -28,6 +28,7 @@ def add_trip(self, trip):
 
         :param trip: Information on trip to be added to rotation
         :type trip: dict
+        :raises Exception: if charging type of trip and rotation differ
         """
         new_trip = Trip(self, **trip)
 
@@ -57,16 +58,21 @@ def add_trip(self, trip):
 
         # set charging type if given
         charging_type = trip.get('charging_type')
+        self.trips.append(new_trip)
         if charging_type in ['depb', 'oppb']:
-            assert self.charging_type is None or self.charging_type == charging_type, (
-                f"Two trips of rotation {self.id} have distinct charging types")
             assert self.schedule.vehicle_types.get(
                 self.vehicle_type, {}).get(charging_type) is not None, (
                 f"The required vehicle type {self.vehicle_type}({charging_type}) "
                 "is not given in the vehicle_types.json file.")
-            self.set_charging_type(charging_type)
-
-        self.trips.append(new_trip)
+            if self.charging_type is None:
+                # set CT for whole rotation
+                self.set_charging_type(charging_type)
+            elif self.charging_type == charging_type:
+                # same CT as other trips: just add trip consumption
+                self.consumption += new_trip.calculate_consumption()
+            else:
+                # different CT than rotation: error
+                raise Exception(f"Two trips of rotation {self.id} have distinct charging types")
 
     def calculate_consumption(self):
         """ Calculate consumption of this rotation and all its trips.

simba/schedule.py

@@ -611,6 +611,8 @@ def generate_scenario(self, args):
                         "estimated_time_of_arrival": arrival_time.isoformat()
                     }
                 })
+                assert trip.delta_soc is not None, (
+                    "Trip delta_soc is None. Did you forget to calculate consumption?")
 
                 # create arrival event
                 events["vehicle_events"].append({

tests/test_schedule.py

@@ -178,7 +178,7 @@ def test_get_negative_rotations(self):
         sched, scen = self.basic_run()
 
         neg_rots = sched.get_negative_rotations(scen)
-        assert '1' in neg_rots
+        assert '2' in neg_rots
 
     def test_rotation_filter(self, tmp_path):
         s = schedule.Schedule(self.vehicle_types, self.electrified_stations, **mandatory_args)