Merge 7044adf into 6b50a92

src/E1_process_results.py

@@ -1,19 +1,45 @@
+r"""
+Module E1 process results
+-------------------------
+Module E1 processes the oemof results.
+- receive time series per bus for all assets
+- write time series to dictionary
+- get optimal capacity of optimized assets
+- add the evaluation of time series
+
+"""
+
 import logging
 import pandas as pd
 
 
 def get_timeseries_per_bus(dict_values, bus_data):
-    """
-    Reads simulation results of all busses.
-    :param dict_values:
-    :param bus_data:
-    :return:
+    r"""
+    Reads simulation results of all busses and stores time series.
+
+    Parameters
+    ----------
+    dict_values : dict
+        Contains all input data of the simulation.
+    bus_data : dict
+        Contains information about all busses in a nested dict.
+        1st level keys: bus names;
+        2nd level keys:
+            'scalars': (pd.Series) (does not exist in all dicts)
+            'sequences': (pd.DataFrame) - contains flows between components and busses
+
+    Returns
+    -------
+    Indirectly updated `dict_values` with 'optimizedFlows' - one data frame for each bus.
+
     """
     bus_data_timeseries = {}
     for bus in bus_data.keys():
         bus_data_timeseries.update(
             {bus: pd.DataFrame(index=dict_values["simulation_settings"]["time_index"])}
         )
+
+        # obtain flows that flow into the bus
         to_bus = {
             key[0][0]: key
             for key in bus_data[bus]["sequences"].keys()
@@ -22,6 +48,7 @@ def get_timeseries_per_bus(dict_values, bus_data):
         for asset in to_bus:
             bus_data_timeseries[bus][asset] = bus_data[bus]["sequences"][to_bus[asset]]
 
+        # obtain flows that flow out of the bus
         from_bus = {
             key[0][1]: key
             for key in bus_data[bus]["sequences"].keys()
@@ -36,25 +63,28 @@ def get_timeseries_per_bus(dict_values, bus_data):
     return
 
 
-def write_bus_timeseries_to_dict_values(dict_asset):
-    """
-
-    :param dict_asset:
-    :return:
-    """
-    logging.debug(
-        "Accessing oemof simulation results for asset %s", dict_asset["label"]
-    )
-    return
-
-
 def get_storage_results(settings, storage_bus, dict_asset):
-    """
+    r"""
+    Reads storage results of simulation and stores them in `dict_asset`.
+
+    Parameters
+    ----------
+    settings : dict
+        Contains simulation settings from `simulation_settings.csv` with
+        additional information like the amount of time steps simulated in the
+        optimization ('periods').
+    storage_bus : dict
+        Contains information about the storage bus. Information about the scalars
+        like investment or initial capacity in key 'scalars' (pd.Series) and the
+        flows between the component and the busses in key 'sequences' (pd.DataFrame).
+    dict_asset : dict
+        Contains information about the storage like capacity, charging power, etc.
+
+    Returns
+    -------
+    Indirectly updates `dict_asset` with simulation results concerning the
+    storage.
 
-    :param settings:
-    :param storage_bus:
-    :param dict_asset:
-    :return:
     """
     power_charge = storage_bus["sequences"][
         ((dict_asset["input_bus_name"], dict_asset["label"]), "flow")
@@ -146,7 +176,7 @@ def get_storage_results(settings, storage_bus, dict_asset):
                 }
             )
 
