Preserving Metadata Across Dataset Versions

pip install "rasterio>=1.3.9" "geopandas>=0.13.2" "pyproj>=3.6.0" "shapely>=2.0.3" "pyyaml>=6.0.1"

project/
├── data/raw/          # original GeoTIFFs, COGs, GeoJSONs
├── data/processed/    # cropped / augmented derivatives
├── metadata/          # YAML sidecars + manifest.json
└── annotations/       # label files with provenance fields

import hashlib
from pathlib import Path
from typing import Any

import rasterio
import geopandas as gpd
import pyproj
import yaml


def normalize_crs(crs_obj: Any) -> dict[str, Any]:
    """Normalize any CRS representation to EPSG code + WKT2 string."""
    try:
        crs = pyproj.CRS.from_user_input(crs_obj)
        return {
            "epsg": crs.to_epsg(),
            "wkt2": crs.to_wkt(version="WKT2_2019"),
            "is_geographic": crs.is_geographic,
            "axis_order": [str(a.direction) for a in crs.axis_info],
        }
    except Exception as exc:
        return {"error": str(exc), "raw": str(crs_obj)}


def extract_raster_metadata(path: Path) -> dict[str, Any]:
    with rasterio.open(path) as src:
        return {
            "driver": src.driver,
            "width": src.width,
            "height": src.height,
            "band_count": src.count,
            "dtype": str(src.dtypes[0]),
            "nodata": src.nodata,
            "crs": normalize_crs(src.crs),
            # Serialize as flat list: [a, b, c, d, e, f] matches GDAL geotransform order
            "affine_transform": list(src.transform)[:6],
            "bounds": dict(src.bounds._asdict()),
            "resolution_m": src.res,
            "compression": src.compression.value if src.compression else None,
        }


def extract_vector_metadata(path: Path) -> dict[str, Any]:
    gdf = gpd.read_file(path)
    geom_types = gdf.geometry.geom_type.value_counts().to_dict() if not gdf.empty else {}
    return {
        "format": path.suffix.lstrip(".").upper(),
        "geometry_types": geom_types,
        "feature_count": len(gdf),
        "crs": normalize_crs(gdf.crs),
        "bounds": gdf.total_bounds.tolist(),  # [minx, miny, maxx, maxy]
        "attribute_schema": {col: str(dtype) for col, dtype in gdf.dtypes.items()},
    }

import rasterio
from rasterio.windows import Window
from rasterio.transform import from_bounds
from pathlib import Path
from typing import Any


def crop_raster_with_metadata(
    src_path: Path,
    dst_path: Path,
    row_off: int,
    col_off: int,
    height: int,
    width: int,
) -> dict[str, Any]:
    """Crop a GeoTIFF tile and return updated metadata dict."""
    with rasterio.open(src_path) as src:
        window = Window(col_off=col_off, row_off=row_off, width=width, height=height)
        transform = src.window_transform(window)
        profile = src.profile.copy()
        profile.update(
            height=height,
            width=width,
            transform=transform,
            # Preserve driver, CRS, dtype, nodata — never drop them
        )
        data = src.read(window=window)

    with rasterio.open(dst_path, "w", **profile) as dst:
        dst.write(data)

    with rasterio.open(dst_path) as verification:
        assert verification.crs == rasterio.open(src_path).crs, (
            "CRS must survive tile crop — check driver write flags"
        )
        return extract_raster_metadata(dst_path)

def write_metadata_sidecar(
    data_path: Path,
    metadata: dict[str, Any],
    sidecar_dir: Path,
    version_tag: str = "v1",
) -> Path:
    """Write a versioned YAML sidecar next to the dataset file."""
    sidecar_dir.mkdir(parents=True, exist_ok=True)
    sidecar_name = f"{data_path.stem}_{version_tag}_metadata.yaml"
    sidecar_path = sidecar_dir / sidecar_name

    payload = {
        "schema_version": "1.0",
        "version_tag": version_tag,
        "source_path": str(data_path),
        "file_size_bytes": data_path.stat().st_size,
        **metadata,
    }

    with open(sidecar_path, "w", encoding="utf-8") as fh:
        yaml.dump(payload, fh, default_flow_style=False, sort_keys=False, allow_unicode=True)

    return sidecar_path

import json
from datetime import datetime, timezone


def compute_sha256(file_path: Path) -> str:
    sha256 = hashlib.sha256()
    with open(file_path, "rb") as fh:
        for chunk in iter(lambda: fh.read(65_536), b""):
            sha256.update(chunk)
    return sha256.hexdigest()


def update_manifest(
    manifest_path: Path,
    data_path: Path,
    sidecar_path: Path,
) -> dict[str, Any]:
    """Append or update a manifest entry for one data/sidecar pair."""
    manifest: dict[str, Any] = {}
    if manifest_path.exists():
        with open(manifest_path, encoding="utf-8") as fh:
            manifest = json.load(fh)

    entry_key = str(data_path.name)
    manifest[entry_key] = {
        "data_path": str(data_path),
        "sidecar_path": str(sidecar_path),
        "data_sha256": compute_sha256(data_path),
        "sidecar_sha256": compute_sha256(sidecar_path),
        "recorded_at": datetime.now(tz=timezone.utc).isoformat(),
    }

    with open(manifest_path, "w", encoding="utf-8") as fh:
        json.dump(manifest, fh, indent=2)

    return manifest[entry_key]

