Implementing DVC for Geospatial Training Data

git init
dvc init

# Add a default remote — S3 shown; substitute gcs:// or azure:// as needed
dvc remote add -d geospatial-remote s3://your-bucket-name/dvc-cache

# Tune for large spatial files
dvc remote modify geospatial-remote credentialpath ~/.aws/credentials
dvc remote modify geospatial-remote upload_concurrency 10
dvc remote modify geospatial-remote download_concurrency 10
dvc remote modify geospatial-remote checksum xxhash

git add .dvc/config
git commit -m "chore: configure DVC remote for geospatial workloads"

data/
├── raw/
│   ├── imagery/          # Source GeoTIFF, NetCDF, or HDF5 scenes
│   └── vectors/          # Reference boundaries, AOIs, road networks
├── processed/
│   ├── normalized/       # CRS-aligned rasters (common grid, unified datum)
│   ├── chips/            # 512×512 or 1024×1024 training patches
│   └── masks/            # Segmentation labels, instance masks
└── annotations/          # COCO, GeoJSON, or custom XML

dvc add data/raw/imagery
dvc add data/processed/chips
dvc add data/annotations
git add data/raw/imagery.dvc data/processed/chips.dvc data/annotations.dvc data/.gitignore
git commit -m "feat: initialize geospatial dataset tracking"

# dvc.yaml
stages:
  normalize_crs:
    cmd: python scripts/normalize_crs.py --input data/raw/imagery --output data/processed/normalized
    deps:
      - data/raw/imagery
      - scripts/normalize_crs.py
    params:
      - params.yaml:
        - crs_target
        - resample_method
    outs:
      - data/processed/normalized

  generate_chips:
    cmd: >-
      python scripts/chip_generator.py
        --input data/processed/normalized
        --output data/processed/chips
        --size ${tile_size}
        --overlap ${overlap}
    deps:
      - data/processed/normalized
      - scripts/chip_generator.py
    params:
      - params.yaml:
        - tile_size
        - overlap
    outs:
      - data/processed/chips

  export_labels:
    cmd: python scripts/label_exporter.py --annotations data/annotations --masks data/processed/masks
    deps:
      - data/annotations
      - scripts/label_exporter.py
    outs:
      - data/processed/masks

# params.yaml
crs_target: "EPSG:32637"   # UTM zone 37N for East Africa example
resample_method: bilinear
tile_size: 512
overlap: 64

# Generate manifest before committing annotation updates
find data/annotations -type f -name "*.geojson" \
  | sort \
  | xargs sha256sum > data/annotations.sha256

dvc add data/annotations
git add data/annotations.dvc data/annotations.sha256
git commit -m "feat: update building footprint annotations — 143 polygons revised"

# scripts/validate_annotations.py
import json
import pathlib
from typing import Iterator

PRECISION_THRESHOLD = 6  # decimal degrees — below 6 introduces >10 cm error at equator

def iter_coordinates(geom: dict) -> Iterator[tuple[float, float]]:
    """Recursively yield (lon, lat) pairs from any GeoJSON geometry."""
    match geom["type"]:
        case "Point":
            yield tuple(geom["coordinates"])
        case "MultiPoint" | "LineString":
            yield from (tuple(c) for c in geom["coordinates"])
        case "Polygon" | "MultiLineString":
            for ring in geom["coordinates"]:
                yield from (tuple(c) for c in ring)
        case "MultiPolygon":
            for polygon in geom["coordinates"]:
                for ring in polygon:
                    yield from (tuple(c) for c in ring)

def check_precision(annotation_dir: pathlib.Path) -> list[str]:
    issues: list[str] = []
    for path in sorted(annotation_dir.glob("**/*.geojson")):
        fc = json.loads(path.read_text())
        for feature in fc.get("features", []):
            for lon, lat in iter_coordinates(feature["geometry"]):
                lon_dec = len(str(lon).split(".")[-1]) if "." in str(lon) else 0
                lat_dec = len(str(lat).split(".")[-1]) if "." in str(lat) else 0
                if lon_dec < PRECISION_THRESHOLD or lat_dec < PRECISION_THRESHOLD:
                    issues.append(f"{path.name}: coordinate ({lon}, {lat}) below {PRECISION_THRESHOLD} decimal places")
    return issues

if __name__ == "__main__":
    found = check_precision(pathlib.Path("data/annotations"))
    if found:
        for msg in found:
            print(f"[PRECISION WARNING] {msg}")
        raise SystemExit(1)
    print("Annotation precision check passed.")

