Fix tests with new consumption calculation and shortened trip data

simba/rotation.py

@@ -79,46 +79,6 @@ def add_trip(self, trip):
                 # 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.
-
-        :return: Consumption of rotation [kWh]
-        :rtype: float
-        """
-        if len(self.trips) == 0:
-            self.consumption = 0
-            return self.consumption
-
-        # get the specific idle consumption of this vehicle type in kWh/h
-        v_info = self.schedule.vehicle_types[self.vehicle_type][self.charging_type]
-
-        rotation_consumption = 0
-
-        # make sure the trips are sorted, so the next trip can be determined
-        self.trips = list(sorted(self.trips, key=lambda trip: trip.arrival_time))
-
-        trip = self.trips[0]
-        for next_trip in self.trips[1:]:
-            # get consumption due to driving
-            driving_consumption, driving_delta_soc = trip.calculate_consumption()
-
-            # get idle consumption of the next break time
-            idle_consumption, idle_delta_soc = get_idle_consumption(trip, next_trip, v_info)
-
-            # set trip attributes
-            trip.consumption = driving_consumption + idle_consumption
-            trip.delta_soc = driving_delta_soc + idle_delta_soc
-
-            rotation_consumption += driving_consumption + idle_consumption
-            trip = next_trip
-
-        # last trip of the rotation has no idle consumption
-        trip.consumption, trip.delta_soc = trip.calculate_consumption()
-        rotation_consumption += trip.consumption
-
-        self.consumption = rotation_consumption
-        return rotation_consumption
-
     def set_charging_type(self, ct):
         """ Change charging type of either all or specified rotations.
 
@@ -186,26 +146,3 @@ def min_standing_time(self):
         return min_standing_time
 
 
-def get_idle_consumption(first_trip: Trip, second_trip: Trip, vehicle_info: dict) -> (float, float):
-    """ Compute consumption while waiting for the next trip
-
-    Calculate the idle consumption between the arrival of the first trip and the departure of the
-    second trip with a vehicle_info containing the keys idle_consumption and capacity.
-    :param first_trip: First trip
-    :type first_trip: Trip
-    :param second_trip: Second trip
-    :type second_trip: Trip
-    :param vehicle_info: Vehicle information
-    :type vehicle_info: dict
-    :return: Consumption of idling [kWh], delta_soc [-]
-    :rtype: float, float
-    """
-    capacity = vehicle_info["capacity"]
-    idle_cons_spec = vehicle_info.get("idle_consumption", 0)
-    if idle_cons_spec == 0:
-        return 0, 0
-
-    break_duration = second_trip.departure_time - first_trip.arrival_time
-    assert break_duration.total_seconds() >= 0
-    idle_consumption = break_duration.total_seconds() / 3600 * idle_cons_spec
-    return idle_consumption, -idle_consumption / capacity

simba/schedule.py

@@ -582,10 +582,43 @@ def calculate_consumption(self):
 
     def calculate_rotation_consumption(self, rotation: Rotation):
         rotation.consumption = 0
-        for trip in rotation.trips:
-            rotation.consumption += self.calculate_trip_consumption(trip)
+        """ Calculate consumption of this rotation and all its trips.
+
+        :return: Consumption of rotation [kWh]
+        :rtype: float
+        """
+        if len(rotation.trips) == 0:
+            rotation.consumption = 0
+            return rotation.consumption
+
+        # get the specific idle consumption of this vehicle type in kWh/h
+        v_info = self.vehicle_types[rotation.vehicle_type][rotation.charging_type]
+        rotation_consumption = 0
+        # make sure the trips are sorted, so the next trip can be determined
+        rotation.trips = list(sorted(rotation.trips, key=lambda trip: trip.arrival_time))
+        trip = rotation.trips[0]
+        for next_trip in rotation.trips[1:]:
+            # get consumption due to driving
+            driving_consumption  = self.calculate_trip_consumption(trip)
+            driving_delta_soc = trip.delta_soc
+            # get idle consumption of the next break time
+            idle_consumption, idle_delta_soc = get_idle_consumption(trip, next_trip, v_info)
+
+            # set trip attributes
+            trip.consumption = driving_consumption + idle_consumption
+            trip.delta_soc = driving_delta_soc + idle_delta_soc
+
+            rotation_consumption += driving_consumption + idle_consumption
+            trip = next_trip
+
+        # last trip of the rotation has no idle consumption
+        trip.consumption = self.calculate_trip_consumption(trip)
+        rotation_consumption += trip.consumption
+
+        rotation.consumption = rotation_consumption
         return rotation.consumption
 
