diff --git a/INTERFACE.yaml b/INTERFACE.yaml index 607248f..bc9b81b 100644 --- a/INTERFACE.yaml +++ b/INTERFACE.yaml @@ -23,10 +23,18 @@ pathvars: description: | Powerplant statistics. See the schemas in `workflow/scripts/_schemas.py` for more information. + statistics_user: + default: "/user/{shapes}/generation.parquet" + description: | + Optional. Annual national hydropower generation statistics. + Used when "generation_adjustement" is set to "user" in the configuration. + See the schemas in `workflow/scripts/_schemas.py` for more information. results: - statistics: + statistics_eia: default: "/{shapes}/statistics/generation.parquet" - description: National hydropower generation statistics for the requested shapes. + description: > + National hydropower generation statistics for the requested shapes. + Generated using EIA data. disaggregated_inflow: default: "/{shapes}/disaggregated/inflow_mwh.parquet" description: > diff --git a/README.md b/README.md index 2a6e347..789f7f2 100644 --- a/README.md +++ b/README.md @@ -22,7 +22,7 @@ and the `snakemake` [documentation](https://snakemake.readthedocs.io/en/stable/s Data processing steps:

- +

1. A global dataset of hydro basins is created for a requested Pfafstetter level using data from the [HydroBASINS](https://www.hydrosheds.org/products/hydrobasins) dataset. @@ -32,9 +32,10 @@ Data processing steps:

-3. User provided shapes, powerplants and configuration are used to construct a data request to the [Copernicus Data Store](https://cds.climate.copernicus.eu/) using the [`atlite` library](https://github.com/PyPSA/atlite). -4. `atlite` is used to construct inflow timeseries per powerplant. -5. Inflow timeseries are combined and aggregated to the requested resolution, using national-level statistics from the [EIA](https://www.eia.gov/international/) as a guiding normalisation heuristic for total generation. +3. If requested, [EIA](https://www.eia.gov/international/) statistics of national hydropower generation are downloaded and harmonised to a standardised schema. +Users may decide to replace this data with their own, or to adjust it slightly (see the [interfacing](#input--output-structure)). +4. User provided shapes, powerplants and configuration are used to construct a data request to the [Copernicus Data Store](https://cds.climate.copernicus.eu/) using the [`atlite` library](https://github.com/PyPSA/atlite) to construct inflow timeseries per powerplant. +5. Inflow timeseries are combined and aggregated to the requested resolution, using the national generation statistics as a guiding normalisation heuristic for total generation. 6. Per-unit inflow timeseries ($PU_{t,r}$) are produced for each region with available capacity. The relation is $InflowMWh_{t,r} = PU_{t,r} \cdot Cap_{r}$, where $Cap_r$ is the capacity per region. - For run of river powerplants, the timeseries are capped so they may not exceed available capacity. @@ -62,7 +63,14 @@ Please consult the configuration [README](./config/README.md) and the [configura As input, you need a file with the polygons to aggregate into (the 'shapes'), and a file specifying national hydropower plants (either RoR or reservoir). These files should follow the `pandera` schemas specified in [_schemas.py](./workflow/scripts/_schemas.py), and can be created using other Modelblocks modules. -Outputs include aggregated timeseries per hydropower plant type, disaggregated inflow timeseries (per point powerplant), and national generation statistics. +If [configured](#configuration), the internally created EIA generation statistics file can be replaced or adjusted by the user. + +

+ +

+ + +Outputs include aggregated timeseries per hydropower plant type, disaggregated inflow timeseries (per point powerplant), and EIA national generation statistics. Please consult the [interface file](./INTERFACE.yaml) for more information. diff --git a/config/config.yaml b/config/config.yaml index 19cc372..28cf87b 100644 --- a/config/config.yaml +++ b/config/config.yaml @@ -1,9 +1,12 @@ -# A minimal example of how to configure this module pfafstetter_level: "06" +crs: + projected: "epsg:3857" + geographic: "epsg:4326" years: # Include 2011, 2012 data start: 2011 end: 2013 powerplants: + generation_adjustement: EIA technology_mapping: run_of_river: "run of river" reservoir: "reservoir" @@ -11,7 +14,4 @@ powerplants: buffer_radius: 1000 # epsg:3857 is in meters max_outside: 50 max_dropped: 30 -crs: - projected: "epsg:3857" - geographic: "epsg:4326" smoothing_hours: "72h" diff --git a/figures/generation.drawio.png b/figures/generation.drawio.png new file mode 100644 index 0000000..00dc979 Binary files /dev/null and b/figures/generation.drawio.png differ diff --git a/figures/rulegraph.png b/figures/rulegraph.png index 35a8827..b676075 100644 Binary files a/figures/rulegraph.png and b/figures/rulegraph.png differ diff --git a/pixi.lock b/pixi.lock index c8b5156..41f2db1 100644 --- a/pixi.lock +++ b/pixi.lock @@ -976,6 +976,7 @@ environments: - conda: https://conda.anaconda.org/conda-forge/noarch/cligj-0.7.2-pyhd8ed1ab_2.conda - conda: https://conda.anaconda.org/conda-forge/noarch/cloudpickle-3.1.2-pyhcf101f3_1.conda - conda: https://conda.anaconda.org/conda-forge/noarch/colorama-0.4.6-pyhd8ed1ab_1.conda + - conda: https://conda.anaconda.org/conda-forge/noarch/country_converter-1.3.2-pyhd8ed1ab_0.conda - conda: https://conda.anaconda.org/conda-forge/noarch/cpython-3.12.13-py312hd8ed1ab_0.conda - conda: https://conda.anaconda.org/conda-forge/noarch/cycler-0.12.1-pyhcf101f3_2.conda - conda: https://conda.anaconda.org/conda-forge/noarch/dask-2026.6.0-pyhc364b38_0.conda @@ -1082,6 +1083,7 @@ environments: - conda: https://conda.anaconda.org/conda-forge/noarch/cligj-0.7.2-pyhd8ed1ab_2.conda - conda: https://conda.anaconda.org/conda-forge/noarch/cloudpickle-3.1.2-pyhcf101f3_1.conda - conda: https://conda.anaconda.org/conda-forge/noarch/colorama-0.4.6-pyhd8ed1ab_1.conda + - conda: https://conda.anaconda.org/conda-forge/noarch/country_converter-1.3.2-pyhd8ed1ab_0.conda - conda: https://conda.anaconda.org/conda-forge/noarch/cpython-3.13.14-py313hd8ed1ab_100.conda - conda: https://conda.anaconda.org/conda-forge/noarch/cycler-0.12.1-pyhcf101f3_2.conda - conda: https://conda.anaconda.org/conda-forge/noarch/dask-2026.6.0-pyhc364b38_0.conda @@ -1353,6 +1355,7 @@ environments: - conda: https://conda.anaconda.org/conda-forge/noarch/cligj-0.7.2-pyhd8ed1ab_2.conda - conda: https://conda.anaconda.org/conda-forge/noarch/cloudpickle-3.1.2-pyhcf101f3_1.conda - conda: https://conda.anaconda.org/conda-forge/noarch/colorama-0.4.6-pyhd8ed1ab_1.conda + - conda: https://conda.anaconda.org/conda-forge/noarch/country_converter-1.3.2-pyhd8ed1ab_0.conda - conda: https://conda.anaconda.org/conda-forge/noarch/cpython-3.13.14-py313hd8ed1ab_100.conda - conda: https://conda.anaconda.org/conda-forge/noarch/cycler-0.12.1-pyhcf101f3_2.conda - conda: https://conda.anaconda.org/conda-forge/noarch/dask-2026.6.0-pyhc364b38_0.conda @@ -6746,6 +6749,17 @@ packages: license_family: MIT size: 8331 timestamp: 1608581999360 +- conda: https://conda.anaconda.org/conda-forge/noarch/country_converter-1.3.2-pyhd8ed1ab_0.conda + sha256: b161a074b92715b9b6a53c35bb84abc0cecea4299ca76b0aaebab0e35b46f3dc + md5: 193a9e54636d8d70781a3e56370f5502 + depends: + - pandas >=1.0.0 + - python >=3.9 + license: GPL-3.0-or-later + license_family: GPL + run_exports: {} + size: 51091 + timestamp: 1761151175246 - conda: https://conda.anaconda.org/conda-forge/noarch/cpython-3.12.13-py312hd8ed1ab_0.conda noarch: generic sha256: d3e9bbd7340199527f28bbacf947702368f31de60c433a16446767d3c6aaf6fe diff --git a/pixi.toml b/pixi.toml index fbf10e6..b2aee05 100644 --- a/pixi.toml +++ b/pixi.toml @@ -32,6 +32,7 @@ pandas = ">=2.2.3" pandera-geopandas = ">=0.24" pyarrow = ">=19.0.1" scipy = ">=1.15.2" +country_converter = ">=1.3.2" [environments] module = { features = ["module"], no-default-feature = true } diff --git a/tests/conftest.py b/tests/conftest.py index 060ce52..71cbd09 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -54,7 +54,7 @@ def user_path(cds_api_file) -> Path: @pytest.fixture(scope="module") -def module_path(user_path: Path): +def module_path(cds_api_file: Path): """Parent directory of the project.""" module_path = Path(__file__).parent.parent @@ -66,7 +66,9 @@ def module_path(user_path: Path): shutil.rmtree(integration_dir / "results/", ignore_errors=True) # copy user inputs needed by the module - dir_to_dir_copy(user_path / "MNE", integration_dir / "resources/inputs/MNE") + dir_to_dir_copy( + module_path / "tests/files/MNE", integration_dir / "resources/inputs/MNE" + ) # copy downloads with unstable Github CI access dir_to_dir_copy( module_path / "tests/files/hydrobasins", diff --git a/tests/files/MNE/generation.parquet b/tests/files/MNE/generation.parquet new file mode 100644 index 0000000..653aee4 Binary files /dev/null and b/tests/files/MNE/generation.parquet differ diff --git a/tests/files/MNE/powerplants.parquet b/tests/files/MNE/powerplants.parquet new file mode 100644 index 0000000..d44188d Binary files /dev/null and b/tests/files/MNE/powerplants.parquet differ diff --git a/tests/files/MNE/shapes.parquet b/tests/files/MNE/shapes.parquet new file mode 100644 index 0000000..7fb118f Binary files /dev/null and b/tests/files/MNE/shapes.parquet differ diff --git a/tests/files/README.md b/tests/files/README.md index db2ae64..b756ceb 100644 --- a/tests/files/README.md +++ b/tests/files/README.md @@ -3,3 +3,4 @@ Used during integration CI tests. - `hydrobasins`: synthetic alternative to the `.zip` files downloaded from the HydroBASINS website. +- `MNE`: testfiles for a small-scale Montenegro run. diff --git a/tests/integration/Snakefile b/tests/integration/Snakefile index a150de8..64354ee 100644 --- a/tests/integration/Snakefile +++ b/tests/integration/Snakefile @@ -11,6 +11,7 @@ module module_hydropower: # Redirect specific user resources (inputs) shapes="resources/inputs/{shapes}/shapes.parquet", powerplants="resources/inputs/{shapes}/powerplants.parquet", + statistics_user="resources/inputs/{shapes}/generation.parquet", # Redirect specific module results (outputs) aggregated_inflow_pu="results/outputs/{shapes}/{plant_type}_pu.parquet", # Redirect module intermediate files diff --git a/tests/integration/test_config.yaml b/tests/integration/test_config.yaml index f553b01..e7c679c 100644 --- a/tests/integration/test_config.yaml +++ b/tests/integration/test_config.yaml @@ -6,6 +6,7 @@ module_hydropower: start: 2015 end: 2016 powerplants: + generation_adjustement: user technology_mapping: run_of_river: "run of river" reservoir: "reservoir" diff --git a/workflow/Snakefile b/workflow/Snakefile index 0acb1e8..8a5e16e 100644 --- a/workflow/Snakefile +++ b/workflow/Snakefile @@ -12,8 +12,9 @@ pathvars: # User resources shapes="/user/{shapes}/shapes.parquet", powerplants="/user/{shapes}/powerplants.parquet", + statistics_user="/user/{shapes}/generation.parquet", # Module results - statistics="/{shapes}/statistics/generation.parquet", + statistics_eia="/{shapes}/statistics/generation.parquet", disaggregated_inflow="/{shapes}/disaggregated/inflow_mwh.parquet", aggregated_inflow_pu="/{shapes}/aggregated/{plant_type}_inflow_pu.parquet", diff --git a/workflow/envs/module.linux-64.pin.txt b/workflow/envs/module.linux-64.pin.txt index 48be256..7b0369e 100644 --- a/workflow/envs/module.linux-64.pin.txt +++ b/workflow/envs/module.linux-64.pin.txt @@ -188,6 +188,7 @@ https://conda.anaconda.org/conda-forge/noarch/decorator-5.3.1-pyhd8ed1ab_0.conda https://conda.anaconda.org/conda-forge/noarch/ipython-9.14.1-pyh53cf698_0.conda#fbd58549b374103c1a80577f09a328ef https://conda.anaconda.org/conda-forge/noarch/ipdb-0.13.13-pyhd8ed1ab_1.conda#044c5249ad8ea18a414d07baa1f369ea https://conda.anaconda.org/conda-forge/linux-64/curl-8.21.0-hcf29cc6_0.conda#35caaabe13a4aa0fe77ea634c71478b8 +https://conda.anaconda.org/conda-forge/noarch/country_converter-1.3.2-pyhd8ed1ab_0.conda#193a9e54636d8d70781a3e56370f5502 https://conda.anaconda.org/conda-forge/linux-64/yaml-0.2.5-h280c20c_3.conda#a77f85f77be52ff59391544bfe73390a https://conda.anaconda.org/conda-forge/noarch/xarray-2026.4.0-pyhc364b38_0.conda#099794df685f800c3f319ff4742dc1bb https://conda.anaconda.org/conda-forge/noarch/tqdm-4.68.3-pyh8f84b5b_0.conda#65094960cb7ed216b6049aab57205902 diff --git a/workflow/envs/module.osx-arm64.pin.txt b/workflow/envs/module.osx-arm64.pin.txt index 4b3c3d5..7c8fe1b 100644 --- a/workflow/envs/module.osx-arm64.pin.txt +++ b/workflow/envs/module.osx-arm64.pin.txt @@ -177,6 +177,7 @@ https://conda.anaconda.org/conda-forge/noarch/decorator-5.3.1-pyhd8ed1ab_0.conda https://conda.anaconda.org/conda-forge/noarch/ipython-9.14.1-pyh53cf698_0.conda#fbd58549b374103c1a80577f09a328ef https://conda.anaconda.org/conda-forge/noarch/ipdb-0.13.13-pyhd8ed1ab_1.conda#044c5249ad8ea18a414d07baa1f369ea https://conda.anaconda.org/conda-forge/osx-arm64/curl-8.21.0-hd5a2499_0.conda#698b465be69a987c1d0f2fc969d7c986 +https://conda.anaconda.org/conda-forge/noarch/country_converter-1.3.2-pyhd8ed1ab_0.conda#193a9e54636d8d70781a3e56370f5502 https://conda.anaconda.org/conda-forge/osx-arm64/yaml-0.2.5-h925e9cb_3.conda#78a0fe9e9c50d2c381e8ee47e3ea437d https://conda.anaconda.org/conda-forge/noarch/xarray-2026.4.0-pyhc364b38_0.conda#099794df685f800c3f319ff4742dc1bb https://conda.anaconda.org/conda-forge/noarch/tqdm-4.68.3-pyh8f84b5b_0.conda#65094960cb7ed216b6049aab57205902 diff --git a/workflow/envs/module.win-64.pin.txt b/workflow/envs/module.win-64.pin.txt index 6cb7e18..51b0b58 100644 --- a/workflow/envs/module.win-64.pin.txt +++ b/workflow/envs/module.win-64.pin.txt @@ -170,6 +170,7 @@ https://conda.anaconda.org/conda-forge/noarch/colorama-0.4.6-pyhd8ed1ab_1.conda# https://conda.anaconda.org/conda-forge/noarch/ipython-9.14.1-pyhe2676ad_0.conda#bf12187c2d1ef0bb63df01ace31ff26b https://conda.anaconda.org/conda-forge/noarch/ipdb-0.13.13-pyhd8ed1ab_1.conda#044c5249ad8ea18a414d07baa1f369ea https://conda.anaconda.org/conda-forge/win-64/curl-8.21.0-h8206538_0.conda#b09f232e463d78e4cc0bec233007441b +https://conda.anaconda.org/conda-forge/noarch/country_converter-1.3.2-pyhd8ed1ab_0.conda#193a9e54636d8d70781a3e56370f5502 https://conda.anaconda.org/conda-forge/win-64/yaml-0.2.5-h6a83c73_3.conda#433699cba6602098ae8957a323da2664 https://conda.anaconda.org/conda-forge/noarch/xarray-2026.4.0-pyhc364b38_0.conda#099794df685f800c3f319ff4742dc1bb https://conda.anaconda.org/conda-forge/noarch/tqdm-4.68.3-pyha7b4d00_0.conda#0b814124391b7e176bcc0cce445a3a36 diff --git a/workflow/envs/module.yaml b/workflow/envs/module.yaml index f3c6175..a534518 100644 --- a/workflow/envs/module.yaml +++ b/workflow/envs/module.yaml @@ -13,3 +13,4 @@ dependencies: - pandera-geopandas >=0.24 - pyarrow >=19.0.1 - scipy >=1.15.2 +- country_converter >=1.3.2 diff --git a/workflow/internal/config.schema.yaml b/workflow/internal/config.schema.yaml index e03d30d..3015ce1 100644 --- a/workflow/internal/config.schema.yaml +++ b/workflow/internal/config.schema.yaml @@ -2,6 +2,12 @@ $schema: "https://json-schema.org/draft/2020-12/schema" description: "Schema for user-provided configuration files." type: object additionalProperties: false +required: + - pfafstetter_level + - years + - powerplants + - crs + - smoothing_hours properties: pfafstetter_level: description: > @@ -30,6 +36,15 @@ properties: description: Powerplant related settings. type: object properties: + generation_adjustement: + description: | + Specifies which file to use for powerplant inflow normalisation. + - EIA: uses the internally generated EIA statistics (results.statistics_eia). + - user: uses a user provided file instead. + type: string + enum: + - EIA + - user technology_mapping: description: > Maps internal names to user-provided names in powerplants files, to make integration smoother. @@ -75,21 +90,22 @@ properties: - max_dropped additionalProperties: false # Ensures no extra properties in "basin_adjustment" required: + - generation_adjustement - basin_adjustment - technology_mapping additionalProperties: false # Ensures no extra properties in "powerplants" crs: - description: Coordinate reference system settings in the form 'epsg:xxxx'. + description: Coordinate reference system settings in the form 'EPSG:xxxx'. type: object properties: projected: description: CRS to use during operations related to distance / area. type: string - pattern: "^epsg:\\d{4,5}$" + pattern: "^[A-Za-z][A-Za-z0-9_-]*:[A-Za-z0-9_.-]+$" geographic: description: CRS to use during operations related to location. type: string - pattern: "^epsg:\\d{4,5}$" + pattern: "^[A-Za-z][A-Za-z0-9_-]*:[A-Za-z0-9_.-]+$" required: - projected - geographic diff --git a/workflow/rules/_utils.smk b/workflow/rules/_utils.smk index 2d7fb27..221d563 100644 --- a/workflow/rules/_utils.smk +++ b/workflow/rules/_utils.smk @@ -24,3 +24,15 @@ def validate_year_config(): f"Start year must be less than end year. Found: '{start},{end}'. " "Remember: end year data is not included!" ) + + +def get_generation_adjustement_file() -> str: + """Identify which generation adjustment file to use.""" + source = config["powerplants"]["generation_adjustement"] + match source: + case "EIA": + return "" + case "user": + return "" + case _: + raise ValueError(f"Invalid generation adjustment option {source}") diff --git a/workflow/rules/powerplants.smk b/workflow/rules/powerplants.smk index 7f3c567..142f0b6 100644 --- a/workflow/rules/powerplants.smk +++ b/workflow/rules/powerplants.smk @@ -50,7 +50,7 @@ rule powerplants_get_inflow_mwh: input: inflow_m3=rules.powerplants_get_inflow_m3.output.inflow, adjusted_powerplants=rules.powerplants_adjust_location.output.adjusted_powerplants, - statistics="", + statistics=get_generation_adjustement_file(), output: inflow_mwh="", log: diff --git a/workflow/rules/statistics.smk b/workflow/rules/statistics.smk index 9559f1b..7b0435b 100644 --- a/workflow/rules/statistics.smk +++ b/workflow/rules/statistics.smk @@ -1,19 +1,32 @@ """Rules for collating and preparing hydropower national statistics.""" +rule harmonise_statistics: + input: + eia_bulk=rules.download_eia.output.zipfile, + output: + generation="/automatic/statistics/baseline.parquet", + log: + "/harmonise_statistics.log", + conda: + "../envs/module.yaml" + message: + "Harmonise EIA annual hydropower generation for all countries and years." + script: + "../scripts/harmonise_statistics.py" + + rule prepare_statistics: input: shapes="", - eia_bulk=rules.download_eia.output.zipfile, + statistics=rules.harmonise_statistics.output.generation, output: - generation="", + generation="", plot="/{shapes}/statistics/generation.pdf", log: "/{shapes}/prepare_statistics.log", conda: "../envs/module.yaml" - params: - years=config["years"], message: "Get EIA annual country hydropower generation statistics." script: diff --git a/workflow/scripts/_schemas.py b/workflow/scripts/_schemas.py index b984aa8..e628fa2 100644 --- a/workflow/scripts/_schemas.py +++ b/workflow/scripts/_schemas.py @@ -12,7 +12,7 @@ class Config: shape_id: Series[str] = Field(unique=True) "Unique ID for this shape." - country_id: Series[str] + country_id: Series[str] = Field(str_length={"min_value": 3, "max_value": 3}) "ISO alpha-3 code." shape_class: Series[str] = Field(isin=["land", "maritime"]) "Shape classifier" @@ -30,13 +30,14 @@ class EIAGenerationSchema(DataFrameModel): class Config: coerce = True strict = True + unique = ["country_id", "year", "category"] country_id: Series[str] - "Country ISO-3 code" + "EIA country or territory code, including historical and non-ISO codes" year: Series[int] "Sample year" category: Series[str] = Field(isin=["hydropower"]) - generation_mwh: Series[float] = Field(ge=0) + generation_mwh: Series[float] = Field(ge=0, nullable=True) "Hydropower generation for the given year" index: Index[int] = Field(unique=True) diff --git a/workflow/scripts/harmonise_statistics.py b/workflow/scripts/harmonise_statistics.py new file mode 100644 index 0000000..67d9696 --- /dev/null +++ b/workflow/scripts/harmonise_statistics.py @@ -0,0 +1,79 @@ +"""Extract annual hydropower generation for all EIA countries and territories.""" + +import io +import logging +import sys +import zipfile +from typing import TYPE_CHECKING, Any + +import _schemas +import country_converter as coco +import pandas as pd + +if TYPE_CHECKING: + snakemake: Any + +CAT_ID = { + "hydropower": 33 + # "pumped storage": 82, TODO: add pumped storage estimations in future updates +} + + +def log_unrecognised_codes(statistics: pd.DataFrame) -> None: + """Log codes that CoCo cannot identify.""" + codes = statistics["country_id"].drop_duplicates() + converter = coco.CountryConverter() + unrecognised = sorted(codes[~codes.isin(converter.data["ISO3"])]) + logging.info( + "coco does not recognise %d of %d country codes: %s", + len(unrecognised), + len(codes), + ", ".join(unrecognised) if unrecognised else "none", + ) + + +def harmonise(input_eia_bulk: str, output_generation: str): + """Extract all available years, retaining EIA country codes and missing values.""" + with zipfile.ZipFile(input_eia_bulk) as zf: + with zf.open("INTL.txt") as f: + eia_stats = pd.read_json(io.TextIOWrapper(f, encoding="utf-8"), lines=True) + + results = [] + for category, identifier in CAT_ID.items(): + series = eia_stats[ + eia_stats["series_id"].str.fullmatch( + rf"INTL\.{identifier}-12-.+-BKWH\.A", na=False + ) + ].copy() + series["country_id"] = series["series_id"].str.split("-").str[2] + # Aggregates and alternative series have a different country_id and geography. + # We force match to retain only countries, historical bodies and non-ISO territories. + series = series[series["country_id"] == series["geography"]] + + for country in series.itertuples(): + data = pd.DataFrame(country.data, columns=["year", "generation_mwh"]) + values = data["generation_mwh"] + data["generation_mwh"] = ( + pd.to_numeric( + values.mask(values.isin(["--", "ie", "NA"])), errors="raise" + ) + * 1e6 + ) # EIA data is in billion kWh. + data["country_id"] = country.country_id + data["category"] = category + results.append(data) + + statistics = _schemas.EIAGenerationSchema.validate( + pd.concat(results, ignore_index=True) + ) + log_unrecognised_codes(statistics) + statistics.to_parquet(output_generation) + + +if __name__ == "__main__": + sys.stderr = open(snakemake.log[0], "w", buffering=1) + logging.basicConfig(level=logging.INFO) + harmonise( + input_eia_bulk=snakemake.input.eia_bulk, + output_generation=snakemake.output.generation, + ) diff --git a/workflow/scripts/powerplants_get_inflow_mwh.py b/workflow/scripts/powerplants_get_inflow_mwh.py index 3b7807f..933d6c8 100644 --- a/workflow/scripts/powerplants_get_inflow_mwh.py +++ b/workflow/scripts/powerplants_get_inflow_mwh.py @@ -81,6 +81,14 @@ def _estimate_bounded_powerplant_inflow( assert annual_national_generation.index.is_unique, ( f"Country data for {year} is not unique!" ) + required_generation = annual_national_generation.reindex( + plants_by_id["country_id"].unique() + ) + missing = required_generation.index[required_generation.isna()].tolist() + if missing: + raise ValueError( + f"Missing or null generation statistics for {year}, countries: {missing}" + ) annual_powerplant_mwh = plants_by_id.apply( lambda x: ( annual_national_generation[x.country_id] diff --git a/workflow/scripts/prepare_statistics.py b/workflow/scripts/prepare_statistics.py index 0a13030..66a09c5 100644 --- a/workflow/scripts/prepare_statistics.py +++ b/workflow/scripts/prepare_statistics.py @@ -1,89 +1,43 @@ """Aggregated hydropower generation data at a country level.""" -import io +import logging import sys -import zipfile from typing import TYPE_CHECKING, Any import _plots import _schemas import geopandas as gpd -import numpy as np import pandas as pd from matplotlib import pyplot as plt if TYPE_CHECKING: snakemake: Any -CAT_ID = { - "hydropower": 33 - # "pumped storage": 82, TODO: add pumped storage estimations in future updates -} - -def _get_id_data(eia_df: pd.DataFrame, code: str) -> pd.DataFrame: - idx = eia_df[eia_df["series_id"] == code].index[0] - df = pd.DataFrame(eia_df.loc[idx, "data"], columns=["year", "value"]) - df = df.replace("NA", np.nan) - return df - - -def _get_generation_id_data( - eia_df: pd.DataFrame, country_a3: str, category_id: int -) -> pd.DataFrame: - """Returns annual generation in Billion KWh.""" - code = f"INTL.{category_id}-12-{country_a3}-BKWH.A" - return _get_id_data(eia_df, code) - - -def _get_country_hydro_generation(eia_df: pd.DataFrame, country_a3: str): - """Parse country generation from the EIA dataset.""" - results = [] - for category, identifier in CAT_ID.items(): - data = _get_generation_id_data(eia_df, country_a3, identifier) - data["category"] = category - results.append(data) - - country_generation = pd.concat(results, ignore_index=True) - country_generation.reset_index(drop=True) - country_generation["generation_mwh"] = ( - pd.to_numeric(country_generation.pop("value"), errors="coerce") * 1e6 - ) # EIA data is in Billion kWh - country_generation["country_id"] = country_a3 - return country_generation - - -def prepare( - input_shapes: str, input_eia_bulk: str, years: dict, output_generation: str -): - """Generate a file with annual hydropower generation statistics per country. +def prepare(input_shapes: str, input_statistics: str, output_generation: str): + """Select all available annual statistics for the countries in the shapes. Args: input_shapes (str): shapes parquet file. - input_eia_bulk (str): EIA bulk zip archive containing INTL.txt. - years (dict): dictionary with start/end years. + input_statistics (str): Harmonised EIA generation parquet file. output_generation (str): hydropower generation parquet file. """ shapes = gpd.read_parquet(input_shapes) shapes = _schemas.ShapeSchema.validate(shapes) - with zipfile.ZipFile(input_eia_bulk) as zf: - with zf.open("INTL.txt") as f: - eia_stats = pd.read_json(io.TextIOWrapper(f, encoding="utf-8"), lines=True) - - results = [] - for country in shapes["country_id"].unique(): - try: - results.append(_get_country_hydro_generation(eia_stats, country)) - except (ValueError, KeyError, IndexError) as ex: - raise ValueError(f"Failed to extract statistics for {country}") from ex - statistics = pd.concat(results, ignore_index=True).reset_index(drop=True) - - # Filter requested data and validate for completeness - statistics["year"] = statistics["year"].astype(int) - statistics = statistics[ - statistics["year"].isin(range(years["start"], years["end"])) - ] + statistics = _schemas.EIAGenerationSchema.validate( + pd.read_parquet(input_statistics) + ) + countries = shapes["country_id"].unique() + missing_countries = sorted(set(countries).difference(statistics["country_id"])) + if missing_countries: + raise ValueError( + f"Missing generation statistics for countries: {missing_countries}" + ) + statistics = statistics[statistics["country_id"].isin(countries)].reset_index( + drop=True + ) + statistics = _schemas.EIAGenerationSchema.validate(statistics) statistics.to_parquet(output_generation) @@ -130,12 +84,8 @@ def plot( _plots.draw_empty(ax, str(country)) continue - pivot = cats.pivot_table( - index="year", - columns="category", - values="generation_mwh", - aggfunc="sum", - fill_value=0, + pivot = cats.pivot( + index="year", columns="category", values="generation_mwh" ).sort_index() pivot.plot.bar(ax=ax, stacked=True, legend=False, zorder=1) @@ -161,10 +111,10 @@ def plot( if __name__ == "__main__": sys.stderr = open(snakemake.log[0], "w", buffering=1) + logging.basicConfig(level=logging.INFO) prepare( input_shapes=snakemake.input.shapes, - input_eia_bulk=snakemake.input.eia_bulk, - years=snakemake.params.years, + input_statistics=snakemake.input.statistics, output_generation=snakemake.output.generation, ) plot(