From d643168e8fd33f1283e5443354fb75426bcc3a38 Mon Sep 17 00:00:00 2001 From: "paul.scheer" Date: Thu, 5 Sep 2024 11:59:27 +0200 Subject: [PATCH] Fix tests with new consumption calculation and shortened trip data --- simba/rotation.py | 63 -------------------------------------- simba/schedule.py | 64 +++++++++++++++++++++++++++++++++++++-- tests/test_consumption.py | 27 +++++++++++++++-- tests/test_rotation.py | 25 --------------- tests/test_schedule.py | 23 ++++++++++++++ tests/test_simulate.py | 17 ++++++----- 6 files changed, 118 insertions(+), 101 deletions(-) diff --git a/simba/rotation.py b/simba/rotation.py index 5bd3c0d..06f927e 100644 --- a/simba/rotation.py +++ b/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 diff --git a/simba/schedule.py b/simba/schedule.py index 2f4883f..4b2b7f7 100644 --- a/simba/schedule.py +++ b/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 diff --git a/tests/test_consumption.py b/tests/test_consumption.py index 0cbda92..bf6e3c2 100644 --- a/tests/test_consumption.py +++ b/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 diff --git a/tests/test_rotation.py b/tests/test_rotation.py index 8fd1c75..c9012aa 100644 --- a/tests/test_rotation.py +++ b/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 diff --git a/tests/test_schedule.py b/tests/test_schedule.py index 28a2605..2eee052 100644 --- a/tests/test_schedule.py +++ b/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 diff --git a/tests/test_simulate.py b/tests/test_simulate.py index 2a7f5ab..36b1a7e 100644 --- a/tests/test_simulate.py +++ b/tests/test_simulate.py @@ -168,8 +168,8 @@ 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, @@ -177,15 +177,18 @@ def test_create_trips_in_report(self, tmp_path): "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()