EIA profile-based powerplant age imputation

config/config.yaml

@@ -118,6 +118,7 @@ imputation:
       utility pv: "land"
   time:
     scenario: "construction" # Options: historical->construction->pre_construction->announced
+    start_year_imputation_method: capacity_profile
     lifetime_years:
       # General lifetime assumptions. Adjust as needed!
       bioenergy turbine: 25
@@ -161,3 +162,80 @@ imputation:
       smr reactor: 50
       steam turbine: 5
       utility pv: 2
+    planned_commissioning_year_windows:
+      bioenergy turbine:
+        construction: [1, 3]
+        pre-construction: [2, 5]
+        announced: [4, 8]
+      ccgt:
+        construction: [1, 3]
+        pre-construction: [2, 5]
+        announced: [4, 8]
+      coal turbine:
+        construction: [1, 3]
+        pre-construction: [2, 5]
+        announced: [4, 8]
+      concentrating solar power:
+        construction: [1, 4]
+        pre-construction: [3, 7]
+        announced: [5, 10]
+      high enthalpy geothermal:
+        construction: [2, 5]
+        pre-construction: [3, 7]
+        announced: [5, 10]
+      igcc:
+        construction: [2, 5]
+        pre-construction: [4, 8]
+        announced: [6, 12]
+      low enthalpy geothermal:
+        construction: [1, 4]
+        pre-construction: [3, 6]
+        announced: [4, 9]
+      nuclear reactor:
+        construction: [3, 7]
+        pre-construction: [5, 13]
+        announced: [8, 18]
+      ocgt:
+        construction: [1, 3]
+        pre-construction: [2, 5]
+        announced: [4, 8]
+      offshore:
+        construction: [2, 5]
+        pre-construction: [3, 8]
+        announced: [5, 12]
+      onshore:
+        construction: [1, 3]
+        pre-construction: [2, 6]
+        announced: [3, 10]
+      pumped storage:
+        construction: [3, 8]
+        pre-construction: [5, 12]
+        announced: [8, 18]
+      reciprocating engine:
+        construction: [1, 3]
+        pre-construction: [2, 5]
+        announced: [4, 8]
+      reservoir:
+        construction: [3, 8]
+        pre-construction: [5, 12]
+        announced: [8, 18]
+      rooftop pv:
+        construction: [1, 2]
+        pre-construction: [2, 3]
+        announced: [3, 5]
+      run of river:
+        construction: [2, 6]
+        pre-construction: [4, 10]
+        announced: [6, 15]
+      smr reactor:
+        construction: [2, 6]
+        pre-construction: [4, 10]
+        announced: [7, 15]
+      steam turbine:
+        construction: [1, 3]
+        pre-construction: [2, 5]
+        announced: [4, 8]
+      utility pv:
+        construction: [1, 3]
+        pre-construction: [2, 5]
+        announced: [3, 8]

workflow/internal/config.schema.yaml

@@ -29,6 +29,43 @@ $defs:
       type: integer
       minimum: 1
 
+  planned_commissioning_year_window:
+    type: array
+    description: >
+      Inclusive minimum and maximum commissioning-year offsets relative
+      to the dataset year.
+    minItems: 2
+    maxItems: 2
+    items:
+      type: integer
+      minimum: 1
+
+  planned_commissioning_status_windows:
+    type: object
+    additionalProperties: false
+    required:
+      - construction
+      - pre-construction
+      - announced
+    properties:
+      construction:
+        $ref: "#/$defs/planned_commissioning_year_window"
+      pre-construction:
+        $ref: "#/$defs/planned_commissioning_year_window"
+      announced:
+        $ref: "#/$defs/planned_commissioning_year_window"
+
+  planned_commissioning_technology_windows:
+    title: Planned commissioning year windows
+    description: |
+      Technology- and status-specific year-offset windows used to impute
+      missing commissioning years for planned powerplants. Each window is
+      expressed as [minimum_year_offset, maximum_year_offset] relative to
+      the dataset year (inclusive).
+    type: object
+    additionalProperties:
+      $ref: "#/$defs/planned_commissioning_status_windows"
+
   technology_mapping_base:
     title: Technology Mapping
     description: >
@@ -423,8 +460,10 @@ properties:
         additionalProperties: false
         required:
           - scenario
