Explicit demand response peak load

assume/common/forecasts.py

@@ -89,21 +89,23 @@ class CsvForecaster(Forecaster):
     """
     This class represents a forecaster that provides timeseries for forecasts derived from existing files.
 
-    It initializes with the provided index. It includes methods to retrieve forecasts for specific columns,
-    availability of units, and prices of fuel types, returning the corresponding timeseries as pandas Series.
+    It initializes with the provided index and configuration data, including power plants, demand units,
+    and market configurations. The forecaster also supports optional inputs like DSM (demand-side management) units,
+    buses, and lines for more advanced forecasting scenarios.
 
-    Attributes:
-        index (pandas.Series): The index of the forecasts.
-        powerplants_units (dict[str, pandas.Series]): The power plants.
+    Methods are included to retrieve forecasts for specific columns, availability of units,
+    and prices of fuel types, returning the corresponding timeseries as pandas Series.
 
     Args:
-        index (pandas.Series): The index of the forecasts.
-        powerplants_units (dict[str, pandas.Series]): The power plants.
-
-    Example:
-        >>> forecaster = CsvForecaster(index=pd.Series([1, 2, 3]))
-        >>> forecast = forecaster['temperature']
-        >>> print(forecast)
+        index (pd.Series): The index of the forecasts.
+        powerplants_units (pd.DataFrame): A DataFrame containing information about power plants.
+        demand_units (pd.DataFrame): A DataFrame with demand unit data.
+        market_configs (dict[str, dict]): Configuration details for the markets.
+        buses (pd.DataFrame | None, optional): A DataFrame of buses information. Defaults to None.
+        lines (pd.DataFrame | None, optional): A DataFrame of line information. Defaults to None.
+        save_path (str, optional): Path where the forecasts should be saved. Defaults to an empty string.
+        *args (object): Additional positional arguments.
+        **kwargs (object): Additional keyword arguments.
 
     """
 
@@ -112,16 +114,21 @@ def __init__(
         index: pd.Series,
         powerplants_units: pd.DataFrame,
         demand_units: pd.DataFrame,
-        market_configs: dict = {},
+        market_configs: dict[str, dict],
+        buses: pd.DataFrame | None = None,
+        lines: pd.DataFrame | None = None,
         save_path: str = "",
-        *args,
-        **kwargs,
+        *args: object,
+        **kwargs: object,
     ):
         super().__init__(index, *args, **kwargs)
         self.logger = logging.getLogger(__name__)
         self.powerplants_units = powerplants_units
         self.demand_units = demand_units
         self.market_configs = market_configs
+        self.buses = buses
+        self.lines = lines
+
         self.forecasts = pd.DataFrame(index=index)
         self.save_path = save_path
 
@@ -191,24 +198,51 @@ def calc_forecast_if_needed(self):
         """
         Calculates the forecasts if they are not already calculated.
 
-        This method calculates price forecast and residual load forecast for available markets, if
-        these don't already exist.
+        This method calculates price forecast and residual load forecast for available markets,
+        and other necessary forecasts if they don't already exist.
         """
+        self.add_missing_availability_columns()
+        self.calculate_market_forecasts()
+
+        # the following forecasts are only calculated if buses and lines are available
+        # and self.demand_units have a node column
+        if self.buses and self.lines:
+            # check if the demand_units have a node column and
+            # if the nodes are available in the buses
+            if (
+                "node" in self.demand_units.columns
+                and self.demand_units["node"].isin(self.buses.index).all()
+            ):
+                self.add_node_congestion_signals()
+                self.add_utilisation_forecasts()
+            else:
+                self.logger.warning(
+                    "Node-specific congestion signals and renewable utilisation forecasts could not be calculated. "
+                    "Either 'node' column is missing in demand_units or nodes are not available in buses."
+                )
 
-        cols = []
-        for pp in self.powerplants_units.index:
-            col = f"availability_{pp}"
-            if col not in self.forecasts.columns:
-                s = pd.Series(1, index=self.forecasts.index)
-                s.name = col
-                cols.append(s)
-        cols.append(self.forecasts)
-        self.forecasts = pd.concat(cols, axis=1).copy()
+    def add_missing_availability_columns(self):
+        """Add missing availability columns to the forecasts."""
+        missing_cols = [
+            f"availability_{pp}"
+            for pp in self.powerplants_units.index
+            if f"availability_{pp}" not in self.forecasts.columns
+        ]
+
+        if missing_cols:
+            # Create a DataFrame with the missing columns initialized to 1
+            missing_data = pd.DataFrame(
+                1, index=self.forecasts.index, columns=missing_cols
+            )
+            # Append the missing columns to the forecasts
+            self.forecasts = pd.concat([self.forecasts, missing_data], axis=1).copy()
 
