Skip to content

Refactor of the local data inference pipeline #82

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
f-PLT wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

Refactor of the local data inference pipeline #82

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 10 additions & 10 deletions docs/local_data_processing.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,15 +13,15 @@ Every operator script lives in
`scripts/developer_scripts/bow_valley_inference_local/`; each is a thin Typer CLI
over package code, run with `uv run python …` (the viewer with `uv run solara run …`). Run the stages top to bottom:

| # | Stage | Script | Output |
| --- | ----------------------------- | ------------------------------------------------------------------------- | ---------------------------------------------------- |
| 0 | Grid + reference patches (§2) | `python -m src.data.local_sources.grid --emit-csv` | `configs/bow_valley/cube_cells.csv`, parity fixtures |
| 1 | Process raw → read roots (§3) | `process_raw_dataset.py process-all` | clipped archive + `sentinel1_snap/` cache |
| 1a | (S1 only, standalone) | `process_raw_dataset.py process-s1` **or** `build_bow_valley_s1_cache.py` | `sentinel1_snap/s1_grd_<granule>.tif` |
| 1b | Audit stage 1 | `process_raw_audit.py` | exit 0 = clean |
| 2 | Assemble 308-band cubes (§5) | `export_bow_valley_cube.py` | `processing_root/cubes/PR_*.tif` |
| 3 | Daily FSC inference (§6) | `infer_bow_valley_daily_fsc.py` | `processing_root/daily_fsc/*.tif` |
| 4 | Inspect / QA (§7) | `solara run data_viewer.py` | Clip / Cube / Daily-FSC tabs |
| # | Stage | Script | Output |
| --- | ------------------------------------------------------ | ------------------------------------------------------------------------- | ---------------------------------------------------- |
| 0 | Grid + reference patches (§2) - DEPRECATED AND REMOVED | `python -m src.data.local_sources.grid --emit-csv` | `configs/bow_valley/cube_cells.csv`, parity fixtures |
| 1 | Process raw → read roots (§3) | `process_raw_dataset.py process-all` | clipped archive + `sentinel1_snap/` cache |
| 1a | (S1 only, standalone) | `process_raw_dataset.py process-s1` **or** `build_bow_valley_s1_cache.py` | `sentinel1_snap/s1_grd_<granule>.tif` |
| 1b | Audit stage 1 | `process_raw_audit.py` | exit 0 = clean |
| 2 | Assemble 308-band cubes (§5) | `export_bow_valley_cube.py` | `processing_root/cubes/PR_*.tif` |
| 3 | Daily FSC inference (§6) | `infer_bow_valley_daily_fsc.py` | `processing_root/daily_fsc/*.tif` |
| 4 | Inspect / QA (§7) | `solara run data_viewer.py` | Clip / Cube / Daily-FSC tabs |

**Key ordering rule (stage 1):** Sentinel-1 is **processed, never clipped** — it
must go through ESA SNAP *before* anything reads it. `process-all` enforces this:
Expand Down Expand Up @@ -75,7 +75,7 @@ Produces the sweep enumeration and parity fixtures every later task consumes.

