Feature/improve optimizer

docs/source/modes.rst

@@ -328,6 +328,10 @@ The functionality of the optimizer is controlled through the optimizer.cfg speci
      - False
      - [True, False]
      - Discard rotations which have SoCs below the threshold, even when every station is electrified
+   * - post_opt_station_pruning
+     - False
+     - [True, False]
+     - Discard electrified stations which are not needed for a fully electrified scenario. This is done after the main optimization loop and respects the min_soc from the configuration. Stations which are electrified at the beginning of the Optimization are not removed, since they are expected to be fixed and confirmed.
    * - check_for_must_stations
      - True
      - [True, False]

pytest.ini

@@ -0,0 +1,3 @@
+[pytest]
+testpaths =
+    tests

simba/optimizer_util.py

@@ -94,6 +94,7 @@ def __init__(self):
         self.eps = 0.0001
 
         self.remove_impossible_rotations = False
+        self.post_opt_station_pruning = False
         self.node_choice = "step-by-step"
         self.max_brute_loop = 20
         self.run_only_neg = True

simba/station_optimization.py

@@ -1,4 +1,5 @@
-""" Try to minimize the amount of electrified stations to achieve full electrification."""
+"""Try to minimize the amount of electrified stations to achieve full electrification."""
+
 from copy import deepcopy
 import json
 import sys
@@ -8,11 +9,9 @@
 from simba.station_optimizer import opt_util
 from spice_ev.report import generate_soc_timeseries
 
-config = opt_util.OptimizerConfig()
-
 
 def setup_logger(conf):
-    """ Setup file and stream logging by config and args arguments.
+    """Setup file and stream logging by config and args arguments.
 
     :param conf: configuration object
     :type conf: simba.optimizer_util.OptimizerConfig
@@ -36,17 +35,16 @@ def setup_logger(conf):
     file_handler_all_opts.setLevel(conf.debug_level)
 
     # and logging to a file which is put in the folder with the other optimizer results
-    file_handler_this_opt = logging.FileHandler(Path(conf.optimizer_output_dir) /
-                                                Path('optimizer.log'))
+    file_handler_this_opt = logging.FileHandler(
+        Path(conf.optimizer_output_dir) / Path("optimizer.log"))
     file_handler_this_opt.setLevel(conf.debug_level)
 
-    formatter = logging.Formatter('%(asctime)s:%(message)s',
-                                  "%m%d %H%M%S")
+    formatter = logging.Formatter("%(asctime)s:%(message)s", "%m%d %H%M%S")
 
     file_handler_all_opts.setFormatter(formatter)
     file_handler_this_opt.setFormatter(formatter)
 
-    formatter = logging.Formatter('%(message)s')
+    formatter = logging.Formatter("%(message)s")
     stream_handler = logging.StreamHandler()
     stream_handler.setFormatter(formatter)
     stream_handler.setLevel(conf.console_level)
@@ -59,7 +57,7 @@ def setup_logger(conf):
 
 
 def prepare_filesystem(args, conf):
-    """ Prepare files and folders in the optimization results folder.
+    """Prepare files and folders in the optimization results folder.
 
     :param conf: configuration
     :type conf:  simba.optimizer_util.OptimizerConfig
@@ -72,8 +70,8 @@ def prepare_filesystem(args, conf):
     conf.optimizer_output_dir.mkdir(parents=True, exist_ok=True)
 
 
-def run_optimization(conf, sched=None, scen=None, args=None):
-    """ Add electrified stations until there are no more negative rotations.
+def run_optimization(conf: opt_util.OptimizerConfig, sched=None, scen=None, args=None):
+    """Add electrified stations until there are no more negative rotations.
 
     Configured with arguments from optimizer config file.
 
@@ -99,7 +97,9 @@ def run_optimization(conf, sched=None, scen=None, args=None):
                          "be provided together")
         assert conf.scenario, error_message
         assert conf.args, error_message
