Merge pull request #196 from rl-institut/feature/cost_calculation_str…

…ategy

Feature/cost calculation strategy

data/examples/simba.cfg

@@ -74,6 +74,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/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/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

@@ -437,6 +437,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