Merge branch 'dev' into feature/#94_extended_output_plots

data/examples/simba.cfg

@@ -76,6 +76,10 @@ strategy_opps = greedy
 strategy_options_deps = {"CONCURRENCY": 1}
 strategy_options_opps = {}
 
+# Cost calculation strategy
+cost_calculation_strategy_deps = balanced
+cost_calculation_strategy_opps = greedy
+
 ##### Physical setup of environment #####
 ### Parametrization of the physical setup ###
 # Default max power [kW] of grid connectors at depot and opp stations,

docs/source/modes.rst

@@ -79,7 +79,7 @@ Now, only rotations are left that are non-negative when viewed alone, but might
 In the end, the largest number of rotations that produce a non-negative result when taken together is returned as the optimized scenario.
 
 Recombination: split negative depot rotations into smaller rotations
--------------
+--------------------------------------------------------------------
 ::
 
     mode = split_negative_depb

docs/source/simulation_parameters.rst

@@ -84,7 +84,22 @@ The example (data/simba.cfg) contains parameter descriptions which are explained
      - false
      - Boolean
      - If activated, plots are displayed with every run of :ref:`report` mode
-
+   * - strategy_deps
+     - balanced
+     - SpiceEV Strategies (greedy, balanced, peak_shaving, peak_load_windows, balanced_market)
+     - Charging strategy used in depots.
+   * - strategy_opps
+     - greedy
+     - SpiceEV Strategies (greedy, balanced, peak_shaving, peak_load_windows, balanced_market)
+     - Charging strategy used in opportunity stations.
+   * - cost_calculation_strategy_deps
+     - strategy_deps value
+     - SpiceEV Strategies (greedy, balanced, peak_shaving, peak_load_windows, balanced_market)
+     - Strategy for cost calculation at depots.
+   * - cost_calculation_strategy_opps
+     - strategy_opps value
+     - SpiceEV Strategies (greedy, balanced, peak_shaving, peak_load_windows, balanced_market)
+     - Strategy for cost calculation at opportunity stations.
    * - preferred_charging_type
      - depb
      - depb, oppb

simba/costs.py

@@ -19,6 +19,7 @@ def calculate_costs(c_params, scenario, schedule, args):
     :type schedule: Schedule
     :param args: Configuration arguments specified in config files contained in configs directory
     :type args: argparse.Namespace
+    :return: cost object
     """
     cost_object = Costs(schedule, scenario, args, c_params)
 
@@ -56,7 +57,7 @@ def calculate_costs(c_params, scenario, schedule, args):
 
     logging.info(cost_object.info())
 
-    setattr(scenario, "costs", cost_object)
+    return cost_object
 
 
 class Costs:
@@ -319,10 +320,21 @@ def set_electricity_costs(self):
                       if pv.parent == gcID])
             timeseries = vars(self.scenario).get(f"{gcID}_timeseries")
 
+            # Get the calculation strategy / method from args.
+            # If no value is set, use the same strategy as the charging strategy
+            default_cost_strategy = vars(self.args)["strategy_" + station.get("type")]
+
+            cost_strategy_name = "cost_calculation_strategy_" + station.get("type")
+            cost_calculation_strategy = (vars(self.args).get(cost_strategy_name)
+                                         or default_cost_strategy)
+
             # calculate costs for electricity
             try:
+                if cost_calculation_strategy == "peak_load_window":
+                    if timeseries.get("window signal [-]") is None:
+                        raise Exception("No peak load window signal provided for cost calculation")
                 costs_electricity = calc_costs_spice_ev(
-                    strategy=vars(self.args)["strategy_" + station.get("type")],
+                    strategy=cost_calculation_strategy,
                     voltage_level=gc.voltage_level,
                     interval=self.scenario.interval,
                     timestamps_list=timeseries.get("time"),

simba/schedule.py

@@ -114,6 +114,7 @@ def from_datacontainer(cls, data: DataContainer, args):
                 trip["departure_name"], trip["arrival_name"])
 
             if trip["level_of_loading"] is None:
+                assert len(data.level_of_loading_data) == 24, "Need 24 entries in level of loading"
                 trip["level_of_loading"] = util.get_mean_from_hourly_dict(
                     data.level_of_loading_data, trip["departure_time"], trip["arrival_time"])
             else:
@@ -122,6 +123,7 @@ def from_datacontainer(cls, data: DataContainer, args):
                     trip["level_of_loading"] = min(1, max(0, trip["level_of_loading"]))
 
             if trip["temperature"] is None:
+                assert len(data.temperature_data) == 24, "Need 24 entries in temperature data"
                 trip["temperature"] = util.get_mean_from_hourly_dict(
                     data.temperature_data, trip["departure_time"], trip["arrival_time"])
 
@@ -209,15 +211,16 @@ def check_consistency(cls, schedule):
     def run(self, args, mode="distributed"):
         """Runs a schedule without assigning vehicles.
 
-        For external usage the core run functionality is accessible through this function. It
-        allows for defining a custom-made assign_vehicles method for the schedule.
+        For external usage the core run functionality is accessible through this function.
+        It allows for defining a custom-made assign_vehicles method for the schedule.
+
         :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 mode: option of "distributed" or "greedy"
+        :param mode: SpiceEV strategy name
         :type mode: str
         :return: scenario