```bash
# Emit the generated cube CSV (cells × inference-window days, full cross-product)
uv run python -m src.data.local_sources.grid --emit-csv --mode A \
uv run python scripts/developer_scripts/bow_valley_inference_local/create_grid_csv.py \
--window-start 2025-04-06 --window-end 2025-05-28

uv run pytest tests/test_local_sources/test_grid.py tests/test_local_sources/test_cube_csv.py -q
Expand Down
99 changes: 3 additions & 96 deletions src/snow_galileo/data/local_sources/grid.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,11 +30,10 @@
from dataclasses import dataclass
from datetime import date, timedelta
from pathlib import Path
from typing import Annotated, Literal
from typing import Literal

import pandas as pd
import structlog
import typer
from pyproj import Transformer
from shapely.geometry import Point, Polygon, box
from shapely.ops import transform as shapely_transform
Expand Down Expand Up @@ -80,7 +79,6 @@
#: grid generator and the clip stage share one source of truth — see
#: data/BOW_VALLEY_DATA_LAYOUT.md.
DEFAULT_AOI_PATH: Path = LocalPaths().aoi_path
DEFAULT_OUTPUT_CSV: Path = Path("configs/bow_valley/cube_cells.csv")
DEFAULT_MANIFEST_PATH: Path = Path("configs/bow_valley/cell_filter_manifest.csv")

KeepRule = Literal["centre_in", "fully_inside"]
Expand Down Expand Up @@ -416,12 +414,12 @@ def _window_days(window_start: date, window_end: date) -> list[date]:
return [window_start + timedelta(days=offset) for offset in range(span + 1)]


def build_cube_csv(
def build_cube_dataframe(
kept: list[CellGeometry],
window_start: date = DEFAULT_WINDOW_START,
window_end: date = DEFAULT_WINDOW_END,
) -> pd.DataFrame:
"""Build the generated cube CSV: full cross-product of cells × window days.
"""Build the generated cube datafrane: full cross-product of cells × window days.

Each row's ``date`` is a window-end day (``YYYYMMDD``); the GEE/export side
derives ``window_start = date - (NUM_TIMESTEPS - 1)``. Cell geometry is
Expand Down Expand Up @@ -459,94 +457,3 @@ def build_cube_csv(
rows=len(frame),
)
return frame


def generate(
legacy_csv: Path = DEFAULT_LEGACY_CSV,
aoi_path: Path = DEFAULT_AOI_PATH,
output_csv: Path = DEFAULT_OUTPUT_CSV,
manifest_path: Path = DEFAULT_MANIFEST_PATH,
keep_rule: KeepRule = "centre_in",
window_start: date = DEFAULT_WINDOW_START,
window_end: date = DEFAULT_WINDOW_END,
) -> pd.DataFrame:
"""Run the full geometry pipeline and write the cube CSV + manifest.

Args:
legacy_csv: Legacy cell-sampling CSV (geometry only).
aoi_path: AOI GeoJSON (authoritative clip/inference boundary).
output_csv: Destination for the generated cube CSV.
manifest_path: Destination for the kept/dropped cell manifest.
keep_rule: AOI containment rule (see :func:`filter_cells`).
window_start: First inference day (inclusive).
window_end: Last inference day (inclusive).

Returns:
The generated cube CSV DataFrame (also written to ``output_csv``).
"""
aoi = load_aoi_polygon(aoi_path)
cells = load_cells(legacy_csv)
kept, dropped = filter_cells(cells, aoi, keep_rule=keep_rule)

manifest = build_manifest(kept, dropped)
manifest_path.parent.mkdir(parents=True, exist_ok=True)
manifest.to_csv(manifest_path, index=False)

cube_csv = build_cube_csv(kept, window_start=window_start, window_end=window_end)
output_csv.parent.mkdir(parents=True, exist_ok=True)
cube_csv.to_csv(output_csv, index=False)

logger.info(
"generate_complete",
kept=len(kept),
dropped=len(dropped),
cube_rows=len(cube_csv),
output_csv=str(output_csv),
manifest=str(manifest_path),
)
return cube_csv


def emit_csv(
legacy_csv: Annotated[
Path, typer.Option(help="Legacy cell-sampling CSV.")
] = DEFAULT_LEGACY_CSV,
aoi_path: Annotated[Path, typer.Option("--aoi", help="AOI GeoJSON.")] = DEFAULT_AOI_PATH,
output_csv: Annotated[
Path, typer.Option(help="Generated cube CSV output.")
] = DEFAULT_OUTPUT_CSV,
manifest_path: Annotated[
Path, typer.Option(help="Kept/dropped manifest output.")
] = DEFAULT_MANIFEST_PATH,
require_fully_inside: Annotated[
bool,
typer.Option("--require-fully-inside", help="Keep only fully-contained cells (→ 338)."),
] = False,
window_start: Annotated[
str, typer.Option(help="First inference day, YYYY-MM-DD.")
] = DEFAULT_WINDOW_START.isoformat(),
window_end: Annotated[
str, typer.Option(help="Last inference day, YYYY-MM-DD.")
] = DEFAULT_WINDOW_END.isoformat(),
) -> None:
"""Emit the generated cube CSV and the kept/dropped cell manifest."""
keep_rule: KeepRule = "fully_inside" if require_fully_inside else "centre_in"
cube_csv = generate(
legacy_csv=legacy_csv,
aoi_path=aoi_path,
output_csv=output_csv,
manifest_path=manifest_path,
keep_rule=keep_rule,
window_start=date.fromisoformat(window_start),
window_end=date.fromisoformat(window_end),
)
typer.echo(f"Wrote {len(cube_csv)} rows to {output_csv}")


def main() -> None:
"""CLI entry point (single command — emits the cube CSV + manifest)."""
typer.run(emit_csv)


if __name__ == "__main__":
main()
36 changes: 7 additions & 29 deletions tests/test_local_sources/test_cube_csv.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,15 +10,13 @@
from datetime import date
from pathlib import Path

import pandas as pd
import pytest

from snow_galileo.data.local_sources.grid import (
CUBE_CSV_COLUMNS,
GRID_MATH_CRS,
build_cube_csv,
build_cube_dataframe,
filter_cells,
generate,
load_aoi_polygon,
load_cells,
)
Expand Down Expand Up @@ -53,25 +51,25 @@ def kept_cells():

def test_schema_is_canonical(kept_cells):
"""CSV has exactly the canonical 8 columns in order (SPEC AC-11b)."""
df = build_cube_csv(kept_cells)
df = build_cube_dataframe(kept_cells)
assert list(df.columns) == CUBE_CSV_COLUMNS


def test_row_count_is_full_cross_product(kept_cells):
"""Row count == kept cells × window days (SPEC AC-11b)."""
df = build_cube_csv(kept_cells)
df = build_cube_dataframe(kept_cells)
assert len(df) == EXPECTED_CENTRE_IN * DEFAULT_WINDOW_DAYS == 18232


def test_crs_column_is_utm11n(kept_cells):
"""Every row carries crs == EPSG:32611 (SPEC AC-11b)."""
df = build_cube_csv(kept_cells)
df = build_cube_dataframe(kept_cells)
assert (df["crs"] == GRID_MATH_CRS).all()


def test_dates_span_full_window(kept_cells):
"""`date` covers every day in the window, each appearing for every cell."""
df = build_cube_csv(kept_cells)
df = build_cube_dataframe(kept_cells)
unique_dates = sorted(df["date"].unique())
assert len(unique_dates) == DEFAULT_WINDOW_DAYS
assert unique_dates[0] == int(date(2025, 4, 6).strftime("%Y%m%d"))
Expand All @@ -83,7 +81,7 @@ def test_dates_span_full_window(kept_cells):

def test_gee_exporter_column_contract(kept_cells):
"""The CSV satisfies export_from_csv_utm's column reads (SPEC AC-11b)."""
df = build_cube_csv(kept_cells)
df = build_cube_dataframe(kept_cells)
assert GEE_REQUIRED_COLUMNS.issubset(set(df.columns))
# date parses as YYYYMMDD (the exporter does strptime(str(date), "%Y%m%d")).
from datetime import datetime
Expand All @@ -98,28 +96,8 @@ def test_filename_matches_gee_pattern(kept_cells):
Reproduces the eo_eval.py:599 filename build and asserts it parses through
the ``PR`` branch of ``LandsatEvalDataset`` (month at parts[1][4:6]).
"""
df = build_cube_csv(kept_cells).iloc[0]
df = build_cube_dataframe(kept_cells).iloc[0]
filename = f"PR_{df['date']}_{df['center_x']:.16f}_{df['center_y']:.16f}.tif"
parts = filename.split("_")
assert parts[0] == "PR"
assert parts[1][4:6] == "04" # month of 2025-04-06 window start day


def test_generate_writes_files(kept_cells, tmp_path):
"""`generate` writes both the cube CSV and the manifest, round-trippable."""
out_csv = tmp_path / "cube_cells.csv"
manifest = tmp_path / "manifest.csv"
df = generate(
legacy_csv=LEGACY_CSV,
aoi_path=AOI_PATH,
output_csv=out_csv,
manifest_path=manifest,
)
assert out_csv.exists()
assert manifest.exists()
reread = pd.read_csv(out_csv)
assert list(reread.columns) == CUBE_CSV_COLUMNS
assert len(reread) == len(df)
# manifest accounts for all 500 cells
man = pd.read_csv(manifest)
assert len(man) == 500
Loading