+    def calculate_market_forecasts(self):
+        """Calculate market-specific price and residual load forecasts."""
         for market_id, config in self.market_configs.items():
             if config["product_type"] != "energy":
                 self.logger.warning(
-                    f"Price forecast could be calculated for {market_id}. It can only be calculated for energy only markets for now"
+                    f"Price forecast could not be calculated for {market_id}. It can only be calculated for energy-only markets for now."
                 )
                 continue
 
@@ -222,6 +256,29 @@ def calc_forecast_if_needed(self):
                     self.calculate_residual_load_forecast(market_id=market_id)
                 )
 
+    def add_node_congestion_signals(self):
+        """Add node-specific congestion signals to the forecasts."""
+        node_congestion_signal_df = self.calculate_node_specific_congestion_forecast()
+        for col in node_congestion_signal_df.columns:
+            if col not in self.forecasts.columns:
+                self.forecasts[col] = node_congestion_signal_df[col]
+
+    def add_utilisation_forecasts(self):
+        """Add renewable utilisation forecasts if missing."""
+        utilisation_columns = [
+            f"{node}_renewable_utilisation"
+            for node in self.demand_units["node"].unique()
+        ]
+        utilisation_columns.append("all_nodes_renewable_utilisation")
+
+        if not all(col in self.forecasts.columns for col in utilisation_columns):
+            renewable_utilisation_forecast = (
+                self.calculate_renewable_utilisation_forecast()
+            )
+            for col in renewable_utilisation_forecast.columns:
+                if col not in self.forecasts.columns:
+                    self.forecasts[col] = renewable_utilisation_forecast[col]
+
     def get_registered_market_participants(self, market_id):
         """
         Retrieves information about market participants to make accurate price forecasts.
@@ -299,7 +356,7 @@ def calculate_market_price_forecast(self, market_id):
 
         """
 
-        # calculate infeed of renewables and residual demand_df
+        # calculate infeed of renewables and residual demand
         # check if max_power is a series or a float
 
         # select only those power plant units, which have a bidding strategy for the specific market_id
@@ -380,6 +437,131 @@ def calculate_marginal_cost(self, pp_series: pd.Series) -> pd.Series:
 
         return marginal_cost
 