def validate_sidecar_consistency(
    data_path: Path,
    sidecar_path: Path,
    manifest_path: Path,
) -> bool:
    """Assert that data file and sidecar still match the recorded manifest hashes."""
    with open(manifest_path, encoding="utf-8") as fh:
        manifest: dict[str, Any] = json.load(fh)

    entry = manifest.get(data_path.name)
    if entry is None:
        raise KeyError(f"{data_path.name} not found in manifest — run update_manifest first")

    actual_data_hash = compute_sha256(data_path)
    actual_sidecar_hash = compute_sha256(sidecar_path)

    if actual_data_hash != entry["data_sha256"]:
        raise ValueError(
            f"Data file hash mismatch for {data_path.name}:\n"
            f"  expected: {entry['data_sha256']}\n"
            f"  actual:   {actual_data_hash}"
        )
    if actual_sidecar_hash != entry["sidecar_sha256"]:
        raise ValueError(
            f"Sidecar hash mismatch for {sidecar_path.name}:\n"
            f"  expected: {entry['sidecar_sha256']}\n"
            f"  actual:   {actual_sidecar_hash}"
        )

    return True

# dvc.yaml
stages:
  extract_metadata:
    cmd: python pipeline/extract_metadata.py
    deps:
      - data/raw/
      - pipeline/extract_metadata.py
    outs:
      - metadata/
      - metadata/manifest.json
    params:
      - params.yaml:
          - metadata.version_tag
          - metadata.schema_version

# .github/workflows/metadata_check.yml
name: Metadata Consistency Check
on: [pull_request]
jobs:
  validate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.11"
      - run: pip install rasterio>=1.3.9 geopandas>=0.13.2 pyproj>=3.6.0 pyyaml>=6.0.1
      - name: Validate manifest consistency
        run: |
          python - <<'EOF'
          from pathlib import Path
          from pipeline.metadata import validate_sidecar_consistency
          import json, sys

          manifest_path = Path("metadata/manifest.json")
          with open(manifest_path) as fh:
              manifest = json.load(fh)

          errors = []
          for name, entry in manifest.items():
              try:
                  validate_sidecar_consistency(
                      Path(entry["data_path"]),
                      Path(entry["sidecar_path"]),
                      manifest_path,
                  )
              except ValueError as exc:
                  errors.append(str(exc))

          if errors:
              print("\n".join(errors), file=sys.stderr)
              sys.exit(1)
          print(f"All {len(manifest)} manifest entries validated successfully.")
          EOF

import json
from pathlib import Path


def inject_provenance(
    label_studio_export: Path,
    manifest_path: Path,
    output_path: Path,
) -> None:
    with open(label_studio_export, encoding="utf-8") as fh:
        tasks: list[dict] = json.load(fh)
    with open(manifest_path, encoding="utf-8") as fh:
        manifest: dict = json.load(fh)

    for task in tasks:
        source_file = Path(task.get("data", {}).get("image", "")).name
        if source_file in manifest:
            task.setdefault("meta", {})["dataset_version"] = manifest[source_file]

    with open(output_path, "w", encoding="utf-8") as fh:
        json.dump(tasks, fh, indent=2)

import pytest
import rasterio
import pyproj
from pathlib import Path
import json


def test_crs_survives_crop(tmp_path: Path, sample_geotiff: Path) -> None:
    """CRS must be identical before and after a tile crop operation."""
    with rasterio.open(sample_geotiff) as src:
        original_epsg = pyproj.CRS(src.crs).to_epsg()

    cropped = tmp_path / "tile.tif"
    crop_raster_with_metadata(sample_geotiff, cropped, 0, 0, 256, 256)

    with rasterio.open(cropped) as dst:
        cropped_epsg = pyproj.CRS(dst.crs).to_epsg()

    assert original_epsg == cropped_epsg, (
        f"CRS changed after crop: {original_epsg} → {cropped_epsg}"
    )


def test_manifest_hash_matches_file(tmp_path: Path, sample_geotiff: Path) -> None:
    """Manifest SHA-256 must match recomputed hash of the file on disk."""
    sidecar_dir = tmp_path / "metadata"
    manifest_path = sidecar_dir / "manifest.json"
    meta = extract_raster_metadata(sample_geotiff)
    sidecar = write_metadata_sidecar(sample_geotiff, meta, sidecar_dir)
    update_manifest(manifest_path, sample_geotiff, sidecar)

    assert validate_sidecar_consistency(sample_geotiff, sidecar, manifest_path)


def test_sidecar_is_valid_yaml(tmp_path: Path, sample_geotiff: Path) -> None:
    """Sidecar file must be parseable YAML with required fields present."""
    import yaml

    sidecar_dir = tmp_path / "metadata"
    meta = extract_raster_metadata(sample_geotiff)
    sidecar = write_metadata_sidecar(sample_geotiff, meta, sidecar_dir)

    with open(sidecar, encoding="utf-8") as fh:
        loaded = yaml.safe_load(fh)

    required_fields = {"schema_version", "version_tag", "crs", "affine_transform", "bounds"}
    missing = required_fields - set(loaded.keys())
    assert not missing, f"Sidecar missing required fields: {missing}"


def test_crs_normalization_roundtrip() -> None:
    """normalize_crs must produce a valid EPSG code for standard projections."""
    import pyproj

    for epsg in [4326, 32633, 3857, 27700]:
        crs = pyproj.CRS.from_epsg(epsg)
        result = normalize_crs(crs)
        assert result["epsg"] == epsg, f"EPSG roundtrip failed for {epsg}: got {result['epsg']}"
        assert "error" not in result