Merge branch 'dev' into feature/logging

data/examples/ebus_toolbox.cfg

@@ -21,6 +21,8 @@ level_of_loading_over_day_path =  data/examples/default_level_of_loading_over_da
 optimizer_config = data/examples/default_optimizer.cfg
 # Cost parameters (needed if cost_calculation flag is set to true, see Flag section below)
 cost_parameters_file = data/examples/cost_params.json
+# Path to rotation filter
+rotation_filter = data/examples/rotation_filter.csv
 
 ##### Modes #####
 ### Specify how you want to simulate the scenario ###
@@ -46,6 +48,11 @@ check_rotation_consistency = false
 skip_inconsistent_rotations = false
 # Show plots for users to view, only valid if generate_report = true (default: false)
 show_plots = true
+# Rotation filter variable, options:
+# "include": include only the rotations from file 'rotation_filter'
+# "exclude": exclude the rotations from file 'rotation_filter' from the schedule
+# null: deactivate function
+rotation_filter_variable = null
 
 ##### Physical setup of environment #####
 ### Parametrization of the physical setup ###

data/examples/rotation_filter.csv

@@ -0,0 +1,2 @@
+1
+2

ebus_toolbox/consumption.py

@@ -109,10 +109,8 @@ def calculate_consumption(self, time, distance, vehicle_type, charging_type, tem
         consumption_function = vehicle_type+"_from_"+consumption_path
         try:
             mileage = self.consumption_files[consumption_function](
-                                                               this_incline=height_diff / distance,
-                                                               this_temp=temp,
-                                                               this_lol=level_of_loading,
-                                                               this_speed=mean_speed)
+                this_incline=height_diff / distance, this_temp=temp,
+                this_lol=level_of_loading, this_speed=mean_speed)
         except KeyError:
             # creating the interpol function from csv file.
             delim = util.get_csv_delim(consumption_path)
@@ -135,10 +133,8 @@ def interpol_function(this_incline, this_temp, this_lol, this_speed):
             self.consumption_files.update({consumption_function: interpol_function})
 
             mileage = self.consumption_files[consumption_function](
-                                                               this_incline=height_diff / distance,
-                                                               this_temp=temp,
-                                                               this_lol=level_of_loading,
-                                                               this_speed=mean_speed)
+               this_incline=height_diff / distance, this_temp=temp,
+               this_lol=level_of_loading, this_speed=mean_speed)
 
         consumed_energy = mileage * distance / 1000  # kWh
         delta_soc = -1 * (consumed_energy / vehicle_info["capacity"])

ebus_toolbox/schedule.py

@@ -386,6 +386,41 @@ def get_negative_rotations(self, scenario):
 
         return list(negative_rotations)
 
+    def rotation_filter(self, args, rf_list=[]):
+        """Edits rotations according to args.rotation_filter_variable.
+
+        :param args: used arguments are rotation_filter, path to rotation ids,
+                     and rotation_filter_variable that sets mode (options: include, exclude)
+        :type args: argparse.Namespace
+        :param rf_list: rotation filter list with strings of rotation ids (default is None)
+        :type rf_list: list
+        """
+        if args.rotation_filter_variable is None:
+            # filtering disabled
+            return
+        # cast rotations in filter to string
+        rf_list = [str(i) for i in rf_list]
+
+        if args.rotation_filter is None and not rf_list:
+            warnings.warn("Rotation filter variable is enabled but file and list are not used.")
+            return
+
+        if args.rotation_filter:
+            # read out rotations from file (one rotation ID per line)
+            try:
+                with open(args.rotation_filter, encoding='utf-8') as f:
+                    for line in f:
+                        rf_list.append(line.strip())
+            except FileNotFoundError:
+                warnings.warn(f"Path to rotation filter {args.rotation_filter} is invalid.")
+                # no file, no change
+                return
+        # filter out rotations in self.rotations
+        if args.rotation_filter_variable == "exclude":
+            self.rotations = {k: v for k, v in self.rotations.items() if k not in rf_list}
+        elif args.rotation_filter_variable == "include":
+            self.rotations = {k: v for k, v in self.rotations.items() if k in rf_list}
+
     def generate_scenario(self, args):
         """ Generate scenario.json for SpiceEV
 

ebus_toolbox/simulate.py

@@ -50,11 +50,15 @@ def simulate(args):
         outside_temperatures=args.outside_temperature_over_day_path,
         level_of_loading_over_day=args.level_of_loading_over_day_path)
 
-    schedule = Schedule.from_csv(args.input_schedule,
-                                 vehicle_types,
-                                 stations,
-                                 **vars(args))
+    # generate schedule from csv
+    schedule = Schedule.from_csv(args.input_schedule, vehicle_types, stations, **vars(args))
+
+    # filter rotations
+    schedule.rotation_filter(args)
+
+    # calculate consumption of all trips
     schedule.calculate_consumption()
+
     # scenario simulated once
     scenario = schedule.run(args)
 

ebus_toolbox/util.py

@@ -328,6 +328,11 @@ def get_args():
     parser.add_argument('--eta', action='store_true',
                         help='Show estimated time to finish simulation after each step, '
                              'instead of progress bar. Not recommended for fast computations.')
+    parser.add_argument('--rotation-filter', default=None,
+                        help='Use json data with rotation ids')
+    parser.add_argument('--rotation-filter-variable', default=None,
+                        choices=[None, 'include', 'exclude'],
+                        help='set mode for filtering schedule rotations')
 
     # #### LOGGING PARAMETERS #### #
     parser.add_argument('--loglevel', default='INFO',

tests/test_schedule.py

@@ -1,3 +1,5 @@
+from argparse import Namespace
+from copy import deepcopy
 from datetime import timedelta
 import pytest
 import sys
@@ -6,7 +8,7 @@
 
 from tests.conftest import example_root, file_root
 from tests.helpers import generate_basic_schedule
-from ebus_toolbox import schedule, trip, consumption, util
+from ebus_toolbox import consumption, rotation, schedule, trip, util
 
 
 mandatory_args = {
@@ -63,9 +65,13 @@ def test_mandatory_options_exit(self):
         """
         Check if the schedule creation properly throws an error in case of missing mandatory options
         """
-        with pytest.raises(Exception):
-            # schedule creation without mandatory args
-            schedule.Schedule(self.vehicle_types, self.electrified_stations)
+        args = mandatory_args.copy()
+        for key in mandatory_args.keys():
+            value = args.pop(key)
+            with pytest.raises(Exception):
+                # schedule creation without mandatory arg
+                schedule.Schedule(self.vehicle_types, self.electrified_stations, **args)
+            args[key] = value
 
     def test_station_data_reading(self):
         """ Test if the reading of the geo station data works and outputs warnings in
