io.ssp

Fetch / read SSP population & GDP proxy datasets used to weight downscaling.

fetch_ssp_variable pulls a country-level IAMC variable directly from the IIASA SSP REST API (no auth/pyam needed); read_ssp_data reads a saved snapshot. Ported from PyPSA-Eur retrieve_ssp_data.py. Output columns: [iso2, year, value].

fetch_ssp_data(scenario, population_model, gdp_model)

Fetch SSP population and GDP|PPP for one scenario; return (population, gdp).

Source code in src/iampypsa/io/ssp.py

def fetch_ssp_data(
    scenario: str,
    population_model: str,
    gdp_model: str,
) -> tuple[pd.DataFrame, pd.DataFrame]:
    """Fetch SSP population and GDP|PPP for one scenario; return ``(population, gdp)``."""
    population = fetch_ssp_variable("Population", population_model, scenario)
    gdp = fetch_ssp_variable("GDP|PPP", gdp_model, scenario)
    return population, gdp

fetch_ssp_variable(variable, model, scenario, *, url=IIASA_URL, page_size=_PAGE_SIZE, timeout=120.0)

Fetch one IAMC variable from the IIASA SSP API as [iso2, year, value].

Source code in src/iampypsa/io/ssp.py

def fetch_ssp_variable(
    variable: str,
    model: str,
    scenario: str,
    *,
    url: str = IIASA_URL,
    page_size: int = _PAGE_SIZE,
    timeout: float = 120.0,
) -> pd.DataFrame:
    """Fetch one IAMC variable from the IIASA SSP API as ``[iso2, year, value]``."""
    import httpx  # optional 'ssp' extra — imported lazily so `import iampypsa` works without it

    body = {
        "join_parameters": True,
        "join_runs": True,
        "join_run_id": False,
        "variable": {"name__like": variable},
        "run": {"default_only": True},
        "model": {"name__like": model},
        "scenario": {"name__like": scenario},
    }
    rows: list = []
    columns: list | None = None
    offset = 0
    while True:
        resp = httpx.patch(
            url, params={"limit": page_size, "offset": offset}, json=body, timeout=timeout
        )
        resp.raise_for_status()
        results = resp.json()["results"]
        columns = columns or results["columns"]
        page = results["data"]
        rows.extend(page)
        offset += len(page)
        if len(page) < page_size:
            break

    df = pd.DataFrame(rows, columns=columns)
    df["iso2"] = coco.CountryConverter().pandas_convert(pd.Series(df["region"]), to="ISO2", not_found=None)
    return (
        df.dropna(subset=["iso2"])[["iso2", "step_year", "value"]]
        .rename(columns={"step_year": "year"})
        .groupby(["iso2", "year"], as_index=False)["value"]
        .sum()
        .sort_values(["iso2", "year"])
        .reset_index(drop=True)
    )

read_ssp_data(path, variables=None)

Read a saved SSP snapshot CSV with columns [iso2, year, value].

Source code in src/iampypsa/io/ssp.py

def read_ssp_data(
    path: str | PathLike,
    variables: Sequence[str] | None = None,  # accepted for API symmetry; snapshot is single-variable
) -> pd.DataFrame:
    """Read a saved SSP snapshot CSV with columns ``[iso2, year, value]``."""
    return pd.read_csv(path)