Add UK-NRFA fetcher, #34

examples/test_uk_nrfa_fetcher.py

@@ -0,0 +1,58 @@
+import logging
+
+import matplotlib.pyplot as plt
+
+from rivretrieve import UKNRFAFetcher, constants
+
+logging.basicConfig(level=logging.INFO)
+
+gauge_ids = [
+    "1001",  # Sample gauge from issue #34
+]
+variable = constants.DISCHARGE
+start_date = "2022-01-01"
+end_date = "2022-01-31"
+
+plt.figure(figsize=(12, 6))
+
+fetcher = UKNRFAFetcher()
+
+# Test get_metadata
+print("Fetching metadata for one gauge...")
+metadata = fetcher.get_metadata()
+if not metadata.empty:
+    print(metadata.loc[gauge_ids[0]])
+else:
+    print("Metadata fetching failed or empty.")
+
+for gauge_id in gauge_ids:
+    print(f"Fetching {variable} for {gauge_id} from {start_date} to {end_date}...")
+    data = fetcher.get_data(gauge_id=gauge_id, variable=variable, start_date=start_date, end_date=end_date)
+    if not data.empty:
+        print(f"Data for {gauge_id}:")
+        print(data.head())
+        print(f"Time series from {data[constants.TIME_INDEX].min()} to {data[constants.TIME_INDEX].max()}")
+        plt.plot(
+            data[constants.TIME_INDEX],
+            data[constants.DISCHARGE],
+            label=gauge_id,
+            marker=".",
+            linestyle="-",
+        )
+    else:
+        print(f"No data found for {gauge_id}")
+
+if "data" in locals() and not data.empty:
+    plt.xlabel(constants.TIME_INDEX)
+    plt.ylabel(f"{constants.DISCHARGE} (m3/s)")
+    plt.title(f"UK NRFA River Discharge ({gauge_ids[0]} - {start_date} to {end_date})")
+    plt.legend()
+    plt.grid(True)
+    plt.tight_layout()
+    plot_path = "uk_nrfa_discharge_plot.png"
+    plt.savefig(plot_path)
+    print(f"Plot saved to {plot_path}")
+else:
+    print("No data to plot.")
+
+print("Test finished.")

requirements.txt

@@ -7,5 +7,6 @@ lxml>=4.8.0
 dataretrieval>=1.0.0
 openpyxl>=3.0.0
 xarray
+parameterized
 tqdm
 zarr>=3.0.7

rivretrieve/__init__.py

@@ -11,6 +11,7 @@
 from .slovenia import SloveniaFetcher
 from .southafrica import SouthAfricaFetcher
 from .uk import UKFetcher
+from .uk_nrfa import UKNRFAFetcher
 from .usa import USAFetcher
 
 __version__ = "0.1.0"

rivretrieve/constants.py

@@ -8,13 +8,14 @@
 DISCHARGE = "discharge"
 STAGE = "stage"
 WATER_TEMPERATURE = "water_temperature"
+CATCHMENT_PRECIPITATION = "catchment_precipitation"
 
 # Attributes
 ALTITUDE = "altitude"
 AREA = "area"
 COUNTRY = "country"
 LATITUDE = "latitude"
-LOCATION = "location"
 LONGITUDE = "longitude"
 RIVER = "river"
 SOURCE = "source"
+STATION_NAME = "station_name"

rivretrieve/uk_nrfa.py

@@ -0,0 +1,135 @@
+"""Fetcher for UK National River Flow Archive (NRFA) data."""
+
+import logging
+from typing import Any, Dict, Optional
+
+import pandas as pd
+import requests
+
+from . import base, constants, utils
+
+logger = logging.getLogger(__name__)
+
+
+class UKNRFAFetcher(base.RiverDataFetcher):
+    """Fetches river gauge data from the UK National River Flow Archive."""
+
+    BASE_URL = "https://nrfaapps.ceh.ac.uk/nrfa/ws"
+    GAUGE_ID_COL = "id"
+
+    METADATA_TRANSLATION_MAPPING = {
+        "name": constants.STATION_NAME,
+        "catchment-area": constants.AREA,
+        "latitude": constants.LATITUDE,
+        "longitude": constants.LONGITUDE,
+        "river": constants.RIVER,
+        # Using the catchment median altitude.
+        "50-percentile-altitude": constants.ALTITUDE,
+    }
+
+    @staticmethod
+    def get_gauge_ids() -> pd.DataFrame:
+        """Retrieves a DataFrame of available NRFA gauge IDs from the cached CSV."""
+        return utils.load_sites_csv("uk_nrfa")
+
+    def get_metadata(self) -> pd.DataFrame:
+        """Fetches site metadata from the NRFA API and renames columns."""
+        query_params = {"station": "*", "format": "json-object", "fields": "all"}
+        try:
+            s = utils.requests_retry_session()
+            response = s.get(f"{UKNRFAFetcher.BASE_URL}/station-info", params=query_params)
+            response.raise_for_status()  # raises an error for non-200 responses
+            data = response.json()
+            df = pd.DataFrame(data["data"])
+
+            # Rename id column to the standard GAUGE_ID
+            df = df.rename(columns={UKNRFAFetcher.GAUGE_ID_COL: constants.GAUGE_ID})
+            df[constants.GAUGE_ID] = df[constants.GAUGE_ID].astype(str)
+
+            # Apply translation mapping for renaming
+            df = df.rename(columns=self.METADATA_TRANSLATION_MAPPING)
+
+            return df.set_index(constants.GAUGE_ID)
+        except requests.exceptions.RequestException as e:
+            logger.error(f"Error fetching NRFA catalogue: {e}")
+            raise
+        except Exception as e:
+            logger.error(f"Error processing NRFA catalogue: {e}")
+            raise
+
+    @staticmethod
+    def get_available_variables() -> tuple[str, ...]:
+        # Based on common NRFA data types, can be expanded
+        return (constants.DISCHARGE, constants.CATCHMENT_PRECIPITATION)
+
+    def _get_nrfa_data_type(self, variable: str) -> str:
+        if variable == constants.DISCHARGE:
+            return "gdf"  # Mean daily flow
+        elif variable == constants.CATCHMENT_PRECIPITATION:
+            return "cdr"  # Catchment daily precipitation.
+        else:
+            raise ValueError(f"Unsupported variable: {variable} for NRFA")
+
+    def _download_data(self, gauge_id: str, variable: str, start_date: str, end_date: str) -> Optional[Dict[str, Any]]:
+        """Downloads the raw time series data from the NRFA API."""
+        data_type = self._get_nrfa_data_type(variable)
+        query_params = {
+            "station": str(gauge_id),
+            "data-type": data_type,
+            "format": "json-object",
+            "start-date": f"{start_date}T00:00:00Z",
+            "end-date": f"{end_date}T23:59:59Z",
+        }
+        s = utils.requests_retry_session()
+        try:
+            response = s.get(f"{self.BASE_URL}/time-series", params=query_params)
+            response.raise_for_status()
+            return response.json()
+        except requests.exceptions.RequestException as e:
+            logger.error(f"Error fetching NRFA time series for {gauge_id} ({data_type}): {e}")
+            return None
+
+    def _parse_data(self, gauge_id: str, raw_data: Optional[Dict[str, Any]], variable: str) -> pd.DataFrame:
+        """Parses the raw JSON time series data."""
+        if not raw_data or "data-stream" not in raw_data or not raw_data["data-stream"]:
+            logger.warning(f"No data stream found for {gauge_id}, variable {variable}")
+            return pd.DataFrame(columns=[constants.TIME_INDEX, variable])
+
+        try:
+            dates = raw_data["data-stream"][0::2]
+            values = raw_data["data-stream"][1::2]
+            df = pd.DataFrame.from_dict({"time": dates, variable: values})
+            df[constants.TIME_INDEX] = pd.to_datetime(df["time"], format="ISO8601").dt.date
+            df[constants.TIME_INDEX] = pd.to_datetime(df[constants.TIME_INDEX])
+            df[variable] = pd.to_numeric(df[variable], errors="coerce")
+            return df[[constants.TIME_INDEX, variable]].dropna().reset_index(drop=True)
+        except Exception as e:
+            logger.error(f"Error parsing NRFA data for {gauge_id}: {e}")
+            return pd.DataFrame(columns=[constants.TIME_INDEX, variable])
+
+    def get_data(
+        self,
+        gauge_id: str,
+        variable: str,
+        start_date: Optional[str] = None,
+        end_date: Optional[str] = None,
+    ) -> pd.DataFrame:
+        """Fetches and parses UK NRFA river gauge data."""
+        if variable not in self.get_available_variables():
+            raise ValueError(f"Unsupported variable: {variable}")
+
+        start_date = utils.format_start_date(start_date)
+        end_date = utils.format_end_date(end_date)
+
+        try:
+            raw_data = self._download_data(gauge_id, variable, start_date, end_date)
+            df = self._parse_data(gauge_id, raw_data, variable)
+
+            # Filter by date range
+            start_date_dt = pd.to_datetime(start_date)
+            end_date_dt = pd.to_datetime(end_date)
+            df = df[(df[constants.TIME_INDEX] >= start_date_dt) & (df[constants.TIME_INDEX] <= end_date_dt)]
+            return df
+        except Exception as e:
+            logger.error(f"Failed to get data for site {gauge_id}, variable {variable}: {e}")
+            return pd.DataFrame(columns=[constants.TIME_INDEX, variable])

