diff --git a/.github/PULL_REQUEST_TEMPLATE/dev_template.md b/.github/PULL_REQUEST_TEMPLATE/dev_template.md deleted file mode 100644 index 2484e2e..0000000 --- a/.github/PULL_REQUEST_TEMPLATE/dev_template.md +++ /dev/null @@ -1,25 +0,0 @@ ---- - -# Description - -Please include a summary of the changes. - -Fixes # (issue) - -# Instructions for Reviewer - -In order to test the code in this PR you need to ... - -Please pay special attention to ... - -# Checklist: - -- [ ] I have checked the code runs -- [ ] I have tested the code -- [ ] I have run `pre-commit` and addressed any issues not automatically fixed -- [ ] I have merged any new changes from `dev` -- [ ] I have documented the code - - [ ] Major functions have docstrings - - [ ] Appropriate information has been added to `README`s -- [ ] I have explained this PR above -- [ ] I have requested a code review diff --git a/.github/PULL_REQUEST_TEMPLATE/prod_template.md b/.github/PULL_REQUEST_TEMPLATE/prod_template.md index a7a1d90..cda089e 100644 --- a/.github/PULL_REQUEST_TEMPLATE/prod_template.md +++ b/.github/PULL_REQUEST_TEMPLATE/prod_template.md @@ -1,13 +1,31 @@ -## Dev β†’ Prod Promotion Checklist + + +--- + +## Description -### Pipeline description +## Type of promotion +- [ ] Existing pipeline - code/data update +- [ ] New pipeline - first promotion to prod + +## Instructions for reviewer + + + +## Checklist + ### Pipeline run -- [ ] Manual pipeline run completed successfully on dev after merging to dev +- [ ] Manual pipeline run completed successfully on dev after merging to dev (link the run below) - [ ] Dev S3 output spot-checked; row counts and values look reasonable - [ ] No unexpected nulls or schema changes in output parquet files +### Image freshness +- [ ] `dev-latest` in ECR was rebuilt from the tip of `dev` that includes this PR's changes + (check the "Build and push Docker Image to ECR" run in Actions, as promotion only re-tags + the existing image, it doesn't rebuild) + ### Code quality - [ ] run-tests workflow passed on this PR - [ ] No hardcoded credentials or environment-specific values @@ -17,5 +35,9 @@ - [ ] No changes to silver/gold dataset column names or types that would break Superset -### Notes +### After merging +- [ ] Trigger the "Run pipeline in prod" workflow to actually refresh prod data β€” merging this + PR only re-tags the image, it does not run the pipeline. Requires approval from at least one of named CODEOWNERS. + +## Notes diff --git a/.github/pull_request_template.md b/.github/pull_request_template.md new file mode 100644 index 0000000..181ab91 --- /dev/null +++ b/.github/pull_request_template.md @@ -0,0 +1,31 @@ + + +--- + +## Description + + + +Fixes # + +## Type of change + +- [ ] New pipeline +- [ ] Change to an existing pipeline +- [ ] Package / infrastructure change +- [ ] Docs only + +## Checklist + +- [ ] `uv run pytest` passes locally +- [ ] `uv run pre-commit run --all-files` passes (ruff, gitleaks, etc.) +- [ ] If this adds a new pipeline: it's registered in `pipelines.yaml` and has its own `README.md` +- [ ] I've merged the latest `dev` into this branch + +## Testing notes + + + +## Instructions for reviewer + + diff --git a/.gitignore b/.gitignore index f7e049a..3734b11 100644 --- a/.gitignore +++ b/.gitignore @@ -49,6 +49,11 @@ infrastructure/.env *.parquet /data/ +# Notebooks +# Scratch notebooks aren't meant to be committed β€” see docs/DEVELOPMENT.md +*.ipynb +.ipynb_checkpoints/ + # Logs *.log diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 67f49b4..6ce1112 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -21,7 +21,7 @@ repos: ### Python Tools ### - repo: https://github.com/astral-sh/ruff-pre-commit - rev: v0.15.20 + rev: v0.16.4 hooks: - id: ruff-check args: [--fix] @@ -30,7 +30,7 @@ repos: name: "🐍 python Β· Format with Ruff" - repo: https://github.com/abravalheri/validate-pyproject - rev: v0.25 + rev: '0.26' hooks: - id: validate-pyproject name: "🐍 python Β· Validate pyproject.toml" @@ -38,7 +38,7 @@ repos: ### Data & Config Validation ### - repo: https://github.com/python-jsonschema/check-jsonschema - rev: 0.37.4 + rev: 0.38.0 hooks: - id: check-github-workflows name: "πŸ™ github-actions Β· Validate gh workflow files" diff --git a/README.md b/README.md index a27a761..26d9d95 100644 --- a/README.md +++ b/README.md @@ -1,8 +1,49 @@ # ASF Mission Data -ETL pipelines for the ASF Policy Dashboard. +Data pipelines for fetching, processing and storing core ASF mission datasets to S3. -## Quick start +## How pipelines work + +Each pipeline has two or three stages: **bronze** (fetch and store raw data), **silver** (clean and transform into Parquet), and optionally **gold** (aggregate into dashboard-ready outputs). + +Stages are implemented as **[Hamilton](https://hamilton.dagworks.io/)** dataflows, where each function defines a transformation and its arguments declare its dependencies. Hamilton then automatically resolves the execution order of functions into a Directed Acyclic Graph (DAG). This gives us a consistent structure for writing pipelines, makes transforms easy to test in isolation, makes data dependencies explicit, and lets Hamilton render the pipeline as a visual graph. It also has built-in decorators for data quality checks. + +In production, pipelines run as Docker containers on AWS ECS, triggered via GitHub Actions. Data is stored in S3 with separate dev and prod buckets. + +## Project structure + +``` +asf_mission_data/ # Python package +β”œβ”€β”€ pipeline/ +β”‚ └── / # one directory per pipeline, each with its own README +β”œβ”€β”€ alerting.py # Slack alerting +β”œβ”€β”€ logging_utils.py # Logging utilities +β”œβ”€β”€ run.py # CLI entrypoint +β”œβ”€β”€ storage.py # Local and S3 read/write utilities +└── utils.py # Shared utility functions +.github/workflows/ # CI/CD workflows +docs/ # Guides and runbooks +infrastructure/ # AWS CDK infrastructure +scripts/ # Utility scripts +tests/ # Test suite +Dockerfile # Container image for running pipelines in AWS +pipelines.yaml # Pipeline registry +pyproject.toml # Project config and dependencies +``` + +## Docs + +| Guide | Covers | +|---|---| +| [Running pipelines](docs/running-pipelines.md) | Local, Docker, GitHub Actions, ad hoc AWS | +| [Development guide](docs/DEVELOPMENT.md) | Editor setup, testing, pre-commit | +| [Adding a pipeline](docs/adding-pipelines.md) | Creating a new ETL pipeline | +| [Contributing](docs/CONTRIBUTING.md) | PR process, code standards | +| [Infrastructure](infrastructure/README.md) | AWS CDK resources and deployment | + +Each pipeline has its own README at `asf_mission_data/pipeline//README.md`. For the full list of existing pipelines, see [`pipelines.yaml`](pipelines.yaml). + +## Setup ### Prerequisites @@ -39,221 +80,32 @@ uv sync # Activate virtual environment source .venv/bin/activate -# Or run commands directly without activating -uv run python -m asf_mission_data.run example --stage all -``` - -### Running pipelines locally - -```bash -# Set local mode (otherwise the default is the dev S3 bucket) -export DATA_MODE=LOCAL -export DATA_ROOT=/tmp/pipeline-dev - -# Run the example pipeline +# Or run commands directly without activating, e.g. uv run python -m asf_mission_data.run example --stage all - -# Check output -ls /tmp/pipeline-dev/ -``` - -Or use the `.env.example` file: - -```bash -cp .env.example .env -source .env -uv run python -m asf_mission_data.run example --stage all -``` - -For the full local-vs-AWS workflow, see [docs/running-pipelines.md](docs/running-pipelines.md). - -## Developer setup - -### Pre-commit and code formatting - -This project uses **pre-commit hooks** and **ruff** to automatically format and lint code. The best experience is to set up your editor to format on save, so issues are fixed before they reach the pre-commit hook. - -#### VS Code setup (recommended) - -1. **Install the Ruff extension** - Search for "Ruff" in VS Code extensions and install the official Astral extension -2. **Add to `.vscode/settings.json`** (You may need to first select the "Preferences: Open Workspace Settings (JSON)" command in the Command Palette (Mac: β‡§βŒ˜P Win: Ctrl + shift + P)): - -```json -{ - "[python]": { - "editor.defaultFormatter": "charliermarsh.ruff", - "editor.formatOnSave": true, - "editor.codeActionsOnSave": { - "source.fixAll.ruff": "explicit" - } - } -} -``` - -This auto-formats your Python files every time you save, so ruff issues are fixed before you commit. - -#### Command line - -You can also manually format/check code: - -```bash -# Auto-fix formatting and common issues -uv run ruff format . - -# Check for remaining linting issues -uv run ruff check . --fix -``` - -#### Pre-commit hooks - -Pre-commit hooks run automatically when you commit. If you see issues at commit time: - -```bash -# Install the pre-commit hooks (one-time setup) -uv run pre-commit install - -# This will run on every git commit and auto-fix what it can -# If it makes changes, stage them and commit again - -# Test hooks without committing -uv run pre-commit run -# or -uv run pre-commit run --all-files - -# Commit without running hooks -git commit --no-verify -``` - -## Project structure - -``` -asf_mission_data/ # Python package (pipeline code) -β”œβ”€β”€ storage.py # Storage abstraction (local/S3) -β”œβ”€β”€ alerting.py # Slack alerting utilities -└── pipeline/ # Pipeline implementations - └── example/ # Template pipeline -infrastructure/ # CDK infrastructure -β”œβ”€β”€ app.py # CDK entry point -β”œβ”€β”€ cdk.json # CDK configuration -β”œβ”€β”€ config/ # Environment configurations -β”‚ β”œβ”€β”€ environments.py # EnvironmentConfig dataclass -β”‚ β”œβ”€β”€ dev.py # Dev environment values -β”‚ └── prod.py # Prod environment values -└── stacks/ # CDK stacks - └── core_stack.py # Shared resources (S3, ECR, IAM) -tests/ # Test suite -docs/ # Documentation and runbooks -scripts/ # Utility scripts ``` ## Infrastructure Infrastructure is managed with [AWS CDK](https://aws.amazon.com/cdk/) (Python). -### Core resources (deployed) - -| Resource | Dev | Prod | -|----------|-----|------| -| S3 bucket | `asf-mission-data-dev` | `asf-mission-data-prod` | -| ECR repository | `asf-mission-data` | `asf-mission-data` | -| GitHub Actions IAM role | `asf-github-actions-dev` | `asf-github-actions-prod` | - -### Deploying infrastructure - -```bash -# Install CDK dependencies -uv sync --extra infrastructure -npm install -g aws-cdk - -# Deploy to dev -cd infrastructure -cdk deploy --context env=dev - -# Preview changes -cdk diff --context env=dev -``` - -See [infrastructure/README.md](infrastructure/README.md) for full documentation. - -## Creating a new pipeline - -*(TODO: Document after Hamilton spike β€” see docs/creating-a-pipeline.md)* - -## Testing pipelines - -For new bronze/silver pipelines in this repo, the recommended baseline is: - -- keep bronze tests focused on fetching and storing raw data correctly -- keep silver tests focused on transformation logic and data contracts -- add one local integration test that proves the real pipeline wiring works end to end without using S3 - -The example pipeline includes a complete test template: +**How this works, in three steps:** -- `tests/pipeline/example/test_bronze.py` shows how to mock an external source and assert the raw file plus metadata are written correctly -- `tests/pipeline/example/test_silver.py` shows how to test silver transform functions, validate a dataframe schema, and run a local integration test against a temporary directory -- `tests/pipeline/example/conftest.py` shows how to share sample input data and set `DATA_MODE=LOCAL` for tests +1. **Build**: our code (all pipelines) gets packaged into a single Docker + container image and pushed to ECR (a private store for container images). -When adding a new pipeline, aim to include at least: +2. **Run**: to run a specific pipeline, someone manually triggers a task on + ECS Fargate (AWS's "run a container without managing a server" service), + telling it which pipeline to run from that shared image. -1. one bronze unit test that mocks the upstream fetch -2. one bronze persistence test that checks the expected file and metadata paths -3. unit tests for each non-trivial silver transform -4. one schema or contract test that rejects invalid data -5. one local integration test that runs the real pipeline against `tmp_path` - -This split matters because different failures happen in different places: - -- bronze tests catch broken downloads, missing metadata, and incorrect storage paths -- silver unit tests catch parsing and transformation bugs -- schema tests catch subtle bad data before it reaches the canonical output -- integration tests catch wiring mistakes between storage, Hamilton, and parquet writes - -Run just the example pipeline tests with: - -```bash -uv run pytest tests/pipeline/example -``` - -Run the full suite with: - -```bash -uv run pytest -``` - -## Pipeline registry - -All pipelines are registered in `pipelines.yaml`. Update this file when adding a new pipeline. - -## Deployment - -For the supported ways to build images and run pipelines in AWS, see [docs/running-pipelines.md](docs/running-pipelines.md). - -## Docker - -Build the image with a canonical tag: - -```bash -docker build -t asf-mission-data . -``` - -Run a pipeline in a local filesystem-backed mode: - -```bash -mkdir -p /tmp/asf-mission-data - -docker run --rm \ - -e DATA_MODE=LOCAL \ - -e DATA_ROOT=/tmp/asf-mission-data \ - -v /tmp/asf-mission-data:/tmp/asf-mission-data \ - asf-mission-data \ - example --stage all -``` +3. **Land**: the pipeline writes its output data to an S3 bucket where it's picked up by downstream tools for analysis. -Notes: +> Note: there's no automatic scheduling yet (as of 31 July 2026), every run is triggered +> manually at the moment. -- The image tag above is `asf-mission-data`, not `asf_mission_data`. -- If you omit `DATA_MODE`/`DATA_ROOT`, the container defaults to the dev S3 location and expects the corresponding AWS runtime configuration. +See [infrastructure/README.md](infrastructure/README.md) for full CDK +documentation, or [docs/running-pipelines.md](docs/running-pipelines.md) +for how to actually trigger a run. -## Runbook +--- -Start with [docs/running-pipelines.md](docs/running-pipelines.md) for the standard local, GitHub UI, and ad hoc AWS run paths. +*Last updated: 31 July 2026 by Elysia Lucas* diff --git a/asf_mission_data/pipeline/energy_price_cap_levels_annex_9/README.md b/asf_mission_data/pipeline/energy_price_cap_levels_annex_9/README.md new file mode 100644 index 0000000..b703682 --- /dev/null +++ b/asf_mission_data/pipeline/energy_price_cap_levels_annex_9/README.md @@ -0,0 +1,78 @@ +# Energy Price Cap Levels Annex 9 + +Ingests Ofgem's published "Final levelised cap rates model (Annex 9)" Excel workbook and transforms it into datasets covering tariff component rates, price ratios, and annual bill contributions across fuel types, payment methods, and quarterly price cap periods. + +**Source:** Ofgem +**Update cadence:** Quarterly (two months before start of each price cap period) +**Pipeline run name:** `energy_price_cap_levels_annex_9` +**Storage prefix:** `energy_price_cap_levels/annex_9` + + +## Pipeline stages + +### Bronze + +- **Source**: [Energy price cap (default tariff) levels webpage](https://www.ofgem.gov.uk/energy-regulation/domestic-and-non-domestic/energy-pricing-rules/energy-price-cap/energy-price-cap-default-tariff-levels) +- **Method**: Web scraping +- **Output**: + - `Annex-9-Levelisation-allowance-methodology-and-levelised-cap-levels-v1.10-July-September-2026.xlsx` + - `Annex-9-Levelisation-allowance-methodology-and-levelised-cap-levels-v1.10-July-September-2026.xlsx.metadata.json` +- **Validators**: + - `LatestPriceCapFileUrlValidator` checks the scraped file URL contains the expected publication month for the current price cap period, guarding against scraping the wrong file. + - `LatestPriceCapValidator` checks the page heading matches the expected price cap period, guarding against scraping the wrong price cap period string. + +### Silver + +#### `1c_consumption_adjusted_levels` +- **Description**: Tidy format. One row per tariff component per fuel type, payment method, consumption type, and charge restriction period. +- **Input**: `1C Consumption adjusted levels` sheet from bronze Excel file +- **Output**: `1c_consumption_adjusted_levels.parquet` +- **Validators**: + - `ChargeRestrictionPeriodValidator`checks charge restriction period string formats are valid. - `PriceCapValidator` checks the price cap period in bronze metadata matches the expected current period. + - Pandera schema check on final silver table. + +### Gold + +#### `1c_consumption_adjusted_levels_with_vat` +- **Description**: Silver tariff component data with VAT added as a separate component and `Total_GB average` uprated to include VAT. Includes period-on-period change columns. +- **Input**: `1c_consumption_adjusted_levels` silver table +- **Output**: `1c_consumption_adjusted_levels_with_vat.parquet` +- **Validators**: + - `TariffComponentsTotalValidator` checks tariff components sum correctly by consumption, fuel, payment method, and period. + - Pandera schema check on final gold table. + +#### `tariff_component_rates` +- **Description**: Standing charges and unit prices for each tariff component, derived from nil and typical consumption values. Includes period-on-period change columns. +- **Input**: `1c_consumption_adjusted_levels` silver table +- **Output**: `tariff_component_rates.parquet` +- **Validators**: + - `TariffComponentsTotalValidator`. + - Pandera schema check on final gold table. + +#### `price_ratios` +- **Description**: Electricity-to-gas unit price ratio by payment method and price cap period. Includes period-on-period change columns. +- **Input**: `1c_consumption_adjusted_levels` silver table +- **Output**: `price_ratios.parquet` +- **Validators**: Pandera schema check on final gold table. + +#### `annual_bill_fixed_and_variable_component_contributions` +- **Description**: Annual bill split into standing charge and consumption-based cost contributions for each tariff component, fuel type, and payment method. Includes period-on-period change columns. +- **Input**: `1c_consumption_adjusted_levels` silver table +- **Output**: `annual_bill_fixed_and_variable_component_contributions.parquet` +- **Validators**: + - `TariffComponentsTotalValidator`. + - Pandera schema check on final gold table. + +--- + +## Notes + +1. All gold tables are derived from the `1c_consumption_adjusted_levels` silver table. +2. VAT is applied at 5% (`VAT = 0.05`). If this rate changes, update `VAT` in `config.py`. +3. Benchmark consumption values (i.e. the medium Typical Domestic Consumption Value, TDCV) used to back-calculate unit prices and standing charges from Annex 9 are defined in `BENCHMARK_CONSUMPTION` in `config.py`. TDCVs are used to set the energy price cap and help show what an average home spends on gas and electricity. Ofgem updates these every few years to make sure they still reflect how much energy people actually use. +4. `price_ratios` may contain null values if gas unit price is zero for a given period β€” this is handled intentionally to avoid misleading spikes in downstream charts. +5. Individual tariff components are often grouped into broader categories for ease of communication. Component-category mapping is defined and can be updated in `COMPONENT_CATEGORY_MAP` in `config.py`. + +--- + +*Last updated: 25 June 2026 by Elysia Lucas* diff --git a/asf_mission_data/pipeline/energy_price_cap_levels_annex_9/config.py b/asf_mission_data/pipeline/energy_price_cap_levels_annex_9/config.py index 5b911bd..a3afd7c 100644 --- a/asf_mission_data/pipeline/energy_price_cap_levels_annex_9/config.py +++ b/asf_mission_data/pipeline/energy_price_cap_levels_annex_9/config.py @@ -1,5 +1,5 @@ """ -Static configuration values for extracting Energy Price Cap Levels Annex 9 data from Ofgem. +Configuration constants for the Energy Price Cap Levels Annex 9 pipeline. """ DATASET_PREFIX = "energy_price_cap_levels/annex_9" @@ -57,9 +57,9 @@ SILVER_TABLES_NODES_MAP = {"1c Consumption adjusted levels": "silver_energy_price_cap_annex_9_1c_consumption_adjusted_levels_parquet"} BENCHMARK_CONSUMPTION = { # MWh per year - "Gas": 11.5, - "Electricity: Single-Rate Metering Arrangement": 2.7, - "Electricity: Multi-Register Metering Arrangement": 3.9, + "Gas": 9.5, # old TDCV before July 26 change was 11.5 + "Electricity: Single-Rate Metering Arrangement": 2.5, # old TDCV before July 26 change was 2.7 + "Electricity: Multi-Register Metering Arrangement": 3.4, # old TDCV before July 26 change was 3.9 } VAT = 0.05 diff --git a/asf_mission_data/pipeline/energy_price_cap_levels_annex_9/gold.py b/asf_mission_data/pipeline/energy_price_cap_levels_annex_9/gold.py index a8f2527..13adbe5 100644 --- a/asf_mission_data/pipeline/energy_price_cap_levels_annex_9/gold.py +++ b/asf_mission_data/pipeline/energy_price_cap_levels_annex_9/gold.py @@ -13,7 +13,6 @@ from asf_mission_data.pipeline.energy_price_cap_levels_annex_9.config import ( BENCHMARK_CONSUMPTION, COMPONENT_CATEGORY_MAP, - VAT, ) from asf_mission_data.pipeline.energy_price_cap_levels_annex_9.schemas import ( GOLD_1C_CONSUMPTION_ADJUSTED_LEVELS_WITH_VAT_SCHEMA, @@ -73,41 +72,70 @@ def consumption_adjusted_levels_with_vat_df( """Add VAT as a tariff component and uprate the total values to include VAT. This function derives VAT-inclusive tariff values from the silver dataset, and - creates a new tariff component representing VAT (calculated as 5% of the - `Total_GB average` component) and adds it as a separate row. It also uprates - the `Total_GB average` values so that they include VAT. + creates a new tariff component representing VAT (calculated as the difference + between the `Total inc VAT` and `Total_GB average` components) and adds it as + a separate row. It also uprates the `Total_GB average` values so that they + include VAT, using the corresponding `Total inc VAT` values. Args: silver_df (pd.DataFrame): Silver-layer Annex 9 DataFrame containing tariff components, consumption levels, and annual values before VAT - adjustments. + adjustments. Must contain both `Total_GB average` and + `Total inc VAT` tariff components for every fuel. Returns: - pd.DataFrame: DataFrame containing the original tariff components, - VAT as a separate component, and updated `Total_GB average` values - that include VAT. + pd.DataFrame: DataFrame containing the original tariff components + (excluding `Total inc VAT`), VAT as a separate component, and updated + `Total_GB average` values that include VAT. """ + for component in ("Total_GB average", "Total inc VAT"): + if not silver_df["Tariff component"].eq(component).any(): + raise ValueError(f"Expected tariff component '{component}' not found in silver_df.") - # Add VAT as individual tariff component - if not silver_df["Tariff component"].eq("Total_GB average").any(): - raise ValueError("Expected tariff component 'Total_GB average' not found in silver_df.") + # Columns that uniquely identify a row aside from "Tariff component", + # "value" and "metadata" + key_cols = [ + "Payment method", + "Fuel", + "Consumption", + "28AD Charge Restriction Period", + "28AD Charge Restriction Period start", + "28AD Charge Restriction Period end", + "28AD Charge Restriction Period interval", + ] - vat_rows = silver_df[silver_df["Tariff component"] == "Total_GB average"].copy() - vat_rows["Tariff component"] = "VAT" - vat_rows["value"] *= VAT + total_gb_avg = silver_df[silver_df["Tariff component"] == "Total_GB average"].copy() + total_inc_vat = silver_df[silver_df["Tariff component"] == "Total inc VAT"].copy() + + merged = total_gb_avg.merge( + total_inc_vat[key_cols + ["value"]], + on=key_cols, + how="left", + suffixes=("", "_inc_vat"), + validate="one_to_one", + ) - # Uprate Total_GB average to include VAT - uprated_silver_df = silver_df.copy() + if merged["value_inc_vat"].isna().any(): + raise ValueError("Some 'Total_GB average' rows have no matching 'Total inc VAT' row.") - uprated_silver_df.loc[ - (uprated_silver_df["Tariff component"] == "Total_GB average"), - "value", - ] *= 1 + VAT + # VAT component = Total inc VAT - Total_GB average + vat_rows = merged.copy() + vat_rows["value"] = vat_rows["value_inc_vat"] - vat_rows["value"] + vat_rows["Tariff component"] = "VAT" + vat_rows = vat_rows.drop(columns="value_inc_vat") + + # Uprated Total_GB average = Total inc VAT value + uprated_total_gb_avg = merged.copy() + uprated_total_gb_avg["value"] = uprated_total_gb_avg["value_inc_vat"] + uprated_total_gb_avg = uprated_total_gb_avg.drop(columns="value_inc_vat") - # Remove now redundant "Total inc VAT" rows that were present only in the Dual fuel table - uprated_silver_df = uprated_silver_df[uprated_silver_df["Tariff component"] != "Total inc VAT"] + # All other rows, unchanged (drop original Total_GB average and Total inc VAT rows) + other_rows = silver_df[~silver_df["Tariff component"].isin(["Total_GB average", "Total inc VAT"])].copy() - return pd.concat([uprated_silver_df, vat_rows], ignore_index=True) + return pd.concat( + [other_rows, uprated_total_gb_avg, vat_rows], + ignore_index=True, + ) @check_output( diff --git a/asf_mission_data/pipeline/energy_price_cap_levels_annex_9/oct_2026_changes_checks.py b/asf_mission_data/pipeline/energy_price_cap_levels_annex_9/oct_2026_changes_checks.py new file mode 100644 index 0000000..e6a6a24 --- /dev/null +++ b/asf_mission_data/pipeline/energy_price_cap_levels_annex_9/oct_2026_changes_checks.py @@ -0,0 +1,120 @@ +# %% [markdown] +# ### Checks for TDCV and electricity VAT changes for Oct - Dec 2026 and Jan - Mar 2027 price cap periods + +# %% +import pandas as pd + +from asf_mission_data import storage +from asf_mission_data.pipeline.energy_price_cap_levels_annex_9.config import ( + BENCHMARK_CONSUMPTION, +) + +# %% +silver_df = storage.read_parquet( + "s3://asf-mission-data-dev/data/silver/energy_price_cap_levels/annex_9/latest/1c_consumption_adjusted_levels/1c_consumption_adjusted_levels.parquet" +) + +# %% +start_date_to_check = "2026-10-01" + +# %% +# Levels table checks +gold_levels_df = storage.read_parquet( + "s3://asf-mission-data-dev/data/gold/energy_price_cap_levels/annex_9/latest/1c_consumption_adjusted_levels_with_vat/1c_consumption_adjusted_levels_with_vat.parquet" +) +df = gold_levels_df + +# %% +# VAT check, should be zero +df[ + (df["28AD Charge Restriction Period start"] == pd.to_datetime(start_date_to_check)) + & (df["Consumption"] == "Typical consumption") + & (df["Payment method"] == "Other Payment Method") + & (df["Fuel"] == "Electricity: Single-Rate Metering Arrangement") + & (df["Tariff component"] == "VAT") +] + +# %% +# Total bill check, should match final dual fuel bill matches what is published on Ofgem page +df = gold_levels_df +df[ + (df["28AD Charge Restriction Period start"] == pd.to_datetime(start_date_to_check)) + & (df["Consumption"] == "Typical consumption") + & (df["Payment method"] == "Other Payment Method") + & (df["Fuel"] == "Dual fuel (implied)") + & (df["Tariff component"] == "Total_GB average") +] + +# %% +# Electricity bill check +df = gold_levels_df +df[ + (df["28AD Charge Restriction Period start"] == pd.to_datetime(start_date_to_check)) + & (df["Consumption"] == "Typical consumption") + & (df["Payment method"] == "Other Payment Method") + & (df["Fuel"] == "Electricity: Single-Rate Metering Arrangement") + & (df["Tariff component"] == "Total_GB average") +] + +# %% +# Component rates and standing charges check +gold_tariff_component_rates_df = storage.read_parquet( + "s3://asf-mission-data-dev/data/gold/energy_price_cap_levels/annex_9/latest/tariff_component_rates/tariff_component_rates.parquet" +) +df = gold_tariff_component_rates_df + +# %% +# Electricity checks +df[ + (df["28AD Charge Restriction Period start"] == pd.to_datetime(start_date_to_check)) + & (df["Payment method"] == "Other Payment Method") + & (df["Fuel"] == "Electricity: Single-Rate Metering Arrangement") + & (df["Tariff component"] == "Total_GB average") +] + +# %% +# Gas checks +df[ + (df["28AD Charge Restriction Period start"] == pd.to_datetime(start_date_to_check)) + & (df["Payment method"] == "Other Payment Method") + & (df["Fuel"] == "Gas") + & (df["Tariff component"] == "Total_GB average") +] + +# %% +# Check annual_bill_fixed_and_variable_contributions_df +# Consumption-based cost (Β£/yr) should match the unit rate (incl VAT) * TDCV +gold_annual_bill_fixed_and_variable_contributions_df = storage.read_parquet( + "s3://asf-mission-data-dev/data/gold/energy_price_cap_levels/annex_9/latest/annual_bill_fixed_and_variable_component_contributions/annual_bill_fixed_and_variable_component_contributions.parquet" +) +df = gold_annual_bill_fixed_and_variable_contributions_df + +# %% +# Electricity checks +df[ + (df["28AD Charge Restriction Period start"] == pd.to_datetime(start_date_to_check)) + & (df["Payment method"] == "Other Payment Method") + & (df["Fuel"] == "Electricity: Single-Rate Metering Arrangement") + & (df["Tariff component"] == "Total_GB average") +] + +# %% +electricity_tdcv = BENCHMARK_CONSUMPTION.get("Electricity: Single-Rate Metering Arrangement") * 1_000 # kWh/year +electricity_unit_rate = 26.322252 # p/kWh +electricity_tdcv * electricity_unit_rate / 100 # should match annual consumption-based cost + +# %% +# Gas checks +df[ + (df["28AD Charge Restriction Period start"] == pd.to_datetime(start_date_to_check)) + & (df["Payment method"] == "Other Payment Method") + & (df["Fuel"] == "Gas") + & (df["Tariff component"] == "Total_GB average") +] + +# %% +gas_tdcv = BENCHMARK_CONSUMPTION.get("Gas") * 1_000 # kWh/year +gas_unit_rate = 7.966458 # p/kWh +gas_tdcv * gas_unit_rate / 100 # should match annual consumption-based cost + +# %% diff --git a/asf_mission_data/pipeline/example/README.md b/asf_mission_data/pipeline/example/README.md new file mode 100644 index 0000000..988e1c0 --- /dev/null +++ b/asf_mission_data/pipeline/example/README.md @@ -0,0 +1,44 @@ + +# Example (UK Bank Holidays) + + +Ingests the UK bank holidays dataset as a simple pipeline example. + +**Source:** GOV.UK +**Update cadence:** Annually +**Pipeline run name:** `example` +**Storage prefix:** `example` + + +## Pipeline stages + +### Bronze + +- **Source**: https://www.gov.uk/bank-holidays.json +- **Method**: HTTP GET request +- **Output**: + - `bank-holidays.json` + - `bank-holidays.json.metadata.json` +- **Validators**: None + +### Silver + +#### `bank_holidays` +- **Description**: One row per bank holiday per UK division. +- **Input**: `bank-holidays.json` +- **Output**: `bank_holidays.parquet` +- **Validators**: Pandera schema check - validates column types and that `division` is one of the three expected values. + +--- + +## Notes + + + +1. There are different bank holidays across England & Wales, Scotland and Northern Ireland. The `division` field therefore corresponds to UK geographical regions and can only be `england-and-wales`, `scotland`, `northern-ireland`. + +--- + + +*Last updated: 25 June 2026 by Elysia Lucas* diff --git a/asf_mission_data/pipeline/example/config.py b/asf_mission_data/pipeline/example/config.py index 1282824..956d4c6 100644 --- a/asf_mission_data/pipeline/example/config.py +++ b/asf_mission_data/pipeline/example/config.py @@ -1,3 +1,3 @@ -# TBC Individual config for pipeline in this file -# Source URLs, versions, table metadata, etc. -# Pipeline-specific config frozen dataclass +""" +Configuration constants for the Bank Holidays example pipeline. +""" diff --git a/asf_mission_data/pipeline/heat_pump_deployment_statistics/README.md b/asf_mission_data/pipeline/heat_pump_deployment_statistics/README.md new file mode 100644 index 0000000..ae591c4 --- /dev/null +++ b/asf_mission_data/pipeline/heat_pump_deployment_statistics/README.md @@ -0,0 +1,51 @@ +# Heat Pump Deployment Statistics + +Ingests data on the number of heat pumps installed in the UK, only includes those installed in existing properties (retrofit). Excludes installations in new builds or retrofit installs that are not MCS certified. + +**Source:** Department for Energy Security and Net Zero, GOV.UK +**Update cadence:** Quarterly +**Pipeline run name:** `heat_pump_deployment_statistics` +**Storage prefix:** `heat_pump_deployment_statistics` + + +## Pipeline stages + +### Bronze + +- **Source**: https://www.gov.uk/api/content/government/collections/heat-pump-deployment-statistics +- **Method**: GOV.UK Content API +- **Output**: + - `Heat_pump_deployment_quarterly_statistics_United_Kingdom_2026_Q1.xlsx` + - `Heat_pump_deployment_quarterly_statistics_United_Kingdom_2026_Q1.xlsx.metadata.json` +- **Validators**: + - `ExcelFileExtensionValidator` checks the downloaded file has a `.xlsx` extension. + - `WithinThreeCalendarMonthsValidator` checks the publication date returned by the API is within the last 3 calendar months, guarding against stale data. + +### Silver + +#### `table_1_1` +- **Description**: Tidy format. One row per installation quarter per heat pump type, for the UK. +- **Input**: Content in tab named `Table 1.1` from bronze Excel file +- **Output**: `table_1_1.parquet ` +- **Validators**: + - `StartStringValidator` on table name and source citation. + - Pandera schema checks on wide and final silver table. + +#### `table_1_2` +- **Description**: Tidy format. One row per installation quarter per country/region, for all heat pump types combined. +- **Input**: Content in tab named `Table 1.2` from bronze Excel file +- **Output**: `table_1_2.parquet ` +- **Validators**: + - `StartStringValidator` on table name and source citation. + - Pandera schema checks on wide and final silver table. + +--- + +## Notes + +1. This dataset has 'Official statistics in development' status meaning it is still undergoing methodological development and subject to change. +2. The source Excel workbook contains a `Notes` sheet with numbered footnotes that are referenced inline in the data tables (e.g. `[note 1]`). The pipeline resolves these into a `notes` column in the silver tables. If the workbook structure changes, e.g., if the `Notes` sheet is renamed or the footnote format changes, this logic may break silently. + +--- + +*Last updated: 25 June 2026 by Elysia Lucas* diff --git a/asf_mission_data/pipeline/heat_pump_deployment_statistics/config.py b/asf_mission_data/pipeline/heat_pump_deployment_statistics/config.py index 660cd1e..7ab1f32 100644 --- a/asf_mission_data/pipeline/heat_pump_deployment_statistics/config.py +++ b/asf_mission_data/pipeline/heat_pump_deployment_statistics/config.py @@ -1,5 +1,5 @@ """ -Static configuration values for extracting Heat Pump Deployment statistics data from DESNZ. +Configuration constants for the Heat Pump Deployment Statistics pipeline. """ DATASET_PREFIX = "heat_pump_deployment_statistics" diff --git a/docs/CONTRIBUTING.md b/docs/CONTRIBUTING.md new file mode 100644 index 0000000..848b69b --- /dev/null +++ b/docs/CONTRIBUTING.md @@ -0,0 +1,82 @@ +# Contributing + +This covers the human process for getting a change into production: branching, PRs, and review. For environment setup, testing, linting, and type checking, see [DEVELOPMENT.md](DEVELOPMENT.md). + +## Issues + +Work is tracked as GitHub issues, labelled along two axes. Apply one `area:` and one `type:` label to every issue. See the [repo's Labels page](https://github.com/nestauk/asf_mission_data/labels) for the full list. + +**`area:*`** (which part of the codebase the issue touches): + +| Label | Use for | +|---|---| +| `area:architecture` | Technical decisions, standards, conventions, operating model questions | +| `area:data-pipeline` | Work on a specific dataset pipeline | +| `area:platform` | The shared setup that packages, deploys, triggers, and runs pipelines | +| `area:shared-framework` | Reusable pipeline code, helpers, patterns, common building blocks | + +**`type:*`** (the nature of the work): + +| Label | Use for | +|---|---| +| `type:bug` | Something isn't working | +| `type:chore` | Maintenance work, dependency updates, config tweaks | +| `type:decision` | Architectural/design choices that need discussion or recording | +| `type:documentation` | Improvements or additions to documentation | +| `type:feature` | New capability | + +Two standalone labels don't fit either axis: `question` (further information requested) and `wontfix` (won't be worked on). + + +## Branching + +Branch off `dev`. If the work is tracked as an issue, name your branch `-` (e.g. `37-etl-code-for-heat-pump-deployment-statistics-data`) and reference the issue in your PR description (`Fixes #`) so it closes automatically on merge. Not every branch needs an issue behind it - a clear, descriptive branch name is what actually matters. + +## Developer workflow + +End-to-end path from a feature branch to a verified change in production, in two phases. + +### Phase 1: land the change on `dev` + +1. Branch off `dev` (e.g. `37-my-pipeline-change`). +2. Test the branch before opening a PR: + - Manually trigger `Build and push Docker Image to ECR` to build an image tagged from your branch. + - Manually trigger `Test pipeline in dev` against that image tag and spot-check the dev S3 output. + - Fix any issues before opening a PR. +3. Open a PR into `dev`. The PR body auto-fills from [`.github/pull_request_template.md`](../.github/pull_request_template.md). + - `run-tests` and `pre-commit` run automatically. + - Can be approved by anyone; self-merge is allowed. +4. Merge to `dev`. This automatically triggers `Build and push Docker Image to ECR`, refreshing the `dev-latest` image. + +### Phase 2: promote `dev` to `prod` + +5. Re-verify on dev: manually trigger `Test pipeline in dev` with `image_tag=dev-latest` and spot-check the S3 output again. If it fails, fix it in a new branch and repeat Phase 1 before continuing. +6. Open a PR from `dev` into `prod`. See [Using the prod PR template](#using-the-prod-pr-template) below. The default template that loads here is the *dev* one, so you must switch it manually. + - `run-tests` and `pre-commit` run automatically. + - No self-merge and requires approval from a CODEOWNER (Elysia, Dan, or Alex - see [`.github/CODEOWNERS`](../.github/CODEOWNERS)). +7. Approver merges to `prod`. This triggers `promote-to-prod.yaml`, which re-tags the existing `dev-latest` image as `prod-latest` (it does not rebuild). +8. Approver triggers `Run pipeline in prod`, spot-checks the prod S3 output, and confirms the Superset datasets look correct. + +There's no separate rollback process. If step 8 turns up a problem, fix it the normal way: branch from `dev`, apply the fix, and repeat both phases to re-promote. + +## Using the prod PR template + +There is a different PR template that should be used when merging `dev` to `prod`. + +[`.github/pull_request_template.md`](../.github/pull_request_template.md) is the repo-wide default and auto-fills *every* new PR, including ones targeting `prod` (GitHub has no way to pick a different template per target branch). + +To load [`prod_template.md`](../.github/PULL_REQUEST_TEMPLATE/prod_template.md) instead, open the PR via this URL, which selects it with a `template` query parameter: + +``` +https://github.com/nestauk/asf_mission_data/compare/prod...dev?quick_pull=1&template=prod_template.md +``` + +**Already opened the PR with the dev checklist by mistake?** Just replace the body with the prod checklist manually - no need to close and reopen. + +## Code standards + +See [DEVELOPMENT.md's Package standards](DEVELOPMENT.md#package-standards): typed, tested, linted, deterministic. Ruff and tests are enforced by CI. + +--- + +*Last updated: 2 July 2026 by Elysia Lucas* diff --git a/docs/DEVELOPMENT.md b/docs/DEVELOPMENT.md index b841b56..055606d 100644 --- a/docs/DEVELOPMENT.md +++ b/docs/DEVELOPMENT.md @@ -1,56 +1,204 @@ # Development Guide +Day-to-day workflow for working in this repo: environment setup, notebooks, testing, linting, and type checking. For project structure and architecture, see the [README](../README.md). For adding a new pipeline, see [adding-pipelines.md](../docs/adding-pipelines.md). For running pipelines locally or in AWS, see [running-pipelines.md](../docs/running-pipelines.md). + +## Contents + +- [Setting up your environment](#setting-up-your-environment) +- [Managing dependencies](#managing-dependencies) +- [Git workflow](#git-workflow) +- [Editor setup](#editor-setup) +- [Linting and formatting](#linting-and-formatting) +- [Type checking](#type-checking) +- [Pre-commit hooks](#pre-commit-hooks) +- [Testing](#testing) +- [Notebooks](#notebooks) +- [Package standards](#package-standards) + ## Setting up your environment ```bash # Install all dependencies including dev tools uv sync --group dev -# Install pre-commit hooks +# Install pre-commit hooks (see "Pre-commit hooks" below) uv run pre-commit install ``` ---- +If you're running pipelines locally, also copy the example environment file: + +```bash +cp .env.example .env +source .env +``` -## Starting a notebook +`.env` is gitignored. `.envrc` will auto-source it for you if you use [direnv](https://direnv.net/). See [`.env.example`](../.env.example) for the available variables, and [running-pipelines.md](running-pipelines.md) for how `DATA_MODE`/`DATA_ROOT` affect where a pipeline reads and writes data. -JupyterLab is included in the dev dependencies. To launch it: +## Managing dependencies + +Dependencies are declared in `pyproject.toml` and pinned in `uv.lock`. Don't manually edit. Use `uv add`/`uv remove`, which update both together. ```bash -uv run jupyter lab +# Add a runtime dependency +uv add + +# Add a dev-only dependency (linting, testing, notebooks, etc.) +uv add --group dev + +# Add an infrastructure-only dependency (CDK, etc.) +uv add --group infrastructure + +# Remove a dependency +uv remove ``` -This opens JupyterLab in your browser. Create your notebook inside the `notebooks/` folder. +These commands update `pyproject.toml`, re-resolve `uv.lock`, and sync your `.venv` in one step. -To import from the package inside a notebook, the package is already available because `uv sync` installs it in editable mode: +To pick up newer versions of dependencies you already have: -```python -from asf_mission_data.transforms import clean_installations +```bash +# Upgrade everything to the latest versions allowed by pyproject.toml +uv lock --upgrade + +# Upgrade a single package +uv lock --upgrade-package + +# Then sync your environment to match +uv sync --group dev ``` -This means your notebook is always running against the real package code rather than reimplementing logic inline, which makes the eventual translation to a `.py` module much easier. +Always commit `uv.lock` alongside any `pyproject.toml` change. CI and other developers install from the lockfile, they don't re-resolve. ---- +## Git workflow -## `asf_mission_data/` β€” the package +- `dev` is the integration branch. Branch off `dev` for new work and open a PR back into it. Direct commits to `dev` are blocked by a pre-commit hook (`no-commit-to-branch`). +- `prod` tracks what's live in production. Merging a PR from `dev` into `prod` triggers `promote-to-prod.yaml`, which re-tags the already-tested `dev-latest` ECR image as `prod-latest`. It does not rebuild the image, so what you tested in dev is exactly what runs in prod. +- CI (`run-tests.yaml`) runs `uv run pytest` on every PR into `dev`. Pre-commit checks (see below) also run in CI via `pre-commit.yaml`. -When you're ready to move code out of a notebook, ideally all code in the package should be: +See [CONTRIBUTING.md](CONTRIBUTING.md) for the PR process itself. -- **Typed** β€” all functions should have type annotations -- **Tested** β€” covered by tests in `tests/` -- **Linted** β€” passes ruff and mypy checks (enforced by pre-commit) -- **Deterministic** β€” no hardcoded local paths, no side effects on import -Once the logic is in the package and tested, any notebooks have served their purpose and can be discarded. +## Editor setup ---- +### VS Code (recommended) -## Running tests and checks +1. Install the **Ruff** extension β€” search for "Ruff" in the VS Code extensions marketplace and install the official Astral extension. +2. Add to `.vscode/settings.json` (Command Palette β†’ "Preferences: Open Workspace Settings (JSON)"): + + ```json + { + "[python]": { + "editor.defaultFormatter": "charliermarsh.ruff", + "editor.formatOnSave": true, + "editor.codeActionsOnSave": { + "source.fixAll.ruff": "explicit" + } + } + } + ``` + + This formats and fixes Python files on save, so most ruff issues are resolved before you commit. + +## Linting and formatting + +This project uses **[ruff](https://docs.astral.sh/ruff/)** for both linting and formatting, configured in `pyproject.toml`. Rules currently enabled are pycodestyle, pyflakes, isort, and flake8-bugbear (`E`, `F`, `I`, `W`, `B`); stricter rule sets like type-annotation and docstring checks are left off for now while pipeline patterns are still settling. ```bash -# Run the test suite -uv run pytest +# Auto-fix formatting and common issues +uv run ruff format . + +# Check for remaining linting issues +uv run ruff check . --fix +``` + +Ruff is enforced in CI and pre-commit. A PR with lint or format issues will fail before it can merge. + + +## Type checking + +Type annotations are expected on all functions in the package (see [Package standards](#package-standards) below), and `mypy` is configured in `pyproject.toml` with a fairly strict profile (`disallow_untyped_defs`, `disallow_any_generics`, `warn_return_any`). + +```bash +uv run mypy asf_mission_data +``` + +**Note:** mypy isn't run by pre-commit or CI, so type errors won't block a PR. Run it yourself before opening one if you want to run those checks. + + +## Pre-commit hooks + +Pre-commit hooks run automatically on `git commit` and cover linting/formatting (ruff), secret scanning (gitleaks), config validation (`pyproject.toml`, GitHub workflow YAML), and basic file hygiene (trailing whitespace, large files, merge conflicts, direct commits to `dev`). The full list is in [`.pre-commit-config.yaml`](../.pre-commit-config.yaml). + +```bash +# One-time setup +uv run pre-commit install + +# Run on staged files only (what happens automatically at commit time) +uv run pre-commit run -# Run all pre-commit checks manually +# Run on the whole repo uv run pre-commit run --all-files + +# Skip hooks for a commit (avoid unless you have a good reason) +git commit --no-verify +``` + +If a hook modifies files (e.g. ruff auto-fixes something), the commit is aborted. Stage the changes it made and commit again. + + +## Testing + +Tests live in `tests/`, mirroring the package layout (e.g. `asf_mission_data/storage.py` β†’ `tests/test_storage.py`, pipeline tests under `tests/pipeline//`). + +```bash +# Run the full test suite with coverage +uv run pytest + +# Run a single file or test +uv run pytest tests/test_storage.py +uv run pytest tests/test_storage.py -k test_get_data_path_defaults_to_dev_bucket +``` + +`pytest.ini_options` in `pyproject.toml` runs with `--cov=asf_mission_data --cov-report=term-missing` by default, so every run prints a coverage summary with the line numbers that aren't covered. + +For anything that reads environment variables, hits S3, or shells out, use `pytest`'s `monkeypatch` fixture (or `pytest-mock`'s `mocker`) rather than mutating real environment/global state. Existing tests in `tests/test_storage.py` and `tests/test_trigger_pipeline.py` are good examples of the pattern: + +```python +def test_get_data_path_defaults_to_dev_bucket(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("DATA_ROOT", raising=False) + assert get_data_path("data/example/file.csv") == "s3://asf-mission-data-dev/data/example/file.csv" ``` + +CI runs the same `uv run pytest` command on every PR into `dev`; a failing test blocks the merge. + + +## Notebooks + +JupyterLab is included in the dev dependencies. To launch it: + +```bash +uv run jupyter lab +``` + +Notebooks (`*.ipynb`) are gitignored wherever they live, so put yours anywhere convenient (a `notebooks/` folder at the repo root is a reasonable default). `marimo` is also available as a dev dependency if you prefer its reactive, git-friendly notebook format (`uv run marimo edit`). + +Because `uv sync` installs the package in editable mode, you can import directly from it in a notebook: + +```python +from asf_mission_data.storage import read_parquet +``` + +Call the real package functions from your notebook instead of retyping the same logic there. That way there's only one copy of the logic, and moving it into a `.py` module later is just cut-and-paste. Once it's in the package and tested, you're done with the notebook. + +## Package standards + +When you're ready to move code out of a notebook and into `asf_mission_data/`, it should be: + +- **Typed** - all functions have type annotations. This is what `mypy` checks (see [Type checking](#type-checking)) β€” run it before opening a PR, since it isn't automated yet. +- **Tested** - covered by tests in `tests/`. Aim to cover fetch, transform, and validation logic, not just the happy path. +- **Linted** - passes `ruff format` and `ruff check`. This part *is* enforced automatically by pre-commit and CI. +- **Deterministic** - no hardcoded local paths, no side effects on import. Anything environment-specific (paths, credentials, endpoints) should come from `asf_mission_data.storage` or environment variables, not be hardcoded. + +--- + +*Last updated: 2 July 2026 by Elysia Lucas* diff --git a/docs/adding-pipelines.md b/docs/adding-pipelines.md new file mode 100644 index 0000000..aeee4a0 --- /dev/null +++ b/docs/adding-pipelines.md @@ -0,0 +1,451 @@ + +# Adding a new pipeline + +This guide walks through adding a new data pipeline to this repo. + +## New pipeline vs. extending an existing one +A new pipeline is needed when data from a new source needs to be ingested: **a new pipeline per distinct data source**, not per dataset variation from a source already ingested. +- Extending an existing pipeline (adding a new table/output) fits inside its existing bronze/silver/gold stages if it comes from the same source and shares fetch logic. +- A genuinely new source (new publisher, new URL, new fetch/parsing logic) warrants a new pipeline directory. + +## General pipeline structure + +A pipeline has two or three stages: + +- **Bronze** – fetch and store raw data +- **Silver** – clean and transform data +- **Gold** (optional) – aggregated outputs + +Every pipeline must include at least the bronze and silver stages. All data written to S3 must be in Parquet format, so it can be read into DuckLake. Each pipeline runs as an ECS task on AWS. + +This guide covers building and registering the pipeline. For running it locally or in AWS once it's built, see [`running-pipelines.md`](../docs/running-pipelines.md). + +### Hamilton + +All pipeline stages are written as **[Hamilton](https://hamilton.dagworks.io/)** dataflows. The core pattern is the same throughout: **each function is a node, and its argument names declare its dependencies.** Hamilton reads the module, matches argument names to other function names, and resolves the execution order automatically. + +e.g., in the `example` pipeline silver stage: + +```python +def bronze_bank_holidays_json(bronze_bank_holidays_uri: str) -> dict: + return storage.read_json(bronze_bank_holidays_uri) + +def flattened_bank_holidays_df(bronze_bank_holidays_json: dict) -> pd.DataFrame: + ... + +def parsed_bank_holidays_df(flattened_bank_holidays_df: pd.DataFrame) -> pd.DataFrame: + ... + +@check_output(schema=SILVER_BANK_HOLIDAYS_SCHEMA, importance="fail") +def validated_bank_holidays_df(parsed_bank_holidays_df: pd.DataFrame) -> pd.DataFrame: + ... +``` + +The `@check_output` decorator on `validated_bank_holidays_df` fails the pipeline if the node's output doesn't match `SILVER_BANK_HOLIDAYS_SCHEMA`. + +Calling for `validated_bank_holidays_df` triggers the full chain: `parsed_bank_holidays_df`, then `flattened_bank_holidays_df`, then `bronze_bank_holidays_json`, and so on. Hamilton works backwards from the target you name, running only what's needed to produce it. + +Some arguments, like `bronze_bank_holidays_uri` above, aren't produced by another function; they're constants such as `dataset_prefix` or `collection_url`. These are supplied via `with_config()` when building the driver in `pipeline.py`: + +```python +dr = driver.Builder().with_modules(silver).with_config({ + "dataset_prefix": DATASET_PREFIX, +}).build() +``` + +Config values are matched by name the same way function outputs are, so `dataset_prefix` in the config satisfies any function argument named `dataset_prefix`. + +The driver is then executed against named output nodes: + +```python +results = driver.execute(["silver_bank_holidays_parquet", "latest_publication_date"]) +``` + +For a bit more background on why we use Hamilton, see the [project README](../README.md). + + +## Pipeline creation steps +Main steps (with further detail under each section below): +1. **Scaffold the pipeline** - create the pipeline directory and its files +2. **Register the pipeline** - add an entry to `pipelines.yaml` +3. **Implement bronze** - fetch and store raw data +4. **Implement silver** - clean, validate and persist transformed data +5. **Implement gold (optional)** - transform silver data further into aggregated outputs +6. **Create pipeline entrypoint** - assemble the stages in `pipeline.py` +7. **Write tests (recommended)** - cover fetch, transform, and validation logic +8. **Verify locally** - run all stages and check outputs +9. **Write a pipeline README** - complete `templates/pipelines/README.md` +10. **Open a PR** - land it on `dev`, then promote to `prod` + +## 1. Scaffold the pipeline + +Create a directory for your pipeline under `asf_mission_data/pipeline/`: + +``` +asf_mission_data/pipeline// +β”œβ”€β”€ __init__.py # empty, marks this as a Python package +β”œβ”€β”€ config.py # constants: URLs, dataset prefix, publisher, table names +β”œβ”€β”€ bronze.py # Hamilton nodes for fetching and storing raw data +β”œβ”€β”€ silver.py # Hamilton nodes for cleaning and transforming data +β”œβ”€β”€ pipeline.py # builds drivers and defines run() entry point +β”œβ”€β”€ schemas.py # pandera schemas for validating silver output +β”œβ”€β”€ validators.py # custom Hamilton validators (if needed) +└── README.md # see step 9 +``` + +`gold.py` follows the same pattern as `silver.py` and is only needed if the pipeline produces aggregated outputs. + +`validators.py` is only needed if you write custom `@check_output_custom` validators. If you only use the built-in `@check_output` decorator with a pandera schema, you don't need it. + +## 2. Register the pipeline + +Add an entry to [`pipelines.yaml`](../pipelines.yaml) at the root of the repo: + +```yaml +pipelines: + your_pipeline_name: + owner: your_name + schedule: + description: One sentence describing what data this pipeline fetches + source_url: https://example.gov.uk/the-source-page + stages: [bronze, silver] +``` + +The pipeline name must be a unique key and must match the directory name you created under `asf_mission_data/pipeline/`. It is also the value you pass to `--pipeline` when running the pipeline via GitHub Actions or the trigger script. + +Include `gold` in `stages` if your pipeline has a gold stage. + +`pipelines.yaml` is config, not documentation. It's read directly by GitHub Actions (and, once implemented, by the EventBridge schedule,) so `schedule` and `stages` must be accurate for the pipeline to run correctly. `description` here should stay to one sentence, since it's surfaced in tooling rather than read as prose. Fuller documentation (what the pipeline does, quirks in the source, update frequency in human terms) belongs in the pipeline's own `README.md` (see step 9). + +**On `schedule`:** all pipeline runs are currently manual β€” nothing reads this field yet. The agreed design (27 July 2026, not yet implemented) is for `pipelines.yaml` to become the source of truth for scheduling: a reconciler script will converge EventBridge Scheduler with the file's `schedule` values, creating/updating/deleting schedules to match. Scheduling will be prod-only; dev runs stay manual by design. Set `schedule` to your intended cadence now if you know it, but don't expect it to have any effect until this ships. See [infrastructure.md](../infrastructure/README.md#i-want-to-schedule-a-pipeline) for details. + + + +## 3. Implement bronze + +Bronze fetches raw data from the source and stores it unchanged, no transformation happens here. + +### `config.py` + +Put all constants here: source URLs, the dataset prefix, publisher name, and any other fixed values the pipeline needs. + +```python +DATASET_PREFIX = "heat_pump_deployment_statistics" +PUBLISHER = "Department for Energy Security and Net Zero" +COLLECTION_URL = "https://www.gov.uk/government/collections/heat-pump-deployment-statistics" +``` + +These are passed into the Hamilton driver via `with_config()` in `pipeline.py`, which makes them available as arguments in any node. + +### `bronze.py` + +The bronze module typically follows this shape: + +1. **Discover the source** - fetch an API response or scrape a page to find where the latest file is +2. **Extract what you need** - file URL, filename, publication date +3. **Fetch the file** - download raw bytes +4. **Build metadata** - a `bronze_metadata` node that records provenance +5. **Persist** - a terminal node that calls `storage.ingest_to_bronze()` which ingests the bronze file itself alongside its accompanying metadata + +**Steps 1–2** sometimes need validation - for example, checking a discovered file has the right extension or that its publication date is recent. Add a custom validator to `validators.py` and apply it with `@check_output_custom`: + +```python +@check_output_custom(ExcelFileExtensionValidator()) +def latest_filename(latest_file_url: str) -> str: + return Path(latest_file_url).name +``` + +See [heat_pump_deployment_statistics/validators.py](../asf_mission_data/pipeline/heat_pump_deployment_statistics/validators.py) for how to implement one. + +**Step 4 - metadata:** The `bronze_metadata` node should capture enough to reconstruct where the data came from. This is the core set of fields, but should be amended to what makes most sense for the pipeline you're writing. This metadata dictionary is expected to flow through to sit alongside downstream silver and gold data too. + +```python +def bronze_metadata( + publisher: str, + collection_url: str, + latest_file_url: str, + latest_filename: str, + latest_publication_date: str, + bronze_ingest_timestamp: str, + pipeline_version: str, +) -> dict[str, str]: + return { + "publisher": publisher, + "collection_url": collection_url, + "file_url": latest_file_url, + "filename": latest_filename, + "publication_date": latest_publication_date, + "bronze_ingest_timestamp": bronze_ingest_timestamp, + "pipeline_version": pipeline_version, + "citation": f"Source: {publisher}, {latest_filename}. Published {latest_publication_date}.", + } +``` + +**Step 5 - persist:** The terminal node is the final ingestion step for the bronze file and its metadata. + +```python +def bronze__file( + dataset_prefix: str, + latest_file_content: bytes, + latest_filename: str, + latest_publication_date: str, + bronze_metadata: dict, +) -> None: + storage.ingest_to_bronze( + layer_prefix="bronze", + dataset_prefix=dataset_prefix, + file=latest_file_content, + filename=latest_filename, + date_stamp=f"published={utils.normalise_date_string(latest_publication_date)}", + metadata=bronze_metadata, + ) +``` + +See [heat_pump_deployment_statistics/bronze.py](../asf_mission_data/pipeline/heat_pump_deployment_statistics/bronze.py) for a complete worked example. + +## 4. Implement silver + +Silver reads raw data from the bronze layer, transforms it into clean, structured tables, and persists them as parquet. The exact structure of nodes will depend on what the source data looks like. + +### `silver.py` + +The silver module typically follows this shape: + +1. **Locate and load bronze** - use `storage.locate_latest()` to find the latest bronze file, then read it with the matching `storage.read_*()` function for its file type (e.g. `storage.read_json()`, `storage.read_excel_sheet()`) +2. **Read publication date from bronze metadata** - use this to date-stamp the silver output +3. **Transform** - parse, clean, reshape the raw data into a tidy DataFrame; metadata is added as a separate column where each row contains the metadata dictionary +4. **Validate** - apply `@check_output` with a schema from `schemas.py` +5. **Persist** - call `storage.ingest_to_silver()` in the terminal node + +### `schemas.py` + +Define a pandera schema for each output table here. These are referenced by `@check_output` in step 4 above, and validated before the DataFrame is persisted: + +```python +import pandera as pa + +SILVER_BANK_HOLIDAYS_SCHEMA = pa.DataFrameSchema({ + "division": pa.Column(str), + "title": pa.Column(str), + "date": pa.Column(pa.DateTime), + ... +}) +``` + +**Step 5 - persist:** The terminal node follows the same pattern as bronze: it returns the DataFrame for convenience, but its main job is writing to storage. Unlike bronze, `storage.ingest_to_silver()` does not take the metadata as an argument as the metadata is expected to be a written into a separate column in the silver data. + +```python +def silver__parquet( + validated_df: pd.DataFrame, + dataset_prefix: str, + latest_publication_date: str, +) -> pd.DataFrame: + storage.ingest_to_silver( + dataset_prefix=dataset_prefix, + df=validated_df, + df_name="", + date_stamp=f"published={utils.normalise_date_string(latest_publication_date)}", + ) + return validated_df +``` + +See [heat_pump_deployment_statistics/silver.py](../asf_mission_data/pipeline/heat_pump_deployment_statistics/silver.py) for a worked example of a pipeline with multiple silver output tables. + +## 5. Implement gold (optional) + +Gold produces aggregated, dashboard-ready outputs derived from the silver layer. Only add a gold stage if the pipeline needs outputs that go beyond the cleaned silver tables; for example, derived metrics, ratios, or reshaped views. + +### `gold.py` + +The pattern is identical to silver, except gold reads from silver storage instead of bronze: + +1. **Load silver** - use `storage.locate_latest()` to find the latest silver file(s), then use `storage.read_parquet()` +2. **Transform** - aggregate, derive metrics, or reshape into the gold output +3. **Validate** - apply `@check_output` with a schema from `schemas.py` +4. **Persist** - call `storage.ingest_to_gold()` in the terminal node + + +Schemas for gold tables go in the same `schemas.py` file as silver schemas. + +See [energy_price_cap_levels_annex_9/gold.py](../asf_mission_data/pipeline/energy_price_cap_levels_annex_9/gold.py) for a worked example of a pipeline with multiple gold output tables. + +## 6. Create pipeline entrypoint + +`pipeline.py` wires the Hamilton modules together into runnable stages. It has three responsibilities: building drivers, running stages, and exposing a `run()` entry point the CLI calls. + +### Building drivers + +Each stage gets its own `build__driver()` function. Pass the stage module to `with_modules()` and all config values to `with_config()`: + +```python +def build_bronze_driver() -> driver.Driver: + return ( + driver.Builder() + .with_modules(bronze) + .with_config({ + "dataset_prefix": DATASET_PREFIX, + "publisher": PUBLISHER, + "collection_url": COLLECTION_URL, + "pipeline_version": version("asf-mission-data"), + "bronze_ingest_timestamp": datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S"), + }) + .build() + ) +``` + +If a stage produces multiple output tables from the same source (e.g. several sheets in one spreadsheet), the driver-builder function can take a parameter and fold it into the config, so the same driver logic can be reused per table: + +```python +def build_silver_driver(sheet_name: str) -> driver.Driver: + return ( + driver.Builder() + .with_modules(silver) + .with_config({"dataset_prefix": DATASET_PREFIX, "sheet_name": sheet_name}) + .build() + ) +``` + +### Running stages + +Each `run__pipeline()` function executes the driver against its target nodes, then generates and saves a DAG visualisation: + +```python +def run_bronze_pipeline() -> None: + dr = build_bronze_driver() + results = dr.execute(["bronze__file", "latest_filename", "latest_publication_date"]) + + dag_png = dr.visualize_execution(["bronze__file"], None, render_kwargs={}).pipe(format="png") + storage.save_dag( + layer_prefix="bronze", + dataset_prefix=DATASET_PREFIX, + accompanying_filename=results["latest_filename"], + dag_image=dag_png, + date_stamp=f"published={utils.normalise_date_string(results['latest_publication_date'])}", + ) +``` + +For a multi-table stage, loop over each table and run the driver once per output node, saving a separate DAG image each time: + +```python +def run_silver_pipeline() -> None: + for sheet_name, output_node in SILVER_TABLES_NODES_MAP.items(): + dr = build_silver_driver(sheet_name=sheet_name) + results = dr.execute([output_node, "latest_publication_date"]) + + dag_png = dr.visualize_execution([output_node]).pipe(format="png") + storage.save_dag( + layer_prefix="silver", + dataset_prefix=DATASET_PREFIX, + accompanying_filename=sheet_name.lower().replace(".", "_").replace(" ", "_"), + dag_image=dag_png, + date_stamp=f"published={utils.normalise_date_string(results['latest_publication_date'])}", + ) +``` + +`SILVER_TABLES_NODES_MAP` (defined in `config.py`) maps each source table to a sheet name, in this example, to its corresponding Hamilton output node name. + +### Entrypoint + +The `run()` function is called by the CLI and routes to the appropriate stage. `logger.info()` calls at each stage so stage boundaries are visible in pipeline logs: + +```python +def run(stage: str = "bronze", extra_args: list[str] | None = None) -> None: + if stage in ("bronze", "all"): + logger.info("Starting bronze stage") + run_bronze_pipeline() + logger.info("Completed bronze stage") + + if stage in ("silver", "all"): + logger.info("Starting silver stage") + run_silver_pipeline() + logger.info("Completed silver stage") +``` + +Add a matching `if stage in ("gold", "all"):` branch if your pipeline has a gold stage. + +## 7. Write tests (recommended) + +Tests are not required to merge a new pipeline, but they guard against regressions as the codebase evolves. Different failures happen in different places, so tests should cover each layer separately: + +| Layer | Catches | Minimum coverage | +|---|---|---| +| Bronze | broken downloads, missing metadata, wrong storage paths | one test mocking the upstream fetch; one test checking the file/metadata land at the expected path | +| Silver | parsing and transformation bugs | one test per non-trivial transform | +| Gold | aggregation and derivation bugs | one test per non-trivial transform (same approach as silver; no worked example yet, for now see silver's test file for the pattern to follow) | +| Schema | bad data reaching the canonical output | one test that rejects invalid data against the schema | +| Integration | wiring mistakes between storage, Hamilton, and parquet writes | one local test running the real pipeline against `tmp_path`, without S3 | + +The example pipeline includes a complete test template to copy from: + +- `tests/pipeline/example/test_bronze.py`: mocking an external source, asserting the raw file and metadata are written correctly +- `tests/pipeline/example/test_silver.py`: testing transform functions, validating a dataframe schema, and running a local integration test against a temporary directory +- `tests/pipeline/example/conftest.py`: sharing sample input data and setting `DATA_MODE=LOCAL` for tests + +Run just the example pipeline tests with: + +```bash +uv run pytest tests/pipeline/example +``` + +Run the full suite with: + +```bash +uv run pytest +``` + +## 8. Verify locally + +Run the full pipeline locally before opening a PR: + +```bash +export DATA_MODE=LOCAL +export DATA_ROOT=/tmp/pipeline-dev + +uv run python -m asf_mission_data.run --stage all +``` + +Check that the expected files landed under `$DATA_ROOT`: + +``` +data/ +β”œβ”€β”€ bronze// +β”‚ β”œβ”€β”€ latest/ +β”‚ β”‚ β”œβ”€β”€ file/ +β”‚ β”‚ └── metadata/ +β”‚ └── historical// +β”‚ β”œβ”€β”€ file/ +β”‚ └── metadata/ +β”œβ”€β”€ silver// +β”‚ β”œβ”€β”€ latest// +β”‚ └── historical/// +└── gold// + β”œβ”€β”€ latest// + └── historical/// +``` + +DAG visualisations are saved alongside each stage's output - inspect them under: + +``` +artifacts/dags/ +β”œβ”€β”€ bronze/// +β”œβ”€β”€ silver/// +└── gold/// +``` + +Run stages individually with `--stage bronze` or `--stage silver` instead of `--stage all`. See [`running-pipelines.md`](../docs/running-pipelines.md) for the full local and AWS run reference. + +## 9. Write a pipeline README + +A pipeline's `README.md` is the human-readable reference for what it produces: a plain-language description of the data, a full list of the files and tables it outputs and what they're called in S3, and any source-specific quirks a maintainer should know about. + +Every pipeline directory needs one. Copy the template from [`docs/templates/pipelines/README.md`](../docs/templates/pipelines/README.md) into your pipeline directory and fill it in. + +`/README.md`is separate from the information in `pipelines.yaml` (step 2) which drives tooling; this README is for maintainers and can go into more depth. + +## 10. Open a PR + +A new pipeline isn't done until it's running in prod. Follow the process in [CONTRIBUTING.md](../docs/CONTRIBUTING.md): a PR into `dev`, then a `dev` β†’ `prod` promotion PR. Since this is the pipeline's first promotion, tick "New pipeline" in the prod PR template's checklist. + +--- + +*Last updated: 1 July 2026 by Elysia Lucas* diff --git a/docs/dag-images/.gitkeep b/docs/dag-images/.gitkeep deleted file mode 100644 index e69de29..0000000 diff --git a/docs/running-pipelines.md b/docs/running-pipelines.md index 2b5a2a4..13c04c6 100644 --- a/docs/running-pipelines.md +++ b/docs/running-pipelines.md @@ -1,84 +1,88 @@ -# Running Pipelines +# Running pipelines -This repo supports a few different ways of running a pipeline, but most people only need two: +This repo supports two main ways to run a pipeline: locally while developing, and via GitHub Actions for AWS runs. -- run locally while developing -- use the GitHub Actions UI for ad hoc runs in AWS - -Everything else should be treated as advanced or debugging-only. - -## What To Use When +## Quick reference | I want to... | Use this | Notes | |---|---|---| | develop or debug pipeline code | local run | fastest feedback, no AWS dependency | -| test my branch in AWS | build an image, then use `Run pipeline` | standard cloud test path | -| run against the default dev image | `Run pipeline` with `image_tag=dev-latest` | easiest ad hoc cloud run | -| launch from a terminal with more control | `scripts/trigger_pipeline.py` | advanced/debug path | - -## Standard Workflow - -For most cloud runs, the process is: +| run a pipeline using code from a branch | `Build and push Docker Image to ECR` (from your branch), then `Test pipeline in dev` | standard dev test path | +| run a pipeline using code already on `dev` | `Test pipeline in dev` with `image_tag=dev-latest` | quickest dev run; no image build needed | +| run a pipeline in prod | `Run pipeline in prod` workflow from `prod` branch | easiest manual pipeline trigger in prod; refreshing data | +| launch a pipeline in AWS from the terminal instead of GitHub UI | `scripts/trigger_pipeline.py` | advanced/debug path | -1. build a Docker image for the code you want to test -2. copy the image tag from the build workflow summary -3. run the `Run pipeline` workflow with that tag -That is the main team workflow. If you are unsure which path to use, use this one. +## Running locally in pipeline development -## Local Development - -Use local mode when developing or debugging pipeline logic. +Use local mode when developing or debugging pipeline logic. Run these commands in your terminal: ```bash +# Set local mode (otherwise the default is the dev S3 bucket) export DATA_MODE=LOCAL export DATA_ROOT=/tmp/pipeline-dev +# Run all stages uv run python -m asf_mission_data.run example --stage all ``` This avoids Docker, ECS, and ECR entirely. -## Build An Image For A Branch +### Stages -Use the `Build and push Docker Image to ECR` workflow when you want to test branch code in AWS. +Pipelines have two or three stages: **bronze** (raw fetch), **silver** (clean and transform), and optionally **gold** (aggregated outputs). Use `--stage` to control which runs: -### How tags are created +- `--stage all` β€” run every available stage in order +- `--stage bronze` β€” run only the fetch stage +- `--stage silver` β€” run only the transform stage +- `--stage gold` β€” run only the aggregation stage (if the pipeline has one) -The workflow tags images from the branch name using: +When developing, you can run a single stage to avoid re-fetching data you already have. -```text -{branch-name-with-/-replaced-by--}-latest +## Advanced: Running locally in pipeline development with Docker + +Use this when you want to test your code inside the container β€” the same environment it runs in on AWS β€” without pushing to ECR or triggering a cloud run. This is useful for catching issues specific to the container, such as missing dependencies in the Dockerfile. + +First, build the image: + +```bash +docker build -t asf-mission-data . ``` -Examples: +Then run a pipeline against your local filesystem: -- `dev` -> `dev-latest` -- `feat/image-check` -> `feat-image-check-latest` -- `alex/test-run` -> `alex-test-run-latest` +```bash +mkdir -p /tmp/asf-mission-data + +docker run --rm \ + -e DATA_MODE=LOCAL \ + -e DATA_ROOT=/tmp/asf-mission-data \ + -v /tmp/asf-mission-data:/tmp/asf-mission-data \ + asf-mission-data \ + example --stage all +``` -The exact image URI is written to the workflow summary at the end of the build. +The `-v` flag mounts your local directory into the container so output is written to `/tmp/asf-mission-data` on your machine. -### Which branches use which tags +Note: the image tag is `asf-mission-data` (hyphen), not `asf_mission_data` (underscore). If you omit `DATA_MODE` and `DATA_ROOT`, the container will try to use the dev S3 bucket instead. -- pushes to `dev` automatically build `dev-latest` -- manual workflow runs build a tag for the branch selected in GitHub +## Running in AWS during pipeline development -Do not guess the tag if you can avoid it. Copy it from the build workflow summary. +Workflows are triggered from the [Actions tab](https://github.com/nestauk/asf_mission_data/actions) in the GitHub repo. -## Run A Pipeline In AWS +Use the `Test pipeline in dev` workflow for dev runs. It takes three inputs: -Use the `Run pipeline` workflow in GitHub Actions for standard ad hoc runs. +- `pipeline` - must match a key in `pipelines.yaml` +- `stage` - one of `all`, `bronze`, `silver`, `gold` +- `image_tag` - ECR tag of the container image to run -Inputs: +Running a pipeline with this workflow will populate data in the `asf-mission-data-dev` S3 bucket. -- `pipeline`: must match a key in `pipelines.yaml` -- `stage`: one of `all`, `bronze`, `silver`, `gold` -- `image_tag`: the ECR tag to use +The workflow summary will show you the resolved task definition, the container image used, and an error if the image tag does not exist in ECR. A missing tag fails before any task is started. -Examples: +### Which image tag to use? -- use the default dev image: +**If you want to test pipeline code that is already on the `dev` branch**, select the `dev` branch from the dropdown under `Use workflow from` and use the `image_tag=dev-latest`. This is built and pushed automatically from the `dev` branch whenever new code is merged into it. ```text pipeline=example @@ -86,7 +90,13 @@ stage=all image_tag=dev-latest ``` -- test a branch image: +**If you want to test pipeline code from a specific branch**, you first need to build an image. Run the `Build and push Docker Image to ECR` workflow and select your branch from the dropdown under `Use workflow from`. The tag is derived from the branch name, e.g.: +- `feat/image-check` β†’ `feat-image-check-latest` +- `alex/test-run` β†’ `alex-test-run-latest` + +Do not guess the tag if you can avoid it. The exact tag is written to the workflow summary, you can copy it from there to use it in `Test pipeline in dev`. + +When running the `Test pipeline in dev`, select the `dev` branch from the dropdown under `Use workflow from`, but the `image_tag` should correspond to the branch image. ```text pipeline=example @@ -94,17 +104,20 @@ stage=all image_tag=feat-image-check-latest ``` -### What the workflow shows you +## Running in AWS for a pipeline in production -The workflow now surfaces the important launch details in the Actions UI: +This writes to the production `asf-mission-data-prod` S3 bucket. Only run this when you intend to refresh production data. -- the resolved task definition -- the container image actually used -- a highlighted error if the requested image tag does not exist in ECR +Use the `Run pipeline in prod` workflow in GitHub Actions, selecting the `prod` branch from the dropdown under `Use workflow from`. It takes two inputs: -That means a missing tag fails before registering a new task definition or starting a task. +- `pipeline` - must match a key in `pipelines.yaml` +- `stage` - one of `all`, `bronze`, `silver`, `gold` -## Advanced: Trigger From The Terminal +Triggering this workflow doesn't run it immediately. It targets the `prod` GitHub Environment, which requires approval from a designated reviewer before the job proceeds. You'll see it sitting in "Waiting" status in the Actions run until someone approves it. + +Data written to the prod bucket is picked up automatically. Infrastructure scans the bucket hourly and runs `CREATE OR REPLACE` on the corresponding DuckLake tables. Those tables are connected to Superset via DuckDB, and changes should appear there after 10 minutes. + +## Advanced: Running in AWS from the terminal You can also launch the same flow directly: @@ -127,22 +140,28 @@ This script: Use this when you need CLI control. For most users, the GitHub UI is simpler. -## What Is Supported vs Advanced +## Troubleshooting -Supported for most users: +### I want to check the Docker image was built and pushed to ECR successfully -- local runs via `python -m asf_mission_data.run` -- cloud runs via `Build and push Docker Image to ECR` -- cloud runs via `Run pipeline` +After the `Build and push Docker Image to ECR` workflow completes, check the `asf-mission-data` [repository in ECR](https://eu-west-2.console.aws.amazon.com/ecr/repositories/private/195787726158/asf-mission-data/_/details?region=eu-west-2) via the AWS console and confirm your tag is listed (either an updated `dev-latest` or a tag corresponding to your feature branch). -Advanced/debug only: +--- -- direct use of `scripts/trigger_pipeline.py` -- trying to run GitHub Actions workflows locally +### I want to watch the logs during a pipeline run in AWS -This repo does not treat "run the GitHub workflow locally" as a standard path. For normal use, prefer the GitHub UI or the Python trigger script. +You can watch the log streams in CloudWatch for each run in the following ECS log groups: -## Troubleshooting +- Dev runs: `asf-mission-data-dev` [ECS log group](https://eu-west-2.console.aws.amazon.com/cloudwatch/home?region=eu-west-2#logsV2:log-groups/log-group/$252Fecs$252Fasf-mission-data-dev). +- Prod runs: `asf-mission-data-prod` [ECS log group](https://eu-west-2.console.aws.amazon.com/cloudwatch/home?region=eu-west-2#logsV2:log-groups/log-group/$252Fecs$252Fasf-mission-data-prod). + +--- + +### I want to check the pipeline output landed in S3 + +After the run completes, check the S3 bucket (`asf-mission-data-dev` for dev runs, or `asf-mission-data-prod` for prod runs) directly in the AWS console to confirm the expected files are present. + +--- ### The image tag does not exist @@ -156,16 +175,14 @@ What to do: 1. run `Build and push Docker Image to ECR` 2. copy the tag from the workflow summary -3. rerun `Run pipeline` with that exact tag +3. rerun `Test pipeline in dev` with that exact tag -### I just want the latest shared dev image +--- -Use: +### I do not know the pipeline name -```text -image_tag=dev-latest -``` +Check `pipelines.yaml`. The `pipeline` input must match one of its keys. Pipeline names are the top-level keys, e.g., `example` or `energy_price_cap_levels_annex_9`. -### I do not know the pipeline name +--- -Check `pipelines.yaml`. The `pipeline` input must match one of its keys. +*Last updated: 29 June 2026 by Elysia Lucas* diff --git a/docs/templates/pipelines/README.md b/docs/templates/pipelines/README.md new file mode 100644 index 0000000..3b92289 --- /dev/null +++ b/docs/templates/pipelines/README.md @@ -0,0 +1,59 @@ +# [Pipeline name] + + + +**Source:** +**Update cadence:** +**Pipeline run name:** +**Storage prefix:** + + +## Pipeline stages + + + +### Bronze + +- **Source**: +- **Method**: +- **Output**: +- **Validators**: + +### Silver + +#### `table_name_1` +- **Description**: +- **Input**: +- **Output**: +- **Validators**: + +#### `table_name_2` +- **Description**: +- **Input**: +- **Output**: +- **Validators**: + +### Gold + +#### `table_name_1` +- **Description**: +- **Input**: +- **Output**: +- **Validators**: + +#### `table_name_2` +- **Description**: +- **Input**: +- **Output**: +- **Validators**: + +--- + +## Notes + + + +--- + +*Last updated: by * diff --git a/docs/test.md b/docs/test.md deleted file mode 100644 index 7697869..0000000 --- a/docs/test.md +++ /dev/null @@ -1 +0,0 @@ -TEST DOC diff --git a/infrastructure/README.md b/infrastructure/README.md index c209587..c6016eb 100644 --- a/infrastructure/README.md +++ b/infrastructure/README.md @@ -1,222 +1,219 @@ # Infrastructure -AWS CDK infrastructure for the ASF Mission Data pipelines. +AWS infrastructure for the ASF Mission Data pipelines, managed with AWS CDK (Python). This guide covers what is deployed, how to change it safely, and how GitHub Actions and pipeline runs connect to AWS. For *running* pipelines see [running-pipelines.md](../docs/running-pipelines.md); for *writing* them see [adding-pipelines.md](../docs/adding-pipelines.md). -## Overview +## Quick reference -This infrastructure supports serverless ETL pipelines that run on AWS Lambda with container images stored in ECR and data stored in S3. +| I want to... | Use this | Notes | +|---|---|---| +| see what a change would do in AWS | `cdk diff --context env=dev` from `infrastructure/` | [Preview or deploy](#i-want-to-preview-or-deploy-an-infrastructure-change) | +| deploy an infrastructure change | `cdk deploy --context env=dev`, verify, then `env=prod` | [Preview or deploy](#i-want-to-preview-or-deploy-an-infrastructure-change) | +| change a config value or add an environment | edit `config/dev.py` / `config/prod.py`, then deploy | [Change configuration](#i-want-to-change-an-environments-configuration) | +| fix a workflow that can't authenticate to AWS | check the OIDC role, `AWS_ACCOUNT_ID`, `id-token` permission | [GitHub auth](#how-github-actions-authenticates-to-aws) | +| understand what actually runs my pipeline | ECS Fargate task from the shared task definition | [How pipelines run](#how-pipelines-run-on-aws) | +| put a pipeline on a schedule | not yet possible - currently, all runs are manual | [Schedule a pipeline](#i-want-to-schedule-a-pipeline) | +| find the logs for a run | CloudWatch log group `/ecs/asf-mission-data-{env}` | [Monitoring](#monitoring-logs-and-alerting) | +| know where the data goes after S3 | DuckLake/Superset, in the `nestauk/de_cdp` repo | [Downstream](#downstream-consumers) | -### Architecture +## What's deployed + +One CDK app ([`app.py`](app.py)) defining a single stack, `CoreStack`, deployed once per environment as `asf-core-dev` and `asf-core-prod`. ```text -β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” -β”‚ GitHub Actions │────▢│ ECR Repo │────▢│ Lambda β”‚ -β”‚ (CI/CD) β”‚ β”‚ (Container β”‚ β”‚ (Pipeline β”‚ -β”‚ β”‚ β”‚ Images) β”‚ β”‚ Execution) β”‚ -β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ - β”‚ β”‚ - β”‚ β–Ό - β”‚ β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” - └──────────────────────────────────────▢│ S3 Bucket β”‚ - β”‚ (Bronze/Silver β”‚ - β”‚ Data) β”‚ - β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ +GitHub Actions ────build & push image───▢ ECR: asf-mission-data + β”‚ β”‚ + β”‚ scripts/trigger_pipeline.py β”‚ image tag {env}-latest + β–Ό β–Ό +ECS Fargate cluster ──runs──▢ task definition asf-mission-data-{env} +asf-mission-data-{env} (container "app") + β”‚ β”‚ + β–Ό β–Ό + CloudWatch Logs S3: asf-mission-data-{env} + /ecs/asf-mission-data-{env} β”‚ silver/gold Parquet + β–Ό + DuckLake β†’ Superset (nestauk/de_cdp) ``` -## Stacks +`CoreStack` creates ([`stacks/core/core_stack.py`](stacks/core/core_stack.py)): -### CoreStack (`asf-core-{env}`) +| Resource | Name | Purpose | +|---|---|---| +| S3 bucket | `asf-mission-data-{env}` | Pipeline data (bronze/silver/gold). SSE-S3, public access blocked, SSL enforced, unversioned | +| ECR repository | `asf-mission-data` | Container images for all pipelines. Scan on push, lifecycle rule keeps the last 10 images | +| ECS cluster | `asf-mission-data-{env}` | Fargate (and Fargate Spot) capacity for pipeline tasks; Container Insights disabled | +| ECS task definition | family `asf-mission-data-{env}` | One shared definition; the pipeline to run is chosen by command override at launch | +| GitHub Actions IAM role | `asf-github-actions-{env}` | OIDC-assumed by workflows β€” see [GitHub auth](#how-github-actions-authenticates-to-aws) | +| Task execution role | `asf-mission-data-{env}-task-execution-role` | Lets the ECS agent pull from ECR and write logs | +| Task role | `asf-mission-data-{env}-task-role` | What pipeline code uses at runtime; read/write on the data bucket | +| Scheduler role | `asf-mission-data-{env}-scheduler-role` | For EventBridge Scheduler to launch tasks β€” reserved for [future scheduling](#i-want-to-schedule-a-pipeline) | +| Security group | `asf-mission-data-{env}-tasks` | Outbound-only (tasks scrape sources and call S3); no inbound access | +| Log group | `/ecs/asf-mission-data-{env}` | Task logs, one-month retention | -Shared infrastructure deployed once per environment. Creates: +Two structural quirks worth knowing: -| Resource | Purpose | -|----------|---------| -| **S3 Bucket** | Pipeline data storage (bronze/silver layers) | -| **ECR Repository** | Container images for pipeline Lambdas | -| **IAM Role** | GitHub Actions OIDC role for CI/CD deployments | +- **The ECR repository is shared.** It is created by the *dev* stack only (tagged `Environment: shared`); the prod stack references it by name. Both `dev-latest` and `prod-latest` tags live in the same repository. +- **Networking is the account default VPC** (`vpc-b556bedd` and its subnets, defined in [`config/environments.py`](config/environments.py)). Tasks launch with a public IP for outbound internet access; nothing accepts inbound traffic. -The GitHub Actions role has permissions to: -- Push/pull images to ECR -- Read/write to the S3 data bucket -- Deploy CloudFormation stacks (for pipeline stacks) -- Create/manage Lambda functions and execution roles -- Create/manage EventBridge schedules -- Create/manage CloudWatch log groups +The stack exports its key values as CloudFormation outputs (bucket, ECR URI, role ARNs, cluster ARN, task definition ARN, security group, subnets). `scripts/trigger_pipeline.py` reads these at run time, so renaming exports is a breaking change for pipeline triggering. ## Environments -| Environment | AWS Account | Region | Stack Name | -|-------------|-------------|--------|------------| -| dev | 195787726158 | eu-west-2 | `asf-core-dev` | -| prod | 195787726158 | eu-west-2 | `asf-core-prod` | +| Environment | Region | Stack | Data bucket | +|---|---|---|---| +| dev | eu-west-2 | `asf-core-dev` | `asf-mission-data-dev` | +| prod | eu-west-2 | `asf-core-prod` | `asf-mission-data-prod` | -### Environment differences +Both environments live in the same account and region; separation is by resource naming, IAM scoping, and the GitHub-side promotion process ([CONTRIBUTING.md](../docs/CONTRIBUTING.md)). -| Setting | Dev | Prod | -|---------|-----|------| -| Removal policy | DESTROY | RETAIN | -| Auto-delete S3 objects | Yes | No | -| Auto-delete ECR images | Yes | No | +| Behaviour | Dev | Prod | +|---|---|---| +| Removal policy on bucket | DESTROY | RETAIN | +| Auto-delete S3 objects on stack teardown | Yes | No | +| ECR repository | Created here (shared) | Referenced by name | +| Container `DATA_MODE` | `DEV` | `PROD` | -## Cost Estimate +## Prerequisites -Estimated monthly costs for the CoreStack resources (eu-west-2 pricing, March 2026). +1. AWS CLI configured with credentials for the account +2. Node.js, for the CDK CLI +3. [uv](https://docs.astral.sh/uv/) (the repo's dependency manager) -### CoreStack resources (always running) +```bash +# Install the CDK CLI globally +npm install -g aws-cdk -| Resource | Unit | Price | Estimated Usage | Monthly Cost | -|----------|------|-------|-----------------|--------------| -| **S3 Storage** | GB/month | $0.023 | 10 GB | $0.23 | -| **S3 Requests** | 1K PUT/GET | $0.005/$0.0004 | 10K PUT, 50K GET | $0.07 | -| **ECR Storage** | GB/month | $0.10 | 5 GB (10 images) | $0.50 | -| **IAM Role** | - | Free | - | $0.00 | +# Install the CDK Python dependencies +uv sync --group infrastructure +``` -**CoreStack total: ~$0.80/month** +All CDK commands must be run from the `infrastructure/` directory (where `cdk.json` lives). -### Pipeline resources (per pipeline, when deployed) +## I want to preview or deploy an infrastructure change -| Resource | Unit | Price | Estimated Usage | Monthly Cost | -|----------|------|-------|-----------------|--------------| -| **Lambda** | 1M requests | $0.20 | 1K invocations | $0.0002 | -| **Lambda Compute** | GB-second | $0.0000167 | 1K x 30s x 1GB | $0.50 | -| **CloudWatch Logs** | GB ingested | $0.57 | 0.5 GB | $0.29 | -| **CloudWatch Logs Storage** | GB/month | $0.03 | 1 GB | $0.03 | -| **EventBridge Scheduler** | 1M invocations | $1.00 | 720 (hourly) | $0.0007 | +Infrastructure changes are shipped by **running `cdk deploy` manually from a developer machine** β€” there is no CI/CD deployment path yet. -**Per pipeline total: ~$0.82/month** (assuming hourly schedule, 30s avg runtime) +```bash +cd infrastructure -### Example scenarios +# See what would change (always do this first) +cdk diff --context env=dev -| Scenario | Pipelines | Schedule | Est. Monthly Cost | -|----------|-----------|----------|-------------------| -| Dev (minimal) | 2 | Daily | ~$1.50 | -| Dev (active) | 5 | Hourly | ~$5.00 | -| Prod | 10 | Mixed | ~$10-20 | +# Deploy to dev, verify behaviour, then deploy to prod +cdk deploy --context env=dev +cdk deploy --context env=prod -### Cost optimization notes +# Generate the CloudFormation template without deploying +cdk synth --context env=dev -- ECR lifecycle policy limits images to 10 per repo (saves ~$0.10/image/month) -- Lambda costs scale with execution time; optimize pipeline code for faster runs -- CloudWatch Logs retention can be reduced from default (never expires) to save costs +# List the stacks CDK knows about +cdk list --context env=dev +``` -## Prerequisites +Deploy to dev first and verify (run a pipeline, check the diff did what you expected) before repeating against prod. If the account/region has never been used with CDK you'll be told to bootstrap: `cdk bootstrap aws://195787726158/eu-west-2` β€” needed once, ever. -1. **AWS CLI** configured with appropriate credentials -2. **Node.js** (for CDK CLI) -3. **Python 3.12+** with project dependencies +## I want to change an environment's configuration -```bash -# Install CDK CLI globally -npm install -g aws-cdk +Configuration lives in [`config/`](config/): -# Install Python dependencies -uv sync --extra infrastructure -``` +- [`environments.py`](config/environments.py) β€” the `EnvironmentConfig` dataclass: shared defaults (VPC, subnets, project prefix, image retention) and derived names (bucket, cluster, role names) as properties +- [`dev.py`](config/dev.py) / [`prod.py`](config/prod.py) β€” per-environment values and tags -## Usage +Change a value, `cdk diff` to confirm the blast radius, then deploy. To add an environment: create `config/{env}.py` with an `EnvironmentConfig`, add it to the `ENVIRONMENTS` dict in `app.py`, and deploy with `--context env={env}`. -All CDK commands should be run from the `infrastructure/` directory (where `cdk.json` lives). +One trap: `task_cpu` and `task_memory` in `EnvironmentConfig` are currently **not wired up** β€” the task definition hardcodes 256 CPU units / 512 MB in `core_stack.py`. Changing the config fields alone does nothing. -### Deploy +## How GitHub Actions authenticates to AWS -```bash -# Deploy to dev -cdk deploy --context env=dev +Workflows authenticate with **OIDC** β€” no long-lived AWS keys are stored in GitHub. Each AWS-touching workflow requests a token (`permissions: id-token: write`) and assumes `asf-github-actions-dev` or `asf-github-actions-prod`. The roles trust the pre-existing GitHub OIDC provider in the account, restricted to this repository (`repo:nestauk/asf_mission_data:*`, any branch/workflow), with a one-hour session cap. Sessions are named `GitHubActions--`, so CloudTrail activity can be traced back to a specific Actions run. -# Deploy to prod -cdk deploy --context env=prod -``` +Required GitHub configuration (Settings β†’ Secrets and variables β†’ Actions): -### Preview changes +| Level | Kind | Name | Used for | +|---|---|---|---| +| Repository | Secret | `AWS_ACCOUNT_ID` | Building the role ARNs in all four AWS workflows | +| Repository | Variable | `AWS_REGION` | Single source of truth for region (`eu-west-2`); wired into each workflow's `env` block | +| Environment (dev, prod) | Variable | `ENV_NAME` | Environment identity; note it only reaches jobs that declare `environment:` | +| Environment (prod) | Protection rule | required reviewers | The approval gate on `Run pipeline in prod` | -```bash -# Show what would change -cdk diff --context env=dev -``` +The `prod` GitHub Environment is what makes prod runs wait for a designated reviewer. The `dev` environment exists for symmetry and future protections; no workflow currently declares it. -### Synthesize CloudFormation +The role's permissions are scoped to `asf-*` resources: ECR push/pull, bucket read/write, ECS task-definition registration and `RunTask` on the cluster, EventBridge Scheduler management (for future scheduling), plus CloudFormation/IAM management for `asf-*` stacks and roles. -```bash -# Generate CloudFormation template without deploying -cdk synth --context env=dev -``` +## How pipelines run on AWS -### List stacks +The model is **one image, many pipelines**: the Docker image (built from the repo [`Dockerfile`](../Dockerfile)) contains every pipeline, and the launch-time command override picks which one runs β€” `["", "--stage", ""]` against the container's CLI entrypoint. -```bash -cdk list --context env=dev -``` +1. **Images**: `Build and push Docker Image to ECR` builds `linux/amd64` images (matching the task definition's platform) and tags them `-latest`. Merges to `dev` refresh `dev-latest`; merging `dev` β†’ `prod` re-tags that exact manifest as `prod-latest` (no rebuild). +2. **Task definition**: the shared family `asf-mission-data-{env}` points at the `{env}-latest` image and injects `DATA_MODE`, `DATA_ROOT=s3://asf-mission-data-{env}` and `ASF_ENVIRONMENT`, so pipeline code needs no per-environment configuration of its own. +3. **Launching**: both run workflows call [`scripts/trigger_pipeline.py`](../scripts/trigger_pipeline.py), which reads the `asf-core-{env}` stack outputs (cluster, subnets, security group), and calls `ecs:RunTask`. For a non-default image tag it first validates the tag exists in ECR, then registers a fresh task-definition revision pointing at it. +4. **Compute**: Fargate by default; the trigger script accepts `--capacity-provider FARGATE_SPOT` for interruptible-but-cheaper runs. -## Configuration +Day-to-day usage (which workflow, which image tag, what to check) is covered in [running-pipelines.md](../docs/running-pipelines.md). -Environment configurations are in `config/`: +## I want to schedule a pipeline -- `config/environments.py` - `EnvironmentConfig` dataclass with shared logic -- `config/dev.py` - Dev environment values -- `config/prod.py` - Prod environment values +**You can't yet.** All pipeline runs are currently manual, via the GitHub workflows or the trigger script. The `schedule` values in [`pipelines.yaml`](../pipelines.yaml) are declarative intent only β€” nothing reads them today. -### Adding a new environment +The agreed design (27 July 2026, not yet implemented): `pipelines.yaml` becomes the source of truth; a reconciler script β€” run by the promote-to-prod workflow after the image re-tag, and manually triggerable β€” converges EventBridge Scheduler with the file, creating, updating and deleting `asf-`-prefixed schedules to match. Schedules will target the shared task definition via the scheduler role that already exists in the core stack. Scheduling is prod-only; dev runs stay manual by design. The `stack_name` field in `pipelines.yaml` is defunct under this design and will be removed. -1. Create `config/{env}.py` with an `EnvironmentConfig` instance -2. Add it to the `ENVIRONMENTS` dict in `app.py` -3. Deploy: `cdk deploy --context env={env}` + -## Project structure +## Monitoring, logs and alerting -```text -infrastructure/ -β”œβ”€β”€ app.py # CDK entry point -β”œβ”€β”€ cdk.json # CDK configuration -β”œβ”€β”€ config/ -β”‚ β”œβ”€β”€ environments.py # EnvironmentConfig dataclass -β”‚ β”œβ”€β”€ dev.py # Dev environment config -β”‚ └── prod.py # Prod environment config -└── stacks/ - └── core/ # Core stack - β”œβ”€β”€ core_stack.py # Stack implementation - └── README.md # Stack documentation -``` +**Logs**: every task logs to CloudWatch log group `/ecs/asf-mission-data-{env}` ([dev](https://eu-west-2.console.aws.amazon.com/cloudwatch/home?region=eu-west-2#logsV2:log-groups/log-group/$252Fecs$252Fasf-mission-data-dev) Β· [prod](https://eu-west-2.console.aws.amazon.com/cloudwatch/home?region=eu-west-2#logsV2:log-groups/log-group/$252Fecs$252Fasf-mission-data-prod)), one stream per run under the `pipeline/` prefix. Retention is one month. -## CI/CD Integration +**Traceability**: workflow-initiated AWS activity is identifiable in CloudTrail by the role session name, which embeds the GitHub Actions run ID. -GitHub Actions authenticates to AWS using OIDC (no long-lived credentials). The trust policy restricts access to: +**Alerting**: none exists yet. `asf_mission_data/alerting.py` is an empty stub, and nothing notifies anyone when a run fails β€” failed runs are only visible in the workflow summary (for manual runs) and CloudWatch. Treat "check the logs after a prod run" as a required manual step until this changes. -- Repository: `nestauk/asf_mission_data` -- Any branch/workflow (configured via `repo:org/repo:*` subject claim) +## Downstream consumers -### Required GitHub secrets/variables +After a pipeline writes silver/gold Parquet to the bucket, consumption is handled by the Data Engineering Core Data Platform ([`nestauk/de_cdp`](https://github.com/nestauk/de_cdp)) β€” not by anything in this repo. Its DuckLake stack runs DuckDB over a PostgreSQL metadata catalogue and registers tables from the source bucket's `data/silver/` and `data/gold/` latest partitions on an **hourly** cron; Superset then queries those tables through the catalogue. Bronze data is not registered. -| Type | Name | Description | -|------|------|-------------| -| Secret | `AWS_ACCOUNT_ID` | AWS account ID | -| Secret | `MISSION_DATA_BUCKET` | S3 bucket name | -| Variable | `AWS_REGION` | AWS region (eu-west-2) | -| Variable | `ENV_NAME` | Environment name (dev/prod) | +> **Known discrepancy** (27 July 2026): all of de_cdp's environment configs β€” including prod β€” currently point at `asf-mission-data-dev`. Data written to the **prod** bucket is *not* being picked up, despite what [running-pipelines.md](../docs/running-pipelines.md) says about prod runs appearing in Superset. Until de_cdp is repointed, treat the dev bucket as the one feeding dashboards. -## Troubleshooting + -### "Unable to assume role" in GitHub Actions +## Costs -1. Verify the OIDC provider exists in the AWS account -2. Check the IAM role trust policy matches the repository name exactly -3. Ensure the workflow has `id-token: write` permission +Approximate, using eu-west-2 list prices as of July 2026 β€” sanity-check against [current AWS pricing](https://aws.amazon.com/fargate/pricing/) before relying on them. There are no always-on compute resources; the standing cost is storage, and the marginal cost is per task-run. -### CDK bootstrap required +| Resource | Driver | Approximate cost | +|---|---|---| +| Fargate task run | 0.25 vCPU / 0.5 GB, ~$0.047 per vCPU-hour + ~$0.005 per GB-hour | ~$0.001–0.002 per 5-minute run | +| S3 storage | ~$0.024/GB-month | ~$0.25/month per 10 GB | +| ECR storage | $0.10/GB-month, capped at 10 images by lifecycle rule | ~$0.50/month | +| CloudWatch Logs | ~$0.59/GB ingested; storage minor at one-month retention | pennies at current volume | +| EventBridge Scheduler | $1.00 per million invocations (once implemented) | negligible | -If you see bootstrap errors, the AWS account needs CDK bootstrapping: +Realistic total at current scale (a handful of pipelines, run manually or daily): **single-digit dollars per month per environment**. The levers that change this are run frequency, run duration, and log volume β€” not the standing infrastructure. -```bash -cdk bootstrap aws://ACCOUNT_ID/eu-west-2 -``` +## Troubleshooting + +### A workflow fails with "Not authorized to perform sts:AssumeRoleWithWebIdentity" + +1. Check the job has `permissions: id-token: write` +2. Check `AWS_ACCOUNT_ID` (repo secret) and `AWS_REGION` (repo variable) are set β€” an empty region makes the credentials action fail obscurely +3. Check the role's trust policy subject matches `repo:nestauk/asf_mission_data:*` exactly (a repo rename breaks this) -Note: This only needs to be done once per account/region. +### `cdk synth`/`deploy` fails complaining about environment or bootstrap -### Stack drift +- "Need to perform AWS calls... but no credentials configured" β€” the VPC lookup (`Vpc.from_lookup`) needs valid credentials the first time; afterwards the result is cached in `cdk.context.json` (committed) +- Bootstrap errors β€” run `cdk bootstrap aws://195787726158/eu-west-2` once -If resources were modified outside CDK: +### A task fails to launch + +The trigger script prints ECS failure reasons. Common ones: the image tag doesn't exist in ECR (see [running-pipelines.md](../docs/running-pipelines.md#the-image-tag-does-not-exist)), or stack outputs changed and the script resolved stale infrastructure β€” check `aws cloudformation describe-stacks --stack-name asf-core-{env}`. + +### Something was changed in the console and CDK is out of sync ```bash -# Check for drift aws cloudformation detect-stack-drift --stack-name asf-core-dev aws cloudformation describe-stack-resource-drifts --stack-name asf-core-dev ``` + +Fix drift by re-deploying from CDK (the code wins), not by editing the template in the console. + +--- + +*Last updated: 27 July 2026 by Alex* diff --git a/infrastructure/stacks/core/README.md b/infrastructure/stacks/core/README.md deleted file mode 100644 index 7556bb0..0000000 --- a/infrastructure/stacks/core/README.md +++ /dev/null @@ -1,159 +0,0 @@ -# Core Stack - -## Overview - -The Core Stack provides shared infrastructure resources for all ASF Mission Data pipelines. It creates the foundational S3 bucket, ECR repository, and GitHub Actions IAM role that pipeline stacks depend on. This stack must be deployed first before any pipeline stacks. - -## Architecture - --- # TO DO: add mermaid diagram - -## Resources Created - -| Resource | Type | Purpose | -|----------|------|---------| -| Data Bucket | `s3.Bucket` | Pipeline data storage (bronze/silver layers) | -| ECR Repository | `ecr.Repository` | Container images for pipeline Lambdas | -| GitHub Actions Role | `iam.Role` | OIDC role for CI/CD deployments | - -### S3 Bucket Features - -- **Encryption**: S3-managed encryption (SSE-S3) -- **Public Access**: Blocked completely -- **SSL**: Enforced for all requests -- **Versioning**: Disabled (pipeline data is reproducible) -- **Removal Policy**: RETAIN in prod, DESTROY in dev - -## Configuration - -| Parameter | Source | Description | -|-----------|--------|-------------| -| `environment` | `config.environment` | Environment name (dev/prod) | -| `aws_account_id` | `config.aws_account_id` | AWS account ID for ARNs | -| `aws_region` | `config.aws_region` | AWS region | -| `github_org` | `config.github_org` | GitHub organization for OIDC | -| `github_repo` | `config.github_repo` | GitHub repository for OIDC | -| `ecr_max_image_count` | `config.ecr_max_image_count` | Max images to retain in ECR | - -### Environment Values - -| Environment | S3 Bucket | ECR Repository | IAM Role | -|-------------|-----------|----------------|----------| -| dev | `asf-mission-data-dev` | `asf-mission-data` | `asf-github-actions-dev` | -| prod | `asf-mission-data-prod` | `asf-mission-data` | `asf-github-actions-prod` | - -## Dependencies - -None - this is the foundation stack that pipeline stacks depend on. - -## Exports - -The stack exports these values via CloudFormation outputs: - -| Output | Export Name | Description | -|--------|-------------|-------------| -| `DataBucketName` | `asf-data-bucket-{env}` | S3 bucket name | -| `DataBucketArn` | `asf-data-bucket-arn-{env}` | S3 bucket ARN | -| `ECRRepositoryUri` | `asf-ecr-uri-{env}` | ECR repository URI | -| `ECRRepositoryArn` | `asf-ecr-arn-{env}` | ECR repository ARN | -| `GitHubActionsRoleArn` | `asf-github-role-arn-{env}` | IAM role ARN | - -## Deployment - -```bash -# From repository root -cd infrastructure - -# Deploy to dev -cdk deploy --context env=dev - -# Deploy to prod -cdk deploy --context env=prod - -# Preview changes -cdk diff --context env=dev -``` - -## Accessing Resources - -### S3 Bucket - -```bash -# List bucket contents -aws s3 ls s3://asf-mission-data-dev/ - -# Bucket structure -# bronze/ - Raw ingested data -# silver/ - Cleaned/transformed data -``` - -### ECR Repository - -```bash -# List images -aws ecr describe-images --repository-name asf-mission-data - -# Login to ECR -aws ecr get-login-password --region eu-west-2 | docker login --username AWS --password-stdin ACCOUNT_ID.dkr.ecr.eu-west-2.amazonaws.com - -# Push an image -docker push ACCOUNT_ID.dkr.ecr.eu-west-2.amazonaws.com/asf-mission-data:latest -``` - -### IAM Role (for debugging) - -```bash -# View role trust policy -aws iam get-role --role-name asf-github-actions-dev --query 'Role.AssumeRolePolicyDocument' - -# List attached policies -aws iam list-role-policies --role-name asf-github-actions-dev -``` - -## Troubleshooting - -| Issue | Cause | Solution | -|-------|-------|----------| -| GitHub Actions can't assume role | OIDC subject mismatch | Verify repo name matches exactly in trust policy | -| ECR push fails with 403 | Missing ECR auth | Run `aws ecr get-login-password` before push | -| S3 access denied | IAM policy issue | Check role has `s3:*` on bucket ARN | -| Stack deletion fails | Bucket not empty | Empty bucket first or set `auto_delete_objects=True` | -| CDK deploy fails | Missing permissions | Ensure deployer has CloudFormation + IAM permissions | - -### OIDC Authentication Fails - -If GitHub Actions can't authenticate: - -1. Verify the OIDC provider exists in AWS IAM -2. Check the trust policy subject claim: - - ``` - repo:nestauk/asf_mission_data:* - ``` - -3. Ensure workflow has `id-token: write` permission -4. Check the audience is `sts.amazonaws.com` - -## Cost Estimate - -| Resource | Monthly Cost | Notes | -|----------|--------------|-------| -| S3 Storage | ~$0.23 | 10 GB at $0.023/GB | -| S3 Requests | ~$0.07 | 10K PUT, 50K GET | -| ECR Storage | ~$0.50 | 5 GB (10 images) at $0.10/GB | -| IAM Role | $0.00 | Free | -| **Total** | **~$0.80/month** | Core stack only | - -See `infrastructure/README.md` for full cost breakdown including pipeline resources. - -## Owner - -TBD - assign a team member as point of contact for this stack. - ---- - -## Code Reference - -- Stack implementation: `stacks/core/core_stack.py` -- Configuration: `config/environments.py` β†’ `EnvironmentConfig` -- CDK entry point: `app.py`