Skip to content

couplers.base

The IAM→PyPSA coupling interface. This is exposed to pypsa models and is the entry point for the coupling workflow.

  • Coupler is the backend-neutral base: it holds the shared, concrete builders (build_co2_prices, discount_rates, downscale_country_demand)
  • it consumes the IAM symbols (resolved via their config) and the region map, and it contains the reference data (population, GDP, etc.) for downscaling.

Concrete subclasses are instantiated directly by the caller, which selects on loader.backend; a new IAM or output format is added as a further Coupler subclass, not a branch here: - RemindGdxCoupler (iampypsa.couplers.remind) - RemindIamcCoupler (iampypsa.couplers.remind)

config keys used: currency_factor, sector_weights, countries, planning_horizons.

Coupler

Backend-neutral base: shared builders + source-specific hook declarations.

Source code in src/iampypsa/couplers/base.py 
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
class Coupler:
    """Backend-neutral base: shared builders + source-specific hook declarations."""

    def __init__(
        self,
        loader,
        symbols: dict[str, Any],
        region_map: dict[str, list[str]],
        config: dict[str, Any],
        *,
        model_regions: list[str] | None = None,
        reference_data: dict[str, pd.DataFrame] | None = None,
        ssp_population: pd.DataFrame | None = None,
        ssp_gdp: pd.DataFrame | None = None,
    ) -> None:
        """Bind the loader, resolved symbol map, region map, config, and reference data."""
        self.loader = loader
        self.symbols = symbols
        self.region_map = region_map
        self.config = config
        self.model_regions = model_regions or list(region_map)
        self.reference_data: dict[str, pd.DataFrame] = dict(reference_data or {})
        if ssp_population is not None:
            self.reference_data.setdefault("population", ssp_population)
        if ssp_gdp is not None:
            self.reference_data.setdefault("gdp", ssp_gdp)

    @property
    def ssp_population(self) -> pd.DataFrame | None:
        return self.reference_data.get("population")

    @property
    def ssp_gdp(self) -> pd.DataFrame | None:
        return self.reference_data.get("gdp")

    # -- Source-specific hooks (must be overridden by subclasses) -----------

    def build_regional_demand(self) -> pd.DataFrame:
        """Read IAM regional sectoral demand as ``[year, region, sector, value, unit]`` (MWh).

        Implemented by ``RemindGdxCoupler`` / ``RemindIamcCoupler``.
        """
        raise NotImplementedError(
            f"{type(self).__name__} must implement build_regional_demand(). "
            "Instantiate RemindGdxCoupler or RemindIamcCoupler (per loader.backend), "
            "or a new Coupler subclass for another IAM."
        )

    def extract_cost_parameters(self, year: int) -> pd.DataFrame:
        """Extract cost parameters as long ``[region, reference, parameter, value, unit]``.

        Implemented by ``RemindGdxCoupler`` / ``RemindIamcCoupler``.
        """
        raise NotImplementedError(
            f"{type(self).__name__} must implement extract_cost_parameters(). "
            "Instantiate RemindGdxCoupler or RemindIamcCoupler (per loader.backend), "
            "or a new Coupler subclass for another IAM."
        )

    # -- Shared concrete builders -------------------------------------------

    def build_co2_prices(self, years: Sequence[int] | None = None) -> pd.DataFrame:
        """Build the per-(region, year) CO2 price pathway.

        Conversion is spec-driven: applies whatever ``(unit, to_unit)`` factor the resolved
        ``co2_price`` symbol's config declares (a no-op when the source already reports t CO2).
        The runtime ``currency_factor`` is always applied.

        ``years`` selects the year set to reindex to (missing filled with 0); when ``None``
        it falls back to ``config["planning_horizons"]``. Callers that derive the coupled-year
        set from the IAM source (e.g. the GDX ``t`` symbol) pass it explicitly.
        """
        raw = load_frame(self.loader, self.symbols["co2_price"])
        years = years if years is not None else self.config.get("planning_horizons")
        prices = extract_co2_prices(raw, regions=self.model_regions, years=years)
        return convert_co2_prices(
            prices, currency_factor=self.config.get("currency_factor", 1.0), carbon_to_co2=False
        )

    def downscale_country_demand(self, regional: pd.DataFrame | None = None) -> pd.DataFrame:
        """Downscale IAM regional demand to per-country annual demand by sector and year.

        The proxy registry is ``self.reference_data`` verbatim — it already holds ``population``/
        ``gdp`` (and any ``heating_demand``/``cooling_demand`` the caller added). Each sector's
        ``sector_weights`` entry names which of those proxies to blend.
        """
        loads = self.build_regional_demand() if regional is None else regional
        if self.config.get("planning_horizons"):
            years = {int(y) for y in self.config["planning_horizons"]}
            loads = loads[loads["year"].isin(years)]
        return disaggregate_demand_to_country(
            loads,
            self.region_map,
            self.reference_data,
            self.config["sector_weights"],
            set(self.config["countries"]),
        )

    def discount_rates(self, year: int) -> pd.Series:
        """Return the discount rate per region for ``year``, indexed by region.

        When a region has no value for ``year`` (e.g. a trailing NaN in the source), the
        most recent earlier year's value is used with a warning.
        """
        p_r = (
            load_frame(self.loader, self.symbols["discount_rate"])
            .pipe(lambda df: df[df["region"].isin(self.model_regions)])
        )
        # Last-value per region: covers both the requested year and any forward-fill fallback.
        last = p_r.sort_values("year").groupby("region")["value"].last()
        missing = set(self.model_regions) - set(last.index)
        if missing:
            raise ValueError(
                f"No discount rate for year {year} or any earlier year, "
                f"regions: {sorted(missing)}"
            )
        exact = p_r[p_r["year"].astype(str) == str(year)].set_index("region")["value"]
        filled = last.index.difference(exact.index)
        if len(filled):
            logger.warning(
                "Discount rate absent for year %d in regions %s; using most-recent value.",
                year, sorted(filled),
            )
        return exact.combine_first(last)