+    def calculate_node_specific_congestion_forecast(self) -> pd.DataFrame:
+        """
+        Calculates a collective node-specific congestion signal by aggregating the congestion severity of all
+        transmission lines connected to each node, taking into account powerplant load based on availability factors.
+
+        Returns:
+            pd.DataFrame: A DataFrame with columns for each node, where each column represents the collective
+                          congestion signal time series for that node.
+        """
+        # Step 1: Calculate powerplant load using availability factors
+        availability_factor_df = pd.DataFrame(
+            index=self.index, columns=self.powerplants_units.index, data=0.0
+        )
+
+        # Calculate load for each powerplant based on availability factor and max power
+        for pp, max_power in self.powerplants_units["max_power"].items():
+            availability_factor_df[pp] = (
+                self.forecasts[f"availability_{pp}"] * max_power
+            )
+
+        # Step 2: Calculate net load for each node (demand - generation)
+        net_load_by_node = {}
+
+        for node in self.demand_units["node"].unique():
+            # Calculate total demand for this node
+            node_demand_units = self.demand_units[
+                self.demand_units["node"] == node
+            ].index
+            node_demand = self.forecasts[node_demand_units].sum(axis=1)
+
+            # Calculate total generation for this node by summing powerplant loads
+            node_generation_units = self.powerplants_units[
+                self.powerplants_units["node"] == node
+            ].index
+            node_generation = availability_factor_df[node_generation_units].sum(axis=1)
+
+            # Calculate net load (demand - generation)
+            net_load_by_node[node] = node_demand - node_generation
+
+        # Step 3: Calculate line-specific congestion severity
+        line_congestion_severity = pd.DataFrame(index=self.index)
+
+        for line_id, line_data in self.lines.iterrows():
+            node1, node2 = line_data["bus0"], line_data["bus1"]
+            line_capacity = line_data["s_nom"]
+
+            # Calculate net load for the line as the sum of net loads from both connected nodes
+            line_net_load = net_load_by_node[node1] + net_load_by_node[node2]
+            congestion_severity = line_net_load / line_capacity
+
+            # Store the line-specific congestion severity in DataFrame
+            line_congestion_severity[f"{line_id}_congestion_severity"] = (
+                congestion_severity
+            )
+
+        # Step 4: Calculate node-specific congestion signal by aggregating connected lines
+        node_congestion_signal = pd.DataFrame(index=self.index)
+
+        for node in self.demand_units["node"].unique():
+            # Find all lines connected to this node
+            connected_lines = self.lines[
+                (self.lines["bus0"] == node) | (self.lines["bus1"] == node)
+            ].index
+
+            # Collect all relevant line congestion severities
+            relevant_lines = [
+                f"{line_id}_congestion_severity" for line_id in connected_lines
+            ]
+
+            # Ensure only existing columns are used to avoid KeyError
+            relevant_lines = [
+                line
+                for line in relevant_lines
+                if line in line_congestion_severity.columns
+            ]
+
+            # Aggregate congestion severities for this node (use max or mean)
+            if relevant_lines:
+                node_congestion_signal[f"{node}_congestion_severity"] = (
+                    line_congestion_severity[relevant_lines].max(axis=1)
+                )
+
+        return node_congestion_signal
+
+    def calculate_renewable_utilisation_forecast(self) -> pd.DataFrame:
+        """
+        Calculates the renewable utilisation forecast by summing the available renewable generation
+        for each node and an overall 'all_nodes' summary.
+
+        Returns:
+            pd.DataFrame: A DataFrame with columns for each node, where each column represents the renewable
+                        utilisation signal time series for that node and a column for total utilisation across all nodes.
+        """
+        # Initialize a DataFrame to store renewable utilisation for each node
+        renewable_utilisation = pd.DataFrame(index=self.index)
+
+        # Identify renewable power plants by filtering `powerplants_units` DataFrame
+        renewable_plants = self.powerplants_units[
+            self.powerplants_units["fuel_type"] == "renewable"
+        ]
+
+        # Calculate utilisation based on availability and max power for each renewable plant
+        for node in self.demand_units["node"].unique():
+            node_renewable_sum = pd.Series(0, index=self.index)
+
+            # Filter renewable plants in this specific node
+            node_renewable_plants = renewable_plants[renewable_plants["node"] == node]
+
+            for pp in node_renewable_plants.index:
+                max_power = node_renewable_plants.loc[pp, "max_power"]
+                availability_col = f"availability_{pp}"
+
+                # Calculate renewable power based on availability and max capacity
+                if availability_col in self.forecasts.columns:
+                    node_renewable_sum += self.forecasts[availability_col] * max_power
+
+            # Store the node-specific renewable utilisation
+            renewable_utilisation[f"{node}_renewable_utilisation"] = node_renewable_sum
+
+        # Calculate the total renewable utilisation across all nodes
+        all_nodes_sum = renewable_utilisation.sum(axis=1)
+        renewable_utilisation["all_nodes_renewable_utilisation"] = all_nodes_sum
+
+        return renewable_utilisation
+
     def save_forecasts(self, path=None):
         """
         Saves the forecasts to a csv file located at the specified path.

assume/scenario/loader_csv.py

@@ -163,6 +163,8 @@ def load_dsm_units(
         "unit_type",
         "node",
         "flexibility_measure",
+        "congestion_threshold",
+        "peak_load_cap",
     ]
     # Filter the common columns to only include those that exist in the DataFrame
     common_columns = [col for col in common_columns if col in dsm_units.columns]
@@ -497,11 +499,16 @@ def load_config_and_create_forecaster(
         path=path, config=config, file_name="temperature", index=index
     )
 
+    buses = load_file(path=path, config=config, file_name="buses")
+    lines = load_file(path=path, config=config, file_name="lines")
+
     forecaster = CsvForecaster(
         index=index,
         powerplants_units=powerplant_units,
         demand_units=demand_units,
         market_configs=config["markets_config"],
+        buses=buses,
+        lines=lines,
     )
 
     forecaster.set_forecast(forecasts_df)
@@ -526,6 +533,8 @@ def load_config_and_create_forecaster(
         "storage_units": storage_units,
         "demand_units": demand_units,
         "dsm_units": dsm_units,
+        "buses": buses,
+        "lines": lines,
         "forecaster": forecaster,
     }
 

assume/strategies/naive_strategies.py

@@ -156,7 +156,7 @@ def calculate_bids(
         **kwargs,
     ) -> Orderbook:
         # calculate the optimal operation of the unit
-        unit.calculate_optimal_operation_if_needed()
+        unit.determine_optimal_operation_with_flex()
 
         bids = []
         for product in product_tuples:
@@ -167,7 +167,7 @@ def calculate_bids(
             start = product[0]
 
             volume = unit.opt_power_requirement.at[start]
-            marginal_price = unit.calculate_marginal_cost(start, volume)
+            marginal_price = 3000
             bids.append(
                 {
                     "start_time": start,
@@ -195,7 +195,7 @@ def calculate_bids(
         **kwargs,
     ) -> Orderbook:
         # calculate the optimal operation of the unit according to the objective function
-        unit.calculate_optimal_operation_if_needed()
+        unit.determine_optimal_operation_with_flex()
 
         bids = []
         for product in product_tuples: