From 587ee85dc6743237bcbbcdd6e29bc8bea27f6002 Mon Sep 17 00:00:00 2001 From: Clara Date: Wed, 14 Oct 2026 22:26:52 +0000 Subject: [PATCH] river.io.usgs: request data from the USGS Water Data APIs (closes #467) WaterServices (waterservices.usgs.gov) is being decommissioned in Q1 2027. request_usgs_data now uses api.waterdata.usgs.gov/ogcapi/v1: the daily collection (statistic 00003) for 'Daily' and continuous for 'Instantaneous', split into yearly queries (continuous serves at most three years per query), following 'next' links, with an optional api_key (default: API_USGS_PAT). Columns keep their legacy names via the parameter-codes collection. read_usgs_file reads both legacy JSON and Water Data API GeoJSON. Also fixes to_pandas=False, which called DataFrame.to_dataset() (no such method) and raised AttributeError; now uses utils.convert_to_dataset. --- mhkit/river/io/usgs.py | 236 +++++++++++++++++++++++------- mhkit/tests/river/test_io_usgs.py | 182 ++++++++++++++--------- 2 files changed, 290 insertions(+), 128 deletions(-) diff --git a/mhkit/river/io/usgs.py b/mhkit/river/io/usgs.py index 2b690ff50..ab79cbbb2 100644 --- a/mhkit/river/io/usgs.py +++ b/mhkit/river/io/usgs.py @@ -3,6 +3,10 @@ Geological Survey (USGS) National Water Information System (NWIS). It enables access to river flow data and related measurements useful for hydrokinetic resource assessment. +Requests go to the USGS Water Data APIs (https://api.waterdata.usgs.gov/ogcapi/v1/), which +replace the legacy WaterServices endpoints (waterservices.usgs.gov) that USGS is +decommissioning. Files saved from either service can still be read. + """ from typing import Dict, Union, Optional @@ -14,6 +18,42 @@ import xarray as xr from pandas import DataFrame from mhkit.utils.cache import handle_caching +from mhkit.utils.type_handling import convert_to_dataset + + +WATER_DATA_API = "https://api.waterdata.usgs.gov/ogcapi/v1" + + +def _read_water_data_features(text: Dict) -> DataFrame: + """ + Process a USGS Water Data API response (GeoJSON features) into a DataFrame + with the same layout as the legacy WaterServices reader: a UTC datetime + index and one numeric column per parameter, named by its description. + + Parameters + ---------- + text : dict + GeoJSON FeatureCollection from the Water Data API. The parameter + description used for the column name is read from + ``text["parameter_description"]`` when present, else the parameter code. + + Returns + ------- + data : pandas.DataFrame + Processed time series data + """ + rows = [feature["properties"] for feature in text.get("features", [])] + if not rows: + return pd.DataFrame() + frame = pd.DataFrame(rows) + name = text.get("parameter_description") or str(frame["parameter_code"].iloc[0]) + data = pd.DataFrame( + {name: pd.to_numeric(frame["value"], errors="coerce").values}, + index=pd.to_datetime(frame["time"], utc=True), + ) + data.index.name = None + data = data[~data.index.duplicated(keep="last")].sort_index() + return data def _read_usgs_json(text: Dict, to_pandas: bool = True) -> Union[DataFrame, xr.Dataset]: @@ -23,7 +63,8 @@ def _read_usgs_json(text: Dict, to_pandas: bool = True) -> Union[DataFrame, xr.D Parameters ---------- text : dict - JSON response from USGS API containing time series data + JSON response from USGS API containing time series data, either + legacy WaterServices JSON or Water Data API GeoJSON to_pandas : bool, optional Flag to output pandas instead of xarray. Default = True. @@ -32,6 +73,10 @@ def _read_usgs_json(text: Dict, to_pandas: bool = True) -> Union[DataFrame, xr.D data : pandas.DataFrame or xarray.Dataset Processed time series data """ + if "features" in text: + data = _read_water_data_features(text) + return data if to_pandas else convert_to_dataset(data) + data = pd.DataFrame() for i in range(len(text["value"]["timeSeries"])): try: @@ -53,7 +98,7 @@ def _read_usgs_json(text: Dict, to_pandas: bool = True) -> Union[DataFrame, xr.D continue if not to_pandas: - data = data.to_dataset() + data = convert_to_dataset(data) return data @@ -62,7 +107,9 @@ def read_usgs_file( file_name: str, to_pandas: bool = True ) -> Union[DataFrame, xr.Dataset]: """ - Reads a USGS JSON data file (from https://waterdata.usgs.gov/nwis) + Reads a USGS JSON data file: legacy WaterServices JSON (from + https://waterdata.usgs.gov/nwis), or Water Data API GeoJSON saved from + https://api.waterdata.usgs.gov/ogcapi/v1/ Parameters ---------- @@ -97,10 +144,13 @@ def request_usgs_data( options: Optional[Dict] = None, ) -> Union[DataFrame, xr.Dataset]: """ - Loads USGS data directly from https://waterdata.usgs.gov/nwis using a - GET request + Loads USGS data from the USGS Water Data APIs + (https://api.waterdata.usgs.gov/ogcapi/v1/) using GET requests: the + ``daily`` collection for 'Daily' and ``continuous`` for 'Instantaneous'. - The request URL prints to the screen. + The request URLs print to the screen. Instantaneous requests are split + into one query per year, as the ``continuous`` collection serves at most + three years per query. Parameters ---------- @@ -111,7 +161,9 @@ def request_usgs_data( start_date : str Start date in the format 'YYYY-MM-DD' (e.g. '2018-01-01') end_date : str - End date in the format 'YYYY-MM-DD' (e.g. '2018-12-31') + End date in the format 'YYYY-MM-DD' (e.g. '2018-12-31'). Both dates + are inclusive. For 'Instantaneous' data they bound the request in UTC; + the legacy WaterServices endpoint read them in the site's local time. options : dict, optional Dictionary containing optional parameters: - data_type: str @@ -127,6 +179,10 @@ def request_usgs_data( Flag to output pandas instead of xarray. Default = True - timeout: int Timeout in seconds for the HTTP request. Default = 30 + - api_key: str or None + USGS Water Data API key for higher rate limits + (https://api.waterdata.usgs.gov/signup/). Default = the + API_USGS_PAT environment variable, else None Returns ------- @@ -169,53 +225,18 @@ def request_usgs_data( return cached_data # If no cached data, proceed with the API request - if data_type == "Daily": - data_url = "https://waterservices.usgs.gov/nwis/dv" - api_query = ( - "/?format=json&sites=" - + station - + "&startDT=" - + start_date - + "&endDT=" - + end_date - + "&statCd=00003" - + "¶meterCd=" - + parameter - + "&siteStatus=all" - ) - else: - data_url = "https://waterservices.usgs.gov/nwis/iv" - api_query = ( - "/?format=json&sites=" - + station - + "&startDT=" - + start_date - + "&endDT=" - + end_date - + "¶meterCd=" - + parameter - + "&siteStatus=all" - ) - - print("Data request URL: ", data_url + api_query) - - max_retries = 3 - retry_count = 0 - while retry_count < max_retries: - try: - response = requests.get( - url=data_url + api_query, proxies=proxy, timeout=timeout, verify=True - ) - text = json.loads(response.text) - break - except requests.exceptions.SSLError as e: - retry_count += 1 - if retry_count == max_retries: - raise e - print( - f"SSL Error occurred, retrying... (Attempt {retry_count}/{max_retries})" - ) - continue + text = _request_water_data( + station, + parameter, + start_date, + end_date, + { + "data_type": data_type, + "proxy": proxy, + "timeout": timeout, + "api_key": options.get("api_key", os.environ.get("API_USGS_PAT")), + }, + ) # handle_caching is only set-up for pandas, so force this data to output as pandas for now data = _read_usgs_json(text, True) @@ -233,6 +254,111 @@ def request_usgs_data( shutil.copy(cache_filepath, write_json) if not to_pandas: - data = data.to_dataset() + data = convert_to_dataset(data) return data + + +def _get_json(url: str, params: Optional[Dict], proxy, timeout) -> Dict: + """GET a URL and parse JSON, retrying SSL errors up to three times.""" + max_retries = 3 + for attempt in range(1, max_retries + 1): + try: + response = requests.get( + url=url, params=params, proxies=proxy, timeout=timeout, verify=True + ) + response.raise_for_status() + return response.json() + except requests.exceptions.SSLError: + if attempt == max_retries: + raise + print(f"SSL Error occurred, retrying... (Attempt {attempt}/{max_retries})") + raise RuntimeError("unreachable") + + +def _year_windows(start_date: str, end_date: str): + """Split [start_date, end_date] into consecutive windows of at most one year.""" + start = pd.Timestamp(start_date, tz="UTC") + end = pd.Timestamp(end_date, tz="UTC") + pd.Timedelta(days=1) + while start < end: + stop = min(start + pd.DateOffset(years=1), end) + yield start, stop + start = stop + + +def _request_water_data( + station: str, parameter: str, start_date: str, end_date: str, settings: Dict +) -> Dict: + """ + Request all features for one station and parameter from the Water Data + API, following "next" links, and return them as one FeatureCollection with + the parameter's description attached for column naming. + """ + proxy, timeout, api_key = ( + settings["proxy"], + settings["timeout"], + settings["api_key"], + ) + base = { + "f": "json", + "monitoring_location_id": f"USGS-{station}", + "parameter_code": parameter, + "limit": 50000, + } + if api_key: + base["api_key"] = api_key + + if settings["data_type"] == "Daily": + collection = "daily" + windows = [(start_date, end_date)] + base["statistic_id"] = "00003" + else: + collection = "continuous" + windows = [ + (a.strftime("%Y-%m-%dT%H:%M:%SZ"), b.strftime("%Y-%m-%dT%H:%M:%SZ")) + for a, b in _year_windows(start_date, end_date) + ] + url = f"{WATER_DATA_API}/collections/{collection}/items" + + features = [] + for begin, finish in windows: + params = dict(base, time=f"{begin}/{finish}") + print( + "Data request URL: ", + requests.Request("GET", url, params=params).prepare().url, + ) + page = _get_json(url, params, proxy, timeout) + features += page.get("features", []) + next_link = _next_link(page) + while next_link: + if api_key and "api_key=" not in next_link: + next_link += f"&api_key={api_key}" + page = _get_json(next_link, None, proxy, timeout) + features += page.get("features", []) + next_link = _next_link(page) + + description = None + try: + code = _get_json( + f"{WATER_DATA_API}/collections/parameter-codes/items/{parameter}", + {"f": "json"}, + proxy, + timeout, + ) + description = code.get("properties", {}).get("parameter_description") + except (requests.exceptions.RequestException, ValueError): + description = None + + return { + "type": "FeatureCollection", + "features": features, + "parameter_description": description, + } + + +def _next_link(page: Dict) -> Optional[str]: + """Return the href of a page's "next" link, if any.""" + for link in page.get("links", []): + if link.get("rel") == "next": + return link.get("href") + return None diff --git a/mhkit/tests/river/test_io_usgs.py b/mhkit/tests/river/test_io_usgs.py index c3929ab0c..92a80616e 100644 --- a/mhkit/tests/river/test_io_usgs.py +++ b/mhkit/tests/river/test_io_usgs.py @@ -45,40 +45,13 @@ def test_load_usgs_data_daily(self): @patch("mhkit.river.io.usgs.requests.get") def test_request_usgs_data_daily(self, mock_get): """ - Test request_usgs_data with daily data + Test request_usgs_data with daily data from the Water Data API """ - # Prepare the mocked HTTP response for daily data - daily_values = [] - start = pd.Timestamp("2009-08-01 00:00:00", tz="UTC") - end = pd.Timestamp("2009-08-10 23:59:59", tz="UTC") - current = start - while current <= end: - daily_values.append( - { - "dateTime": current.strftime("%Y-%m-%dT%H:%M:%S.000Z"), - "value": "1000", - "qualifiers": ["P"], - } - ) - current += pd.Timedelta(days=1) - - mock_payload = { - "value": { - "timeSeries": [ - { - "variable": { - "variableDescription": "Discharge, cubic feet per second" - }, - "values": [{"value": daily_values}], - } - ] - } - } - - mock_resp = MagicMock() - mock_resp.status_code = 200 - mock_resp.text = json.dumps(mock_payload) - mock_get.return_value = mock_resp + days = pd.date_range("2009-08-01", "2009-08-10", freq="D") + mock_get.side_effect = [ + _response(_features([d.strftime("%Y-%m-%d") for d in days])), + _response(PARAMETER_CODE), + ] data = river.io.usgs.request_usgs_data( station="15515500", @@ -88,50 +61,28 @@ def test_request_usgs_data_daily(self, mock_get): options={"data_type": "Daily", "clear_cache": True}, ) - # Verify that we called requests.get - mock_get.assert_called_once() + first = mock_get.call_args_list[0].kwargs + self.assertTrue(first["url"].endswith("/collections/daily/items")) + self.assertEqual(first["params"]["monitoring_location_id"], "USGS-15515500") + self.assertEqual(first["params"]["statistic_id"], "00003") + self.assertEqual(first["params"]["time"], "2009-08-01/2009-08-10") - # Basic functionality checks self.assertIsInstance(data, pd.DataFrame) - self.assertGreater(len(data), 0) # Has data + self.assertEqual(list(data.columns), ["Discharge, cubic feet per second"]) + self.assertEqual(data.shape, (10, 1)) self.assertTrue(data.index.tz is not None) # Timezone aware + self.assertTrue((data.index == days.tz_localize("UTC")).all()) class TestUSGSInstant(unittest.TestCase): @patch("mhkit.river.io.usgs.requests.get") def test_request_usgs_data_instant(self, mock_get): - mock_payload = { - "value": { - "timeSeries": [ - { - "variable": { - "variableDescription": "Discharge, cubic feet per second" - }, - "values": [ - { - "value": [ - { - "dateTime": "2009-08-01T00:00:00.000Z", - "value": "1000", - "qualifiers": ["P"], - }, - { - "dateTime": "2009-08-01T00:15:00.000Z", - "value": "1000", - "qualifiers": ["P"], - }, - ] - } - ], - } - ] - } - } - - mock_resp = MagicMock() - mock_resp.status_code = 200 - mock_resp.text = json.dumps(mock_payload) - mock_get.return_value = mock_resp + mock_get.side_effect = [ + _response( + _features(["2009-08-01T00:00:00+00:00", "2009-08-01T00:15:00+00:00"]) + ), + _response(PARAMETER_CODE), + ] df = river.io.usgs.request_usgs_data( station="15515500", @@ -141,14 +92,99 @@ def test_request_usgs_data_instant(self, mock_get): options={"data_type": "Instantaneous", "clear_cache": True}, ) - # Verify that we called requests.get - mock_get.assert_called_once() + first = mock_get.call_args_list[0].kwargs + self.assertTrue(first["url"].endswith("/collections/continuous/items")) + self.assertEqual( + first["params"]["time"], "2009-08-01T00:00:00Z/2009-08-11T00:00:00Z" + ) - # Basic functionality checks self.assertIsInstance(df, pd.DataFrame) - self.assertGreater(len(df), 0) # Has data + self.assertEqual(df.shape, (2, 1)) self.assertTrue(df.index.tz is not None) # Timezone aware + @patch("mhkit.river.io.usgs.requests.get") + def test_instantaneous_split_by_year_and_pages_followed(self, mock_get): + """The continuous collection serves at most three years per query, and + large results are paged through "next" links""" + next_page = "https://api.waterdata.usgs.gov/ogcapi/v1/collections/continuous/items?cursor=abc" + mock_get.side_effect = [ + _response(_features(["2018-06-01T00:00:00+00:00"], next_link=next_page)), + _response(_features(["2018-06-01T00:15:00+00:00"])), + _response(_features(["2019-06-01T00:00:00+00:00"])), + _response(PARAMETER_CODE), + ] + + df = river.io.usgs.request_usgs_data( + station="08313000", + parameter="00060", + start_date="2018-01-01", + end_date="2019-12-31", + options={ + "data_type": "Instantaneous", + "clear_cache": True, + "api_key": "KEY", + }, + ) + + calls = [c.kwargs for c in mock_get.call_args_list] + self.assertEqual( + calls[0]["params"]["time"], "2018-01-01T00:00:00Z/2019-01-01T00:00:00Z" + ) + self.assertEqual(calls[1]["url"], next_page + "&api_key=KEY") + self.assertEqual( + calls[2]["params"]["time"], "2019-01-01T00:00:00Z/2020-01-01T00:00:00Z" + ) + self.assertEqual(calls[0]["params"]["api_key"], "KEY") + self.assertEqual(df.shape, (3, 1)) + + def test_read_water_data_geojson(self): + """GeoJSON from the Water Data API reads like the legacy JSON""" + text = _features(["2019-01-02", "2019-01-01", "2019-01-01"]) + text["parameter_description"] = "Discharge, cubic feet per second" + df = river.io.usgs._read_usgs_json(text) + self.assertEqual(list(df.columns), ["Discharge, cubic feet per second"]) + self.assertEqual(df.shape, (2, 1)) # duplicate time dropped + self.assertTrue(df.index.is_monotonic_increasing) + + ds = river.io.usgs._read_usgs_json(text, to_pandas=False) + self.assertIn("Discharge, cubic feet per second", ds.data_vars) + + +PARAMETER_CODE = { + "id": "00060", + "properties": {"parameter_description": "Discharge, cubic feet per second"}, +} + + +def _features(times, next_link=None): + page = { + "type": "FeatureCollection", + "features": [ + { + "type": "Feature", + "properties": { + "parameter_code": "00060", + "time": t, + "value": "1000", + "approval_status": "Provisional", + "qualifier": None, + }, + } + for t in times + ], + "links": [], + } + if next_link: + page["links"].append({"rel": "next", "href": next_link}) + return page + + +def _response(payload): + resp = MagicMock() + resp.status_code = 200 + resp.json.return_value = payload + return resp + if __name__ == "__main__": unittest.main()