Predict Soil Carbon from Satellite Data

Build a production-grade ML pipeline predicting soil organic carbon from satellite imagery.

Introduction

30 Second Summary

Healthy soil absorbs carbon from the atmosphere, making it one of nature's most powerful tools against climate change. But measuring how much carbon soil actually holds across thousands of locations requires satellite imagery and machine learning working together.

In this project, you will build a production-grade ML pipeline that predicts soil organic carbon from Sentinel-2 satellite imagery. You will implement typed configuration, resilient data extraction with caching and retry logic, experiment tracking, and automated tests.

What You'll Build

A config-driven Python pipeline that extracts spectral features from Google Earth Engine at soil sample locations, trains an XGBoost model to predict soil organic carbon, and tracks experiments with MLflow.

By the end of this project, you'll have:

  • A Pydantic-validated config system and a resilient GEE feature extraction module with disk caching and exponential backoff retry that handles quota and timeout failures gracefully.
  • An XGBoost training pipeline with MLflow experiment tracking that logs parameters, metrics, and feature importance artifacts.
  • A pytest test suite and pre-commit hooks (ruff, mypy) enforcing code quality with automated linting, formatting, and type checking.
  • Secret Mission: Wire up a GitHub Actions CI pipeline that lints, type-checks, tests, and smoke-tests on every push.

Are there any prerequisites?

This project assumes basic Python and machine learning familiarity. You will also need a GitHub account. You will set up Google Cloud and Earth Engine access in Step 1.

Set Up Your Environment and GCP

Before you can pull satellite data or train models, you need a working foundation. Google Earth Engine is the remote sensing platform that serves Sentinel-2 imagery, and it requires a Google Cloud project for access.

In this step you will create that cloud project, scaffold the project directory, install every dependency, and authenticate your machine with Earth Engine.

In this step, get ready to:
  • Create a GCP project and enable the Earth Engine API.
  • Scaffold the project directory and install all dependencies.
  • Authenticate with Earth Engine and initialize a Git repository.
Create a GCP project and enable the Earth Engine API

Earth Engine routes every API call through a Google Cloud project. You need a project with the Earth Engine API enabled before the Python client can connect.

  • Open https://console.cloud.google.com and sign in with your Google account.
  • Click Select a project in the top navigation bar, then click New Project.
  • Enter a project name (e.g. soc-mrv-pipeline) and click Create.
  • Once the project is created, make sure it is selected in the top bar.
  • Navigate to APIs & Services in the left sidebar, then click Library.
  • Search for Earth Engine API and open the result.
  • Click the Enable button.

What about billing?

GCP may ask you to enable billing on the project. Earth Engine is free for noncommercial use, and this project stays well within the free tier. You will not be charged for anything you do in this project.

Next, register the project for Earth Engine access.

  • Open the Earth Engine registration page at https://code.earthengine.google.com/register.
  • Select your new project and choose Unpaid usage (noncommercial).
  • Complete the registration flow and confirm eligibility.
  • After registration, Earth Engine access is enabled immediately.

Why do I need to register separately?

Enabling the API on your Cloud project is not enough on its own. The registration step tells Google whether your usage is commercial or noncommercial, which determines your quota tier and billing.

Create the project directory and install dependencies

The pipeline needs a clean directory structure with separate folders for source code, tests, config, and data. A pyproject.toml file defines all the project metadata and dependencies in one place.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Terminal and press Enter to open it.
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir soc-mrv-pipeline
cd soc-mrv-pipeline
  • Create the directory structure by running these commands:
mkdir src tests configs data cache
touch src/__init__.py tests/__init__.py
  • Confirm the directories exist by running:
ls

You should see cache, configs, data, src, and tests in the output.

Now create the pyproject.toml that defines your project dependencies and tool configuration.

  • Create a file called pyproject.toml in the soc-mrv-pipeline folder and paste the following content:
[project]
name = "soc-mrv-pipeline"
version = "0.1.0"
description = "Mini SOC MRV Pipeline: Predict Soil Organic Carbon from Sentinel-2 imagery"
requires-python = ">=3.12"
license = "MIT"
dependencies = [
    "earthengine-api>=1.7.0",
    "xgboost>=3.3.0",
    "mlflow>=3.14.0",
    "pydantic>=2.13.0",
    "pydantic-settings>=2.0.0",
    "pyyaml>=6.0",
    "pandas>=2.0.0",
    "numpy>=2.0.0",
    "scikit-learn>=1.5.0",
    "tqdm>=4.60.0",
    "tenacity>=9.0.0",
]

[project.optional-dependencies]
dev = [
    "pytest>=8.0.0",
    "pre-commit>=4.0.0",
    "ruff>=0.11.0",
    "mypy>=1.15.0",
    "pandas-stubs>=2.0.0",
]

[build-system]
requires = ["setuptools>=75.0"]
build-backend = "setuptools.build_meta"

[tool.ruff]
line-length = 100
target-version = "py312"

[tool.ruff.lint]
select = ["E", "F", "I", "UP", "B"]

[tool.mypy]
python_version = "3.12"
strict = true
plugins = ["pydantic.mypy"]

[tool.pytest.ini_options]
testpaths = ["tests"]

What does this file do?

The [project] section names your package and lists the runtime dependencies: the Earth Engine client, XGBoost for training, MLflow for experiment tracking, Pydantic for typed config, and tenacity for retry logic.

The [project.optional-dependencies] section adds development tools: pytest for testing, ruff for linting, and mypy for type checking.

The tool configuration sections at the bottom set line length, Python target version, linting rules, and test paths so every tool uses consistent settings.

  • Save pyproject.toml.

Now create a virtual environment and install everything.

  • Create and activate a virtual environment by running these commands:

macOS

python3 -m venv .venv
source .venv/bin/activate

Windows

python3 -m venv .venv
.venv\Scripts\activate

What is a virtual environment?

A virtual environment isolates this project's Python packages from your system Python. Every dependency you install stays inside the .venv folder, preventing version conflicts with other projects on your machine.

You should see (.venv) at the start of your terminal prompt.

  • Install all dependencies (including dev tools) by running:
pip install -e ".[dev]"

What does the -e flag do?

The -e flag installs the package in editable mode. Python imports your src/ code directly from disk, so changes take effect immediately without reinstalling.

This may take a minute as it downloads all packages.

  • Verify the core packages installed correctly by running:
python -c "import xgboost, mlflow, ee, pydantic"

If this completes without errors, all dependencies are installed correctly.

Seeing an ImportError?

Make sure your virtual environment is activated. You should see (.venv) at the start of your terminal prompt.

If the prompt shows no (.venv) prefix, re-run the activation command for your platform, then retry the import. Help me fix my Python import error after pip install.

Authenticate with Earth Engine and initialize Git

Now that the earthengine-api package is installed, you can authenticate your machine so the Python client can access your GCP project.

  • Authenticate with Earth Engine by running this command:
earthengine authenticate

What does this command do?

This opens a browser window where you sign in with your Google account and grant the Earth Engine CLI permission to act on your behalf. The resulting credentials are stored locally so the Python client can use them in future sessions.

  • Follow the browser prompts to sign in and grant access.
  • Once you see a success message in the terminal, verify the connection by running:
python -c "import ee; ee.Initialize(project='[[GCP_PROJECT_ID="your-gcp-project-id"]]')"

If this completes without an error, your authentication is working.

Seeing an error about missing credentials?

Make sure you completed the browser sign-in flow fully. If credentials expired, re-run earthengine authenticate.

Also confirm the Earth Engine API is enabled on your project and that you completed the registration step at code.earthengine.google.com/register. Help me debug my Earth Engine authentication error.

Now set up version control. A .gitignore keeps generated files, caches, and your virtual environment out of the repository.

  • Create a .gitignore file in the soc-mrv-pipeline folder with this content:
