diff --git a/data/examples/simba.cfg b/data/examples/simba.cfg index f62b5dee..27912033 100644 --- a/data/examples/simba.cfg +++ b/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, diff --git a/docs/source/modes.rst b/docs/source/modes.rst index 12640d35..85ed29dd 100644 --- a/docs/source/modes.rst +++ b/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 diff --git a/docs/source/simulation_parameters.rst b/docs/source/simulation_parameters.rst index b45ca8e3..cbfba443 100644 --- a/docs/source/simulation_parameters.rst +++ b/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 diff --git a/simba/costs.py b/simba/costs.py index ec4db73d..5d1a998d 100644 --- a/simba/costs.py +++ b/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"), diff --git a/simba/schedule.py b/simba/schedule.py index c469a054..60dd8354 100644 --- a/simba/schedule.py +++ b/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 diff --git a/simba/simulate.py b/simba/simulate.py index 205d1146..01139e86 100644 --- a/simba/simulate.py +++ b/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: diff --git a/simba/util.py b/simba/util.py index b8d88e0c..07e41dc9 100644 --- a/simba/util.py +++ b/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') diff --git a/tests/test_cost_calculation.py b/tests/test_cost_calculation.py new file mode 100644 index 00000000..110b0b2c --- /dev/null +++ b/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