__init__(loader, symbols, region_map, config, *, model_regions=None, reference_data=None, ssp_population=None, ssp_gdp=None)

Bind the loader, resolved symbol map, region map, config, and reference data.

Source code in src/iampypsa/couplers/base.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
def __init__(
    self,
    loader,
    symbols: dict[str, Any],
    region_map: dict[str, list[str]],
    config: dict[str, Any],
    *,
    model_regions: list[str] | None = None,
    reference_data: dict[str, pd.DataFrame] | None = None,
    ssp_population: pd.DataFrame | None = None,
    ssp_gdp: pd.DataFrame | None = None,
) -> None:
    """Bind the loader, resolved symbol map, region map, config, and reference data."""
    self.loader = loader
    self.symbols = symbols
    self.region_map = region_map
    self.config = config
    self.model_regions = model_regions or list(region_map)
    self.reference_data: dict[str, pd.DataFrame] = dict(reference_data or {})
    if ssp_population is not None:
        self.reference_data.setdefault("population", ssp_population)
    if ssp_gdp is not None:
        self.reference_data.setdefault("gdp", ssp_gdp)

build_co2_prices(years=None)

Build the per-(region, year) CO2 price pathway.

Conversion is spec-driven: applies whatever (unit, to_unit) factor the resolved co2_price symbol's config declares (a no-op when the source already reports t CO2). The runtime currency_factor is always applied.

years selects the year set to reindex to (missing filled with 0); when None it falls back to config["planning_horizons"]. Callers that derive the coupled-year set from the IAM source (e.g. the GDX t symbol) pass it explicitly.

Source code in src/iampypsa/couplers/base.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
def build_co2_prices(self, years: Sequence[int] | None = None) -> pd.DataFrame:
    """Build the per-(region, year) CO2 price pathway.

    Conversion is spec-driven: applies whatever ``(unit, to_unit)`` factor the resolved
    ``co2_price`` symbol's config declares (a no-op when the source already reports t CO2).
    The runtime ``currency_factor`` is always applied.

    ``years`` selects the year set to reindex to (missing filled with 0); when ``None``
    it falls back to ``config["planning_horizons"]``. Callers that derive the coupled-year
    set from the IAM source (e.g. the GDX ``t`` symbol) pass it explicitly.
    """
    raw = load_frame(self.loader, self.symbols["co2_price"])
    years = years if years is not None else self.config.get("planning_horizons")
    prices = extract_co2_prices(raw, regions=self.model_regions, years=years)
    return convert_co2_prices(
        prices, currency_factor=self.config.get("currency_factor", 1.0), carbon_to_co2=False
    )