+
     def calculate_trip_consumption(self, trip: Trip):
         """ Compute consumption for this trip.
 
@@ -1329,7 +1362,6 @@ def generate_event_list_from_prices(
             logging.info(f"{gc_name} price csv does not cover simulation time")
     return events
 
-
 def get_charge_delta_soc(charge_curves: dict, vt: str, ct: str, max_power: float,
                          duration_min: float, start_soc: float) -> float:
     """ Get the delta soc of a charge event for a given vehicle and charge type
@@ -1353,6 +1385,32 @@ def get_charge_delta_soc(charge_curves: dict, vt: str, ct: str, max_power: float
     return optimizer_util.get_delta_soc(charge_curve, start_soc, duration_min=duration_min)
 
 
+def get_idle_consumption(first_trip: Trip, second_trip: Trip, vehicle_info: dict) -> (float, float):
+    """ Compute consumption while waiting for the next trip
+
+    Calculate the idle consumption between the arrival of the first trip and the departure of the
+    second trip with a vehicle_info containing the keys idle_consumption and capacity.
+    :param first_trip: First trip
+    :type first_trip: Trip
+    :param second_trip: Second trip
+    :type second_trip: Trip
+    :param vehicle_info: Vehicle information
+    :type vehicle_info: dict
+    :return: Consumption of idling [kWh], delta_soc [-]
+    :rtype: float, float
+    """
+    capacity = vehicle_info["capacity"]
+    idle_cons_spec = vehicle_info.get("idle_consumption", 0)
+    if idle_cons_spec == 0:
+        return 0, 0
+
+    break_duration = second_trip.departure_time - first_trip.arrival_time
+    assert break_duration.total_seconds() >= 0
+    idle_consumption = break_duration.total_seconds() / 3600 * idle_cons_spec
+    return idle_consumption, -idle_consumption / capacity
+
+
+
 def soc_at_departure_time(v_id, deps, departure_time, vehicle_data, stations, charge_curves, args):
     """ Get the possible SoC of the vehicle at a specified departure_time
 

tests/test_consumption.py

@@ -1,7 +1,7 @@
 from tests.test_schedule import BasicSchedule
 from tests.conftest import example_root
-from datetime import datetime, timedelta
-from simba.rotation import get_idle_consumption
+from datetime import timedelta
+from simba.schedule import get_idle_consumption
 import pandas as pd
 
 
@@ -46,11 +46,30 @@ def test_calculate_idle_consumption(self, tmp_path):
         idle_consumption, idle_delta_soc = get_idle_consumption(first_trip, second_trip, v_info)
         assert idle_consumption == 0
 
+
+        # make all rotations depb
+        for r in schedule.rotations.values():
+            r.set_charging_type("depb")
+        # make all trips of all rotations consecutive
+        last_trip = schedule.rotations["1"].trips[0]
+
+        for r in schedule.rotations.values():
+            r.departure_time = last_trip.arrival_time + timedelta(minutes=10)
+            for t in r.trips:
+                t.departure_time = last_trip.arrival_time + timedelta(minutes=10)
+                t.arrival_time = t.departure_time + timedelta(minutes=1)
+                last_trip = t
+            r.arrival_time = t.arrival_time
+
         # Check that assignment of vehicles changes due to increased consumption. Only works
         # with adaptive_soc assignment
         for vt in schedule.vehicle_types.values():
             for ct in vt:
                 vt[ct]["idle_consumption"] = 0
+                vt[ct]["mileage"] = 0
+        schedule.calculate_consumption()
+        assert schedule.consumption == 0
+
         schedule.assign_vehicles_w_adaptive_soc(args)
         no_idle_consumption = schedule.vehicle_type_counts.copy()
 
@@ -59,7 +78,9 @@ def test_calculate_idle_consumption(self, tmp_path):
                 vt[ct]["idle_consumption"] = 9999
         schedule.calculate_consumption()
         schedule.assign_vehicles_w_adaptive_soc(args)
-        assert no_idle_consumption["AB_depb"] * 2 == schedule.vehicle_type_counts["AB_depb"]
+        # Without consumption, single vehicle can service all rotations.
+        # With high idling, every rotation needs its own vehicle
+        assert no_idle_consumption["AB_depb"] * 4 == schedule.vehicle_type_counts["AB_depb"]
 
     def test_calculate_consumption(self, tmp_path):
         """Various tests to trigger errors and check if behaviour is as expected

tests/test_rotation.py

@@ -30,28 +30,3 @@ def test_set_charging_type():
     consumption_depb = s.calculate_rotation_consumption(rot)
     assert consumption_depb * 2 == pytest.approx(consumption_oppb)
 
-
-def test_calculate_consumption():
-    s = generate_basic_schedule()
-    vt = next(iter(s.vehicle_types))
-
-    r = Rotation("my_rot", vt, s)
-    # consumption without trips
-    assert r.calculate_consumption() == 0
-
-    some_rot = next(iter(s.rotations.values()))
-    first_trip = some_rot.trips[0]
-    first_trip.charging_type = "depb"
-    del first_trip.rotation
-    r.add_trip(vars(first_trip))
-    r.trips[0].consumption = None
-    r.calculate_consumption()
-    assert r.trips[0].consumption is not None
-    second_trip = some_rot.trips[1]
-    del second_trip.rotation
-    r.add_trip(vars(second_trip))
-    for trip in r.trips:
-        trip.consumption = None
-    r.calculate_consumption()
-    for trip in r.trips:
-        assert trip.consumption is not None

tests/test_schedule.py

@@ -686,3 +686,26 @@ def test_generate_event_list_from_prices(self):
             price_interval_s=3600
         )
         assert len(events) == 0
