Add fimbox.nextgen sub-package

pyproject.toml

@@ -81,6 +81,11 @@ include-package-data = true
 where = ["src"]
 exclude = ["tests*"]
 
+[tool.setuptools.package-data]
+# Ship the NextGen per-VPU bounding-box index so AOI->VPU resolution needs no
+# network round-trip. (Other runtime data is fetched on demand via pooch/S3.)
+"fimbox.nextgen" = ["data/*.json"]
+
 [tool.ruff]
 line-length = 88
 target-version = "py310"

src/fimbox/__init__.py

@@ -284,3 +284,29 @@
 # On-the-fly access to the bundled reference/calibration lookup tables, fetched
 # anonymously from the public SDML S3 bucket via pooch (imported at module top).
 __all__ += ["fetch_data"]
+
+# NextGen-in-a-Box bridge: resolve an AOI to the NextGen v2.2 hydrofabric
+# catchments and pull ngen/t-route discharge from the public CIROH community
+# datastream bucket. Heavy/optional deps (s3fs, xarray, geopandas) are imported
+# lazily inside the functions, so guard the import to keep a partial install
+# from breaking `import fimbox`.
+try:
+    from .nextgen import (
+        AOIHydrofabric,
+        NextGenAOI,
+        NextGenDatastream,
+        NextGenHydrofabric,
+        NextGenResult,
+        getNextGenAOI,
+    )
+
+    __all__ += [
+        "getNextGenAOI",
+        "NextGenAOI",
+        "NextGenResult",
+        "NextGenHydrofabric",
+        "AOIHydrofabric",
+        "NextGenDatastream",
+    ]
+except Exception:
+    pass

src/fimbox/nextgen/README.md