build_regional_demand()

Read IAM regional sectoral demand as [year, region, sector, value, unit] (MWh).

Implemented by RemindGdxCoupler / RemindIamcCoupler.

Source code in src/iampypsa/couplers/base.py 
71
72
73
74
75
76
77
78
79
80
def build_regional_demand(self) -> pd.DataFrame:
    """Read IAM regional sectoral demand as ``[year, region, sector, value, unit]`` (MWh).

    Implemented by ``RemindGdxCoupler`` / ``RemindIamcCoupler``.
    """
    raise NotImplementedError(
        f"{type(self).__name__} must implement build_regional_demand(). "
        "Instantiate RemindGdxCoupler or RemindIamcCoupler (per loader.backend), "
        "or a new Coupler subclass for another IAM."
    )

discount_rates(year)

Return the discount rate per region for year, indexed by region.

When a region has no value for year (e.g. a trailing NaN in the source), the most recent earlier year's value is used with a warning.

Source code in src/iampypsa/couplers/base.py 
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
def discount_rates(self, year: int) -> pd.Series:
    """Return the discount rate per region for ``year``, indexed by region.

    When a region has no value for ``year`` (e.g. a trailing NaN in the source), the
    most recent earlier year's value is used with a warning.
    """
    p_r = (
        load_frame(self.loader, self.symbols["discount_rate"])
        .pipe(lambda df: df[df["region"].isin(self.model_regions)])
    )
    # Last-value per region: covers both the requested year and any forward-fill fallback.
    last = p_r.sort_values("year").groupby("region")["value"].last()
    missing = set(self.model_regions) - set(last.index)
    if missing:
        raise ValueError(
            f"No discount rate for year {year} or any earlier year, "
            f"regions: {sorted(missing)}"
        )
    exact = p_r[p_r["year"].astype(str) == str(year)].set_index("region")["value"]
    filled = last.index.difference(exact.index)
    if len(filled):
        logger.warning(
            "Discount rate absent for year %d in regions %s; using most-recent value.",
            year, sorted(filled),
        )
    return exact.combine_first(last)

downscale_country_demand(regional=None)

Downscale IAM regional demand to per-country annual demand by sector and year.

The proxy registry is self.reference_data verbatim — it already holds population/ gdp (and any heating_demand/cooling_demand the caller added). Each sector's sector_weights entry names which of those proxies to blend.

Source code in src/iampypsa/couplers/base.py 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
def downscale_country_demand(self, regional: pd.DataFrame | None = None) -> pd.DataFrame:
    """Downscale IAM regional demand to per-country annual demand by sector and year.

    The proxy registry is ``self.reference_data`` verbatim — it already holds ``population``/
    ``gdp`` (and any ``heating_demand``/``cooling_demand`` the caller added). Each sector's
    ``sector_weights`` entry names which of those proxies to blend.
    """
    loads = self.build_regional_demand() if regional is None else regional
    if self.config.get("planning_horizons"):
        years = {int(y) for y in self.config["planning_horizons"]}
        loads = loads[loads["year"].isin(years)]
    return disaggregate_demand_to_country(
        loads,
        self.region_map,
        self.reference_data,
        self.config["sector_weights"],
        set(self.config["countries"]),
    )

extract_cost_parameters(year)

Extract cost parameters as long [region, reference, parameter, value, unit].

Implemented by RemindGdxCoupler / RemindIamcCoupler.

Source code in src/iampypsa/couplers/base.py 
82
83
84
85
86
87
88
89
90
91
def extract_cost_parameters(self, year: int) -> pd.DataFrame:
    """Extract cost parameters as long ``[region, reference, parameter, value, unit]``.

    Implemented by ``RemindGdxCoupler`` / ``RemindIamcCoupler``.
    """
    raise NotImplementedError(
        f"{type(self).__name__} must implement extract_cost_parameters(). "
        "Instantiate RemindGdxCoupler or RemindIamcCoupler (per loader.backend), "
        "or a new Coupler subclass for another IAM."
    )