diff --git a/powerplantmatching/collection.py b/powerplantmatching/collection.py index 5e462f5c..fa47abcb 100644 --- a/powerplantmatching/collection.py +++ b/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"] diff --git a/powerplantmatching/data.py b/powerplantmatching/data.py index 5dfba6f0..8ed0a595 100644 --- a/powerplantmatching/data.py +++ b/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 diff --git a/powerplantmatching/heuristics.py b/powerplantmatching/heuristics.py index d48b3794..414e38a8 100644 --- a/powerplantmatching/heuristics.py +++ b/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, diff --git a/test/test_collection.py b/test/test_collection.py index 1d54a99e..ce49e430 100644 --- a/test/test_collection.py +++ b/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()