"""Retrieve hydrologic network features from the Network Linked Data Index (NLDI).
The getters below navigate the hydrologic network from an origin -- a feature
source and id, a ``comid``, or a lat/long point -- and return flowlines, basins,
or registered features as a ``geopandas.GeoDataFrame``, or as raw JSON when
``as_json=True``. This module requires geopandas.
See https://api.water.usgs.gov/nldi/linked-data for the API reference.
"""
from __future__ import annotations
from json import JSONDecodeError
from typing import Any, Literal, cast
from dataretrieval._querying import _query_with_retry
__all__ = [
"get_flowlines",
"get_basin",
"get_features",
"get_features_by_data_source",
"search",
]
try:
import geopandas as gpd
except ImportError as err:
raise ImportError("Install geopandas to use the NLDI module.") from err
NLDI_API_BASE_URL = "https://api.water.usgs.gov/nldi/linked-data"
_AVAILABLE_DATA_SOURCES = None
_CRS = "EPSG:4326"
_VALID_NAVIGATION_MODES = ("UM", "DM", "UT", "DD")
def _query_nldi(
url: str,
query_params: dict[str, str],
) -> dict[str, Any] | list[Any]:
# A helper function to query the NLDI API. ``query()`` already raises a
# typed ``DataRetrievalError`` for any HTTP error response, so a returned
# response is a success that we only need to parse.
response = _query_with_retry(url, payload=query_params)
response_data: dict[str, Any] | list[Any] = {}
try:
response_data = response.json()
except JSONDecodeError:
# even with a 200 status code, the response sometimes does not return JSON
# data which causes a JSONDecodeError
pass
return response_data
def _features_to_gdf(feature_collection: dict[str, Any]) -> gpd.GeoDataFrame:
"""Build a GeoDataFrame from an NLDI FeatureCollection, tolerating empties.
NLDI can legitimately return no features (e.g. a feature with nothing
upstream), and :func:`_query_nldi` returns ``{}`` when a 200 response
carries no JSON body. ``GeoDataFrame.from_features`` raises on both cases
(there's no geometry column to attach the CRS to), so return an empty
GeoDataFrame with the correct CRS instead of crashing.
"""
features = feature_collection.get("features") if feature_collection else None
if not features:
return gpd.GeoDataFrame(geometry=[], crs=_CRS)
return gpd.GeoDataFrame.from_features(feature_collection, crs=_CRS)
def _query_features(
url: str, query_params: dict[str, str], as_json: bool
) -> gpd.GeoDataFrame | dict[str, Any]:
"""Run an NLDI query and return the raw FeatureCollection or a GeoDataFrame."""
feature_collection = cast("dict[str, Any]", _query_nldi(url, query_params))
return feature_collection if as_json else _features_to_gdf(feature_collection)
[docs]
def get_flowlines(
navigation_mode: str,
distance: int = 5,
feature_source: str | None = None,
feature_id: str | None = None,
comid: int | None = None,
stop_comid: int | None = None,
trim_start: bool = False,
as_json: bool = False,
) -> gpd.GeoDataFrame | dict[str, Any]:
"""Get the flowlines for a navigation, either by comid or by feature source.
Flowlines are returned in WGS84 lat/long coordinates as a GeoDataFrame
containing a polyline geometry.
Parameters
----------
navigation_mode: string, navigation mode, allowed values are 'UM', 'DM', 'UT', 'DD'
distance: int, distance in kilometers, default is 5
feature_source: string, optional, name of the feature source,
required if comid is not provided
feature_id: string, optional, identifier of the feature,
required if comid is not provided
comid: integer, optional, comid, required if feature source is not provided
stop_comid: integer, optional, stop comid
trim_start: bool, trim start, default is False
as_json: bool, optional, return flowlines as JSON if set to True,
otherwise return as GeoDataFrame, default is False
Returns
-------
gdf: GeoDataFrame or dict
GeoDataFrame/dict of flowlines
Examples
--------
.. doctest::
>>> # Get flowlines for a feature source: WQP and
>>> # feature id: USGS-01031500 in the upstream main
>>> gdf = dataretrieval.nldi.get_flowlines(
... feature_source="WQP",
... feature_id="USGS-01031500",
... navigation_mode="UM",
... )
>>> # Get flowlines for comid: 13294314 in the upstream main
>>> gdf = dataretrieval.nldi.get_flowlines(
... comid=13294314, navigation_mode="UM"
... )
"""
navigation_mode = _validate_navigation_mode(navigation_mode)
_validate_feature_source_comid(feature_source, feature_id, comid)
if feature_source:
_validate_data_source(feature_source)
url, query_params = _navigation_request(
feature_source=feature_source,
feature_id=feature_id,
comid=comid,
navigation_mode=navigation_mode,
distance=distance,
tail="flowlines",
)
query_params["trimStart"] = str(trim_start).lower()
if stop_comid is not None:
query_params["stopComid"] = str(stop_comid)
return _query_features(url, query_params, as_json)
[docs]
def get_basin(
feature_source: str,
feature_id: str,
simplified: bool = True,
split_catchment: bool = False,
as_json: bool = False,
) -> gpd.GeoDataFrame | dict[str, Any]:
"""Get the aggregated basin for the specified feature.
The basin is returned in WGS84 lat/lon as a GeoDataFrame or as JSON,
containing a polygon geometry.
Parameters
----------
feature_source: string, name of the feature source
feature_id: string, identifier of the feature
simplified: bool, simplified, default is True
split_catchment: bool, split catchment, default is False
as_json: bool, return basin as JSON if set to True, otherwise return
as GeoDataFrame, default is False
Returns
-------
gdf: GeoDataFrame or dict
GeoDataFrame/dict of basin
Examples
--------
.. doctest::
>>> # Get basin for a feature source: WQP and feature id: USGS-01031500
>>> gdf = dataretrieval.nldi.get_basin(
... feature_source="WQP", feature_id="USGS-01031500"
... )
"""
# validate the feature source
_validate_data_source(feature_source)
if not feature_id:
raise ValueError("feature_id is required")
url = f"{NLDI_API_BASE_URL}/{feature_source}/{feature_id}/basin"
simplified_str = str(simplified).lower()
split_catchment_str = str(split_catchment).lower()
query_params = {
"simplified": simplified_str,
"splitCatchment": split_catchment_str,
}
return _query_features(url, query_params, as_json)
[docs]
def get_features(
data_source: str | None = None,
navigation_mode: str | None = None,
distance: int = 50,
feature_source: str | None = None,
feature_id: str | None = None,
comid: int | None = None,
lat: float | None = None,
long: float | None = None,
stop_comid: int | None = None,
as_json: bool = False,
) -> gpd.GeoDataFrame | dict[str, Any]:
"""Get all features along a navigation, either by comid or by feature source.
Features are returned as points in WGS84 lat/long coordinates - a
GeoDataFrame containing a point geometry.
Parameters
----------
feature_source: string, optional, name of the feature source,
required if comid is not provided
feature_id: string, optional, identifier of the feature,
required if comid is not provided
navigation_mode: string, navigation mode, allowed values are 'UM', 'DM', 'UT', 'DD'
data_source: string, data source
distance: int, distance in kilometers, default is 50
comid: integer, optional, comid, required if feature source is not provided
lat: float, optional, latitude, if provided, long is also required
long: float, optional, longitude, if provided, lat is also required
stop_comid: integer, optional, stop comid
as_json: bool, optional, return features as JSON if set to True,
otherwise return as GeoDataFrame, default is False
Returns
-------
gdf: GeoDataFrame or dict
GeoDataFrame/dict of features
Examples
--------
.. doctest::
>>> # Get registered features for a feature source: WQP,
>>> # feature id: USGS-01031500
>>> gdf = dataretrieval.nldi.get_features(
... feature_source="WQP", feature_id="USGS-01031500"
... )
>>> # Get features for a feature source: WQP, feature id: USGS-01031500,
>>> # and data source: nwissite in the upstream main
>>> gdf = dataretrieval.nldi.get_features(
... feature_source="WQP",
... feature_id="USGS-01031500",
... navigation_mode="UM",
... data_source="nwissite",
... distance=50,
... )
>>> # Get features for a comid: 13294314, and data source: nwissite
>>> # in the upstream main
>>> gdf = dataretrieval.nldi.get_features(
... comid=13294314,
... navigation_mode="UM",
... data_source="nwissite",
... distance=50,
... )
>>> # Get features for a latitude: 43.073051 and longitude: -89.401230
>>> gdf = dataretrieval.nldi.get_features(lat=43.073051, long=-89.401230)
"""
url, query_params = _get_features_request(
data_source=data_source,
navigation_mode=navigation_mode,
distance=distance,
feature_source=feature_source,
feature_id=feature_id,
comid=comid,
lat=lat,
long=long,
stop_comid=stop_comid,
)
return _query_features(url, query_params, as_json)
def _navigation_request(
*,
feature_source: str | None,
feature_id: str | None,
comid: int | None,
navigation_mode: str,
distance: int,
tail: str,
) -> tuple[str, dict[str, str]]:
"""URL and query params for an NLDI navigation from a validated origin.
The single home for the navigation path grammar — ``{origin}/navigation/
{mode}/{tail}`` — and its ``distance`` knob. Callers add the knobs specific
to their endpoint (``trimStart``, ``stopComid``) afterwards, so the query
string keeps its documented parameter order.
"""
origin = f"{feature_source}/{feature_id}" if feature_source else f"comid/{comid}"
url = f"{NLDI_API_BASE_URL}/{origin}/navigation/{navigation_mode}/{tail}"
return url, {"distance": str(distance)}
def _get_features_request(
*,
data_source: str | None,
navigation_mode: str | None,
distance: int,
feature_source: str | None,
feature_id: str | None,
comid: int | None,
lat: float | None,
long: float | None,
stop_comid: int | None,
) -> tuple[str, dict[str, str]]:
"""Validate a feature origin and build its NLDI request parameters."""
if (lat is None) != (long is None):
raise ValueError("Both lat and long are required")
if lat is not None:
if comid is not None:
raise ValueError(
"Provide only one origin type - comid cannot be provided"
" with lat or long"
)
if feature_source is not None or feature_id is not None:
raise ValueError(
"Provide only one origin type - feature_source and feature_id cannot"
" be provided with lat or long"
)
return f"{NLDI_API_BASE_URL}/comid/position", {"coords": f"POINT({long} {lat})"}
if (comid is not None or data_source is not None) and navigation_mode is None:
raise ValueError(
"navigation_mode is required if comid or data_source is provided"
)
_validate_feature_source_comid(feature_source, feature_id, comid)
if data_source is not None:
_validate_data_source(data_source)
if feature_source is not None:
_validate_data_source(feature_source)
if not navigation_mode:
return f"{NLDI_API_BASE_URL}/{feature_source}/{feature_id}", {}
navigation_mode = _validate_navigation_mode(navigation_mode)
url, query_params = _navigation_request(
feature_source=feature_source,
feature_id=feature_id,
comid=comid,
navigation_mode=navigation_mode,
distance=distance,
tail=f"{data_source}",
)
if stop_comid is not None:
query_params["stopComid"] = str(stop_comid)
return url, query_params
# TODO: This function can cause a timeout error for some data sources
# - maybe we shouldn't provide this function?
[docs]
def get_features_by_data_source(data_source: str) -> gpd.GeoDataFrame:
"""Get all features for the specified data source.
Features are returned as points in WGS84 lat/long coordinates as a
GeoDataFrame containing a point geometry.
Parameters
----------
data_source: string, data source
Returns
-------
gdf: GeoDataFrame
GeoDataFrame of features
Examples
--------
.. doctest::
>>> # "nwissite" returns every NWIS site nationwide, so this example is
>>> # skipped in the doctest build to avoid the (very large) download.
>>> gdf = dataretrieval.nldi.get_features_by_data_source( # doctest: +SKIP
... data_source="nwissite"
... )
"""
# validate the data source
_validate_data_source(data_source)
url = f"{NLDI_API_BASE_URL}/{data_source}"
feature_collection = cast("dict[str, Any]", _query_nldi(url, {}))
gdf = _features_to_gdf(feature_collection)
return gdf
[docs]
def search(
feature_source: str | None = None,
feature_id: str | None = None,
navigation_mode: str | None = None,
data_source: str | None = None,
find: Literal["basin", "flowlines", "features"] = "features",
comid: int | None = None,
lat: float | None = None,
long: float | None = None,
distance: int = 50,
) -> dict[str, Any]:
"""Search NLDI for the specified feature and return the results as a dict.
Parameters
----------
feature_source: string, name of the feature source
feature_id: string, identifier of the feature
navigation_mode: string, optional, navigation mode,
allowed values are 'UM', 'DM', 'UT', 'DD'
data_source: string, optional, data source
find: string, search for 'basin', 'flowlines', or 'features', default is 'features'
comid: int, optional, comid, default is None
lat: float, optional, latitude, default is None
long: float, optional, longitude, default is None
distance: int, optional, distance in kilometers, default is 50
Returns
-------
dict: search results
Examples
--------
.. doctest::
>>> # Search for aggregated basin for feature source: WQP
>>> # and feature id: USGS-01031500
>>> search_results = dataretrieval.nldi.search(
... feature_source="WQP", feature_id="USGS-01031500", find="basin"
... )
>>> # Search for flowlines for feature source: WQP and
>>> # feature id: USGS-01031500 in the upstream main
>>> search_results = dataretrieval.nldi.search(
... feature_source="WQP",
... feature_id="USGS-01031500",
... navigation_mode="UM",
... find="flowlines",
... )
>>> # Get registered features for a feature source: WQP,
>>> # feature id: USGS-01031500
>>> gdf = dataretrieval.nldi.get_features(
... feature_source="WQP", feature_id="USGS-01031500"
... )
>>> # Search for features for feature source: WQP, feature id: USGS-01031500,
>>> # and data source: census2020-nhdpv2 in the upstream main
>>> search_results = dataretrieval.nldi.search(
... feature_source="WQP",
... feature_id="USGS-01031500",
... data_source="census2020-nhdpv2",
... navigation_mode="UM",
... find="features",
... )
>>> # Search for features for comid: 13294314,
>>> # and data source: census2020-nhdpv2 in the upstream main
>>> search_results = dataretrieval.nldi.search(
... comid=13294314,
... data_source="census2020-nhdpv2",
... navigation_mode="UM",
... )
>>> # Search for flowlines for comid: 13294314 in the upstream main
>>> search_results = dataretrieval.nldi.search(
... comid=13294314, navigation_mode="UM", find="flowlines"
... )
>>> # Search for features for latitude: 43.073051 and longitude: -89.401230
>>> search_results = dataretrieval.nldi.search(
... lat=43.073051, long=-89.401230
... )
"""
if (lat is None) != (long is None):
raise ValueError("Both lat and long are required")
find = cast("Literal['basin', 'flowlines', 'features']", find.lower())
if find not in ("basin", "flowlines", "features"):
raise ValueError(
f"Invalid value for find: {find} - allowed values are:"
f" 'basin', 'flowlines', or 'features'"
)
if lat is not None and find != "features":
raise ValueError(
f"Invalid value for find: {find} - lat/long is to get features not {find}"
)
if comid is not None and find == "basin":
raise ValueError(
"Invalid value for find: basin - comid is to get features"
" or flowlines not basin"
)
if lat is not None:
return get_features(lat=lat, long=long, as_json=True)
if find == "basin":
if feature_source is None or feature_id is None:
raise ValueError(
"feature_source and feature_id are required to find a basin"
)
return get_basin(
feature_source=feature_source, feature_id=feature_id, as_json=True
)
if find == "flowlines":
if navigation_mode is None:
raise ValueError(
"navigation_mode is required for find='flowlines';"
f" allowed values are {_VALID_NAVIGATION_MODES}"
)
return get_flowlines(
navigation_mode=navigation_mode,
distance=distance,
feature_source=feature_source,
feature_id=feature_id,
comid=comid,
as_json=True,
)
# here find == 'features'
return get_features(
data_source=data_source,
navigation_mode=navigation_mode,
distance=distance,
feature_source=feature_source,
feature_id=feature_id,
comid=comid,
as_json=True,
)
def _validate_data_source(data_source: str) -> None:
# A helper function to validate user specified data source/feature source
global _AVAILABLE_DATA_SOURCES
# get the available data/feature sources - if not already cached
if _AVAILABLE_DATA_SOURCES is None:
url = f"{NLDI_API_BASE_URL}/"
available_data_sources = _query_nldi(url, {})
if not isinstance(available_data_sources, list) or not all(
isinstance(ds, dict) and "source" in ds for ds in available_data_sources
):
raise ValueError(
"NLDI data-source catalog returned an unexpected shape; "
"expected a list of {'source': ..., ...} objects, got: "
f"{available_data_sources!r}"
)
_AVAILABLE_DATA_SOURCES = [ds["source"] for ds in available_data_sources]
if data_source not in _AVAILABLE_DATA_SOURCES:
err_msg = (
f"Invalid data source '{data_source}'."
f" Available data sources are: {_AVAILABLE_DATA_SOURCES}"
)
raise ValueError(err_msg)
def _validate_navigation_mode(navigation_mode: str | None) -> str:
if navigation_mode is None:
raise ValueError(
f"navigation_mode is required; allowed values are {_VALID_NAVIGATION_MODES}"
)
normalized = navigation_mode.upper()
if normalized not in _VALID_NAVIGATION_MODES:
raise ValueError(
f"Invalid navigation mode '{navigation_mode}';"
f" allowed values are {_VALID_NAVIGATION_MODES}"
)
return normalized
def _validate_feature_source_comid(
feature_source: str | None, feature_id: str | None, comid: int | None
) -> None:
if feature_source is not None and feature_id is None:
raise ValueError("feature_id is required if feature_source is provided")
if feature_id is not None and feature_source is None:
raise ValueError("feature_source is required if feature_id is provided")
if comid is not None and feature_source is not None:
raise ValueError(
"Specify only one origin type - comid and feature_source"
" cannot be provided together"
)
if comid is None and feature_source is None:
raise ValueError(
"Specify one origin type - comid or feature_source is required"
)