Add min/max load transformer and reducable demand as assets

CHANGELOG.md

@@ -23,8 +23,13 @@ Here is a template for new release sections
 
 ### Added
 
+- Introduce reducable demand. It should be listed within sinks, and provided an efficiency (number between 0 and 1). This efficiency correspond to the percent of the demand which must be provided (critical demand). The oemof-solph sinks which models the  non-critical part of the demand has very small variable_costs such that it should not influence the costs calculations but should be fulfilled rather than dumping energy into excess sinks. Developed for the server version. (#969)
+
+
 ### Changed
 
+- Add costs to excess sinks of busses. If the dictionary containing the information about the bus contains a key "price", its value will be applied to the variable costs of the sink (unit of the price is currency/energy unit, default currency/kWh). Developed for the server version. (#969)
+
 ### Fixed
 
 ## [1.1.0] - 2024-04-27

src/multi_vector_simulator/C0_data_processing.py

@@ -290,10 +290,12 @@ def define_excess_sinks(dict_values):
     for bus in dict_values[ENERGY_BUSSES]:
         excess_sink_name = bus + EXCESS_SINK
         energy_vector = dict_values[ENERGY_BUSSES][bus][ENERGY_VECTOR]
+        # TODO make this official if needed
+        excess_price = dict_values[ENERGY_BUSSES][bus].get("price", 0)
         define_sink(
             dict_values=dict_values,
             asset_key=excess_sink_name,
-            price={VALUE: 0, UNIT: CURR + "/" + UNIT},
+            price={VALUE: excess_price, UNIT: CURR + "/" + UNIT},
             inflow_direction=bus,
             energy_vector=energy_vector,
             asset_type="excess",

src/multi_vector_simulator/D0_modelling_and_optimization.py

@@ -348,6 +348,7 @@ def simulating(dict_values, model, local_energy_system):
         logging.info("Starting simulation.")
         # turn warnings into errors
         warnings.filterwarnings("error")
+        warnings.filterwarnings("always", category=FutureWarning)
         try:
             local_energy_system.solve(
                 solver="cbc",

src/multi_vector_simulator/D1_model_components.py

@@ -47,6 +47,7 @@
     MAXIMUM_ADD_CAP,
     MAXIMUM_ADD_CAP_NORMALIZED,
     DISPATCHABILITY,
+    TYPE_ASSET,
     OEMOF_ASSET_TYPE,
     OEMOF_GEN_STORAGE,
     OEMOF_SINK,
@@ -57,8 +58,13 @@
     EMISSION_FACTOR,
     BETA,
     INVESTMENT_BUS,
+    REDUCABLE_DEMAND,
+)
+from multi_vector_simulator.utils.helpers import (
+    get_item_if_list,
+    get_length_if_list,
+    reducable_demand_name,
 )
-from multi_vector_simulator.utils.helpers import get_item_if_list, get_length_if_list
 from multi_vector_simulator.utils.exceptions import (
     MissingParameterError,
     WrongParameterFormatError,
@@ -354,7 +360,10 @@ def sink(model, dict_asset, **kwargs):
 
     """
     if TIMESERIES in dict_asset:
-        sink_non_dispatchable(model, dict_asset, **kwargs)
+        if dict_asset.get(TYPE_ASSET) == REDUCABLE_DEMAND:
+            sink_demand_reduction(model, dict_asset, **kwargs)
+        else:
+            sink_non_dispatchable(model, dict_asset, **kwargs)
 
     else:
         sink_dispatchable_optimize(model, dict_asset, **kwargs)
@@ -641,6 +650,23 @@ def transformer_constant_efficiency_fix(model, dict_asset, **kwargs):
     else:
         # single input and single output
 
+        min_load_opts = {"min": 0, "max": 1}
+        min_load = dict_asset.get(SOC_MIN, None)
+        if min_load is not None:
+            if min_load[VALUE] != 0:
+                logging.warning(
+                    f"Minimal load of {min_load[VALUE]} was set to asset {dict_asset[LABEL]}"
+                )
+            min_load_opts["min"] = min_load[VALUE]
+        max_load = dict_asset.get(SOC_MAX, None)
+        if max_load is not None:
+            if max_load[VALUE] != 1:
+                logging.warning(
+                    f"Maximal load of {max_load[VALUE]} was set to asset {dict_asset[LABEL]}"
+                )
+
+            min_load_opts["max"] = max_load[VALUE]
+
         check_list_parameters_transformers_single_input_single_output(
             dict_asset, model.timeindex.size
         )
@@ -650,6 +676,7 @@ def transformer_constant_efficiency_fix(model, dict_asset, **kwargs):
             kwargs[OEMOF_BUSSES][dict_asset[OUTFLOW_DIRECTION]]: solph.Flow(
                 nominal_value=dict_asset[INSTALLED_CAP][VALUE],
                 variable_costs=dict_asset[DISPATCH_PRICE][VALUE],
+                **min_load_opts,
             )
         }
         efficiencies = {
@@ -691,10 +718,14 @@ def transformer_constant_efficiency_optimize(model, dict_asset, **kwargs):
     missing_dispatch_prices_or_efficiencies = None
 
     investment_bus = dict_asset.get(INVESTMENT_BUS)
+    invest_opts = {}
+    if dict_asset[MAXIMUM_ADD_CAP][VALUE] is not None:
+        invest_opts["maximum"] = dict_asset[MAXIMUM_ADD_CAP][VALUE]
+
     investment = solph.Investment(
         ep_costs=dict_asset[SIMULATION_ANNUITY][VALUE],
-        maximum=dict_asset[MAXIMUM_ADD_CAP][VALUE],
         existing=dict_asset[INSTALLED_CAP][VALUE],
+        **invest_opts,
     )
 
     # check if the transformer has multiple input or multiple output busses
@@ -801,6 +832,32 @@ def transformer_constant_efficiency_optimize(model, dict_asset, **kwargs):
 
         # single input and single output
 
+        min_load_opts = {"min": 0, "max": 1}
+        min_load = dict_asset.get(SOC_MIN, None)
+        if min_load is not None:
+            if min_load[VALUE] != 0:
+                logging.warning(
+                    f"Minimal load of {min_load[VALUE]} was set to asset {dict_asset[LABEL]}"
+                )
+                min_load_opts["nonconvex"] = solph.NonConvex()
+            min_load_opts["min"] = min_load[VALUE]
+
+        max_load = dict_asset.get(SOC_MAX, None)
+        if max_load is not None:
+            if max_load[VALUE] != 1:
+                logging.warning(
+                    f"Maximal load of {max_load[VALUE]} was set to asset {dict_asset[LABEL]}"
+                )
+                min_load_opts["nonconvex"] = solph.NonConvex()
+
+            min_load_opts["max"] = max_load[VALUE]
+
+        if "nonconvex" in min_load_opts:
+            if invest_opts.get("maximum", None) is None:
+                raise ValueError(
+                    f"You need to provide a maximum_capacity to the asset {dict_asset[LABEL]}, if you set a minimal/maximal load different from 0/1"
+                )
+
         if investment_bus is None:
             investment_bus = dict_asset[OUTFLOW_DIRECTION]
 
@@ -826,6 +883,7 @@ def transformer_constant_efficiency_optimize(model, dict_asset, **kwargs):
                 kwargs[OEMOF_BUSSES][bus]: solph.Flow(
                     investment=investment if bus == investment_bus else None,
                     variable_costs=dict_asset[DISPATCH_PRICE][VALUE],
+                    **min_load_opts,
                 )
             }
 
@@ -842,7 +900,6 @@ def transformer_constant_efficiency_optimize(model, dict_asset, **kwargs):
             outputs=outputs,
             conversion_factors=efficiencies,
         )
-
         model.add(t)
         kwargs[OEMOF_TRANSFORMER].update({dict_asset[LABEL]: t})
 
@@ -1312,6 +1369,86 @@ def sink_non_dispatchable(model, dict_asset, **kwargs):
     )
 
 
+def sink_demand_reduction(model, dict_asset, **kwargs):
+    r"""
+    Defines a non dispatchable sink to serve critical and non-critical demand.
+
+    See :py:func:`~.sink` for more information, including parameters.
+
+    Notes
+    -----
+    Tested with:
+    - test_sink_non_dispatchable_single_input_bus()
+    - test_sink_non_dispatchable_multiple_input_busses()
+
+    Returns
+    -------
+    Indirectly updated `model` and dict of asset in `kwargs` with the sink object.
+
+    """
+    demand_reduction_factor = 1 - dict_asset[EFFICIENCY][VALUE]
+    tot_demand = dict_asset[TIMESERIES]
+    non_critical_demand_ts = tot_demand * demand_reduction_factor
+    non_critical_demand_peak = non_critical_demand_ts.max()
+    if non_critical_demand_peak == 0:
+        max_non_critical = 1
+    else:
+        max_non_critical = non_critical_demand_ts / non_critical_demand_peak
+    critical_demand_ts = tot_demand * dict_asset[EFFICIENCY][VALUE]
+
+    # check if the sink has multiple input busses
+    if isinstance(dict_asset[INFLOW_DIRECTION], list):
+        raise (
+            ValueError(
+                f"The reducable demand {dict_asset[LABEL]} does not support multiple input busses"
+            )
+        )
+        # inputs_noncritical = {}
+        # inputs_critical = {}
+        # index = 0
+        # for bus in dict_asset[INFLOW_DIRECTION]:
+        #     inputs_critical[kwargs[OEMOF_BUSSES][bus]] = solph.Flow(
+        #         fix=dict_asset[TIMESERIES], nominal_value=1
+        #     )
+        #     index += 1
+    else:
+        inputs_noncritical = {
+            kwargs[OEMOF_BUSSES][dict_asset[INFLOW_DIRECTION]]: solph.Flow(
+                min=0,
+                max=max_non_critical,
+                nominal_value=non_critical_demand_peak,
+                variable_costs=-1e-15,
+            )
+        }
+        inputs_critical = {
+            kwargs[OEMOF_BUSSES][dict_asset[INFLOW_DIRECTION]]: solph.Flow(
+                fix=critical_demand_ts, nominal_value=1
+            )
+        }
+
+    non_critical_demand = solph.components.Sink(
+        label=reducable_demand_name(dict_asset[LABEL]), inputs=inputs_noncritical,
+    )
+    critical_demand = solph.components.Sink(
+        label=reducable_demand_name(dict_asset[LABEL], critical=True),
+        inputs=inputs_critical,
+    )
+
+    # create and add demand sink and critical demand sink
+
+    model.add(critical_demand)
+    model.add(non_critical_demand)
+    kwargs[OEMOF_SINK].update(
+        {reducable_demand_name(dict_asset[LABEL]): non_critical_demand}
+    )
+    kwargs[OEMOF_SINK].update(
+        {reducable_demand_name(dict_asset[LABEL], critical=True): critical_demand}
+    )
+    logging.debug(
+        f"Added: Reducable Non-dispatchable sink {dict_asset[LABEL]} to bus {dict_asset[INFLOW_DIRECTION]}"
+    )
+
+
 def chp_fix(model, dict_asset, **kwargs):
     r"""
     Extraction turbine chp from Oemof solph. Extraction turbine must have one input and two outputs

src/multi_vector_simulator/E1_process_results.py

@@ -12,7 +12,7 @@
 import logging
 import copy
 import pandas as pd
-
+from multi_vector_simulator.utils.helpers import reducable_demand_name
 from multi_vector_simulator.utils.constants import TYPE_NONE, TOTAL_FLOW
 from multi_vector_simulator.utils.constants_json_strings import (
     ECONOMIC_DATA,
@@ -70,6 +70,7 @@
     FIX_COST,
     LIFETIME_PRICE_DISPATCH,
     AVERAGE_SOC,
+    TYPE_ASSET,
 )
 
 # Oemof.solph variables
@@ -734,8 +735,21 @@ def get_flow(settings, bus, dict_asset, flow_tuple, multi_bus=None):
     the flow ('average_flow').
 
     """
-    flow = bus[OEMOF_SEQUENCES][(flow_tuple, OEMOF_FLOW)]
-    flow = cut_below_micro(flow, dict_asset[LABEL] + FLOW)
+
+    if dict_asset.get(TYPE_ASSET) == "reducable_demand":
+        flow_tuple = (flow_tuple[0], reducable_demand_name(dict_asset[LABEL]))
+        flow = bus[OEMOF_SEQUENCES][(flow_tuple, OEMOF_FLOW)]
+        flow = cut_below_micro(flow, dict_asset[LABEL] + FLOW)
+        flow_tuple = (flow_tuple[0], reducable_demand_name(dict_asset[LABEL], critical=True))
+
+        flow_crit = bus[OEMOF_SEQUENCES][(flow_tuple, OEMOF_FLOW)]
+        flow_crit = cut_below_micro(flow_crit, dict_asset[LABEL] + FLOW)
+        flow = flow + flow_crit
+
+    else:
+        flow = bus[OEMOF_SEQUENCES][(flow_tuple, OEMOF_FLOW)]
+        flow = cut_below_micro(flow, dict_asset[LABEL] + FLOW)
+
     add_info_flows(
         evaluated_period=settings[EVALUATED_PERIOD][VALUE],
         dict_asset=dict_asset,

src/multi_vector_simulator/server.py

@@ -160,7 +160,9 @@ def __init__(self, results, busses_info=None, asset_types=None):
             ts_index = pd.to_datetime(js["index"][:-1], unit="ms")
             investments = df.iloc[-1]
             ts_df.index = ts_index
-
+        for extra_var in ["status", "status_nominal"]:
+            if extra_var in ts_df:
+                ts_df.drop(extra_var, axis=1, inplace=True)
         super().__init__(
             data=ts_df.T.to_dict(orient="split")["data"],
             index=mindex,
@@ -189,7 +191,7 @@ def asset_optimized_capacity(self, asset_name):
         return optimized_capacity
 
 
-def run_simulation(json_dict, epa_format=True, **kwargs):
+def run_simulation(json_dict, epa_format=True, verbatim=False, **kwargs):
     r"""
      Starts MVS tool simulation from an input json file
 
@@ -309,7 +311,9 @@ def run_simulation(json_dict, epa_format=True, **kwargs):
     logging.debug("Convert results to json")
 
     if epa_format is True:
-        epa_dict_values = data_parser.convert_mvs_params_to_epa(dict_values)
+        epa_dict_values = data_parser.convert_mvs_params_to_epa(
+            dict_values, verbatim=verbatim
+        )
 
         json_values = F0.store_as_json(epa_dict_values)
         answer = json.loads(json_values)

src/multi_vector_simulator/utils/constants_json_strings.py

@@ -192,6 +192,10 @@
 )
 CONNECTED_FEEDIN_SINK = "connected_feedin_sink"
 
+SUFFIX_CRITICAL = "critical"
+SUFFIX_NONCRITICAL = "noncritical"
+REDUCABLE_DEMAND = "reducable_demand"
+
 # Autogenerated assets
 DISPATCHABILITY = "dispatchable"
 AVAILABILITY_DISPATCH = "availability_timeseries"

src/multi_vector_simulator/utils/helpers.py

@@ -8,6 +8,7 @@
 - find_valvue_by_key(): Finds value of a key in a nested dictionary.
 """
 
+from copy import deepcopy
 import os
 
 from multi_vector_simulator.utils.constants_json_strings import (
@@ -28,6 +29,9 @@
     INFLOW_DIRECTION,
     OUTFLOW_DIRECTION,
     ENERGY_VECTOR,
+    SUFFIX_CRITICAL,
+    SUFFIX_NONCRITICAL,
+    REDUCABLE_DEMAND,
 )
 
 
@@ -120,6 +124,17 @@ def get_length_if_list(list_or_float):
     return answer
 
 
+def reducable_demand_name(demand_name: str, critical: bool = False):
+    """Name for auto created bus related to peak demand pricing period"""
+
+    if critical is False:
+        suffix = SUFFIX_NONCRITICAL
+    else:
+        suffix = SUFFIX_CRITICAL
+
+    return f"{demand_name}_{suffix} {AUTO_CREATED_HIGHLIGHT}"
+
+
 def peak_demand_bus_name(dso_name: str, feedin: bool = False):
     """Name for auto created bus related to peak demand pricing period"""
 
@@ -182,5 +197,16 @@ def get_asset_types(dict_values):
             for bus in input_bus + output_bus:
                 asset_busses[bus] = dict_values[ENERGY_BUSSES][bus].get(ENERGY_VECTOR)
             asset_type["busses"] = asset_busses
-            asset_types.append(asset_type)
+            if asset_type[TYPE_ASSET] == REDUCABLE_DEMAND:
+
+                asset_label = asset_type["label"]
+                asset_type["label"] = reducable_demand_name(asset_label)
+                asset_types.append(asset_type)
+                crit_asset_type = deepcopy(asset_type)
+                crit_asset_type["label"] = reducable_demand_name(
+                    asset_label, critical=True
+                )
+                asset_types.append(crit_asset_type)
+            else:
+                asset_types.append(asset_type)
     return asset_types