+          - start_year_imputation_method
           - lifetime_years
           - retirement_delay_years
+          - planned_commissioning_year_windows
         properties:
           scenario:
             title: Scenario
@@ -444,3 +483,16 @@ properties:
             $ref: "#/$defs/year_map"
           retirement_delay_years:
             $ref: "#/$defs/year_map"
+          start_year_imputation_method:
+            title: Start year imputation method
+            description: |
+              Method used to impute missing start years when backfilling from known
+              end years is not possible.
+
+              - 'capacity_profile' assigns missing start years deterministically using
+              capacity-weighted commissioning profiles from annual commissioning statistics.
+            type: string
+            enum:
+              - capacity_profile
+          planned_commissioning_year_windows:
+            $ref: "#/$defs/planned_commissioning_technology_windows"

workflow/report/impute_time_profile.rst

@@ -0,0 +1,5 @@
+Annual commissioning-capacity imputation profile.
+
+Bars show commissioning capacity by start year, split into observed and imputed
+powerplants. The line shows the target annual commissioning profile used during
+imputation.

workflow/rules/_utils.smk

@@ -31,7 +31,7 @@ IMPUTED_CAT_WITHOUT_ADJUSTMENT = {"large_solar"}
 
 
 def additional_config_validation():
-    """Ensures technology mapping and lifetime-related names match."""
+    """Ensure technology mappings and time-imputation settings match."""
     lifetime_set = set(config["imputation"]["time"]["lifetime_years"].keys())
     mismatch = lifetime_set ^ set(
         config["imputation"]["time"]["retirement_delay_years"]
@@ -54,6 +54,25 @@ def additional_config_validation():
             f"Technology mapping does not match lifetime technologies for {mismatch}"
         )
 
+    planned_windows = config["imputation"]["time"]["planned_commissioning_year_windows"]
+    mismatch = tech_map_set ^ set(planned_windows)
+    if mismatch:
+        raise ValueError(
+            "Technology mapping does not match planned commissioning windows "
+            f"for {mismatch}"
+        )
+    reversed_windows = {
+        f"{technology}/{status}": window
+        for technology, status_windows in planned_windows.items()
+        for status, window in status_windows.items()
+        if window[0] > window[1]
+    }
+    if reversed_windows:
+        raise ValueError(
+            "Planned commissioning windows must be ordered [minimum, maximum]: "
+            f"{reversed_windows}"
+        )
+
 
 def get_excluded_powerplant_ids(category):
     """Handle cases where the naming between files and configuration mismatch."""

workflow/rules/impute.smk

@@ -45,6 +45,10 @@ rule impute_location:
 rule impute_time:
     input:
         relocated=rules.impute_location.output.relocated,
+        category_capacity=(
+            "<resources>/automatic/shapes/{shapes}/"
+            "statistics/category_capacity.parquet"
+        ),
     output:
         aged=workflow.pathvars.apply("<powerplants>").format(
             shapes="{shapes}",
@@ -63,10 +67,19 @@ rule impute_time:
             category="Powerplants module",
             subcategory="{category}",
         ),
+        capacity_date_events=(
+            "<resources>/automatic/shapes/{shapes}/impute_time/{category}_capacity_date_events.parquet"
+        ),
+        capacity_date_plot=report(
+            "<results>/{shapes}/powerplants/unadjusted/{category}_capacity_date_events.pdf",
+            caption="../report/impute_time_profile.rst",
+            category="Powerplants module",
+            subcategory="{category}",
+        ),
     log:
         "<logs>/{shapes}/{category}/impute_time.log",
     wildcard_constraints:
-        dataset="|".join(IMPUTED_CAT),
+        category="|".join(IMPUTED_CAT),
     conda:
         "../envs/module.yaml"
     params:

workflow/scripts/_plots.py

@@ -1,10 +1,13 @@
 """Plot functions used in one or more rules."""
 
+from collections.abc import Collection
+
 import _schemas
 import _utils
 import geopandas as gpd
 import numpy as np
 import pandas as pd
+from cmap import Colormap
 from matplotlib import pyplot as plt
 from matplotlib import ticker as mticker
 from matplotlib.axes import Axes