-    dict_asset.update(
+    dict_asset.update(  # todo: this could be a separate function for testing.
         {
             "timeseries_soc": dict_asset["storage capacity"]["flow"]
             / (
@@ -159,12 +189,28 @@ def get_storage_results(settings, storage_bus, dict_asset):
 
 
 def get_results(settings, bus_data, dict_asset):
-    """
+    r"""
+    Reads results of the asset defined in `dict_asset` and stores them in `dict_asset`.
+
+    Parameters
+    ----------
+    settings : dict
+        Contains simulation settings from `simulation_settings.csv` with
+        additional information like the amount of time steps simulated in the
+        optimization ('periods').
+    bus_data : dict
+        Contains information about all busses in a nested dict.
+        1st level keys: bus names;
+        2nd level keys:
+            'scalars': (pd.Series) (does not exist in all dicts)
+            'sequences': (pd.DataFrame) - contains flows between components and busses
+    dict_asset : dict
+        Contains information about the asset.
+
+    Returns
+    -------
+    Indirectly updates `dict_asset` with results.
 
-    :param settings:
-    :param bus_data:
-    :param dict_asset:
-    :return:
     """
     # Check if the component has multiple input or output busses
     if "input_bus_name" in dict_asset:
@@ -220,13 +266,33 @@ def get_results(settings, bus_data, dict_asset):
 
 
 def get_optimal_cap(bus, dict_asset, bus_name, direction):
-    """
+    r"""
+    Retrieves optimized capacity of asset specified in `dict_asset`.
+
+    Parameters
+    ----------
+    bus : dict
+        Contains information about the busses linked to the asset specified in
+        `dict_asset`. Information about the scalars like investment or initial
+        capacity in key 'scalars' (pd.Series) and the flows between the
+        component and the busses in key 'sequences' (pd.DataFrame).
+    dict_asset : dict
+        Contains information about the asset.
+    bus_name : str
+        Name of `bus`.
+    direction : str
+        Direction of flow. Options: 'input', 'output'.
+
+    possible todos
+    --------------
+    * direction as optimal parameter or with default value None (direction is
+        not needed if 'optimizeCap' is not in `dict_asset` or if it's value is False
+
+    Returns
+    -------
+    Indirectly updated `dict_asset` with optimal capacity to be added
+    ('optimizedAddCap').
 
-    :param bus:
-    :param dict_asset:
-    :param bus_name:
-    :param direction:
-    :return:
     """
     if "optimizeCap" in dict_asset:
         if dict_asset["optimizeCap"]["value"] == True:
@@ -239,8 +305,8 @@ def get_optimal_cap(bus, dict_asset, bus_name, direction):
                     ((dict_asset["label"], bus_name), "invest")
                 ]
             else:
-                logging.error(
-                    "Function get_optimal_cap has invalid value of parameter direction."
+                raise ValueError(
+                    f"`direction` should be 'input' or 'output' but is {direction}."
                 )
 
             if "timeseries_peak" in dict_asset:
@@ -283,22 +349,44 @@ def get_optimal_cap(bus, dict_asset, bus_name, direction):
 
 
 def get_flow(settings, bus, dict_asset, bus_name, direction):
-    """
+    r"""
+    Adds flow of `bus` and total flow amongst other information to `dict_asset`.
+
+    Depending on `direction` the input or the output flow is used.
+
+    Parameters
+    ----------
+    settings : dict
+        Contains simulation settings from `simulation_settings.csv` with
+        additional information like the amount of time steps simulated in the
+        optimization ('periods').
+    bus : dict
+        Contains information about a specific bus. Information about the scalars, if they exist,
+            like investment or initial capacity in key 'scalars' (pd.Series) and the
+            flows between the component and the bus(ses) in key 'sequences' (pd.DataFrame).
+    dict_asset : dict
+        Contains information about the asset.
+    bus_name : str
+        Name of `bus`.
+    direction : str
+        Direction of flow. Options: 'input', 'output'.
+
+    Returns
+    -------
+    Indirectly updates `dict_asset` with the flow of `bus`, the total flow, the annual
+    total flow, the maximum of the flow ('peak_flow') and the average value of
+    the flow ('average_flow').
 
-    :param settings:
-    :param bus:
-    :param dict_asset:
-    :param bus_name:
-    :param direction:
-    :return:
     """
     if direction == "input":
         flow = bus["sequences"][((bus_name, dict_asset["label"]), "flow")]
     elif direction == "output":
         flow = bus["sequences"][((dict_asset["label"], bus_name), "flow")]
 
     else:
-        logging.warning('Value %s not "input" or "output"!', direction)
+        raise ValueError(
+            f"`direction` should be 'input' or 'output' but is {direction}."
+        )
     add_info_flows(settings, dict_asset, flow)
 
     logging.debug(
@@ -310,12 +398,26 @@ def get_flow(settings, bus, dict_asset, bus_name, direction):
 
 
 def add_info_flows(settings, dict_asset, flow):
-    """
+    r"""
+    Adds `flow` and total flow amongst other information to `dict_asset`.
+
+    Parameters
+    ----------
+    settings : dict
+        Contains simulation settings from `simulation_settings.csv` with
+        additional information like the amount of time steps simulated in the
+        optimization ('periods').
+    dict_asset : dict
+        Contains information about the asset `flow` belongs to.
+    flow : pd.Series
+        Time series of the flow.
+
+    Returns
+    -------
+    Indirectly updates `dict_asset` with the `flow`, the total flow, the annual
+    total flow, the maximum of the flow ('peak_flow') and the average value of
+    the flow ('average_flow').
 
-    :param settings:
-    :param dict_asset:
-    :param flow:
-    :return:
     """
     total_flow = sum(flow)
     dict_asset.update(

tests/test_E1_process_results.py

@@ -0,0 +1,109 @@
+import os
+import pytest
+
+# from .constants import JSON_PATH
+import src.E1_process_results as E1
+
+# Note: test functions might be summed up in classes..
+
+
+def test_get_timeseries_per_bus_one_bus():
+    pass
+    # check updated dict_values
+
+
+def test_get_timeseries_per_bus_three_busses():
+    pass
+
+
+def test_get_storage_results_optimize():
+    pass
+    # check dict_asset updated. updated are for all functions:
+    # charging_power (see add_info_flow,),
+    #  discharging_power (see add_info_flow,),
+    #  capacity, (see add_info_flow),
+    # timeseries_soc
+
+    # additionally only optimize is true. (false: 0 is added as optimizedAddCap)
+    #  optimizedAddCap of charging_power, discharging_power and capacity
+
+
+def test_get_storage_results_no_optimization():
+    pass
+    # NOTE: optimizedAddCap = 0 for no optimization
+
+
+def test_get_storage_results_optimizeCap_not_in_dict_asset():
+    pass
+    # optimizedAddCap not added to dict_asset
+
+
+def test_get_results_only_input_bus_single():
+    pass
+
+
+def test_get_results_only_output_bus_single():
+    pass
+
+
+def test_get_results_input_and_output_bus_single():
+    pass
+
+
+def test_get_results_multiple_input_busses():
+    pass
+
+
+def test_get_results_multiple_output_busses():
+    pass
+    # check if dict_asset is updated. see add_info_flows for keys
+    # check optimal capacity
+
+
+def test_get_optimal_cap_optimize_input_flow_timeseries_peak_provided():
+    pass
+
+
+def test_get_optimal_cap_optimize_input_flow_timeseries_peak_not_provided():
+    pass
+
+
+def test_get_optimal_cap_optimize_output_flow():
+    pass
+    # check dict_asset updated with optimizedAddCap
+
+
+def test_get_optimal_cap_optimize_invalid_direction_raises_value_error():
+    pass
+
+
+def test_get_optimal_cap_optimize_negative_time_series_peak_logging_warning():
+    pass
+
+
+def test_get_optimal_cap_optimize_time_series_peak_is_zero_logging_warning():
+    pass
+
+
+def test_get_opitmal_cap_no_optimization():
+    pass
+    # check dict_asset updated with optimizedAddCap = 0
+
+
+def test_get_optimal_cap_optimizeCap_not_in_dict_asset():
+    pass
+    # check that dict_asset did not change
+
+
+# NOTE: I decided to not test get_flow() and add_info_flow() as they are tested by other functions extensively.
+#       Please comment if you are of another opinion.
+
+# def test_get_flow_input():
+#     pass
+#
+# def test_get_flow_output():
+#     pass
+#
+# def test_get_flow_invalid_direction_raises_value_error():
+#     pass
+# same tests as add_info_flow() just that bus and direction is provided.