+
+    def test_rotation_consumption_calc(self):
+        s, args = generate_basic_schedule()
+        rot_iter = iter(s.rotations.values())
+        r = next(rot_iter)
+        r.trips = []
+        assert s.calculate_rotation_consumption(r) == 0
+
+        some_rot =   next(rot_iter)
+        first_trip = some_rot.trips[0]
+        del first_trip.rotation
+        r.add_trip(vars(first_trip))
+        r.trips[0].consumption = None
+        s.calculate_rotation_consumption(r)
+        assert r.trips[0].consumption is not None
+        second_trip = some_rot.trips[1]
+        del second_trip.rotation
+        r.add_trip(vars(second_trip))
+        for trip in r.trips:
+            trip.consumption = None
+        s.calculate_rotation_consumption(r)
+        for trip in r.trips:
+            assert trip.consumption is not None

tests/test_simulate.py

@@ -168,24 +168,27 @@ def test_empty_report(self, tmp_path):
 
     def test_create_trips_in_report(self, tmp_path):
         # create_trips_in_report option: must generate valid input trips.csv
-        values = self.DEFAULT_VALUES.copy()
-        values.update({
+        args_dict = vars(self.get_args())
+        update_dict={
             "mode": ["report"],
             "desired_soc_deps": 0,
             "ALLOW_NEGATIVE_SOC": True,
             "cost_calculation": False,
             "output_directory": tmp_path,
             "show_plots": False,
             "create_trips_in_report": True,
-        })
+        }
+        args_dict.update(update_dict)
+
+
         # simulate base scenario, report generates new trips.csv in (tmp) output
         with warnings.catch_warnings():
             warnings.simplefilter("ignore")
-            simulate(Namespace(**values))
+            simulate(Namespace(**args_dict))
         # new simulation with generated trips.csv
-        values = self.DEFAULT_VALUES.copy()
-        values["input_schedule"] = tmp_path / "report_1/trips.csv"
-        simulate(Namespace(**(values)))
+        args = vars(self.get_args())
+        args_dict["input_schedule"] = tmp_path / "report_1/trips.csv"
+        simulate(Namespace(**(args_dict)))
 
     def test_mode_recombination(self):
         args = self.get_args()