tests/test_data/uk_nrfa_1001_discharge_20220101.json

@@ -0,0 +1,84 @@
+{
+  "timestamp": "2025-10-14T08:22:36",
+  "interval": "R31/2022-01-01/P1D",
+  "station": {
+    "id": 1001,
+    "name": "Wick at Tarroul",
+    "easting": 326202.0,
+    "northing": 954915.0,
+    "latitude": 58.4761957805334,
+    "longitude": -3.2670605529772914
+  },
+  "data-type": {
+    "id": "gdf",
+    "name": "Gauged Daily Flow",
+    "parameter": "Flow",
+    "units": "m3/s",
+    "measurement-type": "Mean",
+    "period": "P1D"
+  },
+  "data-stream": [
+    "2022-01-01",
+    1.552,
+    "2022-01-02",
+    1.461,
+    "2022-01-03",
+    2.035,
+    "2022-01-04",
+    7.232,
+    "2022-01-05",
+    6.539,
+    "2022-01-06",
+    3.901,
+    "2022-01-07",
+    2.886,
+    "2022-01-08",
+    3.556,
+    "2022-01-09",
+    2.618,
+    "2022-01-10",
+    2.336,
+    "2022-01-11",
+    2.173,
+    "2022-01-12",
+    1.919,
+    "2022-01-13",
+    1.694,
+    "2022-01-14",
+    2.088,
+    "2022-01-15",
+    2.138,
+    "2022-01-16",
+    2.185,
+    "2022-01-17",
+    1.378,
+    "2022-01-18",
+    1.355,
+    "2022-01-19",
+    2.313,
+    "2022-01-20",
+    1.624,
+    "2022-01-21",
+    1.316,
+    "2022-01-22",
+    1.177,
+    "2022-01-23",
+    1.054,
+    "2022-01-24",
+    1.06,
+    "2022-01-25",
+    1.159,
+    "2022-01-26",
+    1.573,
+    "2022-01-27",
+    1.211,
+    "2022-01-28",
+    1.642,
+    "2022-01-29",
+    1.999,
+    "2022-01-30",
+    1.982,
+    "2022-01-31",
+    3.905
+  ]
+}

tests/test_data/uk_nrfa_1001_precipitation_20220101.json

@@ -0,0 +1,84 @@
+{
+  "timestamp": "2025-10-14T17:58:16",
+  "interval": "R31/2022-01-01/P1D",
+  "station": {
+    "id": 1001,
+    "name": "Wick at Tarroul",
+    "easting": 326202.0,
+    "northing": 954915.0,
+    "latitude": 58.4761957805334,
+    "longitude": -3.2670605529772914
+  },
+  "data-type": {
+    "id": "cdr",
+    "name": "Catchment Daily Rainfall",
+    "parameter": "Rainfall",
+    "units": "mm",
+    "measurement-type": "Accumulation",
+    "period": "P1D"
+  },
+  "data-stream": [
+    "2022-01-01",
+    0.2,
+    "2022-01-02",
+    1.7,
+    "2022-01-03",
+    4.4,
+    "2022-01-04",
+    9.1,
+    "2022-01-05",
+    0.2,
+    "2022-01-06",
+    1,
+    "2022-01-07",
+    3.8,
+    "2022-01-08",
+    0.3,
+    "2022-01-09",
+    0.6,
+    "2022-01-10",
+    1,
+    "2022-01-11",
+    0.1,
+    "2022-01-12",
+    0,
+    "2022-01-13",
+    3.6,
+    "2022-01-14",
+    0.8,
+    "2022-01-15",
+    0.7,
+    "2022-01-16",
+    1,
+    "2022-01-17",
+    0,
+    "2022-01-18",
+    3.8,
+    "2022-01-19",
+    1,
+    "2022-01-20",
+    0.6,
+    "2022-01-21",
+    0,
+    "2022-01-22",
+    0,
+    "2022-01-23",
+    0.7,
+    "2022-01-24",
+    1.1,
+    "2022-01-25",
+    0.6,
+    "2022-01-26",
+    2.6,
+    "2022-01-27",
+    2.1,
+    "2022-01-28",
+    1.6,
+    "2022-01-29",
+    2.6,
+    "2022-01-30",
+    7.7,
+    "2022-01-31",
+    6.3
+  ]
+}