Soc renaming (#709)

## Related Issue
Closes #525 

## Description
This pull request standardizes the handling of energy storage units
across the codebase by consistently using a normalized state of charge
(`soc` between 0 and 1) and a new `capacity` field (in MWh) instead of
the previously used `max_soc`. It updates storage models, strategies,
and scenario loaders to use these new conventions, ensuring calculations
and constraints are based on normalized values and absolute capacity.
This improves clarity, correctness, and interoperability between
different modules.

**Core data model updates:**
* Replaced `max_soc` with `capacity` in the `SupportsMinMaxCharge` class
and related storage classes, and updated all references accordingly. The
`soc` is now always a float between 0 and 1, and all calculations are
adjusted to use `capacity` for absolute values.
* Updated docstrings and function arguments to clarify that `soc` is
normalized and to document the new `capacity` field.

**Calculation and logic changes:**
* Adjusted all calculations involving state of charge,
charging/discharging, and energy deltas to use normalized `soc` and
scale by `capacity`. This affects dispatch planning, bid calculations,
and reward functions.
* Updated optimization models and constraints to use `capacity` and
normalized `soc`, including Pyomo model bounds and initializations.

**Scenario and loader updates:**
* Modified all scenario loaders (`loader_amiris.py`, `loader_csv.py`,
`loader_oeds.py`, `loader_pypsa.py`, `oeds/infrastructure.py`) to output
`capacity` and normalized `initial_soc`, and to require the `capacity`
field in input data.

These changes make the codebase more robust and less error-prone by
enforcing a clear separation between normalized and absolute values for
storage units.

## Checklist
- [x] Documentation updated (docstrings, READMEs, user guides, inline
comments, `doc` folder updates etc.)
- New unit/integration tests added (not applicable)
- [x] Changes noted in release notes (if any)
- [x] Consent to release this PR's code under the GNU Affero General
Public License v3.0

---------

Co-authored-by: mthede <[email protected]>
Co-authored-by: kim-mskw <[email protected]>

Author: 3 people

assume/common/base.py

@@ -473,6 +473,7 @@ class SupportsMinMaxCharge(BaseUnit):
     """
 
     initial_soc: float
+    # float between 0 and 1 - initial state of charge
     min_power_charge: float
     # negative float - if this storage is charging, what is the minimum charging power (the negative non-zero power closest to zero) (resulting in negative current power)
     max_power_charge: float
@@ -489,7 +490,7 @@ class SupportsMinMaxCharge(BaseUnit):
     # negative
     ramp_down_charge: float | None
     # ramp_down_charge is negative
-    max_soc: float
+    capacity: float
     efficiency_charge: float
     efficiency_discharge: float
 
@@ -509,7 +510,7 @@ def calculate_min_max_charge(
         Args:
             start (datetime.datetime): The start time of the dispatch.
             end (datetime.datetime): The end time of the dispatch.
-            soc (float, optional): The current state-of-charge. Defaults to None.
+            soc (float, optional): The current state-of-charge (between 0 and 1). Defaults to None.
 
         Returns:
             tuple[np.ndarray, np.ndarray]: The min and max charging power for the given time period.
@@ -524,7 +525,7 @@ def calculate_min_max_discharge(
         Args:
             start (datetime.datetime): The start time of the dispatch.
             end (datetime.datetime): The end time of the dispatch.
-            soc (float, optional): The current state-of-charge. Defaults to None.
+            soc (float, optional): The current state-of-charge (between 0 and 1). Defaults to None.
 
         Returns:
             tuple[np.ndarray, np.ndarray]: The min and max discharging power for the given time period.
@@ -660,7 +661,9 @@ def set_dispatch_plan(
                 if current_power > max_soc_discharge:
                     current_power = max_soc_discharge
 
-                delta_soc = -current_power * time_delta / self.efficiency_discharge
+                delta_soc = (
+                    -current_power * time_delta / self.efficiency_discharge
+                ) / self.capacity
 
             # charging
             elif current_power < 0:
@@ -669,7 +672,9 @@ def set_dispatch_plan(
                 if current_power < max_soc_charge:
                     current_power = max_soc_charge
 
-                delta_soc = -current_power * time_delta * self.efficiency_charge
+                delta_soc = (
+                    -current_power * time_delta * self.efficiency_charge
+                ) / self.capacity
 
             # update the values of the state of charge and the energy
             self.outputs["soc"].at[next_t] = soc + delta_soc

assume/scenario/loader_amiris.py

@@ -285,8 +285,8 @@ def add_agent_to_world(
                 market_prices={"eom": forecast_price},
             )
 
-            max_soc = device["EnergyToPowerRatio"] * device["InstalledPowerInMW"]
-            initial_soc = device["InitialEnergyLevelInMWH"]
+            capacity = device["EnergyToPowerRatio"] * device["InstalledPowerInMW"]
+            initial_soc = device["InitialEnergyLevelInMWH"] / capacity
             # TODO device["SelfDischargeRatePerHour"]
             world.add_unit(
                 f"StorageTrader_{agent['Id']}",
@@ -298,7 +298,7 @@ def add_agent_to_world(
                     "efficiency_charge": device["ChargingEfficiency"],
                     "efficiency_discharge": device["DischargingEfficiency"],
                     "initial_soc": initial_soc,
-                    "max_soc": max_soc,
+                    "capacity": capacity,
                     "bidding_strategies": storage_strategies,
                     "technology": "hydro",  # PSPP? Pump-Storage Power Plant
                     "emission_factor": 0,

assume/scenario/loader_csv.py

@@ -515,6 +515,8 @@ def load_config_and_create_forecaster(
             storage_units["max_power_charge"] = -abs(storage_units["max_power_charge"])
         if "min_power_charge" in storage_units.columns:
             storage_units["min_power_charge"] = -abs(storage_units["min_power_charge"])
+        if "capacity" not in storage_units.columns:
+            raise ValueError("No capacity column provided for storage units!")
 
     # Initialize an empty dictionary to combine the DSM units
     dsm_units = {}

assume/scenario/loader_oeds.py

@@ -302,8 +302,9 @@ def load_oeds(
                     {
                         "max_power_charge": storage["max_power_charge"] / 1e3,
                         "max_power_discharge": storage["max_power_discharge"] / 1e3,
-                        "max_soc": storage["max_soc"] / 1e3,
-                        "min_soc": storage["min_soc"] / 1e3,
+                        "capacity": storage["capacity"] / 1e3,
+                        "max_soc": storage["max_soc"],
+                        "min_soc": storage["min_soc"],
                         "efficiency_charge": storage["efficiency_charge"],
                         "efficiency_discharge": storage["efficiency_discharge"],
                         "bidding_strategies": bidding_strategies["storage"],

assume/scenario/loader_pypsa.py

@@ -159,10 +159,10 @@ def load_pypsa(
                 "efficiency_charge": storage.efficiency_store,
                 "efficiency_discharge": storage.efficiency_dispatch,
                 "initial_soc": storage.state_of_charge_initial,
-                "max_soc": storage.p_nom,
+                "capacity": storage.p_nom * storage.max_hours,
                 "bidding_strategies": bidding_strategies[unit_type][storage.name],
-                "technology": "hydro",
-                "emission_factor": 0,
+                "technology": storage.carrier,
+                "emission_factor": storage.emission_factor or 0,
                 "node": storage.bus,
             },
             UnitForecaster(index),

assume/scenario/oeds/infrastructure.py

@@ -676,16 +676,18 @@ def get_water_storage_systems(
                 "startDate": pd.to_datetime(data["startDate"].to_numpy()[0]),
                 "max_power_discharge": data["PMinus_max"].sum(),
                 "max_power_charge": -data["PPlus_max"].sum(),
-                "max_soc": data["VMax"].to_numpy()[0],
+                "capacity": data["VMax"].to_numpy()[0],
+                "max_soc": 1,
                 "min_soc": 0,
-                "V0": data["VMax"].to_numpy()[0] / 2,
+                "initial_soc": 0.5,
+                "V0": 0.5 * data["VMax"].to_numpy()[0],
                 "lat": data["lat"].to_numpy()[0],
                 "lon": data["lon"].to_numpy()[0],
                 "efficiency_charge": 0.88,
                 "efficiency_discharge": 0.92,
             }
             # https://energie.ch/pumpspeicherkraftwerk/
-            if storage["max_soc"] > 0:
+            if storage["capacity"] > 0:
                 storages.append(storage)
         return storages
 

assume/strategies/dmas_storage.py

@@ -139,7 +139,7 @@ def build_model(
             time_range, within=pyo.Reals, bounds=(0, unit.max_power_discharge)
         )
         self.model.volume = pyo.Var(
-            time_range, within=pyo.NonNegativeReals, bounds=(0, unit.max_soc)
+            time_range, within=pyo.NonNegativeReals, bounds=(0, unit.capacity)
         )
 
         self.power = np.array(
@@ -151,7 +151,7 @@ def build_model(
         )
 
         self.model.vol_con = pyo.ConstraintList()
-        v0 = unit.outputs["soc"].at[start]
+        v0 = unit.outputs["soc"].at[start] * unit.capacity
 
         for t in time_range:
             if t == 0:
@@ -162,7 +162,7 @@ def build_model(
                 )
 
         # always end with half full SoC
-        self.model.vol_con.add(self.model.volume[hour_count - 1] == unit.max_soc / 2)
+        self.model.vol_con.add(self.model.volume[hour_count - 1] == unit.capacity / 2)
         return self.power
 
     def optimize(

assume/strategies/flexable.py

@@ -503,7 +503,12 @@ def calculate_EOM_price_if_off(
     # if we split starting_cost across av_operating_time
     # we are never adding the other parts of the cost to the following hours
 
-    markup = starting_cost / avg_operating_time / bid_quantity_inflex
+    # if unit never operated before and min_operating_time is 0, set avg_operating_time is considered to be 1 and hence neglected to avoid division by zero
+    # this lets the power plant only start if it can recover the starting costs in the first hour, which is quite restrictive
+    if avg_operating_time == 0:
+        markup = starting_cost / bid_quantity_inflex
+    else:
+        markup = starting_cost / avg_operating_time / bid_quantity_inflex
 
     bid_price_inflex = min(marginal_cost_inflex + markup, 3000.0)
 
@@ -546,7 +551,13 @@ def calculate_EOM_price_if_on(
     # check the starting cost if the unit were turned off for min_down_time
     starting_cost = unit.get_starting_costs(-unit.min_down_time)
 
-    price_reduction_restart = starting_cost / unit.min_down_time / bid_quantity_inflex
+    # disregard unit.min_down_time of 0 to avoid division by zero
+    if unit.min_down_time == 0:
+        price_reduction_restart = starting_cost / bid_quantity_inflex
+    else:
+        price_reduction_restart = (
+            starting_cost / unit.min_down_time / bid_quantity_inflex
+        )
 
     if unit.outputs["heat"].at[start] > 0:
         heat_gen_cost = (

assume/strategies/flexable_storage.py

@@ -152,14 +152,22 @@ def calculate_bids(
             # calculate theoretic SOC
             time_delta = (end - start) / timedelta(hours=1)
             if bid_quantity + current_power > 0:
-                delta_soc = -(
-                    (bid_quantity + current_power)
-                    * time_delta
-                    / unit.efficiency_discharge
+                delta_soc = (
+                    -(
+                        (bid_quantity + current_power)
+                        * time_delta
+                        / unit.efficiency_discharge
+                    )
+                    / unit.capacity
                 )
             elif bid_quantity + current_power < 0:
-                delta_soc = -(
-                    (bid_quantity + current_power) * time_delta * unit.efficiency_charge
+                delta_soc = (
+                    -(
+                        (bid_quantity + current_power)
+                        * time_delta
+                        * unit.efficiency_charge
+                    )
+                    / unit.capacity
                 )
             else:
                 delta_soc = 0
@@ -329,10 +337,13 @@ def calculate_bids(
                 )
                 # calculate theoretic SOC
                 time_delta = (end - start) / timedelta(hours=1)
-                delta_soc = -(
-                    (bid_quantity + current_power)
-                    * time_delta
-                    / unit.efficiency_discharge
+                delta_soc = (
+                    -(
+                        (bid_quantity + current_power)
+                        * time_delta
+                        / unit.efficiency_discharge
+                    )
+                    / unit.capacity
                 )
                 theoretic_SOC += delta_soc
                 previous_power = bid_quantity + current_power
@@ -440,7 +451,7 @@ def calculate_bids(
                 time_delta = (end - start) / timedelta(hours=1)
                 delta_soc = (
                     (bid_quantity + current_power) * time_delta * unit.efficiency_charge
-                )
+                ) / unit.capacity
                 theoretic_SOC += delta_soc
                 previous_power = bid_quantity + current_power
             else:

assume/strategies/learning_strategies.py

@@ -906,12 +906,12 @@ def get_individual_observations(
         the agent's action selection.
         """
         # get the current soc and energy cost value
-        soc_scaled = unit.outputs["soc"].at[start] / unit.max_soc
+        soc = unit.outputs["soc"].at[start]
         cost_stored_energy_scaled = (
             unit.outputs["cost_stored_energy"].at[start] / self.max_bid_price
         )
 
-        individual_observations = np.array([soc_scaled, cost_stored_energy_scaled])
+        individual_observations = np.array([soc, cost_stored_energy_scaled])
 
         return individual_observations
 
@@ -1069,15 +1069,17 @@ def calculate_reward(
 
         # Calculate and clip the energy cost for the start time
         # cost_stored_energy = average volume-weighted procurement costs of the currently stored energy
-        if next_soc < 1:
+        if next_soc * unit.capacity < 1:
             unit.outputs["cost_stored_energy"].at[next_time] = 0
         elif accepted_volume < 0:
             # increase costs of current SoC by price for buying energy
             # not fully representing the true cost per MWh (e.g. omitting discharge efficiency losses), but serving as a proxy for it
             unit.outputs["cost_stored_energy"].at[next_time] = (
-                unit.outputs["cost_stored_energy"].at[start] * current_soc
+                unit.outputs["cost_stored_energy"].at[start]
+                * current_soc
+                * unit.capacity
                 - (accepted_price + marginal_cost) * accepted_volume * duration_hours
-            ) / next_soc
+            ) / (next_soc * unit.capacity)
         else:
             unit.outputs["cost_stored_energy"].at[next_time] = unit.outputs[
                 "cost_stored_energy"