@@ -0,0 +1,135 @@
+### NextGen-in-a-Box Integration
+
+This part of fimbox connects your **area of interest** (a map outline of the place you care about) to NextGen streamflow data, so that flood maps can be produced for that area.
+
+In simple terms, you hand it a shapefile or geopackage of an area, and it figures out which rivers and drainage areas fall inside that shape, then goes and fetches how much water is flowing in each of them. 
+
+**What you give it**
+
+- An **area of interest** — a boundary file (a `.shp` shapefile or a `.gpkg` GeoPackage) outlining the region you want. For example, a city, a watershed, or a study site.
+
+**What you get back**, saved in a folder named after your area:
+
+- A **catchments** map file — the drainage areas that fall inside your boundary.
+- A **discharge** table — how much water is flowing in each of those areas (in cubic metres per second).
+- The **stream IDs** — the identifiers that link each drainage area to its river reach.
+
+Everything is downloaded from a [free public s3 bucket](https://ciroh-community-ngen-datastream.s3.amazonaws.com/index.html).
+
+### How to run it
+
+The simplest way, from a terminal:
+
+```bash
+python -m fimbox.nextgen my_area.shp --out-dir out
+```
+
+Replace `my_area.shp` with the path to your boundary file. Results land in `out/my_area/`.
+
+Or from Python, in one line:
+
+```python
+import fimbox
+
+result = fimbox.getNextGenAOI("my_area.shp", out_dir="out")
+
+print(result.catchments_path)   # the catchments map file
+print(result.discharge_csvs)    # the water-flow table(s)
+print(result.feature_ids)       # the stream IDs
+```
+
+By default it grabs the **most recent** available forecast.
+
+**Output folder structure**
+
+```
+out/my_area/
+  feature_id.csv                 the list of stream IDs
+  hydrofabric/
+    aoi_catchments.gpkg          the catchments (drainage areas) map
+    aoi_flowpaths.gpkg           the rivers/streams inside your area
+    network_crosswalk.csv        a lookup table linking IDs together
+  discharge-inputs/
+    nextgen_..._maximum.csv      the water-flow numbers for each stream
+```
+
+### A few useful options
+
+You can add these to the command above:
+
+- `--no-discharge` — just find the catchments and IDs, skip downloading the water-flow data (faster).
+- `--date 20260721 --cycle 09` — use a specific day/time instead of the latest.
+- `--forecast medium_range` — use a longer-range forecast (default is `short_range`).
+- `--sortby maximum` — for each stream, keep the **peak** flow over the forecast (the default, best for flood extent). Other choices: `minimum`, `mean`, or `none` to keep every hour.
+
+
+---
+
+### Developer reference
+
+`fimbox.nextgen` maps an AOI to the NOAA/OWP **NextGen v2.2 hydrofabric** and the ngen/t-route **discharge** published on the public [CIROH community NextGen DataStream](https://ciroh-community-ngen-datastream.s3.amazonaws.com) S3 bucket.
+
+**How the crosswalk works.** Each NextGen divide (`divide_id` = `cat-<n>`) drains to exactly one flowpath (`id` = `wb-<n>`). The integer `<n>` is the `feature_id` used in the ngen/t-route outputs, so catchment → flowpath → discharge is a direct join. The `network` layer also carries `hf_id` (the NWM COMID), available as an optional crosswalk (`nwm_crosswalk=True`, slower because it scans that table over S3).
+
+```mermaid
+flowchart LR
+    A[AOI shp/gpkg] --> B[Resolve VPU<br/>cached bbox index]
+    B --> C[Read intersecting<br/>divides from S3 gpkg]
+    C --> D[cat-* -> wb-* -> feature_id<br/>+ optional NWM hf_id]
+    D --> E[Locate ngen run<br/>model/forecast/date/cycle/VPU]
+    E --> F[Read t-route flow<br/>parquet or tar.gz/netCDF]
+    F --> G[FIM-ready CSVs<br/>feature_id, discharge_cms]
+    D --> H[aoi_catchments.gpkg<br/>network_crosswalk.csv]
+```
+
+Newer runs store discharge as `ngen-run/outputs/troute/troute_output_*.parquet`; older runs as an `ngen-run.tar.gz` containing `outputs/troute/*.nc`. Both are handled transparently.
+
+**Module contents**
+
+| File | What it contains |
+|---|---|
+| `hydrofabric.py` | `NextGenHydrofabric` / `AOIHydrofabric`: AOI → VPU resolution, intersecting `divides`, and the `cat`/`wb`/`feature_id`(+`hf_id`) crosswalk. `build_vpu_index()` regenerates the cached VPU bbox index. |
+| `datastream.py` | `NextGenDatastream` / `DischargeRun`: locate an ngen run on the bucket (default `cfe_nom` / `short_range` / latest), read t-route `flow` for the feature ids, and write FIM-ready CSVs. |
+| `pipeline.py` | `NextGenAOI` and `getNextGenAOI`: AOI → catchments + discharge in one call, written into the standard fimbox AOI layout. Also the module CLI. |
+| `_common.py` | Bucket constants, anonymous S3 handle, AOI-layout helpers, the cached VPU bbox index, and `wb`→`feature_id` parsing. |
+| `data/vpu_bbox.json` | Per-VPU bounding boxes (EPSG:5070) so AOI→VPU resolution needs no network round-trip. |
+
+**Full Python API**
+
+```python
+import fimbox
+
+res = fimbox.getNextGenAOI(
+    "my_aoi.shp",                 # AOI shapefile / GeoPackage / GeoJSON
+    out_dir="out",                # AOI folder created at out/<aoi_stem>
+    # aoi_layer=None,             # layer name for a multi-layer gpkg
+    # predicate="intersects",     # or "within" (catchments fully inside the AOI)
+    # nwm_crosswalk=False,        # also resolve NWM hf_id/COMID (slower S3 read)
+    # model="cfe_nom",            # ngen model output set
+    # forecast="short_range",     # short_range | medium_range | analysis_assim_extend
+    # date=None, cycle=None,      # YYYYMMDD + cycle hour; default latest available
+    # sortby="maximum",           # horizon aggregation: maximum|minimum|mean|None
+    # at_time=None,               # single timestamp instead of aggregating
+    # fetch_discharge=True,       # False -> resolve catchments/ids only
+)
+
+res.catchments_path   # <AOI>/hydrofabric/aoi_catchments.gpkg
+res.feature_ids       # NextGen feature_ids == streamflow network ids
+res.network_ids       # wb-* flowpath ids
+res.discharge         # long DataFrame [feature_id, time, flow] (flow in cms)
+res.discharge_csvs    # FIM-ready CSVs in <AOI>/discharge-inputs/
+res.run               # e.g. "cfe_nom/short_range/ngen.20260721/09/VPU_03N"
+
+# Step-by-step (class-based)
+from fimbox.nextgen import NextGenHydrofabric, NextGenDatastream
+
+hf = NextGenHydrofabric("my_aoi.shp").resolve()      # AOIHydrofabric
+ds = NextGenDatastream(hf.vpus[0])                   # latest cfe_nom short_range run
+flow = ds.read_discharge(hf.feature_ids)             # [feature_id, time, flow]
+csvs = ds.to_fim_inputs("out/my_aoi", hf.feature_ids, sortby="maximum")
+```
+
+**Notes**
+
+- The bucket is read anonymously; no AWS credentials are required.
+- `feature_id.csv` and `discharge-inputs/` match the `fimbox.streamflow` layout, so NextGen discharge is a drop-in alternative to the NWM/GEOGLOWS sources for FIM generation.

src/fimbox/nextgen/__init__.py

@@ -0,0 +1,47 @@
+"""
+Author: Manjila Singh
+Date Updated: July 2026
+
+NextGen-in-a-Box integration for fimbox.
+
+Map an AOI to the NOAA/OWP NextGen v2.2 hydrofabric and the
+ngen/t-route discharge published on the public CIROH community NextGen
+DataStream bucket (https://ciroh-community-ngen-datastream.s3.amazonaws.com),
+so the resulting streamflow can drive FIM generation.
+
+Resolution
+----------
+NextGenHydrofabric   AOI (shp/gpkg) -> intersecting catchments + network-id crosswalk
+NextGenDatastream    NextGen network ids -> ngen/t-route discharge (S3)
+
+Orchestration
+-------------
+NextGenAOI / getNextGenAOI   AOI -> catchments + discharge in one call, written
+                             into the standard fimbox AOI layout.
+
+Example
+-------
+    import fimbox
+
+    res = fimbox.getNextGenAOI("my_aoi.shp", out_dir="out")
+    res.catchments_path   # <AOI>/hydrofabric/aoi_catchments.gpkg
+    res.feature_ids       # NextGen feature_ids (== streamflow network ids)
+    res.discharge_csvs    # FIM-ready <AOI>/discharge-inputs/*.csv
+"""
+
+from __future__ import annotations
+
+from .datastream import DischargeRun, NextGenDatastream
+from .hydrofabric import AOIHydrofabric, NextGenHydrofabric, build_vpu_index
+from .pipeline import NextGenAOI, NextGenResult, getNextGenAOI
+
+__all__ = [
+    "NextGenHydrofabric",
+    "AOIHydrofabric",
+    "NextGenDatastream",
+    "DischargeRun",
+    "NextGenAOI",
+    "NextGenResult",
+    "getNextGenAOI",
+    "build_vpu_index",
+]

src/fimbox/nextgen/__main__.py

@@ -0,0 +1,17 @@
+"""
+Author: Manjila Singh
+Date Created: July 2026
+
+Package entry point so ``python -m fimbox.nextgen`` runs the pipeline CLI.
+
+Kept separate from ``pipeline.py`` (and not imported by ``__init__``) so that
+``python -m fimbox.nextgen`` does not trip the runpy "found in sys.modules"
+double-import warning.
+"""
+
+from __future__ import annotations
+
+from .pipeline import _main
+
+if __name__ == "__main__":
+    _main()

src/fimbox/nextgen/_common.py

@@ -0,0 +1,113 @@
+"""
+Shared helpers for the ``fimbox.nextgen`` subpackage: constants for the public
+CIROH community NextGen DataStream bucket, an anonymous s3fs handle, AOI-layout
+resolution, and the cached per-VPU bounding-box index used to resolve an area
+of interest to its hydrofabric VPU(s) without opening every GeoPackage.
+
+The bucket is read anonymously (``--no-sign-request`` equivalent) so it works
+for any caller regardless of whether AWS credentials are configured.
+"""
+
+from __future__ import annotations
+
+import importlib
+import json
+import logging
+from functools import lru_cache
+from pathlib import Path
+from typing import Optional, Union
+
+from ..logging_utils import aoi_root, attach_case_log
+
+log = logging.getLogger(__name__)
+
+PathLike = Union[str, Path]
+
+BUCKET = "ciroh-community-ngen-datastream"
+S3_BUCKET_URL = f"https://{BUCKET}.s3.amazonaws.com"
+
+HF_VERSION = "v2.2"
+HF_GEOPACKAGES_PREFIX = f"{BUCKET}/resources/{HF_VERSION}_hydrofabric/geopackages"
+
+# outputs/<model>/<HF_VERSION>_hydrofabric/ngen.<YYYYMMDD>/<forecast>/<cycle>/VPU_<id>/
+OUTPUTS_PREFIX = f"{BUCKET}/outputs"
+
+HF_EPSG = 5070  # CONUS Albers — all spatial ops happen in this CRS
+DISCHARGE_COL = "discharge_cms"  # column name the FIM Inundator expects
+
+HYDROFABRIC_DIR = "hydrofabric"
+DISCHARGE_INPUTS_DIR = "discharge-inputs"
+
+
+def require(module: str):
+    """Import an optional dependency, raising a clear install hint if missing."""
+    try:
+        return importlib.import_module(module)
+    except ImportError as exc:  # pragma: no cover - trivial
+        raise ImportError(
+            f"'{module}' is required for fimbox.nextgen. "
+            f"Install it with: pip install {module}"
+        ) from exc
+
+
+@lru_cache(maxsize=1)
+def s3() -> "object":
+    """Anonymous s3fs filesystem for the public bucket (cached)."""
+    s3fs = require("s3fs")
+    return s3fs.S3FileSystem(anon=True)
+
+
+def resolve_aoi(aoi_dir: PathLike) -> Path:
+    """Return the AOI root for any directory the caller passes (the root itself
+    or its ``watershed-data`` subfolder)."""
+    return aoi_root(Path(aoi_dir))
+
+
+def hydrofabric_dir(aoi_dir: PathLike) -> Path:
+    """``<AOI>/hydrofabric`` — subset catchments / flowpaths / crosswalk land
+    here. Created on demand."""
+    d = resolve_aoi(aoi_dir) / HYDROFABRIC_DIR
+    d.mkdir(parents=True, exist_ok=True)
+    return d
+
+
+def discharge_inputs_dir(aoi_dir: PathLike) -> Path:
+    """``<AOI>/discharge-inputs`` — the FIM-ready discharge CSVs the generator
+    iterates (shared with the streamflow subpackage)."""
+    d = resolve_aoi(aoi_dir) / DISCHARGE_INPUTS_DIR
+    d.mkdir(parents=True, exist_ok=True)
+    return d
+
+
+def attach_log(aoi_dir: PathLike) -> None:
+    """Route nextgen log records into the AOI's combined processing.log."""
+    attach_case_log(aoi_dir)
+
+
+_DATA_DIR = Path(__file__).parent / "data"
+_VPU_INDEX_PATH = _DATA_DIR / "vpu_bbox.json"
+
+
+@lru_cache(maxsize=1)
+def vpu_bbox_index() -> dict[str, list[float]]:
+    """Load the cached ``{VPU_id: [minx, miny, maxx, maxy]}`` index (EPSG:5070).
+
+    Shipped as package data so AOI->VPU resolution needs no network round-trip.
+    """
+    payload = json.loads(_VPU_INDEX_PATH.read_text())
+    return {k: list(map(float, v)) for k, v in payload["bboxes"].items()}
+
+
+def vpu_gpkg_uri(vpu: str) -> str:
+    """``/vsis3`` URI for a VPU's NextGen GeoPackage (read anonymously)."""
+    return f"/vsis3/{HF_GEOPACKAGES_PREFIX}/{vpu}/nextgen_{vpu}.gpkg"
+
+
+def feature_id_from_wb(wb_id: str) -> Optional[int]:
+    """``"wb-2855078"`` -> ``2855078``. Returns None for non-``wb`` ids."""
+    if not isinstance(wb_id, str) or "-" not in wb_id:
+        return None
+    prefix, _, num = wb_id.partition("-")
+    if prefix != "wb" or not num.isdigit():
+        return None
+    return int(num)

src/fimbox/nextgen/data/vpu_bbox.json

@@ -0,0 +1,28 @@
+{
+  "_comment": "NextGen v2.2 hydrofabric per-VPU bounding boxes in EPSG:5070 [minx, miny, maxx, maxy]. Derived from the divides layer total_bounds of each nextgen_VPU_<id>.gpkg in s3://ciroh-community-ngen-datastream/resources/v2.2_hydrofabric/geopackages/. Used to resolve an AOI to its VPU(s) without opening every gpkg. Regenerate with fimbox.nextgen.hydrofabric.build_vpu_index().",
+  "crs": "EPSG:5070",
+  "hydrofabric_version": "v2.2",
+  "bboxes": {
+    "VPU_01": [1824824.9997, 2215544.9958, 2258234.9955, 3086925.0003],
+    "VPU_02": [1350645.0003, 1687635.0, 1991054.9979, 2720864.9997],
+    "VPU_03N": [1115894.9997, 1023914.9979, 1833435.0045, 1740465.0],
+    "VPU_03S": [1061444.9997, 340785.0, 1598235.0003, 1331925.0003],
+    "VPU_03W": [578895.0003, 788805.0, 1161915.0003, 1402994.9997],
+    "VPU_04": [213255.0, 1990604.9997, 1744665.0003, 2824454.9997],
+    "VPU_05": [588854.9997, 1413285.0003, 1488224.9997, 2292855.0003],
+    "VPU_06": [645014.9997, 1272255.0003, 1294065.0, 1672544.9997],
+    "VPU_07": [-106125.0003, 1565505.0, 837735.0003, 2754015.0003],
+    "VPU_08": [150975.0, 673394.9976, 689775.0003, 1661985.0],
+    "VPU_09": [-646274.9997, 2498744.9997, 455654.9997, 3083384.9997],
+    "VPU_10L": [-1074915.0, 1556235.0, 505784.9997, 2300715.0],
+    "VPU_10U": [-1389615.0003, 2117324.9997, 31545.0, 3039464.9997],
+    "VPU_11": [-907575.0003, 908984.9997, 518355.0, 1866134.9997],
+    "VPU_12": [-729254.9997, 312795.0, 271244.9997, 1319204.9997],
+    "VPU_13": [-1206315.0, 209715.0003, -115845.0003, 1759995.0],
+    "VPU_14": [-1425105.0, 1456994.9997, -811784.9997, 2356695.0],
+    "VPU_15": [-1772655.0003, 975944.9997, -1081784.9997, 1971974.9997],
+    "VPU_16": [-2063565.0, 1528364.9997, -1192725.0, 2368545.0003],
+    "VPU_17": [-2294145.0, 2191574.9997, -1102155.0003, 3506235.0003],
+    "VPU_18": [-2355824.9997, 1202325.0003, -1696155.0003, 2538315.0]
+  }
+}