Reconcile retirement capacity

powerplantmatching/collection.py

@@ -14,7 +14,11 @@
 
 from .cleaning import aggregate_units
 from .core import _data_out, get_config
-from .heuristics import extend_by_non_matched, extend_by_VRE
+from .heuristics import (
+    extend_by_non_matched,
+    extend_by_VRE,
+    reconcile_retirement_capacity,
+)
 from .matching import combine_multiple_datasets, reduce_matched_dataframe
 from .utils import (
     parmap,
@@ -119,6 +123,8 @@ def powerplants(
     extend_by_kwargs=None,
     fill_geopositions=True,
     filter_missing_geopositions=True,
+    reconcile_retirements=True,
+    as_of=None,
     **collection_kwargs,
 ):
     """
@@ -160,6 +166,13 @@ def powerplants(
             if `update` is true.
     **collection_kwargs : kwargs
             Arguments passed to powerplantmatching.collection.Collection.
+    reconcile_retirements : bool, default True
+            Split reduced plants into retired and operational portions after matching.
+            Also applies to local cached results, but not from_url downloads.
+            Requires loading the original source units.
+    as_of : int, optional
+            Retirement snapshot year, default current year. Only retirement
+            years strictly before the snapshot year count as retired.
 
     """
     from . import latest_release
@@ -169,6 +182,12 @@ def powerplants(
             config = get_config()
         else:
             config = get_config(**config_update)
+
+    def reconcile(df):
+        if reconcile_retirements and collection_kwargs.get("reduced", True):
+            df = reconcile_retirement_capacity(df, config=config, as_of=as_of)
+        return df
+
     if extendby_kwargs is None:
         extendby_kwargs = {}
     if extend_by_kwargs is None:
@@ -214,6 +233,7 @@ def powerplants(
             .pipe(parse_string_to_dict, ["projectID", "EIC"])
             .pipe(set_column_name, "Matched Data")
         )
+        df = reconcile(df)
         if extend_by_vres:
             return df.pipe(
                 extend_by_VRE, config=config, base_year=config["opsd_vres_base_year"]
@@ -245,6 +265,7 @@ def powerplants(
 
     matched.reset_index(drop=True).to_csv(fn, index_label="id", encoding="utf-8")
 
+    matched = reconcile(matched.pipe(set_column_name, "Matched Data"))
     if extend_by_vres:
         matched = extend_by_VRE(
             matched, config=config, base_year=config["opsd_vres_base_year"]

powerplantmatching/data.py

@@ -2664,7 +2664,8 @@ def _assign_PP_or_CHP(df):
     df_processed["Name"] = df_processed.apply(
         lambda x: (
             f"{x.Name} {x.NameKraftwerksblock.replace(x.Name, '').strip()}"
-            if x.NameKraftwerksblock
+            if pd.notna(x.NameKraftwerksblock)
+            and pd.notna(x.Name)
             and x.NameKraftwerksblock != x.Name
             and x.Fueltype in config["clean_name"]["fueltypes_with_blocks"]
             else x.Name

powerplantmatching/heuristics.py

@@ -7,6 +7,8 @@
 """
 
 import logging
+from datetime import date
+from numbers import Integral
 
 import numpy as np
 import pandas as pd
@@ -21,6 +23,186 @@
 logger = logging.getLogger(__name__)
 
 
+def reconcile_retirement_capacity(df, config=None, as_of=None):
+    """
+    Reconcile retirement status in a reduced dataset after matching.
+
+    For each plant, use the original units of its highest-priority capacity
+    source to separate retired and operational capacity. Units are retired
+    only when DateOut is strictly before as_of; unknown retirement dates count
+    as operational. Mixed plants produce two rows, otherwise one row is kept.
+    No units are dropped and no new cross-source matches are made.
+
+    Allocate the final plant's capacity, stored energy and reservoir volume
+    proportionally to each portion, preserving their totals. Recompute
+    commissioning and retirement dates from the selected units,
+    retaining compatible matched-date fallbacks. Operational retirement dates
+    may be extended by the matched plant's date. Matched retrofit dates are
+    retained, except when later than a retired portion's retirement.
+    Efficiency is capacity-weighted
+    over units with known efficiency; otherwise the matched value is retained.
+
+    Selected-source IDs are partitioned between the portions. Other-source IDs
+    remain on the operational row, or the sole row if there is no split.
+    Incomplete or invalid capacity evidence leaves the plant unchanged and
+    emits a warning. Output columns are unchanged.
+
+    Parameters
+    ----------
+    df : pandas.DataFrame
+        Reduced matched plants with source unit IDs in projectID dictionaries.
+    config : dict, optional
+        Source configuration and reliability scores; defaults to get_config().
+    as_of : int, optional
+        Snapshot year, default current year. Units retiring in this year remain
+        operational. Future commissioning dates are not filtered here.
+
+    Returns
+    -------
+    pandas.DataFrame
+        Reconciled plants, with at most two rows per input plant.
+    """
+    from . import data
+
+    config = get_config() if config is None else config
+    year = date.today().year if as_of is None else as_of
+    if not isinstance(year, Integral):
+        raise ValueError("as_of must be an integer year")
+    if df.empty:
+        return df.copy()
+    # Load original units once per source; conflicting IDs cannot be reconciled.
+    sources = sorted(
+        {source for ids in df.projectID if isinstance(ids, dict) for source in ids}
+    )
+    lookups = {}
+    for source in sources:
+        units = getattr(data, source)(config=config)
+        units = units.reindex(
+            columns=[
+                "projectID",
+                "Capacity",
+                "DateIn",
+                "DateOut",
+                "Efficiency",
+            ]
+        ).drop_duplicates()
+        lookups[source] = units.loc[~units.projectID.duplicated(keep=False)].set_index(
+            "projectID"
+        )
+
+    rows = []
+    for _, plant in df.iterrows():
+        if not isinstance(plant.projectID, dict) or not plant.projectID:
+            logger.warning(
+                "No unit provenance for %s; retaining original row.", plant.Name
+            )
+            rows.append(plant)
+            continue
+        capacity_source = None
+        # Select the capacity provider without changing the original matches.
+        # Stable sorting preserves the reducer's provenance order for tied scores.
+        source_order = sorted(
+            plant.projectID,
+            key=lambda source: config[source]["reliability_score"],
+            reverse=True,
+        )
+        for source in source_order:
+            ids = plant.projectID[source]
+            if not ids:
+                continue
+            lookup = lookups[source]
+            ids = list(ids)
+            if not all(unit_id in lookup.index for unit_id in ids):
+                break
+            units = lookup.loc[ids]
+            # aggregate_units converts zero sums (including all-missing) to NaN.
+            if units.Capacity.notna().any() and (
+                config[source].get("aggregated_units", False)
+                or units.Capacity.sum() != 0
+            ):
+                capacity_source = source
+                break
+        if capacity_source is None:
+            logger.warning(
+                "Incomplete unit evidence for %s; retaining original row.",
+                plant.Name,
+            )
+            rows.append(plant)
+            continue
+        total = units.Capacity.sum()
+        if (
+            units.Capacity.isna().any()
+            or (units.Capacity < 0).any()
+            or not np.isfinite(total)
+            or total <= 0
+            or not np.isfinite(plant.Capacity)
+            or plant.Capacity < 0
+        ):
+            logger.warning(
+                "Invalid capacity evidence for %s; retaining original row.", plant.Name
+            )
+            rows.append(plant)
+            continue
+        if plant.Capacity == 0:
+            rows.append(plant)
+            continue
+        # Unknown retirement dates and the snapshot year remain operational.
+        retired = units.DateOut.lt(year)
+        portions = [
+            (True, units.loc[retired]),
+            (False, units.loc[~retired]),
+        ]
+        portions = [
+            (status, part) for status, part in portions if part.Capacity.sum() > 0
+        ]
+        for is_retired, portion in portions:
+            row = plant.copy()
+            share = portion.Capacity.sum() / total
+            # Split extensive quantities; duration and dam height stay unchanged.
+            for column in ("Capacity", "StorageCapacity_MWh", "Volume_Mm3"):
+                if column in df:
+                    row[column] = plant[column] * share
+            # Recompute dates, keeping only compatible matched-date fallbacks.
+            row.DateIn = portion.DateIn.min()
+            row.DateOut = portion.DateOut.max()
+            if not is_retired and pd.notna(plant.DateOut) and plant.DateOut >= year:
+                row.DateOut = (
+                    max(row.DateOut, plant.DateOut)
+                    if pd.notna(row.DateOut)
+                    else plant.DateOut
+                )
+            if pd.isna(row.DateIn) and (
+                pd.isna(row.DateOut) or plant.DateIn <= row.DateOut
+            ):
+                row.DateIn = plant.DateIn
+            if is_retired and "DateRetrofit" in df and row.DateRetrofit > row.DateOut:
+                row.DateRetrofit = np.nan
+            if "Efficiency" in df:
+                # Missing efficiencies must not contribute to the denominator.
+                known = portion.dropna(subset=["Efficiency"])
+                weight = known.Capacity.sum()
+                if weight > 0:
+                    row.Efficiency = (known.Efficiency * known.Capacity).sum() / weight
+            row.projectID = {
+                capacity_source: set(
+                    units.index if len(portions) == 1 else portion.index
+                )
+            }
+            # Unmapped sources describe the plant, not verified retirement portions.
+            if not is_retired or len(portions) == 1:
+                row.projectID.update(
+                    {
+                        key: value
+                        for key, value in plant.projectID.items()
+                        if key != capacity_source
+                    }
+                )
+            rows.append(row)
+    result = pd.DataFrame(rows).reset_index(drop=True).reindex(columns=df.columns)
+    result.columns.name = df.columns.name
+    return result
+
+
 def extend_by_non_matched(
     df,
     extend_by,

test/test_collection.py

@@ -23,6 +23,11 @@ def test_powerplants_keeps_units_with_same_name(monkeypatch):
     monkeypatch.setattr(collection, "collect", lambda *args, **kwargs: turbines)
     config = pm.get_config(fully_included_sources=[], variant="test-same-name")
 
-    df = pm.powerplants(config=config, update=True, fill_geopositions=False)
+    df = pm.powerplants(
+        config=config,
+        update=True,
+        fill_geopositions=False,
+        reconcile_retirements=False,
+    )
 
     assert df.Capacity.sum() == turbines.Capacity.sum()