-        sched, scen, args = opt_util.toolbox_from_pickle(conf.schedule, conf.scenario, conf.args)
+        sched, scen, args = opt_util.toolbox_from_pickle(
+            conf.schedule, conf.scenario, conf.args
+        )
 
     original_schedule = deepcopy(sched)
 
@@ -120,9 +120,13 @@ def run_optimization(conf, sched=None, scen=None, args=None):
     # filter out depot chargers if option is set
     if conf.run_only_oppb:
         optimizer.config.exclusion_rots = optimizer.config.exclusion_rots.union(
-            r for r in sched.rotations if "depb" == sched.rotations[r].charging_type)
-        sched.rotations = {r: sched.rotations[r] for r in sched.rotations
-                           if "oppb" == sched.rotations[r].charging_type}
+            r for r in sched.rotations if "depb" == sched.rotations[r].charging_type
+        )
+        sched.rotations = {
+            r: sched.rotations[r]
+            for r in sched.rotations
+            if "oppb" == sched.rotations[r].charging_type
+        }
         if len(sched.rotations) == 0:
             raise Exception("No rotations left after removing depot chargers")
 
@@ -151,13 +155,15 @@ def run_optimization(conf, sched=None, scen=None, args=None):
         neg_rots = optimizer.get_negative_rotations_all_electrified()
         optimizer.config.exclusion_rots.update(neg_rots)
         optimizer.schedule.rotations = {
-            r: optimizer.schedule.rotations[r] for r in optimizer.schedule.rotations if
-            r not in optimizer.config.exclusion_rots}
+            r: optimizer.schedule.rotations[r]
+            for r in optimizer.schedule.rotations
+            if r not in optimizer.config.exclusion_rots
+        }
 
         logger.warning(f"{len(neg_rots)} negative rotations {neg_rots} were removed from schedule "
                        "because they cannot be electrified")
         assert len(optimizer.schedule.rotations) > 0, (
-            "Schedule cannot be optimized, since rotations cannot be electrified.")
+           "Schedule cannot be optimized, since rotations cannot be electrified.")
 
     # if the whole network can not be fully electrified if even just a single station is not
     # electrified, this station must be included in a fully electrified network
@@ -184,6 +190,9 @@ def run_optimization(conf, sched=None, scen=None, args=None):
     logger.debug("%s total stations", len(ele_stations))
     logger.debug("These rotations could not be electrified: %s", optimizer.could_not_be_electrified)
 
+    if conf.post_opt_station_pruning:
+        ele_station_set, ele_stations = optimizer.prune_stations(ele_station_set)
+
     # remove none values from socs in the vehicle_socs so timeseries_calc can work
     optimizer.replace_socs_from_none_to_value()
 
@@ -212,7 +221,9 @@ def run_optimization(conf, sched=None, scen=None, args=None):
 
     # Restore original rotations
     for rotation_id in original_schedule.rotations:
-        optimizer.schedule.rotations[rotation_id] = original_schedule.rotations[rotation_id]
+        optimizer.schedule.rotations[rotation_id] = original_schedule.rotations[
+            rotation_id
+        ]
 
     # remove exclusion since internally these would not be simulated
     optimizer.config.exclusion_rots = set()

simba/station_optimizer.py