@@ -173,6 +179,67 @@ def test_get_negative_rotations(self):
         neg_rots = sched.get_negative_rotations(scen)
         assert '1' in neg_rots
 
+    def test_rotation_filter(self, tmp_path):
+        s = schedule.Schedule(self.vehicle_types, self.electrified_stations, **mandatory_args)
+        args = Namespace(**{
+            "rotation_filter_variable": None,
+            "rotation_filter": None,
+        })
+        # add dummy rotations
+        s.rotations = {
+            str(i): rotation.Rotation(id=str(i), vehicle_type="", schedule=None)
+            for i in range(6)
+        }
+        s.original_rotations = deepcopy(s.rotations)
+        # filtering disabled
+        args.rotation_filter_variable = None
+        s.rotation_filter(args)
+        assert s.rotations.keys() == s.original_rotations.keys()
+
+        # filtering not disabled, but neither file nor list given -> warning
+        args.rotation_filter_variable = "include"
+        args.rotation_filter = None
+        with pytest.warns(UserWarning):
+            s.rotation_filter(args)
+        assert s.rotations.keys() == s.original_rotations.keys()
+
+        # filter file not found -> warning
+        args.rotation_filter = tmp_path / "filter.txt"
+        with pytest.warns(UserWarning):
+            s.rotation_filter(args)
+        assert s.rotations.keys() == s.original_rotations.keys()
+
+        # filter (include) from JSON file
+        args.rotation_filter.write_text("3 \n 4\n16")
+        s.rotation_filter(args)
+        assert sorted(s.rotations.keys()) == ['3', '4']
+
+        # filter (exclude) from given list
+        args.rotation_filter_variable = "exclude"
+        args.rotation_filter = None
+        s.rotations = deepcopy(s.original_rotations)
+        s.rotation_filter(args, rf_list=['3', '4'])
+        assert sorted(s.rotations.keys()) == ['0', '1', '2', '5']
+
+        # filter (include) from integer list
+        s.rotations = deepcopy(s.original_rotations)
+        args.rotation_filter_variable = "include"
+        s.rotation_filter(args, rf_list=[3, 4])
+        assert sorted(s.rotations.keys()) == ['3', '4']
+
+        # filter nothing
+        s.rotations = deepcopy(s.original_rotations)
+        args.rotation_filter = tmp_path / "filter.txt"
+        args.rotation_filter.write_text('')
+        args.rotation_filter_variable = "exclude"
+        s.rotation_filter(args, rf_list=[])
+        assert s.rotations.keys() == s.original_rotations.keys()
+
+        # filter all (is this intended?)
+        args.rotation_filter_variable = "include"
+        s.rotation_filter(args, rf_list=[])
+        assert not s.rotations
+
     def test_scenario_with_feed_in(self):
         """ Check if running a example with an extended electrified stations file
          with feed in, external load and battery works and if a scenario object is returned"""

tests/test_simulate.py

@@ -32,6 +32,7 @@ class TestSimulate:
         "days": None,
         "signal_time_dif": 10,
         "include_price_csv": None,
+        "rotation_filter_variable": None,
         "seed": None,
         "default_buffer_time_opps": 0,
         "desired_soc_opps": 1,