@@ -238,3 +241,24 @@ def plot_capacity_aggregation(
         ax.set_axis_off()
         ax.set_title(title + f" in year {agg.attrs['year']}")
         fig.savefig(output_file, bbox_inches="tight")
+
+
+def get_colour_dict(
+    sources: Collection[str],
+    colormap: str,
+    *,
+    value_range: tuple[float, float] = (0.0, 1.0),
+) -> dict[str, str]:
+    """Return deterministic colours for a collection of source types."""
+    sorted_sources = sorted(sources)
+
+    if not sorted_sources:
+        return {}
+
+    cmap = Colormap(colormap).to_mpl()
+    values = np.linspace(value_range[0], value_range[1], len(sorted_sources))
+
+    return {
+        source: colour
+        for source, colour in zip(sorted_sources, cmap(values), strict=True)
+    }

workflow/scripts/_schemas.py

@@ -3,6 +3,7 @@
 # ruff: noqa: UP007
 from typing import Literal
 
+import _utils
 import pandera.pandas as pa
 from pandera.pandas import DataFrameModel, Field, check
 from pandera.typing.geopandas import GeoSeries
@@ -87,6 +88,14 @@ class Config:
     "Expected decommissioning year."
     status: Series[str]
     "Known state of the project."
+    start_year_source_type: Series[str] | None = Field(
+        isin=_utils.date_source_types_for("start_year")
+    )
+    "Source/provenance label for the start year."
+    end_year_source_type: Series[str] | None = Field(
+        isin=_utils.date_source_types_for("end_year")
+    )
+    "Source/provenance label for the end year."
     # Location / size
     geometry: GeoSeries[Point] = Field()
     "Powerplant point data."

workflow/scripts/_utils.py

@@ -49,6 +49,74 @@ def listify(item) -> list:
 }
 EIA_CAT_MAPPING = {k: listify(v) for k, v in EIA_CAT_MAPPING.items()}
 
+DATE_SOURCE_METADATA = {
+    "observed": {
+        "label": "Observed date (from powerplant data)",
+        "applies_to": {"start_year", "end_year"},
+    },
+    "derived_from_end_year": {
+        "label": "Start date derived from observed end date",
+        "applies_to": {"start_year"},
+    },
+    "imputed_capacity_profile": {
+        "label": "Start date imputed from historical commissioning-profile",
+        "applies_to": {"start_year"},
+    },
+    "imputed_construction_window": {
+        "label": "Start date imputed within construction window",
+        "applies_to": {"start_year"},
+    },
+    "imputed_pre_construction_window": {
+        "label": "Start date imputed within pre-construction window",
+        "applies_to": {"start_year"},
+    },
+    "imputed_announced_window": {
+        "label": "Start date imputed within announced window",
+        "applies_to": {"start_year"},
+    },
+    "derived_from_imputed_retirement_end_year": {
+        "label": "Start date derived from retirement-profile end date",
+        "applies_to": {"start_year"},
+    },
+    "derived_from_start_year_lifetime": {
+        "label": "End date derived from start date and lifetime",
+        "applies_to": {"end_year"},
+    },
+    "imputed_retirement_capacity_profile": {
+        "label": "End date imputed from retirement-profile",
+        "applies_to": {"end_year"},
+    },
+    "derived_from_start_year_lifetime_capped_to_retired_status": {
+        "label": "End date derived from start date but capped to retired status",
+        "applies_to": {"end_year"},
+    },
+    "derived_from_start_year_lifetime_with_retirement_delay": {
+        "label": "End date derived from start date, lifetime, and retirement delay",
+        "applies_to": {"end_year"},
+    },
+    "observed_adjusted_with_retirement_delay": {
+        "label": "Observed end date adjusted with retirement delay",
+        "applies_to": {"end_year"},
+    },
+}
+
+
+def date_source_types_for(year_column: str) -> set[str]:
+    """Return source types valid for a date column."""
+    return {
+        source_type
+        for source_type, metadata in DATE_SOURCE_METADATA.items()
+        if year_column in metadata["applies_to"]
+    }
+
+
+def date_source_labels() -> dict[str, str]:
+    """Return date-source display labels."""
+    return {
+        source_type: metadata["label"]
+        for source_type, metadata in DATE_SOURCE_METADATA.items()
+    }
+
 
 def get_eia_stats_in_cat_yr(
     stats: pd.DataFrame, year: int, category: str