@@ -1199,6 +1199,52 @@ def lift_and_clip_positive_gradient(self, start_idx: int, soc: np.array,
         soc = np.hstack((soc_pre, soc))
         return soc
 
+    def prune_stations(self, electrified_station_set):
+        """Prune electrified stations not needed for full electrification
+
+        This uses a single greedy approach, iterating over stations one by one,
+        removing stations without reducing the level of electrification.
+        If the removal of a station leads to low_socs,
+        the station is added again and not removed again.
+        :param electrified_station_set: set of stations which are checked for remomval
+        :type electrified_station_set: set[str]
+        :return: None
+        """
+        # These stations were given in the config to be electrified or where electrified before
+        # optimizing.
+        # They are not removed, since its assumed, they are "set in stone"
+        pre_electrified = self.config.inclusion_stations.union(self.base_schedule.stations.keys())
+
+        # Without the must_include_stations the scenario can not be fully electrified.
+        # This was checked earlier
+        not_removable_stations = pre_electrified.union(self.must_include_set)
+
+        removable_stations = electrified_station_set.difference(not_removable_stations)
+        self.logger.log(msg="Searching for stations not needed for a full electrification scenario",
+                        level=100)
+        self.logger.log(msg=f"Deelectrifying {len(removable_stations)} stations one by one.",
+                        level=100)
+        for station in sorted(removable_stations):
+            electrified_station_set = electrified_station_set.difference([station])
+            electrified_stations = not_removable_stations.union(electrified_station_set)
+            vehicle_socs = self.timeseries_calc(electrified_stations)
+            min_soc = 1
+            for rot in self.schedule.rotations:
+                soc, start, end = self.get_rotation_soc(rot, vehicle_socs)
+                soc_min = np.min(soc[start:end])
+                min_soc = min(min_soc, soc_min)
+                if soc_min < self.config.min_soc:
+                    break
+            if min_soc < self.config.min_soc:
+                self.logger.info("%s , can't be deelectrified. SoC would drop to: %s",
+                                 station, min_soc)
+                electrified_station_set.add(station)
+                continue
+            self.logger.info("%s can be removed. SoC drops to: %s", station, min_soc)
+            self.electrified_station_set.remove(station)
+            del self.electrified_stations[station]
+        return self.electrified_station_set, self.electrified_stations
+
 
 def get_min_soc_and_index(soc_idx, mask):
     """ Returns the minimal SoC and the corresponding index of a masked soc_idx.

tests/test_station_optimization.py

@@ -69,7 +69,7 @@ def setup_test(self, tmp_path):
         self.vehicle_types = adjust_vehicle_file(vehicles_dest, capacity=50, mileage=10)
 
         # remove escape characters from string
-        vehicles_dest_str = str(vehicles_dest).replace('\\', '/')
+        vehicles_dest_str = str(vehicles_dest).replace("\\", "/")
         # replace line which defines vehicle_types up to line break. line break is concatenated in
         # the replacement, to keep format
         src_text = re.sub(
@@ -78,36 +78,38 @@ def setup_test(self, tmp_path):
 
         # Use the default electrified stations from example folder but change some values
         stations_path = example_root / "electrified_stations/electrified_stations.json"
-        with open(stations_path, "r", encoding='utf-8') as file:
+        with open(stations_path, "r", encoding="utf-8") as file:
             self.electrified_stations = util.uncomment_json_file(file)
         # only keep Station-0 electrified and remove the other staitons
-        self.electrified_stations = {"Station-0": self.electrified_stations["Station-0"]}
+        self.electrified_stations = {
+            "Station-0": self.electrified_stations["Station-0"]
+        }
         del self.electrified_stations["Station-0"]["external_load"]
         del self.electrified_stations["Station-0"]["battery"]
         del self.electrified_stations["Station-0"]["energy_feed_in"]
 
         # store the adjusted electrified_stations temporarily and use them in the config file
         electrified_stations_dest = tmp_path / "electrified_stations.json"
-        with open(electrified_stations_dest, "w", encoding='utf-8') as file:
+        with open(electrified_stations_dest, "w", encoding="utf-8") as file:
             json.dump(self.electrified_stations, file)
 
         # remove escape characters from string. \1 refers to the replacement of the first group
         # in the regex expression, i.e. not replacing the newline characters
-        electrified_stations_dest_str = str(electrified_stations_dest).replace('\\', '/')
+        electrified_stations_dest_str = str(electrified_stations_dest).replace("\\", "/")
         src_text = re.sub(
             r"(electrified_stations\s=.*)(:=\r\n|\r|\n)",
-            "electrified_stations = " + electrified_stations_dest_str + r"\g<2>", src_text)
+            "electrified_stations = " + electrified_stations_dest_str + r"\g<2>", src_text,)
 
         src_text = re.sub(
             r"(preferred_charging_type\s=.*)(:=\r\n|\r|\n)",
-            "preferred_charging_type = oppb"r"\g<2>", src_text)
+            "preferred_charging_type = oppb" r"\g<2>", src_text)
 
         # change config file with adjusted temporary paths to vehicles and electrified stations
         dst = tmp_path / "simba.cfg"
         dst.write_text(src_text)
 
     def generate_datacontainer_args(self, trips_file_name="trips.csv"):
-        """ Check if running a basic example works and return data container.
+        """Check if running a basic example works and return data container.
 
         :param trips_file_name: file name of the trips file. Has to be inside the test_input_file
             folder
@@ -151,7 +153,7 @@ def test_join_all_subsets(self):
             assert subset in joined_subsets2
 
     def test_fast_calculations_and_events(self):
-        """ Test if the base optimization finishes without raising errors"""
+        """Test if the base optimization finishes without raising errors"""
         trips_file_name = "trips_for_optimizer.csv"
         data_container, args = self.generate_datacontainer_args(trips_file_name)
         data_container.stations_data = {}
@@ -166,8 +168,9 @@ def test_fast_calculations_and_events(self):
         generate_soc_timeseries(scen)
 
         config = opt_util.OptimizerConfig()
-        sopt = station_optimizer.StationOptimizer(sched, scen, args, config=config,
-                                                  logger=logging.getLogger())
+        sopt = station_optimizer.StationOptimizer(
+            sched, scen, args, config=config, logger=logging.getLogger()
+        )
 
         # create charging dicts which contain soc over time, which is numerically calculated
         sopt.create_charging_curves()
@@ -186,8 +189,10 @@ def test_fast_calculations_and_events(self):
         assert len(events) == 1
         e = events[0]
         e1 = copy(e)
-        vehicle_socs_reduced = {vehicle: [soc - 1 for soc in socs] for vehicle, socs in
-                                scen.vehicle_socs.items()}
+        vehicle_socs_reduced = {
+            vehicle: [soc - 1 for soc in socs]
+            for vehicle, socs in scen.vehicle_socs.items()
+        }
         # The vehicle socs were reduced. Now both vehicles have socs below 0
         assert 2 == sum(min(socs) < 0 for socs in vehicle_socs_reduced.values())
 
@@ -213,22 +218,23 @@ def test_fast_calculations_and_events(self):
         # Higher socs are increased.
         # Since the soc stays at 1 for longer, the start index should change
         vehicle_socs_increased = {
-            vehicle: [min(soc + abs(e1.min_soc) + new_low_soc, 1) for soc in socs] for
-            vehicle, socs in scen.vehicle_socs.items()}
+            vehicle: [min(soc + abs(e1.min_soc) + new_low_soc, 1) for soc in socs]
+            for vehicle, socs in scen.vehicle_socs.items()}
         events = sopt.get_low_soc_events(soc_data=vehicle_socs_increased, rel_soc=True)
         e3 = events[0]
         assert e1.start_idx != e3.start_idx
         assert e1.end_idx == e3.end_idx
         assert e1.min_soc != e3.min_soc
 
         vehicle_socs_more_increased = {
-            vehicle: [min(soc + abs(e1.min_soc) + 0.1, 1) for soc in socs] for vehicle, socs in
-            scen.vehicle_socs.items()}
-        events = sopt.get_low_soc_events(soc_data=vehicle_socs_more_increased, rel_soc=True)
+            vehicle: [min(soc + abs(e1.min_soc) + 0.1, 1) for soc in socs]
+            for vehicle, socs in scen.vehicle_socs.items()}
+        events = sopt.get_low_soc_events(
+            soc_data=vehicle_socs_more_increased, rel_soc=True)
         assert len(events) == 0
 
     def test_basic_optimization(self):
-        """ Test if the base optimization finishes without raising errors"""
+        """Test if the base optimization finishes without raising errors"""
         trips_file_name = "trips_for_optimizer.csv"
         data_container, args = self.generate_datacontainer_args(trips_file_name)
         data_container.stations_data = {}
@@ -242,7 +248,7 @@ def test_basic_optimization(self):
         opt_sched, opt_scen = run_optimization(conf, sched=sched, scen=scen, args=args)
 
     def test_schedule_consistency(self):
-        """ Test if the optimization returns all rotations even when some filters are active"""
+        """Test if the optimization returns all rotations even when some filters are active"""
         trips_file_name = "trips_for_optimizer.csv"
         data_container, args = self.generate_datacontainer_args(trips_file_name)
         args.preferred_charging_type = "oppb"
@@ -305,6 +311,36 @@ def test_deep_optimization(self, solver, node_choice):
         assert "Station-2" in opt_sched.stations
         assert "Station-3" in opt_sched.stations
 
+    def test_greedy_optimization_with_post_opt_station_removal(self):
+        """Identical setup to test_deep_optimization.
+        Uses greedy optimization, and use a post optimization module to remove stations which
+        can be removed without moving the soc below the minimal soc.
+        """
+        trips_file_name = "trips_for_optimizer_deep.csv"
+        data_container, args = self.generate_datacontainer_args(trips_file_name)
+        data_container.stations_data = {}
+        args.preferred_charging_type = "oppb"
+        for trip_d in data_container.trip_data:
+            trip_d["distance"] *= 15
+        sched, scen = self.generate_schedule_scenario(args, data_container)
+        config_path = example_root / "default_optimizer.cfg"
+        conf = opt_util.read_config(config_path)
+
+        assert len(sched.get_negative_rotations(scen)) == 2
+
+        conf.opt_type = "greedy"
+        # conf.solver = "spiceev"  # solver
+        # conf.node_choice = "brute"  # node_choice
+        conf.solver = "quick"
+        conf.node_choice = "node-choice"
+        conf.post_opt_station_pruning = True
+        opt_sched, opt_scen = run_optimization(conf, sched=sched, scen=scen, args=args)
+
+        assert len(opt_sched.get_negative_rotations(opt_scen)) == 0
+        assert "Station-1" not in opt_sched.stations
+        assert "Station-2" in opt_sched.stations
+        assert "Station-3" in opt_sched.stations
+
     def test_deep_optimization_extended(self):
         trips_file_name = "trips_extended.csv"
         data_container, args = self.generate_datacontainer_args(trips_file_name)
@@ -355,13 +391,13 @@ def test_critical_stations_optimization(self, caplog):
         conf.solver = "quick"
         conf.node_choice = "step-by-step"
         opt_sched, opt_scen = run_optimization(conf, sched=sched, scen=scen, args=args)
-        assert ("must stations {'Station-3', 'Station-2'}" in caplog.text or
-                "must stations {'Station-2', 'Station-3'}" in caplog.text)
+        assert ("must stations {'Station-3', 'Station-2'}" in caplog.text
+                or "must stations {'Station-2', 'Station-3'}" in caplog.text)
 
 
 def adjust_vehicle_file(source, capacity=None, mileage=0):
     # use the default vehicles from example folder
-    with open(source, "r", encoding='utf-8') as file:
+    with open(source, "r", encoding="utf-8") as file:
         vehicle_types = util.uncomment_json_file(file)
 
     for vehicle in vehicle_types:
@@ -370,6 +406,6 @@ def adjust_vehicle_file(source, capacity=None, mileage=0):
                 vehicle_types[vehicle][vtype]["capacity"] = capacity
             if mileage is not None:
                 vehicle_types[vehicle][vtype]["mileage"] = mileage
-    with open(source, "w", encoding='utf-8') as file:
+    with open(source, "w", encoding="utf-8") as file:
         json.dump(vehicle_types, file)
     return vehicle_types