-        :rtype spice_ev.Scenario
+        :rtype: spice_ev.Scenario
         """
         # Make sure all rotations have an assigned vehicle
         assert all([rot.vehicle_id is not None for rot in self.rotations.values()])
@@ -348,8 +351,9 @@ def assign_vehicles_w_min_recharge_soc(self):
     def assign_vehicles_custom(self, vehicle_assigns: Iterable[dict]):
         """ Assign vehicles on a custom basis.
 
-        Assign vehicles based on a datasource, containing all rotations, their vehicle_ids and
-        desired start socs.
+        Assign vehicles based on a datasource,
+        containing all rotations, their vehicle_ids and desired start socs.
+
         :param vehicle_assigns: Iterable of dict with keys rotation_id, vehicle_id and start_soc
             for each rotation
         :type vehicle_assigns: Iterable[dict]
@@ -1125,6 +1129,7 @@ def update_csv_file_info(file_info, gc_name):
     - set grid_connector_id
     - update csv_file path
     - set start_time and step_duration_s from CSV information if not given
+
     :param file_info: csv information from electrified station
     :type file_info: dict
     :param gc_name: station name

simba/simulate.py

@@ -268,7 +268,7 @@ def report(schedule, scenario, args, i):
             # cost calculation part of report
             try:
                 cost_parameters = schedule.data_container.cost_parameters_data
-                calculate_costs(cost_parameters, scenario, schedule, args)
+                scenario.costs = calculate_costs(cost_parameters, scenario, schedule, args)
             except Exception:
                 logging.warning(f"Cost calculation failed due to {traceback.format_exc()}")
                 if args.propagate_mode_errors:

simba/util.py

@@ -439,6 +439,13 @@ def get_parser():
                         help='strategy to use in depot')
     parser.add_argument('--strategy-opps', default='greedy', choices=STRATEGIES,
                         help='strategy to use at station')
+
+    # #### Cost calculation strategy #####
+    parser.add_argument('--cost-calculation-strategy-deps', choices=STRATEGIES,
+                        help='Strategy for cost calculation to use in depot')
+    parser.add_argument('--cost-calculation-strategy-opps', choices=STRATEGIES,
+                        help='Strategy for cost calculation to use at station')
+
     parser.add_argument('--strategy-options-deps', default={},
                         type=lambda s: s if type(s) is dict else json.loads(s),
                         help='special strategy options to use in depot')

tests/test_cost_calculation.py

@@ -0,0 +1,63 @@
+from tests.test_schedule import BasicSchedule
+from simba.util import uncomment_json_file
+from simba.costs import calculate_costs
+
+
+class TestCostCalculation:
+    def test_cost_calculation(self):
+        schedule, scenario, args = BasicSchedule().basic_run()
+        file = args.cost_parameters_path
+        with open(file, "r") as file:
+            cost_params = uncomment_json_file(file)
+
+        assert args.strategy_deps == "balanced"
+        assert args.strategy_opps == "greedy"
+
+        args.cost_calculation_strategy_deps = None
+        args.cost_calculation_strategy_opps = None
+
+        costs_vanilla = calculate_costs(cost_params, scenario, schedule, args)
+
+        assert args.strategy_deps == "balanced"
+        assert args.strategy_opps == "greedy"
+
+        args.cost_calculation_strategy_deps = "balanced"
+        args.cost_calculation_strategy_opps = "greedy"
+        costs_with_same_strat = calculate_costs(cost_params, scenario, schedule, args)
+
+        # assert all costs are the same
+        for station in costs_vanilla.costs_per_gc:
+            for key in costs_vanilla.costs_per_gc[station]:
+                assert (costs_vanilla.costs_per_gc[station][key] ==
+                        costs_with_same_strat.costs_per_gc[station][key]), station
+
+        args.cost_calculation_strategy_opps = "balanced_market"
+        args.cost_calculation_strategy_deps = "balanced_market"
+        costs_with_other_strat = calculate_costs(cost_params, scenario, schedule, args)
+        print(costs_vanilla.costs_per_gc["cumulated"]["c_total_annual"])
+        print(costs_with_other_strat.costs_per_gc["cumulated"]["c_total_annual"])
+        station = "cumulated"
+        for key in costs_vanilla.costs_per_gc[station]:
+            if "el_energy" not in key:
+                continue
+            assert (costs_vanilla.costs_per_gc[station][key] !=
+                    costs_with_other_strat.costs_per_gc[station][key]), key
+
+        args.cost_calculation_strategy_opps = "peak_load_window"
+        args.cost_calculation_strategy_deps = "peak_load_window"
+        costs_with_other_strat = calculate_costs(cost_params, scenario, schedule, args)
+        station = "cumulated"
+        for key in costs_vanilla.costs_per_gc[station]:
+            if "el_energy" not in key:
+                continue
+            assert (costs_vanilla.costs_per_gc[station][key] !=
+                    costs_with_other_strat.costs_per_gc[station][key]), key
+
+        args.cost_calculation_strategy_opps = "peak_shaving"
+        args.cost_calculation_strategy_deps = "peak_shaving"
+        costs_with_other_strat = calculate_costs(cost_params, scenario, schedule, args)
+        # assert all costs are the same
+        for station in costs_vanilla.costs_per_gc:
+            for key in costs_vanilla.costs_per_gc[station]:
+                assert (costs_vanilla.costs_per_gc[station][key] ==
+                        costs_with_other_strat.costs_per_gc[station][key]), station