# Inspect the history of a specific tracked directory
git log --oneline -- data/processed/chips.dvc

# Example output:
# a3f91c2 feat: regenerate chips at 1024px after AOI expansion
# 7b82e10 feat: initial chip generation at 512px
# 4cd3a01 feat: initialize geospatial dataset tracking

# Roll back to the 512px chip set
git checkout 7b82e10 -- data/processed/chips.dvc
dvc checkout data/processed/chips.dvc

# Pull cache metadata without downloading data files
dvc fetch --jobs 4

# Compare local workspace against tracked state
dvc status

# If mismatches exist, pull the correct files
dvc pull data/processed/chips.dvc

# Remove corrupted outputs and force full re-execution
dvc repro --force

# Build a temporal stack VRT from COGs already in S3
gdalbuildvrt \
  -separate \
  -input_file_list scene_list.txt \
  data/processed/temporal_stack.vrt

dvc add data/processed/temporal_stack.vrt
git add data/processed/temporal_stack.vrt.dvc
git commit -m "feat: add multi-temporal VRT index for 2024 Q1 acquisition series"

import rasterio
from rasterio.windows import from_bounds
from pyproj import Transformer

def read_chip_from_vrt(
    vrt_path: str,
    lon_min: float,
    lat_min: float,
    lon_max: float,
    lat_max: float,
    target_crs: str = "EPSG:32637",
) -> tuple:
    """Read a spatial chip from a VRT without loading the full scene."""
    with rasterio.open(vrt_path) as src:
        transformer = Transformer.from_crs("EPSG:4326", src.crs.to_epsg(), always_xy=True)
        x_min, y_min = transformer.transform(lon_min, lat_min)
        x_max, y_max = transformer.transform(lon_max, lat_max)
        window = from_bounds(x_min, y_min, x_max, y_max, src.transform)
        data = src.read(window=window)
        transform = src.window_transform(window)
    return data, transform

# .github/workflows/spatial-data-ci.yml
name: Spatial Data Validation

on:
  push:
    paths:
      - 'data/**/*.dvc'
      - 'params.yaml'
      - 'dvc.yaml'

jobs:
  validate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.10"

      - name: Install dependencies
        run: |
          pip install "dvc[s3]==3.50.1" rasterio==1.3.9 geopandas==0.14.4 pyproj==3.6.1

      - name: Configure DVC remote credentials
        env:
          AWS_ACCESS_KEY_ID: $
          AWS_SECRET_ACCESS_KEY: $
        run: dvc remote modify geospatial-remote profile ci

      - name: Pull changed datasets
        run: dvc pull --jobs 4

      - name: Validate annotation precision
        run: python scripts/validate_annotations.py

      - name: Reproduce pipeline (changed stages only)
        run: dvc repro --dry

# scripts/ls_export_hook.py
import subprocess
import datetime
import pathlib

def commit_annotation_export(export_path: str, task_name: str) -> None:
    """Add exported annotations to DVC and commit the metadata update."""
    export_dir = pathlib.Path(export_path)
    subprocess.run(["dvc", "add", str(export_dir)], check=True)
    subprocess.run(["git", "add", f"{export_dir}.dvc"], check=True)
    date_str = datetime.date.today().isoformat()
    subprocess.run([
        "git", "commit", "-m",
        f"feat: label-studio export — {task_name} — {date_str}"
    ], check=True)

# 1. Verify the remote is reachable and credentials are valid
dvc remote list
dvc push --dry-run data/raw/imagery.dvc

# 2. Confirm pipeline DAG is acyclic and all deps resolve
dvc dag

# 3. Dry-run the full pipeline to preview which stages will execute
dvc repro --dry

# 4. After reproducing, verify outputs match expected checksums
dvc status

import rasterio
import pyproj

def assert_raster_contract(
    path: str,
    expected_crs: str,
    expected_bands: int,
    expected_width: int,
    expected_height: int,
) -> None:
    with rasterio.open(path) as src:
        actual_crs = pyproj.CRS.from_user_input(src.crs)
        target_crs = pyproj.CRS.from_epsg(int(expected_crs.split(":")[1]))
        assert actual_crs.equals(target_crs), f"CRS mismatch: {src.crs} != {expected_crs}"
        assert src.count == expected_bands, f"Band count: {src.count} != {expected_bands}"
        assert src.width == expected_width, f"Width: {src.width} != {expected_width}"
        assert src.height == expected_height, f"Height: {src.height} != {expected_height}"
    print(f"Contract passed: {path}")