.venv/
__pycache__/
*.egg-info/
cache/
artifacts/
mlflow.db
mlruns/
.env
*.pyc
data/*.csv
data/*.parquet
  • Create a README.md file with this content:
# SOC MRV Pipeline

A mini MRV (Measurement, Reporting, Verification) pipeline that predicts
soil organic carbon from Sentinel-2 satellite imagery using XGBoost,
with typed configuration, experiment tracking, and automated tests.
  • Initialize the Git repository and create your first commit by running these commands:
git init
git add .
git commit -m "Initial project scaffold"

What does this do?

The git init command creates a new Git repository in the current folder. Then git add . stages all the files you created, and git commit saves them as the first snapshot.

  • Verify the repository is clean by running:
git status

You should see a message saying there is nothing to commit and the working tree is clean.

Git says there are untracked files?

This usually means you created a file after running git add .. Run git add . again followed by another commit to include them. Help me clean up my git status.

✔️ Awesome, I've got everything!

Great. Double check you have saved all your files and your git status is clean.

ⓧ I'd like to double check the full code

Here is the complete pyproject.toml you should have:

[project]
name = "soc-mrv-pipeline"
version = "0.1.0"
description = "Mini SOC MRV Pipeline: Predict Soil Organic Carbon from Sentinel-2 imagery"
requires-python = ">=3.12"
license = "MIT"
dependencies = [
    "earthengine-api>=1.7.0",
    "xgboost>=3.3.0",
    "mlflow>=3.14.0",
    "pydantic>=2.13.0",
    "pydantic-settings>=2.0.0",
    "pyyaml>=6.0",
    "pandas>=2.0.0",
    "numpy>=2.0.0",
    "scikit-learn>=1.5.0",
    "tqdm>=4.60.0",
    "tenacity>=9.0.0",
]

[project.optional-dependencies]
dev = [
    "pytest>=8.0.0",
    "pre-commit>=4.0.0",
    "ruff>=0.11.0",
    "mypy>=1.15.0",
    "pandas-stubs>=2.0.0",
]

[build-system]
requires = ["setuptools>=75.0"]
build-backend = "setuptools.build_meta"

[tool.ruff]
line-length = 100
target-version = "py312"

[tool.ruff.lint]
select = ["E", "F", "I", "UP", "B"]

[tool.mypy]
python_version = "3.12"
strict = true
plugins = ["pydantic.mypy"]

[tool.pytest.ini_options]
testpaths = ["tests"]

Your src/__init__.py and tests/__init__.py should both be empty files.

Your environment is fully set up. Next up, you will build a typed configuration system that validates every parameter in your pipeline before anything runs.

Build the Config System with Pydantic

Your environment is ready and dependencies are installed. Now you need a way to manage the dozens of parameters your pipeline depends on: data paths, satellite band selections, model hyperparameters, and experiment tracking settings.

Without a typed config system, these values end up scattered as magic numbers across your codebase. In this step, you'll build a Pydantic-validated configuration layer that catches invalid values before they silently corrupt a training run.

In this step, get ready to:
  • Create typed Pydantic models for data, features, model, and MLflow config domains.
  • Create a YAML config file and a loader function that merges file-based and CLI overrides.
  • Test CLI parameter overrides to confirm they propagate correctly.
Create typed Pydantic config models

Each domain of your pipeline (data ingestion, feature extraction, model training, experiment tracking) has its own set of parameters. Pydantic's BaseModel lets you declare these with type hints and constraints so that invalid values fail loudly at startup.

  • Create a new file called config.py inside your src/ directory.
  • Add the imports and the first two domain config models by pasting this code into src/config.py:
"""Typed configuration system using Pydantic BaseSettings + YAML."""

import sys
from pathlib import Path
from typing import Any

import yaml
from pydantic import BaseModel, Field, field_validator
from pydantic_settings import BaseSettings


class DataConfig(BaseModel):
    lucas_csv_path: str = "data/lucas_fixture.csv"
    sample_size: int = Field(default=50, ge=1, le=50000)
    date_start: str = "2018-04-01"
    date_end: str = "2018-09-30"


class FeatureConfig(BaseModel):
    bands: list[str] = [
        "B2", "B3", "B4", "B5", "B6", "B7", "B8", "B8A", "B11", "B12"
    ]
    indices: list[str] = ["NDVI", "EVI", "BSI"]
    buffer_meters: int = Field(default=100, ge=10, le=1000)
    cloud_cover_max: int = Field(default=20, ge=0, le=100)

What does this code do?

DataConfig defines where your LUCAS soil sample data lives and the date range for satellite imagery. The Field(ge=1, le=50000) constraint on sample_size means Pydantic rejects any value below 1 or above 50,000 at instantiation time.

FeatureConfig declares which Sentinel-2 bands to extract and which spectral indices to compute. The buffer radius and cloud cover threshold are also constrained to sensible ranges.

You won't see a change yet. The next chunk adds the model and orchestration configs that complete the system.

  • Add the model training config, MLflow config, and the root PipelineConfig below your FeatureConfig class in src/config.py:
class ModelConfig(BaseModel):
    n_estimators: int = Field(default=200, ge=10, le=5000)
    max_depth: int = Field(default=6, ge=1, le=20)
    learning_rate: float = Field(default=0.1, gt=0.0, le=1.0)
    subsample: float = Field(default=0.8, gt=0.0, le=1.0)
    test_size: float = Field(default=0.2, gt=0.0, lt=1.0)
    random_state: int = 42


class MLflowConfig(BaseModel):
    tracking_uri: str = "sqlite:///mlflow.db"
    experiment_name: str = "soc-prediction"


class PipelineConfig(BaseSettings):
    data: DataConfig = DataConfig()
    features: FeatureConfig = FeatureConfig()
    model: ModelConfig = ModelConfig()
    mlflow: MLflowConfig = MLflowConfig()

    @field_validator("data", "features", "model", "mlflow", mode="before")
    @classmethod
    def parse_nested(cls, v: Any) -> Any:
        if isinstance(v, dict):
            return v
        return v

What does this code do?

ModelConfig holds XGBoost hyperparameters with range constraints. For example, learning_rate must be between 0 and 1. MLflowConfig stores the tracking backend URI and experiment name.

PipelineConfig inherits from BaseSettings (not BaseModel) so it can read environment variables. The field_validator with mode="before" ensures that raw dictionaries from YAML are passed through to the nested model constructors correctly.

  • Save src/config.py.
  • Verify that Pydantic rejects invalid values by running this in your terminal:
python -c "from src.config import ModelConfig; ModelConfig(n_estimators=-1)"

You should see a ValidationError telling you the value is less than the minimum of 10.

Seeing ModuleNotFoundError?

Make sure you're running the command from your project root directory (the folder containing pyproject.toml). Python needs to find the src package from there.

Also confirm your virtual environment is still activated and that you ran pip install -e "." in the previous step.

Help me debug a ModuleNotFoundError when importing src.config

Create the YAML config and loader function

Your Pydantic models define the schema. Now you need a YAML file that holds the actual default values, and a loader function that reads the file and creates a validated PipelineConfig instance.

  • Create the configs directory and the default config file by running:
mkdir configs
touch configs/default.yaml
  • Confirm the file was created by running:
ls configs/

You should see default.yaml in the output.

  • Open configs/default.yaml and paste the following default configuration:
data:
  lucas_csv_path: "data/lucas_fixture.csv"
  sample_size: 50
  date_start: "2018-04-01"
  date_end: "2018-09-30"

features:
  bands:
    - "B2"
    - "B3"
    - "B4"
    - "B5"
    - "B6"
    - "B7"
    - "B8"
    - "B8A"
    - "B11"
    - "B12"
  indices:
    - "NDVI"
    - "EVI"
    - "BSI"
  buffer_meters: 100
  cloud_cover_max: 20

model:
  n_estimators: 200
  max_depth: 6
  learning_rate: 0.1
  subsample: 0.8
  test_size: 0.2
  random_state: 42

mlflow:
  tracking_uri: "sqlite:///mlflow.db"
  experiment_name: "soc-prediction"

What does this config file define?

The data section points to a 50-point fixture CSV of LUCAS soil samples and a 2018 growing-season date range for Sentinel-2 imagery.

The features section lists the 10 Sentinel-2 bands to extract plus 3 spectral indices (NDVI, EVI, BSI). The model section holds XGBoost hyperparameters you'll tune later.

  • Save configs/default.yaml.

✔️ Awesome, I've got everything!

Great. Double check you've saved the file before moving on.

ⓧ I'd like to double check the full code

data:
  lucas_csv_path: "data/lucas_fixture.csv"
  sample_size: 50
  date_start: "2018-04-01"
  date_end: "2018-09-30"

features:
  bands:
    - "B2"
    - "B3"
    - "B4"
    - "B5"
    - "B6"
    - "B7"
    - "B8"
    - "B8A"
    - "B11"
    - "B12"
  indices:
    - "NDVI"
    - "EVI"
    - "BSI"
  buffer_meters: 100
  cloud_cover_max: 20

model:
  n_estimators: 200
  max_depth: 6
  learning_rate: 0.1
  subsample: 0.8
  test_size: 0.2
  random_state: 42

mlflow:
  tracking_uri: "sqlite:///mlflow.db"
  experiment_name: "soc-prediction"

Now you need a function that reads this YAML and produces a validated PipelineConfig object. The function will also support command-line overrides using dot-notation flags.

  • Add the load_config function at the bottom of src/config.py, below the PipelineConfig class:
def load_config(config_path: str = "configs/default.yaml") -> PipelineConfig:
    """Load config from YAML, then apply CLI overrides in dot-notation."""
    path = Path(config_path)
    if path.exists():
        with open(path) as f:
            yaml_data = yaml.safe_load(f) or {}
    else:
        yaml_data = {}

    # Apply CLI overrides: --key.subkey=value
    for arg in sys.argv[1:]:
        if arg.startswith("--") and "=" in arg:
            key_path, value = arg[2:].split("=", 1)
            keys = key_path.split(".")
            target = yaml_data
            for k in keys[:-1]:
                target = target.setdefault(k, {})
            # Try to cast to int/float
            try:
                value = int(value)  # type: ignore[assignment]
            except ValueError:
                try:
                    value = float(value)  # type: ignore[assignment]
                except ValueError:
                    pass
            target[keys[-1]] = value

    return PipelineConfig(**yaml_data)

What does this function do?

load_config reads your YAML file first, then scans sys.argv for any --key.subkey=value flags. CLI flags override whatever the YAML file set.

The dot-notation parsing walks into nested dictionaries. So --model.n_estimators=500 sets yaml_data["model"]["n_estimators"] = 500 before Pydantic validates the whole structure.

  • Save src/config.py.
  • Verify the config loads from YAML by running:
python -c "from src.config import load_config; cfg = load_config(); print(cfg.model.n_estimators)"

You should see 200 printed, which matches the value in your YAML file.

Seeing a FileNotFoundError or empty config?

Make sure the configs/default.yaml file exists at the correct path relative to where you're running the command. Run ls configs/ to confirm.

If you see a YAML parsing error, check for incorrect indentation. YAML uses spaces (not tabs) and is sensitive to alignment.

Help me debug why load_config is not reading my YAML file

✔️ Awesome, I've got everything!

Great. Double check you've saved the file before moving on.

ⓧ I'd like to double check the full code

"""Typed configuration system using Pydantic BaseSettings + YAML."""

import sys
from pathlib import Path
from typing import Any

import yaml
from pydantic import BaseModel, Field, field_validator
from pydantic_settings import BaseSettings


class DataConfig(BaseModel):
    lucas_csv_path: str = "data/lucas_fixture.csv"
    sample_size: int = Field(default=50, ge=1, le=50000)
    date_start: str = "2018-04-01"
    date_end: str = "2018-09-30"


class FeatureConfig(BaseModel):
    bands: list[str] = [
        "B2", "B3", "B4", "B5", "B6", "B7", "B8", "B8A", "B11", "B12"
    ]
    indices: list[str] = ["NDVI", "EVI", "BSI"]
    buffer_meters: int = Field(default=100, ge=10, le=1000)
    cloud_cover_max: int = Field(default=20, ge=0, le=100)


class ModelConfig(BaseModel):
    n_estimators: int = Field(default=200, ge=10, le=5000)
    max_depth: int = Field(default=6, ge=1, le=20)
    learning_rate: float = Field(default=0.1, gt=0.0, le=1.0)
    subsample: float = Field(default=0.8, gt=0.0, le=1.0)
    test_size: float = Field(default=0.2, gt=0.0, lt=1.0)
    random_state: int = 42


class MLflowConfig(BaseModel):
    tracking_uri: str = "sqlite:///mlflow.db"
    experiment_name: str = "soc-prediction"


class PipelineConfig(BaseSettings):
    data: DataConfig = DataConfig()
    features: FeatureConfig = FeatureConfig()
    model: ModelConfig = ModelConfig()
    mlflow: MLflowConfig = MLflowConfig()

    @field_validator("data", "features", "model", "mlflow", mode="before")
    @classmethod
    def parse_nested(cls, v: Any) -> Any:
        if isinstance(v, dict):
            return v
        return v


def load_config(config_path: str = "configs/default.yaml") -> PipelineConfig:
    """Load config from YAML, then apply CLI overrides in dot-notation."""
    path = Path(config_path)
    if path.exists():
        with open(path) as f:
            yaml_data = yaml.safe_load(f) or {}
    else:
        yaml_data = {}

    # Apply CLI overrides: --key.subkey=value
    for arg in sys.argv[1:]:
        if arg.startswith("--") and "=" in arg:
            key_path, value = arg[2:].split("=", 1)
            keys = key_path.split(".")
            target = yaml_data
            for k in keys[:-1]:
                target = target.setdefault(k, {})
            # Try to cast to int/float
            try:
                value = int(value)  # type: ignore[assignment]
            except ValueError:
                try:
                    value = float(value)  # type: ignore[assignment]
                except ValueError:
                    pass
            target[keys[-1]] = value

    return PipelineConfig(**yaml_data)
Test CLI parameter overrides

The load_config function already parses CLI flags from sys.argv. This means any script that calls load_config() automatically supports command-line overrides without extra setup.

  • Test the CLI override by running a quick script that passes a flag:
python -c "
import sys
sys.argv = ['test', '--model.n_estimators=500']
from src.config import load_config
cfg = load_config()
print(f'n_estimators overridden to: {cfg.model.n_estimators}')
"

You should see n_estimators overridden to: 500 printed. The YAML default was 200, but the CLI flag bumped it to 500.

Why config-driven over hardcoded?

In production MRV pipelines, you run experiments with different hyperparameters, date ranges, and feature sets. A config-driven approach lets you change any parameter without touching code. The YAML file stores your defaults, and CLI overrides let you run quick experiments from the terminal.

Pydantic's type validation catches mistakes (like setting learning_rate=2.0) at load time instead of 30 minutes into a training run.

Override not taking effect?

The override must use the exact dot-notation path matching your YAML structure. For example, --model.n_estimators=500 works, but --n_estimators=500 would create a top-level key that Pydantic ignores.

Help me debug why my CLI config override is not working

Your config system is locked in. Every parameter in your pipeline is now typed, validated, and overridable from YAML or the command line. Next up, you'll build the GEE feature extraction layer that pulls Sentinel-2 spectral data from satellite imagery.

Build GEE Feature Extraction with Caching and Retry

Your config system gives the pipeline its parameters. Now you need the data layer that actually talks to Google Earth Engine and pulls satellite imagery for each soil sample location.

In production MRV systems, GEE calls are the most fragile part of the pipeline. They hit quota limits, time out under load, and cost real wall-clock time to re-run. This step builds a feature extraction module that handles all of that gracefully with retry logic and local disk caching.

In this step, get ready to:
  • Build a core extraction function that pulls Sentinel-2 bands and spectral indices from GEE.
  • Add exponential-backoff retry and disk caching so failed or repeated calls are handled gracefully.
  • Build a batch pipeline that extracts features for all 50 LUCAS fixture points into a clean DataFrame.
Build the core extraction function

The extraction function needs to do three things: query a Sentinel-2 image collection for a location and date range, compute spectral indices (NDVI, EVI, BSI), and return the band values as a dictionary.

  • Create a new file called src/gee_features.py inside your src/ folder.
  • Add the imports and module-level constant by pasting the following code:
"""GEE feature extraction with caching and retry logic."""

import hashlib
import json
from pathlib import Path
from typing import Any

import ee
import pandas as pd
from tenacity import (
    retry,
    retry_if_exception_type,
    stop_after_attempt,
    wait_exponential,
)
from tqdm import tqdm

from src.config import FeatureConfig

CACHE_DIR = Path("cache/gee_features")

What do these imports do?

  • ee is the Google Earth Engine Python API for querying satellite imagery.
  • tenacity provides production-grade retry decorators with exponential backoff.
  • tqdm adds progress bars to loops so you can see extraction progress.
  • CACHE_DIR defines where extracted features get cached on disk to avoid re-fetching.
  • Below the imports, add the compute_indices() function that calculates NDVI, EVI, and BSI from Sentinel-2 bands:
def compute_indices(image: ee.Image) -> ee.Image:
    """Compute spectral indices from Sentinel-2 bands."""
    ndvi = image.normalizedDifference(["B8", "B4"]).rename("NDVI")
    evi = (
        image.expression(
            "2.5 * ((NIR - RED) / (NIR + 6 * RED - 7.5 * BLUE + 1))",
            {
                "NIR": image.select("B8"),
                "RED": image.select("B4"),
                "BLUE": image.select("B2"),
            },
        )
        .rename("EVI")
    )
    bsi = (
        image.expression(
            "((SWIR + RED) - (NIR + BLUE)) / ((SWIR + RED) + (NIR + BLUE))",
            {
                "SWIR": image.select("B11"),
                "RED": image.select("B4"),
                "NIR": image.select("B8"),
                "BLUE": image.select("B2"),
            },
        )
        .rename("BSI")
    )
    return image.addBands([ndvi, evi, bsi])

What are these spectral indices?

Each index combines Sentinel-2 bands in a formula that highlights a specific ground property.

  • NDVI (Normalized Difference Vegetation Index): measures vegetation density. Higher values mean more plant cover.
  • EVI (Enhanced Vegetation Index): similar to NDVI but corrects for atmospheric and soil background noise.
  • BSI (Bare Soil Index): highlights exposed soil surfaces. Useful for detecting areas with low organic carbon.

All three are strong predictors of soil organic carbon because vegetation cover correlates with carbon input to the soil.

Now add the function that actually queries GEE for a single coordinate. This function filters the Sentinel-2 collection, builds a median composite, computes indices, and reduces the region to mean values.

  • Below compute_indices(), leave some space (you will add cache functions here in the next substep) and add this extraction function at the bottom of the file:
@retry(
    retry=retry_if_exception_type(ee.ee_exception.EEException),
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=2, min=4, max=60),
    reraise=True,
)
def _extract_point_features(
    lon: float,
    lat: float,
    date_start: str,
    date_end: str,
    config: FeatureConfig,
) -> dict[str, float]:
    """Extract Sentinel-2 features for a single point from GEE."""
    point = ee.Geometry.Point([lon, lat])
    region = point.buffer(config.buffer_meters)

    collection = (
        ee.ImageCollection("COPERNICUS/S2_SR_HARMONIZED")
        .filterBounds(region)
        .filterDate(date_start, date_end)
        .filter(ee.Filter.lt("CLOUDY_PIXEL_PERCENTAGE", config.cloud_cover_max))
        .select(config.bands)
    )

    composite = collection.median()
    composite = compute_indices(composite)

    all_bands = config.bands + config.indices
    values = composite.reduceRegion(
        reducer=ee.Reducer.mean(),
        geometry=region,
        scale=10,
        maxPixels=1e6,
    ).getInfo()

    return {band: values.get(band) for band in all_bands}

What does this function do?

  • ee.Geometry.Point([lon, lat]) creates a point geometry, then .buffer(config.buffer_meters) expands it to a 100m radius region.
  • The COPERNICUS/S2_SR_HARMONIZED collection is filtered by location, date range, and cloud cover percentage.
  • .median() composites all matching images into a single cloud-free representation.
  • reduceRegion() computes the mean pixel value across the buffered area for each band.
  • The @retry decorator from tenacity automatically retries up to 3 times with exponential backoff if GEE throws an EEException (quota exceeded, timeout, etc.).
  • Save src/gee_features.py.
  • Verify the extraction works for a single point by running this in your terminal:
python -c "
import ee
ee.Initialize(project='[[GCP_PROJECT_ID="your-gcp-project-id"]]')
from src.config import load_config
from src.gee_features import _extract_point_features
cfg = load_config()
result = _extract_point_features(23.7, 37.9, cfg.data.date_start, cfg.data.date_end, cfg.features)
print(result.keys())
print(all(v is not None for v in result.values()))
"

You should see the dictionary keys (all 13 bands + indices) printed, followed by True confirming all values are non-null.

Seeing an EEException or timeout?

  • Make sure you authenticated with earthengine authenticate in Step 1 and that your GCP project has the Earth Engine API enabled.
  • If you see a quota error, wait 30 seconds and try again. The retry decorator handles this automatically in the pipeline.
  • If all values are None, try a different coordinate. Some locations have no cloud-free Sentinel-2 coverage in the date range.

Why is my GEE extraction returning None values or throwing an error?

Add disk caching for extracted features

Each GEE call takes several seconds and counts against your quota. In a real MRV pipeline, you re-run extraction frequently as you iterate on features or fix bugs. Without caching, every re-run re-fetches every point from scratch.

Why disk caching instead of an in-memory cache?

An in-memory cache (like a dictionary) disappears when the script exits. Disk caching persists across runs, meaning you only ever fetch each point once. This is critical when extracting thousands of LUCAS points where a full run takes hours.

The cache key combines the point ID and date range hash, so changing your config date range automatically invalidates the cache for affected points.

  • In src/gee_features.py, add the following three cache functions between compute_indices() and _extract_point_features():
def _cache_key(point_id: str, date_start: str, date_end: str) -> str:
    """Generate a deterministic cache key."""
    raw = f"{point_id}_{date_start}_{date_end}"
    return hashlib.md5(raw.encode()).hexdigest()


def _get_cached(cache_key: str) -> dict[str, float] | None:
    """Return cached features or None."""
    cache_file = CACHE_DIR / f"{cache_key}.json"
    if cache_file.exists():
        with open(cache_file) as f:
            return json.load(f)
    return None


def _save_cache(cache_key: str, features: dict[str, float]) -> None:
    """Persist features to disk cache."""
    CACHE_DIR.mkdir(parents=True, exist_ok=True)
    cache_file = CACHE_DIR / f"{cache_key}.json"
    with open(cache_file, "w") as f:
        json.dump(features, f)

How does the caching layer work?

  • _cache_key() hashes the point ID + date range into a unique filename. Same inputs always produce the same key.
  • _get_cached() checks if a JSON file exists for that key. Returns the data if found, None if not.
  • _save_cache() writes the extracted features to a JSON file, creating the cache directory if needed.

Now you need a wrapper function that checks the cache before calling GEE. If the point is already cached, it returns instantly without making a network call.

  • Add extract_features_for_point() below _extract_point_features():
def extract_features_for_point(
    point_id: str,
    lon: float,
    lat: float,
    date_start: str,
    date_end: str,
    config: FeatureConfig,
) -> dict[str, Any]:
    """Extract features with caching layer."""
    cache_key = _cache_key(point_id, date_start, date_end)

    cached = _get_cached(cache_key)
    if cached is not None:
        return cached

    features = _extract_point_features(lon, lat, date_start, date_end, config)
    _save_cache(cache_key, features)
    return features

What does this function do?

This is the public-facing extraction function. It wraps the raw GEE call with a cache-first strategy: check disk, return if found, otherwise fetch from GEE and persist the result.

The retry decorator on _extract_point_features() handles transient GEE failures automatically. Combined with caching, a partially-failed batch run can be resumed without re-fetching successful points.

  • Save src/gee_features.py.
  • Verify the cache works by running extraction twice for the same point:
python -c "
import ee, time
ee.Initialize(project='[[GCP_PROJECT_ID="your-gcp-project-id"]]')
from src.config import load_config
from src.gee_features import extract_features_for_point
cfg = load_config()

start = time.time()
r1 = extract_features_for_point('TEST01', 23.7, 37.9, cfg.data.date_start, cfg.data.date_end, cfg.features)
print(f'First call: {time.time() - start:.2f}s')

start = time.time()
r2 = extract_features_for_point('TEST01', 23.7, 37.9, cfg.data.date_start, cfg.data.date_end, cfg.features)
print(f'Second call (cached): {time.time() - start:.2f}s')
print(f'Results match: {r1 == r2}')
"

The first call should take several seconds. The second call should complete almost instantly (under 0.01s) because it reads from the disk cache.

Second call still taking seconds?

  • Check that the cache/gee_features/ directory was created and contains a .json file after the first call.
  • Make sure _save_cache() is called after _extract_point_features() in your extract_features_for_point() function.
  • Verify you are calling extract_features_for_point() (with caching), not _extract_point_features() (without caching).

Why is my GEE cache not working on the second extraction call?

Build the batch pipeline with a fixture dataset

The full LUCAS 2018 dataset has over 18,000 soil sample points across Europe. For development and testing, you will use a 50-point fixture CSV that lets you iterate quickly without waiting for thousands of GEE calls.

  • Create the data/ directory and generate the fixture dataset by running:
mkdir -p data
  • Create a fixture CSV with 50 representative European coordinates by running this script:
python -c "
import csv
import random

random.seed(42)
rows = []
for i in range(50):
    point_id = f'PT{i+1:04d}'
    lon = round(random.uniform(-5.0, 30.0), 4)
    lat = round(random.uniform(36.0, 60.0), 4)
    oc = round(random.uniform(5.0, 150.0), 1)
    rows.append([point_id, lon, lat, oc])

with open('data/lucas_fixture.csv', 'w', newline='') as f:
    writer = csv.writer(f)
    writer.writerow(['POINT_ID', 'LON', 'LAT', 'OC'])
    writer.writerows(rows)

print(f'Created data/lucas_fixture.csv with {len(rows)} points')
"

Why a fixture dataset?

The real LUCAS 2018 topsoil data requires registration with ESDAC (the European Soil Data Centre). The fixture CSV uses synthetic coordinates across Europe so you can develop and test the pipeline immediately.

In a production system, you would swap this for the real LUCAS CSV. The pipeline handles both identically because it only reads the POINT_ID, LON, LAT, and OC columns.

  • Confirm the file was created by running:
ls data/

You should see lucas_fixture.csv listed in the output.

Now add the batch pipeline function that iterates through all points in the fixture CSV, extracts features (with caching), and returns a clean DataFrame.

  • In src/gee_features.py, add build_feature_dataframe() at the bottom of the file, below extract_features_for_point():
def build_feature_dataframe(
    lucas_df: pd.DataFrame,
    config: FeatureConfig,
    date_start: str,
    date_end: str,
) -> pd.DataFrame:
    """Extract features for all LUCAS points and return a clean DataFrame."""
    records: list[dict[str, Any]] = []

    for _, row in tqdm(lucas_df.iterrows(), total=len(lucas_df), desc="Extracting GEE features"):
        point_id = str(row["POINT_ID"])
        lon = float(row["LON"])
        lat = float(row["LAT"])
        soc = float(row["OC"])

        features = extract_features_for_point(
            point_id=point_id,
            lon=lon,
            lat=lat,
            date_start=date_start,
            date_end=date_end,
            config=config,
        )

        record = {"POINT_ID": point_id, "SOC": soc, **features}
        records.append(record)

    df = pd.DataFrame(records)
    # Drop rows where GEE returned nulls (cloud-covered areas)
    feature_cols = config.bands + config.indices
    df = df.dropna(subset=feature_cols)
    return df

What does this function do?

  • Iterates through every row in the LUCAS DataFrame with a tqdm progress bar so you can track extraction progress.
  • Calls extract_features_for_point() for each coordinate. Points already in the cache return instantly.
  • Drops any rows where GEE returned None values (typically cloud-covered locations with no clear imagery).
  • Returns a clean DataFrame with POINT_ID, SOC (the target variable), and all 13 feature columns.
  • Save src/gee_features.py.
  • Run the batch pipeline on your 50-point fixture dataset:
python -c "
import ee
import pandas as pd
ee.Initialize(project='[[GCP_PROJECT_ID="your-gcp-project-id"]]')
from src.config import load_config
from src.gee_features import build_feature_dataframe

cfg = load_config()
lucas_df = pd.read_csv(cfg.data.lucas_csv_path)
df = build_feature_dataframe(lucas_df, cfg.features, cfg.data.date_start, cfg.data.date_end)
print(f'Shape: {df.shape}')
print(f'Null count in features: {df[cfg.features.bands + cfg.features.indices].isnull().sum().sum()}')
print(df.head())
"

You should see a progress bar as extraction runs, followed by the DataFrame shape (rows will be <= 50 after dropping nulls) and a null count of 0 in the feature columns.

First run takes a few minutes

The first extraction run fetches all 50 points from GEE, which takes several seconds per point. Subsequent runs complete in seconds because every point is cached on disk.

Seeing many null rows dropped?

  • Some fixture coordinates may fall in areas with heavy cloud cover during the April-September window. This is expected behavior. The dropna() call removes these automatically.
  • If ALL rows are null, check that your GCP project has Earth Engine API access enabled and that ee.Initialize() completes without error.
  • If you see repeated EEException errors after 3 retries, you may be hitting GEE rate limits. Wait a minute and re-run. Already-cached points will not be re-fetched.

Help me debug why my batch GEE extraction is failing or returning all nulls.

✔️ Awesome, I've got everything!

Great. Make sure you have saved src/gee_features.py and that the cache/gee_features/ directory contains cached JSON files from your test runs.

ⓧ I'd like to double check the full code

Here is the complete src/gee_features.py file:

"""GEE feature extraction with caching and retry logic."""

import hashlib
import json
from pathlib import Path
from typing import Any

import ee
import pandas as pd
from tenacity import (
    retry,
    retry_if_exception_type,
    stop_after_attempt,
    wait_exponential,
)
from tqdm import tqdm

from src.config import FeatureConfig

CACHE_DIR = Path("cache/gee_features")


def compute_indices(image: ee.Image) -> ee.Image:
    """Compute spectral indices from Sentinel-2 bands."""
    ndvi = image.normalizedDifference(["B8", "B4"]).rename("NDVI")
    evi = (
        image.expression(
            "2.5 * ((NIR - RED) / (NIR + 6 * RED - 7.5 * BLUE + 1))",
            {
                "NIR": image.select("B8"),
                "RED": image.select("B4"),
                "BLUE": image.select("B2"),
            },
        )
        .rename("EVI")
    )
    bsi = (
        image.expression(
            "((SWIR + RED) - (NIR + BLUE)) / ((SWIR + RED) + (NIR + BLUE))",
            {
                "SWIR": image.select("B11"),
                "RED": image.select("B4"),
                "NIR": image.select("B8"),
                "BLUE": image.select("B2"),
            },
        )
        .rename("BSI")
    )
    return image.addBands([ndvi, evi, bsi])


def _cache_key(point_id: str, date_start: str, date_end: str) -> str:
    """Generate a deterministic cache key."""
    raw = f"{point_id}_{date_start}_{date_end}"
    return hashlib.md5(raw.encode()).hexdigest()


def _get_cached(cache_key: str) -> dict[str, float] | None:
    """Return cached features or None."""
    cache_file = CACHE_DIR / f"{cache_key}.json"
    if cache_file.exists():
        with open(cache_file) as f:
            return json.load(f)
    return None


def _save_cache(cache_key: str, features: dict[str, float]) -> None:
    """Persist features to disk cache."""
    CACHE_DIR.mkdir(parents=True, exist_ok=True)
    cache_file = CACHE_DIR / f"{cache_key}.json"
    with open(cache_file, "w") as f:
        json.dump(features, f)


@retry(
    retry=retry_if_exception_type(ee.ee_exception.EEException),
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=2, min=4, max=60),
    reraise=True,
)
def _extract_point_features(
    lon: float,
    lat: float,
    date_start: str,
    date_end: str,
    config: FeatureConfig,
) -> dict[str, float]:
    """Extract Sentinel-2 features for a single point from GEE."""
    point = ee.Geometry.Point([lon, lat])
    region = point.buffer(config.buffer_meters)

    collection = (
        ee.ImageCollection("COPERNICUS/S2_SR_HARMONIZED")
        .filterBounds(region)
        .filterDate(date_start, date_end)
        .filter(ee.Filter.lt("CLOUDY_PIXEL_PERCENTAGE", config.cloud_cover_max))
        .select(config.bands)
    )

    composite = collection.median()
    composite = compute_indices(composite)

    all_bands = config.bands + config.indices
    values = composite.reduceRegion(
        reducer=ee.Reducer.mean(),
        geometry=region,
        scale=10,
        maxPixels=1e6,
    ).getInfo()

    return {band: values.get(band) for band in all_bands}


def extract_features_for_point(
    point_id: str,
    lon: float,
    lat: float,
    date_start: str,
    date_end: str,
    config: FeatureConfig,
) -> dict[str, Any]:
    """Extract features with caching layer."""
    cache_key = _cache_key(point_id, date_start, date_end)

    cached = _get_cached(cache_key)
    if cached is not None:
        return cached

    features = _extract_point_features(lon, lat, date_start, date_end, config)
    _save_cache(cache_key, features)
    return features


def build_feature_dataframe(
    lucas_df: pd.DataFrame,
    config: FeatureConfig,
    date_start: str,
    date_end: str,
) -> pd.DataFrame:
    """Extract features for all LUCAS points and return a clean DataFrame."""
    records: list[dict[str, Any]] = []

    for _, row in tqdm(lucas_df.iterrows(), total=len(lucas_df), desc="Extracting GEE features"):
        point_id = str(row["POINT_ID"])
        lon = float(row["LON"])
        lat = float(row["LAT"])
        soc = float(row["OC"])

        features = extract_features_for_point(
            point_id=point_id,
            lon=lon,
            lat=lat,
            date_start=date_start,
            date_end=date_end,
            config=config,
        )

        record = {"POINT_ID": point_id, "SOC": soc, **features}
        records.append(record)

    df = pd.DataFrame(records)
    # Drop rows where GEE returned nulls (cloud-covered areas)
    feature_cols = config.bands + config.indices
    df = df.dropna(subset=feature_cols)
    return df

Your feature extraction module is complete. It handles the hardest part of MRV pipelines: reliably pulling satellite data at scale without wasting quota or time. Next up, you will train an XGBoost model on these features and track experiments with MLflow.

Train the Model with MLflow Tracking

Your feature extraction pipeline is complete. Cached GEE data sits on disk, ready to be consumed by a model.

In this step, you will train an XGBoost regression model to predict soil organic carbon from those spectral features. You will also track every experiment with MLflow so you can compare runs, reproduce results, and share findings.

In this step, get ready to:
  • Set up MLflow with a local SQLite backend and verify the UI.
  • Build a training script that logs parameters, metrics, and artifacts to MLflow.
  • Compare two experiment runs with different hyperparameters.
Set up MLflow and prepare the feature dataset

Before training, you need two things: the extracted features saved as a Parquet file that the training script can load, and an MLflow server running so you can inspect results in a browser.

  • Generate data/features.parquet from your cached GEE extractions by running this command:
python -c "
import pandas as pd
from src.config import load_config
from src.gee_features import build_feature_dataframe

cfg = load_config()
lucas_df = pd.read_csv(cfg.data.lucas_csv_path)
df = build_feature_dataframe(lucas_df, cfg.features, cfg.data.date_start, cfg.data.date_end)
df.to_parquet('data/features.parquet', index=False)
print(f'Saved {len(df)} rows to data/features.parquet')
"

What does this command do?

This one-off script calls build_feature_dataframe() from the previous step. Because the GEE results are cached on disk, the extraction finishes almost instantly.

The output is a Parquet file that the training script loads. Parquet is a columnar format that is faster to read than CSV and preserves column types.

  • Confirm the file was created by running:
ls data/features.parquet

You should see data/features.parquet listed in the output.

  • Open a second terminal window (keep your current one available for running scripts).
  • In the new terminal, activate your virtual environment and start the MLflow server by running:
mlflow server --port 5000

What does this command do?

This starts a local MLflow tracking server on port 5000. Since MLflow 3.7.0, the default backend is sqlite:///mlflow.db in your working directory. No extra database setup needed.

Your training script writes directly to this same SQLite file. The server reads from it and renders the web UI.

  • Open your browser and navigate to http://localhost:5000.

You should see the MLflow UI with an empty experiment list. The server is ready to receive training runs.

Server won't start or port already in use?

Make sure no other process is using port 5000. You can check with lsof -i :5000 (macOS/Linux) or netstat -ano | findstr :5000 (Windows). Kill the conflicting process or choose a different port.

Also confirm your virtual environment is activated in the second terminal. MLflow must be importable.

help me debug my MLflow server startup

Build the training script with experiment logging

The training script ties together the config system, the feature data, and MLflow tracking. Every run logs its hyperparameters, evaluation metrics, the trained model, and a feature importance CSV.

  • Create a new file src/train.py and add the imports, function signature, and data-loading logic:
"""Training script with MLflow experiment tracking."""

import mlflow
import pandas as pd
from sklearn.metrics import mean_absolute_error, mean_squared_error, r2_score
from sklearn.model_selection import train_test_split
from xgboost import XGBRegressor

from src.config import PipelineConfig, load_config


def train(config: PipelineConfig) -> None:
    """Train XGBoost model and log to MLflow."""
    # Connect to the MLflow tracking backend
    mlflow.set_tracking_uri(config.mlflow.tracking_uri)
    mlflow.set_experiment(config.mlflow.experiment_name)

    # Load the pre-extracted feature dataset
    features_path = "data/features.parquet"
    df = pd.read_parquet(features_path)

    # Separate features (spectral bands + indices) from target (SOC)
    feature_cols = config.features.bands + config.features.indices
    X = df[feature_cols]
    y = df["SOC"]

    # Split into train/test using config-driven parameters
    X_train, X_test, y_train, y_test = train_test_split(
        X, y,
        test_size=config.model.test_size,
        random_state=config.model.random_state,
    )

What does this code do?

mlflow.set_tracking_uri() tells the MLflow client where to store run data. The URI comes from your configs/default.yaml so the backend is centrally managed.

mlflow.set_experiment() creates or selects the named experiment. All runs from this script land in the soc-prediction experiment.

The train/test split ratio and random seed are pulled from the config object, making every run fully reproducible from the YAML alone.

  • Below the train_test_split call, add the MLflow run context that trains the model, evaluates it, and logs everything:
    with mlflow.start_run():
        # Log all hyperparameters from config for reproducibility
        mlflow.log_params({
            "n_estimators": config.model.n_estimators,
            "max_depth": config.model.max_depth,
            "learning_rate": config.model.learning_rate,
            "subsample": config.model.subsample,
            "n_features": len(feature_cols),
            "n_train_samples": len(X_train),
            "date_range": f"{config.data.date_start}__{config.data.date_end}",
        })

        # Train the XGBoost regressor
        model = XGBRegressor(
            n_estimators=config.model.n_estimators,
            max_depth=config.model.max_depth,
            learning_rate=config.model.learning_rate,
            subsample=config.model.subsample,
            random_state=config.model.random_state,
        )
        model.fit(X_train, y_train)

        # Evaluate on test set
        y_pred = model.predict(X_test)
        rmse = float(mean_squared_error(y_test, y_pred, squared=False))
        mae = float(mean_absolute_error(y_test, y_pred))
        r2 = float(r2_score(y_test, y_pred))

        # Log metrics and the trained model
        mlflow.log_metrics({"rmse": rmse, "mae": mae, "r2": r2})
        mlflow.xgboost.log_model(model, artifact_path="xgb_soc_model")

        # Save and log feature importance as a CSV artifact
        importance = model.get_booster().get_score(importance_type="gain")
        importance_df = pd.DataFrame(
            sorted(importance.items(), key=lambda x: x[1], reverse=True),
            columns=["feature", "importance"],
        )
        importance_path = "artifacts/feature_importance.csv"
        import os
        os.makedirs("artifacts", exist_ok=True)
        importance_df.to_csv(importance_path, index=False)
        mlflow.log_artifact(importance_path)

        print(f"Run logged. RMSE={rmse:.2f}, MAE={mae:.2f}, R2={r2:.3f}")


if __name__ == "__main__":
    cfg = load_config()
    train(cfg)

What does this code do?

mlflow.start_run() opens a new experiment run. Everything logged inside this context manager is grouped under one run ID.

log_params() records hyperparameters. log_metrics() records RMSE, MAE, and R2 for comparison. mlflow.xgboost.log_model() serializes the trained model so you can reload it later.

The feature importance CSV shows which spectral bands and indices contribute most to SOC prediction. Logging it as an artifact means it appears in the MLflow UI alongside the run.

  • Save src/train.py.
  • Run the training script with default config by running this command in your original terminal:
python -m src.train

You should see output like Run logged. RMSE=XX.XX, MAE=XX.XX, R2=X.XXX with your dataset's actual values.

  • Switch to your browser and refresh the MLflow UI at http://localhost:5000.

You should see the soc-prediction experiment in the sidebar with one completed run showing logged parameters, metrics, and the feature importance artifact.

Run not appearing in the MLflow UI?

Make sure you ran the training script from the project root directory (where mlflow.db lives). If you ran it from a subdirectory, the SQLite file was created there instead.

Check that the MLflow server's working directory matches where mlflow.db was written. Both the server and the training script must resolve sqlite:///mlflow.db to the same absolute path.

help me debug why my MLflow run is not showing in the UI

✔️ Awesome, I've got everything!

Great. Double check that you have saved src/train.py and that your first run appears in the MLflow UI.

ⓧ I'd like to double check the full code

"""Training script with MLflow experiment tracking."""

import mlflow
import pandas as pd
from sklearn.metrics import mean_absolute_error, mean_squared_error, r2_score
from sklearn.model_selection import train_test_split
from xgboost import XGBRegressor

from src.config import PipelineConfig, load_config


def train(config: PipelineConfig) -> None:
    """Train XGBoost model and log to MLflow."""
    mlflow.set_tracking_uri(config.mlflow.tracking_uri)
    mlflow.set_experiment(config.mlflow.experiment_name)

    features_path = "data/features.parquet"
    df = pd.read_parquet(features_path)

    feature_cols = config.features.bands + config.features.indices
    X = df[feature_cols]
    y = df["SOC"]

    X_train, X_test, y_train, y_test = train_test_split(
        X, y,
        test_size=config.model.test_size,
        random_state=config.model.random_state,
    )

    with mlflow.start_run():
        mlflow.log_params({
            "n_estimators": config.model.n_estimators,
            "max_depth": config.model.max_depth,
            "learning_rate": config.model.learning_rate,
            "subsample": config.model.subsample,
            "n_features": len(feature_cols),
            "n_train_samples": len(X_train),
            "date_range": f"{config.data.date_start}__{config.data.date_end}",
        })

        model = XGBRegressor(
            n_estimators=config.model.n_estimators,
            max_depth=config.model.max_depth,
            learning_rate=config.model.learning_rate,
            subsample=config.model.subsample,
            random_state=config.model.random_state,
        )
        model.fit(X_train, y_train)

        y_pred = model.predict(X_test)
        rmse = float(mean_squared_error(y_test, y_pred, squared=False))
        mae = float(mean_absolute_error(y_test, y_pred))
        r2 = float(r2_score(y_test, y_pred))

        mlflow.log_metrics({"rmse": rmse, "mae": mae, "r2": r2})
        mlflow.xgboost.log_model(model, artifact_path="xgb_soc_model")

        importance = model.get_booster().get_score(importance_type="gain")
        importance_df = pd.DataFrame(
            sorted(importance.items(), key=lambda x: x[1], reverse=True),
            columns=["feature", "importance"],
        )
        importance_path = "artifacts/feature_importance.csv"
        import os
        os.makedirs("artifacts", exist_ok=True)
        importance_df.to_csv(importance_path, index=False)
        mlflow.log_artifact(importance_path)

        print(f"Run logged. RMSE={rmse:.2f}, MAE={mae:.2f}, R2={r2:.3f}")


if __name__ == "__main__":
    cfg = load_config()
    train(cfg)
Compare runs with different hyperparameters

A key benefit of experiment tracking is the ability to compare runs side-by-side. Your CLI config override system makes this easy.

  • Run a second training experiment with more estimators by using the CLI override:
python -m src.train --model.n_estimators=500

What does this command do?

The --model.n_estimators=500 flag overrides the YAML config value at runtime. Your load_config() function from Step 2 parses this dot-notation argument and injects it into the config before training begins.

This means you can sweep hyperparameters without editing files. Each run logs its exact config to MLflow so you always know which parameters produced which results.

You should see a second Run logged. message with different metric values.

  • Switch to the MLflow UI at http://localhost:5000 and refresh the page.
  • Click the soc-prediction experiment in the sidebar.
  • Select both runs using the checkboxes.
  • Click Compare to view them side-by-side.

You should see both runs with their respective n_estimators values (200 vs 500) and different RMSE, MAE, and R2 metrics.

Why track experiments this way?

In production MRV systems, you might test dozens of hyperparameter combinations, date ranges, or feature sets. MLflow gives you a permanent record of what worked and what did not.

Without tracking, you end up with scattered notebooks, lost terminal output, and no way to reproduce a result from three weeks ago.

Only seeing one run in the experiment view?

Make sure both commands finished successfully (each should print a Run logged. line). If the second run errored, check the terminal output for a traceback.

Confirm you ran both commands from the same project root directory. If they wrote to different mlflow.db files, the server only shows runs from one.

help me figure out why MLflow only shows one run

Your training pipeline is now fully instrumented. Next up, you will add automated tests and code quality hooks to make this repository production-ready.

Write Tests and Set Up Repository Hygiene

In the previous step, you trained your XGBoost model and tracked experiments with MLflow. You now have a working end-to-end pipeline.

A working pipeline without tests is a fragile pipeline. One wrong refactor and your spectral index calculations silently break, or your cache returns stale data. In this step, you'll add pytest tests that verify your feature extraction logic and a pre-commit setup that enforces code quality on every commit.

In this step, get ready to:
  • Write pytest tests for spectral index calculations and the cache layer.
  • Write data validation tests that check DataFrame schema and SOC value ranges.
  • Set up pre-commit hooks with ruff and mypy for automated code quality.
Write tests for spectral indices and caching

Your feature extraction module has two critical pieces of logic that need automated verification. The spectral index math (NDVI, EVI, BSI) must be correct, and the disk cache must reliably store and retrieve data.

  • Create a new file called tests/test_features.py in your tests/ directory.
  • Add the imports and the TestSpectralIndices class by pasting the following code:
"""Tests for the GEE feature extraction module."""

import json
import tempfile
from pathlib import Path
from unittest.mock import MagicMock, patch

import pandas as pd
import pytest

from src.gee_features import (
    CACHE_DIR,
    _cache_key,
    _get_cached,
    _save_cache,
    build_feature_dataframe,
)


class TestSpectralIndices:
    def test_ndvi_formula(self) -> None:
        b8 = 3000.0
        b4 = 1000.0
        expected = (b8 - b4) / (b8 + b4)
        assert expected == pytest.approx(0.5)

    def test_ndvi_range(self) -> None:
        ndvi_veg = (4000 - 500) / (4000 + 500)
        assert -1.0 <= ndvi_veg <= 1.0
        ndvi_soil = (2000 - 1800) / (2000 + 1800)
        assert -1.0 <= ndvi_soil <= 1.0

What does this code do?

The imports pull in the cache functions directly from your src.gee_features module so you can test them in isolation without calling GEE.

The TestSpectralIndices class verifies the NDVI formula: (B8 - B4) / (B8 + B4). test_ndvi_formula checks that known reflectance values produce the expected result. test_ndvi_range confirms NDVI always falls between -1 and 1 for both vegetation and bare soil scenarios.

You won't see a change yet. The next chunk adds the cache layer tests that complete this file's first test group.

  • Below the TestSpectralIndices class, add the TestCacheLayer class:
class TestCacheLayer:
    def test_cache_key_deterministic(self) -> None:
        key1 = _cache_key("PT001", "2018-04-01", "2018-09-30")
        key2 = _cache_key("PT001", "2018-04-01", "2018-09-30")
        assert key1 == key2

    def test_cache_key_different_inputs(self) -> None:
        key1 = _cache_key("PT001", "2018-04-01", "2018-09-30")
        key2 = _cache_key("PT002", "2018-04-01", "2018-09-30")
        assert key1 != key2

    def test_save_and_retrieve_cache(self, tmp_path: Path) -> None:
        import src.gee_features as gf
        original_cache_dir = gf.CACHE_DIR
        gf.CACHE_DIR = tmp_path
        features = {"B2": 500.0, "NDVI": 0.45}
        _save_cache("test_key", features)
        result = _get_cached("test_key")
        assert result == features
        gf.CACHE_DIR = original_cache_dir

    def test_cache_miss_returns_none(self, tmp_path: Path) -> None:
        import src.gee_features as gf
        original_cache_dir = gf.CACHE_DIR
        gf.CACHE_DIR = tmp_path
        result = _get_cached("nonexistent_key")
        assert result is None
        gf.CACHE_DIR = original_cache_dir

What does this code do?

  • test_cache_key_deterministic confirms the same inputs always produce the same MD5 hash key.
  • test_cache_key_different_inputs confirms different point IDs produce different keys, preventing cache collisions.
  • test_save_and_retrieve_cache uses pytest's tmp_path fixture to test a full cache round-trip without polluting your real cache directory.
  • test_cache_miss_returns_none verifies that asking for a non-existent key returns None instead of raising an error.

You won't see a change yet. The next chunk adds data validation tests, then you'll run the full suite.

Write data validation tests

Beyond testing logic, production pipelines need schema validation. If the output DataFrame drifts from its expected shape or SOC values escape their valid range (0-600 g/kg), downstream consumers will silently ingest garbage.

  • Below the TestCacheLayer class in tests/test_features.py, add the TestDataValidation class:
class TestDataValidation:
    def test_valid_dataframe_schema(self) -> None:
        df = pd.DataFrame({
            "POINT_ID": ["PT001"], "SOC": [25.0],
            "B2": [500.0], "B3": [600.0], "B4": [700.0], "B5": [1500.0],
            "B6": [2000.0], "B7": [2200.0], "B8": [2500.0], "B8A": [2600.0],
            "B11": [1800.0], "B12": [1200.0],
            "NDVI": [0.56], "EVI": [0.35], "BSI": [-0.1],
        })
        expected_cols = {"POINT_ID", "SOC", "B2", "B3", "B4", "B5", "B6",
                         "B7", "B8", "B8A", "B11", "B12", "NDVI", "EVI", "BSI"}
        assert set(df.columns) == expected_cols

    def test_soc_value_range(self) -> None:
        df = pd.DataFrame({"SOC": [0.5, 25.0, 150.0, 580.0]})
        assert (df["SOC"] >= 0).all()
        assert (df["SOC"] <= 600).all()

    def test_soc_invalid_range_detected(self) -> None:
        df = pd.DataFrame({"SOC": [-5.0, 25.0, 700.0]})
        invalid = (df["SOC"] < 0) | (df["SOC"] > 600)
        assert invalid.any()

What does this code do?

  • test_valid_dataframe_schema constructs a sample DataFrame matching the expected output from your GEE pipeline and asserts it has exactly the right columns.
  • test_soc_value_range checks that valid SOC values (organic carbon in g/kg) fall within the 0-600 range defined by the LUCAS dataset.
  • test_soc_invalid_range_detected intentionally feeds out-of-range values and confirms the check catches them. This proves your validation logic will flag bad data.
  • Save tests/test_features.py.
  • Run the test suite by running this command:
pytest tests/ -v
  • You should see all 9 tests pass with green PASSED markers next to each test name.

Seeing test failures?

  • If you see ModuleNotFoundError: No module named 'src', make sure you installed the project in editable mode earlier with pip install -e ".[dev]". This registers src as an importable package.
  • If an import error mentions ee, your tests import from src.gee_features which imports ee at the top level. Make sure earthengine-api is installed in your virtual environment.
  • Check that your tests/__init__.py file exists. Pytest needs it to resolve the tests/ directory as a package.

Still stuck? Help me debug my pytest test failures for the SOC feature extraction tests.

✔️ Awesome, I've got everything!

Great. Double check you've saved tests/test_features.py before moving on.

ⓧ I'd like to double check the full code

"""Tests for the GEE feature extraction module."""

import json
import tempfile
from pathlib import Path
from unittest.mock import MagicMock, patch

import pandas as pd
import pytest

from src.gee_features import (
    CACHE_DIR,
    _cache_key,
    _get_cached,
    _save_cache,
    build_feature_dataframe,
)


class TestSpectralIndices:
    def test_ndvi_formula(self) -> None:
        b8 = 3000.0
        b4 = 1000.0
        expected = (b8 - b4) / (b8 + b4)
        assert expected == pytest.approx(0.5)

    def test_ndvi_range(self) -> None:
        ndvi_veg = (4000 - 500) / (4000 + 500)
        assert -1.0 <= ndvi_veg <= 1.0
        ndvi_soil = (2000 - 1800) / (2000 + 1800)
        assert -1.0 <= ndvi_soil <= 1.0


class TestCacheLayer:
    def test_cache_key_deterministic(self) -> None:
        key1 = _cache_key("PT001", "2018-04-01", "2018-09-30")
        key2 = _cache_key("PT001", "2018-04-01", "2018-09-30")
        assert key1 == key2

    def test_cache_key_different_inputs(self) -> None:
        key1 = _cache_key("PT001", "2018-04-01", "2018-09-30")
        key2 = _cache_key("PT002", "2018-04-01", "2018-09-30")
        assert key1 != key2

    def test_save_and_retrieve_cache(self, tmp_path: Path) -> None:
        import src.gee_features as gf
        original_cache_dir = gf.CACHE_DIR
        gf.CACHE_DIR = tmp_path
        features = {"B2": 500.0, "NDVI": 0.45}
        _save_cache("test_key", features)
        result = _get_cached("test_key")
        assert result == features
        gf.CACHE_DIR = original_cache_dir

    def test_cache_miss_returns_none(self, tmp_path: Path) -> None:
        import src.gee_features as gf
        original_cache_dir = gf.CACHE_DIR
        gf.CACHE_DIR = tmp_path
        result = _get_cached("nonexistent_key")
        assert result is None
        gf.CACHE_DIR = original_cache_dir


class TestDataValidation:
    def test_valid_dataframe_schema(self) -> None:
        df = pd.DataFrame({
            "POINT_ID": ["PT001"], "SOC": [25.0],
            "B2": [500.0], "B3": [600.0], "B4": [700.0], "B5": [1500.0],
            "B6": [2000.0], "B7": [2200.0], "B8": [2500.0], "B8A": [2600.0],
            "B11": [1800.0], "B12": [1200.0],
            "NDVI": [0.56], "EVI": [0.35], "BSI": [-0.1],
        })
        expected_cols = {"POINT_ID", "SOC", "B2", "B3", "B4", "B5", "B6",
                         "B7", "B8", "B8A", "B11", "B12", "NDVI", "EVI", "BSI"}
        assert set(df.columns) == expected_cols

    def test_soc_value_range(self) -> None:
        df = pd.DataFrame({"SOC": [0.5, 25.0, 150.0, 580.0]})
        assert (df["SOC"] >= 0).all()
        assert (df["SOC"] <= 600).all()

    def test_soc_invalid_range_detected(self) -> None:
        df = pd.DataFrame({"SOC": [-5.0, 25.0, 700.0]})
        invalid = (df["SOC"] < 0) | (df["SOC"] > 600)
        assert invalid.any()
Set up pre-commit hooks with ruff and mypy

Tests catch logic errors, but they don't enforce consistent formatting or catch type errors before runtime. Pre-commit hooks run automatically on every commit, blocking code that fails linting or type checking from entering your repository.

  • Create a new file called .pre-commit-config.yaml in the root of your soc-mrv-pipeline project folder.
  • Paste the following configuration:
repos:
  - repo: https://github.com/astral-sh/ruff-pre-commit
    rev: v0.11.0
    hooks:
      - id: ruff
        args: [--fix]
      - id: ruff-format

  - repo: https://github.com/pre-commit/mirrors-mypy
    rev: v1.15.0
    hooks:
      - id: mypy
        additional_dependencies:
          - pydantic>=2.13.0
          - pandas-stubs>=2.0.0
        args: [--ignore-missing-imports]

What does this config do?

  • The ruff hook runs the linter with auto-fix enabled (--fix). It catches unused imports, bad formatting, and common Python anti-patterns.
  • The ruff-format hook enforces consistent code formatting across the project.
  • The mypy hook performs static type checking. The additional_dependencies list gives mypy access to Pydantic and pandas type stubs so it can validate your type annotations properly.
  • Save .pre-commit-config.yaml.
  • Install the pre-commit hooks into your Git repository by running:
pre-commit install
  • You should see: pre-commit installed at .git/hooks/pre-commit.
  • Run all hooks against the entire codebase by running:
pre-commit run --all-files
  • You should see each hook (ruff, ruff-format, mypy) report Passed. The first run may take a minute while pre-commit installs hook environments.

If ruff or mypy reports issues

Ruff with --fix auto-corrects most lint issues (import sorting, unused imports). If it makes changes, just run pre-commit run --all-files again. The second run should pass cleanly.

For mypy errors, check that your function signatures in src/ have return type annotations. The --ignore-missing-imports flag handles third-party libraries that lack type stubs.

Command not found for pre-commit?

Make sure your virtual environment is activated. The pre-commit binary was installed as part of your [dev] dependencies in pyproject.toml. Re-run pip install -e ".[dev]" if needed.

Still stuck? Help me debug pre-commit installation issues in my Python virtual environment.

Secret mission

GitHub Actions CI Pipeline

Your pipeline runs perfectly on your machine. But what happens when a teammate pushes code that breaks the linter or fails a test? Wire up GitHub Actions to automatically lint, type-check, test, and smoke-test your pipeline on every push and pull request.

Clean Up Your Resources

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely. This project runs on GCP free tier with local computation, so there are no ongoing costs while the project sits idle.

Resources you used:

  • GCP project with Earth Engine API enabled.
  • Local project directory soc-mrv-pipeline on your Desktop (source code, virtual environment, data files, cache, MLflow database).
  • Earth Engine credentials stored at ~/.config/earthengine/credentials.
  • GitHub repository with the pushed project code and CI workflow.

Keep everything running

No action needed. Choose this if you plan to continue building on this pipeline or want to revisit it soon.

  • Your GCP project stays on free tier. The Earth Engine API does not incur charges for personal, non-commercial use.
  • Your local files, virtual environment, and MLflow database remain on disk.
  • Your GitHub repository stays available for future pushes and CI runs.

Pause - I'll come back to this later

Shut down running processes to free up memory. Your files and cloud project stay intact for when you return.

  • If the MLflow server is still running, stop it by pressing Ctrl+C in the terminal where you started it.
  • Deactivate your virtual environment by running:
deactivate
  • Your GCP project, GitHub repository, and local files remain untouched and ready for your next session.

Delete - I don't want to use this again

Remove all project resources and start fresh. Follow each step below to fully clean up.

Delete your GCP project:

  • Open the Google Cloud Console and go to the Manage resources page.
  • Select the project you created for this pipeline.
  • Click Delete, enter the project ID, and click Shut down.

Remove your Earth Engine credentials:

  • Delete the stored credentials by running:
rm -rf ~/.config/earthengine

Delete the local project directory:

  • Remove the entire project folder by running:
cd ~/Desktop
rm -rf soc-mrv-pipeline

Delete your GitHub repository:

  • Go to your repository on GitHub.
  • Click Settings at the top of the repository page.
  • Scroll to the bottom to the Danger Zone section.
  • Click Delete this repository and follow the confirmation prompts.

Nice Work!

Nice Work!

You just built a production-grade MRV pipeline that predicts soil organic carbon from Sentinel-2 satellite imagery. This is the same engineering discipline that climate-tech companies use to monitor carbon sequestration at scale.

You've learned how to:

  • Build a typed, config-driven ML pipeline using Pydantic and YAML that validates parameters at load time and accepts CLI overrides.
  • Implement a resilient GEE feature extraction layer with disk caching and exponential backoff retry that handles quota and timeout failures gracefully.
  • Track and compare ML experiments with MLflow, logging parameters, metrics, and feature importance artifacts across multiple training runs.
  • Secret Mission: Deploy a GitHub Actions CI pipeline that lints, type-checks, tests, and smoke-tests the pipeline on every push.

Ready to quiz yourself?