USGS Water — API reference

USGS NWIS / Water Data source subpackage — earthlens.usgs_water. Background, usage, available data, and credentials are covered under the other pages in this section; this page is the rendered API.

earthlens.usgs_water

USGS NWIS / Water Data backend.

Thin wrapper over the official dataretrieval SDK that pulls per-site water observations from the U.S. Geological Survey's National Water Information System — stream gauges, groundwater wells, and water-quality sites across the United States — and returns them as a long-format :class:pandas.DataFrame.

This is a tabular backend: the result is a table of per-site observations, not a gridded array, so :data:USGSWater.OUTPUT_KIND is "tabular" and the :class:earthlens.earthlens.EarthLens facade rejects an aggregate= argument for it (use service="statistics" for a server-side temporal rollup instead).

Request shape: variables is a list[str] of NWIS parameter codes or friendly names — variables=["00060"], variables=["discharge", "gage_height"] — resolved to 5-digit codes via the bundled catalog. The NWIS / Water Data service plane is selected by service= (default "daily"); the modern / legacy endpoint by api= (default "auto"). Authentication is an optional Personal Access Token.

Public surface (re-exported from this package):

• :class:USGSWater — the backend; instantiate with a date range, a bbox (or sites=), and variables=[code_or_name, ...], then call :meth:USGSWater.download.
• :class:Catalog — pydantic-backed loader for the bundled usgs_water_data_catalog.yaml parameter-code table.
• :class:Parameter — one parameter code's catalog row (code, name, units, group, services).
• :class:UsgsWaterAuth — AbstractAuth implementation resolving the optional API_USGS_PAT.
• :class:UsgsWaterCredentials — frozen pydantic value object the auth class binds to (the optional token).
• :data:CATALOG_PATH — path to the bundled parameter YAML.

Catalog

Bases: AbstractCatalog

Parameter-code catalog for the USGS Water backend.

Reads the bundled usgs_water_data_catalog.yaml (shipped as package data) and exposes its parameters: block as a map of :class:Parameter rows keyed by friendly name. Instantiate with no arguments (Catalog()). Resolve a name or raw code with :meth:resolve, or a single row with :meth:get_parameter.

Attributes:

Name	Type	Description
parameters	dict[str, Parameter]	Map from the friendly parameter name to its :class:Parameter row.

Examples:

• Resolve a friendly name and a raw code:

  >>> from earthlens.usgs_water import Catalog
  >>> cat = Catalog()
  >>> cat.resolve("discharge")
  '00060'
  >>> cat.resolve("00060")
  '00060'
  

• An unknown but close name raises with a did-you-mean hint:

  >>> from earthlens.usgs_water import Catalog
  >>> Catalog().resolve("dischrge")
  Traceback (most recent call last):
      ...
  ValueError: 'dischrge' is not in the USGS Water parameter catalog. Known parameters: [...]. Did you mean 'discharge'?
  

Source code in src/earthlens/usgs_water/catalog.py

class Catalog(AbstractCatalog):
    """Parameter-code catalog for the USGS Water backend.

    Reads the bundled `usgs_water_data_catalog.yaml` (shipped as
    package data) and exposes its `parameters:` block as a map of
    :class:`Parameter` rows keyed by friendly name. Instantiate with no
    arguments (`Catalog()`). Resolve a name or raw code with
    :meth:`resolve`, or a single row with :meth:`get_parameter`.

    Attributes:
        parameters: Map from the friendly parameter name to its
            :class:`Parameter` row.

    Examples:
        - Resolve a friendly name and a raw code:
            ```python
            >>> from earthlens.usgs_water import Catalog
            >>> cat = Catalog()
            >>> cat.resolve("discharge")
            '00060'
            >>> cat.resolve("00060")
            '00060'

            ```
        - An unknown but close name raises with a did-you-mean hint:
            ```python
            >>> from earthlens.usgs_water import Catalog
            >>> Catalog().resolve("dischrge")
            Traceback (most recent call last):
                ...
            ValueError: 'dischrge' is not in the USGS Water parameter catalog. Known parameters: [...]. Did you mean 'discharge'?

            ```
    """

    _catalog_kind: str = "USGS Water parameter catalog"
    _entry_noun: str = "parameters"

    #: The parameter rows live in the base :attr:`datasets` field so the
    #: inherited dict surface (`len`, `in`, `[]`, iteration) and
    #: :meth:`get_dataset`'s did-you-mean hint work unchanged. The field
    #: is narrowed here to :class:`Parameter` values; :attr:`parameters`
    #: is the domain-named read alias.
    datasets: dict[str, Parameter] = Field(default_factory=dict)

    @model_validator(mode="before")
    @classmethod
    def _accept_parameters_alias(cls, data: Any) -> Any:
        """Accept the legacy `parameters=` kwarg as an alias for `datasets`.

        Older callers (and tests) construct `Catalog(parameters={...})`.
        The rows now live in the base `datasets` field, so rewrite that
        key on the way in. An explicit `datasets=` always wins.

        Args:
            data: The raw model input (a mapping when constructed with
                keyword arguments).

        Returns:
            The input with `parameters` renamed to `datasets`, untouched
            otherwise.
        """
        if isinstance(data, dict) and "parameters" in data and "datasets" not in data:
            data = dict(data)
            data["datasets"] = data.pop("parameters")
        return data

    @property
    def parameters(self) -> dict[str, Parameter]:
        """The parameter map — alias for the base :attr:`datasets` field.

        Returns:
            dict[str, Parameter]: The same mapping stored in
                :attr:`datasets`.

        Examples:
            - The alias and the base field are the same object:
                ```python
                >>> from earthlens.usgs_water import Catalog
                >>> cat = Catalog()
                >>> cat.parameters is cat.datasets
                True
                >>> cat.parameters["discharge"].code
                '00060'

                ```
        """
        return self.datasets

    def model_post_init(self, __context: Any) -> None:
        """Auto-load the bundled catalog when no parameters were supplied.

        `Catalog()` with no args reads :data:`CATALOG_PATH`; passing
        `parameters=...` (or `datasets=...`) skips the disk read.

        Raises:
            ValueError: Propagated from :meth:`load` when the YAML is
                missing, empty, or has a malformed row.
        """
        if not self.datasets:
            self.datasets = Catalog.load().datasets
        super().model_post_init(__context)

    @classmethod
    def load(cls, catalog_path: Path | None = None) -> Catalog:
        """Read the USGS Water parameter catalog from disk.

        Args:
            catalog_path: Path to the catalog YAML. Defaults to the
                module-level :data:`CATALOG_PATH`.

        Returns:
            A fully-populated :class:`Catalog`.

        Raises:
            ValueError: If the file has no `parameters:` block, or a row
                fails :class:`Parameter` validation.
        """
        catalog_path = catalog_path if catalog_path is not None else CATALOG_PATH
        return cls(datasets=dict(_load_catalog_data(catalog_path)))

    def get_catalog(self) -> dict[str, Parameter]:
        """Return the parameter map (satisfies the abstract contract).

        Returns:
            dict[str, Parameter]: Same object as :attr:`datasets` /
                :attr:`parameters`.
        """
        return self.datasets

    @property
    def available_parameters(self) -> list[str]:
        """The sorted list of curated friendly parameter names.

        Returns:
            list[str]: Every curated catalog key, sorted.

        Examples:
            - List the curated names and check membership:
                ```python
                >>> from earthlens.usgs_water import Catalog
                >>> names = Catalog().available_parameters
                >>> "discharge" in names
                True
                >>> names == sorted(names)
                True

                ```
        """
        return sorted(self.datasets)

    def get_parameter(self, name: str) -> Parameter:
        """Resolve a friendly name to its :class:`Parameter` row.

        Thin wrapper over the inherited :meth:`get_dataset`, which raises
        a `ValueError` with a did-you-mean hint on an unknown name.

        Args:
            name: A friendly parameter name (`"discharge"`).

        Returns:
            Parameter: The matching catalog row.

        Raises:
            ValueError: If `name` is not a known parameter; the message
                names the catalog kind and, when a close match exists,
                adds a did-you-mean hint.

        Examples:
            - Resolve a row and read its fields:
                ```python
                >>> from earthlens.usgs_water import Catalog
                >>> p = Catalog().get_parameter("gage_height")
                >>> p.code
                '00065'
                >>> p.units
                'ft'

                ```
        """
        return self.get_dataset(name)

    def resolve(self, code_or_name: str) -> str:
        """Resolve a friendly name or a raw 5-digit code to a code.

        A raw 5-digit code (`"00060"`) passes through unmapped — the
        full USGS code table is far larger than the curated catalog, so
        any valid code is accepted directly. A non-code string is
        looked up as a friendly name via :meth:`get_parameter`.

        Args:
            code_or_name: A friendly name (`"discharge"`) or a raw
                5-digit NWIS code (`"00060"`).

        Returns:
            str: The 5-digit parameter code.

        Raises:
            ValueError: If `code_or_name` is neither a 5-digit code nor
                a known friendly name (with a did-you-mean hint).
        """
        if _CODE_RE.match(code_or_name):
            return code_or_name
        return self.get_parameter(code_or_name).code

available_parameters property

The sorted list of curated friendly parameter names.

Returns:

Type	Description
list[str]	list[str]: Every curated catalog key, sorted.

Examples:

• List the curated names and check membership:

  >>> from earthlens.usgs_water import Catalog
  >>> names = Catalog().available_parameters
  >>> "discharge" in names
  True
  >>> names == sorted(names)
  True
  

parameters property

The parameter map — alias for the base :attr:datasets field.

Returns:

Type	Description
dict[str, Parameter]	dict[str, Parameter]: The same mapping stored in :attr:datasets.

Examples:

• The alias and the base field are the same object:

  >>> from earthlens.usgs_water import Catalog
  >>> cat = Catalog()
  >>> cat.parameters is cat.datasets
  True
  >>> cat.parameters["discharge"].code
  '00060'
  

get_catalog()

Return the parameter map (satisfies the abstract contract).

Returns:

Type	Description
dict[str, Parameter]	dict[str, Parameter]: Same object as :attr:datasets / :attr:parameters.

Source code in src/earthlens/usgs_water/catalog.py

def get_catalog(self) -> dict[str, Parameter]:
    """Return the parameter map (satisfies the abstract contract).

    Returns:
        dict[str, Parameter]: Same object as :attr:`datasets` /
            :attr:`parameters`.
    """
    return self.datasets

get_parameter(name)

Resolve a friendly name to its :class:Parameter row.

Thin wrapper over the inherited :meth:get_dataset, which raises a ValueError with a did-you-mean hint on an unknown name.

Parameters:

Name	Type	Description	Default
name	str	A friendly parameter name ("discharge").	required

Returns:

Name	Type	Description
Parameter	Parameter	The matching catalog row.

Raises:

Type	Description
ValueError	If name is not a known parameter; the message names the catalog kind and, when a close match exists, adds a did-you-mean hint.

Examples:

• Resolve a row and read its fields:

  >>> from earthlens.usgs_water import Catalog
  >>> p = Catalog().get_parameter("gage_height")
  >>> p.code
  '00065'
  >>> p.units
  'ft'
  

Source code in src/earthlens/usgs_water/catalog.py

def get_parameter(self, name: str) -> Parameter:
    """Resolve a friendly name to its :class:`Parameter` row.

    Thin wrapper over the inherited :meth:`get_dataset`, which raises
    a `ValueError` with a did-you-mean hint on an unknown name.

    Args:
        name: A friendly parameter name (`"discharge"`).

    Returns:
        Parameter: The matching catalog row.

    Raises:
        ValueError: If `name` is not a known parameter; the message
            names the catalog kind and, when a close match exists,
            adds a did-you-mean hint.

    Examples:
        - Resolve a row and read its fields:
            ```python
            >>> from earthlens.usgs_water import Catalog
            >>> p = Catalog().get_parameter("gage_height")
            >>> p.code
            '00065'
            >>> p.units
            'ft'

            ```
    """
    return self.get_dataset(name)

load(catalog_path=None) classmethod

Read the USGS Water parameter catalog from disk.

Parameters:

Name	Type	Description	Default
catalog_path	Path | None	Path to the catalog YAML. Defaults to the module-level :data:CATALOG_PATH.	None

Returns:

Type	Description
Catalog	A fully-populated :class:Catalog.

Raises:

Type	Description
ValueError	If the file has no parameters: block, or a row fails :class:Parameter validation.

Source code in src/earthlens/usgs_water/catalog.py

@classmethod
def load(cls, catalog_path: Path | None = None) -> Catalog:
    """Read the USGS Water parameter catalog from disk.

    Args:
        catalog_path: Path to the catalog YAML. Defaults to the
            module-level :data:`CATALOG_PATH`.

    Returns:
        A fully-populated :class:`Catalog`.

    Raises:
        ValueError: If the file has no `parameters:` block, or a row
            fails :class:`Parameter` validation.
    """
    catalog_path = catalog_path if catalog_path is not None else CATALOG_PATH
    return cls(datasets=dict(_load_catalog_data(catalog_path)))

model_post_init(__context)

Auto-load the bundled catalog when no parameters were supplied.

Catalog() with no args reads :data:CATALOG_PATH; passing parameters=... (or datasets=...) skips the disk read.

Raises:

Type	Description
ValueError	Propagated from :meth:load when the YAML is missing, empty, or has a malformed row.

Source code in src/earthlens/usgs_water/catalog.py

def model_post_init(self, __context: Any) -> None:
    """Auto-load the bundled catalog when no parameters were supplied.

    `Catalog()` with no args reads :data:`CATALOG_PATH`; passing
    `parameters=...` (or `datasets=...`) skips the disk read.

    Raises:
        ValueError: Propagated from :meth:`load` when the YAML is
            missing, empty, or has a malformed row.
    """
    if not self.datasets:
        self.datasets = Catalog.load().datasets
    super().model_post_init(__context)

resolve(code_or_name)

Resolve a friendly name or a raw 5-digit code to a code.

A raw 5-digit code ("00060") passes through unmapped — the full USGS code table is far larger than the curated catalog, so any valid code is accepted directly. A non-code string is looked up as a friendly name via :meth:get_parameter.

Parameters:

Name	Type	Description	Default
code_or_name	str	A friendly name ("discharge") or a raw 5-digit NWIS code ("00060").	required

Returns:

Name	Type	Description
str	str	The 5-digit parameter code.

Raises:

Type	Description
ValueError	If code_or_name is neither a 5-digit code nor a known friendly name (with a did-you-mean hint).

Source code in src/earthlens/usgs_water/catalog.py

def resolve(self, code_or_name: str) -> str:
    """Resolve a friendly name or a raw 5-digit code to a code.

    A raw 5-digit code (`"00060"`) passes through unmapped — the
    full USGS code table is far larger than the curated catalog, so
    any valid code is accepted directly. A non-code string is
    looked up as a friendly name via :meth:`get_parameter`.

    Args:
        code_or_name: A friendly name (`"discharge"`) or a raw
            5-digit NWIS code (`"00060"`).

    Returns:
        str: The 5-digit parameter code.

    Raises:
        ValueError: If `code_or_name` is neither a 5-digit code nor
            a known friendly name (with a did-you-mean hint).
    """
    if _CODE_RE.match(code_or_name):
        return code_or_name
    return self.get_parameter(code_or_name).code

Parameter

Bases: BaseModel

One NWIS parameter code's catalog row.

The user-facing name is the parent key in :attr:Catalog.parameters and is also stored on the row as :attr:name so a resolved :class:Parameter is self-describing.

Attributes:

Name	Type	Description
code	str	The 5-digit NWIS parameter code ("00060").
name	str	Human-readable label ("Discharge").
units	str	The reporting units ("ft3/s", "degC").
group	ParameterGroup	Coarse USGS classification ("Physical", "Nutrients", …).
services	list[str]	The :data:earthlens.usgs_water.backend.SERVICES this code is typically available on (["daily", "instantaneous"]).

Examples:

• Build a row directly:

  >>> from earthlens.usgs_water import Parameter
  >>> p = Parameter(code="00060", name="Discharge", units="ft3/s")
  >>> p.code
  '00060'
  >>> p.group
  'Physical'
  

Source code in src/earthlens/usgs_water/catalog.py

class Parameter(BaseModel):
    """One NWIS parameter code's catalog row.

    The user-facing name is the parent key in
    :attr:`Catalog.parameters` and is also stored on the row as
    :attr:`name` so a resolved :class:`Parameter` is self-describing.

    Attributes:
        code: The 5-digit NWIS parameter code (`"00060"`).
        name: Human-readable label (`"Discharge"`).
        units: The reporting units (`"ft3/s"`, `"degC"`).
        group: Coarse USGS classification (`"Physical"`, `"Nutrients"`,
            …).
        services: The :data:`earthlens.usgs_water.backend.SERVICES`
            this code is typically available on (`["daily",
            "instantaneous"]`).

    Examples:
        - Build a row directly:
            ```python
            >>> from earthlens.usgs_water import Parameter
            >>> p = Parameter(code="00060", name="Discharge", units="ft3/s")
            >>> p.code
            '00060'
            >>> p.group
            'Physical'

            ```
    """

    model_config = ConfigDict(frozen=True, extra="forbid")

    code: str = Field(pattern=r"^\d{5}$")
    name: str = ""
    units: str = ""
    group: ParameterGroup = "Physical"
    services: list[str] = Field(default_factory=list)

USGSWater

Bases: AbstractDataSource

USGS NWIS / Water Data backend (long-format tabular output).

Fetches per-site water observations for a bbox / explicit sites / date window / parameter-code list through the same download() shape every other earthlens backend uses, and returns a long-format :class:pandas.DataFrame (one row per observation). The query is a search/fetch split: :meth:_search enumerates the monitoring locations to pull, and :meth:_fetch pulls each service's observations and normalises them to one tidy long schema.

The service= argument selects the NWIS / Water Data plane (see :data:SERVICES); api= selects the modern / legacy endpoint with a 429 auto-fallback (see the module docstring). Authentication is an optional Personal Access Token.

Attributes:

Name	Type	Description
OUTPUT_KIND	OutputKind	"tabular" — the result is per-row site observations, so the facade rejects aggregate= with NotImplementedError.

Source code in src/earthlens/usgs_water/backend.py

class USGSWater(AbstractDataSource):
    """USGS NWIS / Water Data backend (long-format tabular output).

    Fetches per-site water observations for a bbox / explicit sites /
    date window / parameter-code list through the same `download()`
    shape every other earthlens backend uses, and returns a
    long-format :class:`pandas.DataFrame` (one row per observation).
    The query is a search/fetch split: :meth:`_search` enumerates the
    monitoring locations to pull, and :meth:`_fetch` pulls each
    service's observations and normalises them to one tidy long schema.

    The `service=` argument selects the NWIS / Water Data plane (see
    :data:`SERVICES`); `api=` selects the modern / legacy endpoint with
    a 429 auto-fallback (see the module docstring). Authentication is
    an optional Personal Access Token.

    Attributes:
        OUTPUT_KIND: `"tabular"` — the result is per-row site
            observations, so the facade rejects `aggregate=` with
            `NotImplementedError`.
    """

    OUTPUT_KIND: OutputKind = "tabular"

    def __init__(
        self,
        start: str,
        end: str,
        variables: list[str],
        lat_lim: list[float],
        lon_lim: list[float],
        temporal_resolution: str = "daily",
        path: Path | str = "",
        fmt: str = "%Y-%m-%d",
        api_token: str | None = None,
        service: str = "daily",
        sites: list[str] | str | None = None,
        api: ApiFlavour = "auto",
        output_format: OutputFormat = "csv",
        stat_type: str = "daily",
        limit: int | None = None,
    ):
        """Initialise a USGS Water backend instance.

        Args:
            start: Inclusive start of the window, parsed with `fmt`.
            end: Inclusive end of the window.
            variables: List of NWIS parameter codes or friendly names
                (`["00060"]`, `["discharge", "gage_height"]`), resolved
                to 5-digit codes via the catalog. An empty list defaults
                to discharge (`["00060"]`). Ignored by the site-keyed
                services (`peaks`, `ratings`).
            lat_lim: `[lat_min, lat_max]` bounding-box latitudes.
            lon_lim: `[lon_min, lon_max]` bounding-box longitudes.
            temporal_resolution: Convenience alias mapped onto
                `service` when `service` is left at its default —
                `"daily"` keeps `daily`, a sub-daily value selects
                `instantaneous`. An explicit `service=` always wins.
            path: Output directory for the written table.
            fmt: `strptime` format for `start` / `end`.
            api_token: Optional USGS Personal Access Token; falls back
                to the `API_USGS_PAT` env var, then anonymous access.
            service: The NWIS / Water Data plane to query — one of
                :data:`SERVICES`. Defaults to `"daily"`.
            sites: Explicit USGS site number(s) to query, bypassing the
                bbox site discovery. **Required** for the services that
                have no spatial (bbox) filter — `peaks`, `ratings`, and
                `statistics`; a bbox-only request for one of those raises
                `ValueError`.
            api: Endpoint selector — `"auto"` (default; modern with a
                429 fallback to legacy), `"waterdata"` (force modern),
                or `"legacy"` (force the deprecated `nwis` endpoint).
            output_format: On-disk format — `"csv"` (default) or
                `"parquet"`.
            stat_type: For `service="statistics"` on the **legacy**
                endpoint (`api="legacy"`), the `get_stats` rollup period
                — `"daily"`, `"monthly"`, or `"annual"`. Ignored by the
                modern endpoint, whose `get_stats_date_range` returns its
                own intervals over the `start`/`end` window.
            limit: Optional cap on the rows pulled per request (passed
                through to the modern endpoint's `limit=`). `None`
                means the SDK default.

        Raises:
            ValueError: When `service`, `api`, or `output_format` is
                not a recognised value.
            TypeError: When `variables` is a mapping (this backend takes
                a flat list of parameter codes / names).
        """
        if isinstance(variables, dict):
            raise TypeError(
                "USGSWater `variables` must be a list of NWIS parameter "
                "codes or friendly names (e.g. ['00060', 'gage_height']), "
                "not a mapping. Query filters are explicit USGSWater(...) "
                "keyword arguments (service=, sites=, api=, ...)."
            )
        if service not in SERVICES:
            raise ValueError(
                f"service must be one of {list(SERVICES)}, got {service!r}."
            )
        if api not in API_FLAVOURS:
            raise ValueError(f"api must be one of {list(API_FLAVOURS)}, got {api!r}.")
        if output_format not in OUTPUT_FORMATS:
            raise ValueError(
                f"output_format must be one of {list(OUTPUT_FORMATS)}, "
                f"got {output_format!r}."
            )

        # `service` wins; otherwise honour the temporal_resolution alias —
        # but only for explicitly sub-daily tokens. Any other value
        # (e.g. "monthly", "yearly", "") leaves the default daily service
        # rather than silently selecting instantaneous.
        if service == "daily" and temporal_resolution.lower() in _SUBDAILY_RESOLUTIONS:
            service = "instantaneous"

        self._api_token = api_token
        self._service = service
        self._sites = [sites] if isinstance(sites, str) else sites
        self._api_flavour = api
        self._output_format: OutputFormat = output_format
        self._stat_type = stat_type
        self._limit = limit
        self._auth: UsgsWaterAuth | None = None
        self._catalog = Catalog()
        self._used_legacy_fallback = False
        super().__init__(
            start=start,
            end=end,
            variables=list(variables) or list(_DEFAULT_CODES),
            temporal_resolution=temporal_resolution,
            lat_lim=lat_lim,
            lon_lim=lon_lim,
            fmt=fmt,
            path=path,
        )

    def _initialize(self):
        """Build :class:`UsgsWaterAuth` and resolve the optional token.

        Returns `None` (the SDK has no global client object; the token,
        when present, is exported to `API_USGS_PAT` by the auth).
        """
        self._auth = UsgsWaterAuth(UsgsWaterCredentials(api_token=self._api_token))
        self._auth.configure()
        return None

    def _create_grid(self, lat_lim: list, lon_lim: list) -> SpatialExtent:
        """Wrap the WGS84 bbox into a :class:`SpatialExtent` (no snapping).

        Args:
            lat_lim: `[lat_min, lat_max]` in degrees.
            lon_lim: `[lon_min, lon_max]` in degrees.

        Returns:
            SpatialExtent: Validated, frozen bbox.
        """
        return SpatialExtent.from_pairs(lat_lim=lat_lim, lon_lim=lon_lim)

    def _check_input_dates(
        self,
        start: str,
        end: str,
        temporal_resolution: str,
        fmt: str,
    ) -> TemporalExtent:
        """Parse the `[start, end]` window into a :class:`TemporalExtent`.

        The whole window is fetched per site in one call (NWIS takes a
        start/end range), so there is no per-date loop; `dates`
        collapses to the two endpoints.

        Args:
            start: Inclusive start date string.
            end: Inclusive end date string.
            temporal_resolution: Recorded as the resolution label.
            fmt: `strptime` format applied to `start` and `end`.

        Returns:
            TemporalExtent: Frozen model with the parsed endpoints.

        Raises:
            ValueError: If `start` parses to a date later than `end`.
        """
        start_dt = dt.datetime.strptime(start, fmt)
        end_dt = dt.datetime.strptime(end, fmt)
        return TemporalExtent(
            start_date=start_dt,
            end_date=end_dt,
            resolution=temporal_resolution,
            dates=pd.DatetimeIndex([start_dt, end_dt]),
        )

    def _resolved_codes(self) -> list[str]:
        """Resolve `variables` to 5-digit NWIS parameter codes, order-stable.

        Returns:
            list[str]: The resolved codes (de-duplicated, first-wins).

        Raises:
            ValueError: If a name is neither a known catalog entry nor a
                raw 5-digit code (with a did-you-mean hint).
        """
        codes: list[str] = []
        for name in self.vars:
            code = self._catalog.resolve(name)
            if code not in codes:
                codes.append(code)
        return codes

    def _bbox_list(self) -> list[float]:
        """Return the request bbox as modern `[west, south, east, north]`.

        The legacy `"west,south,east,north"` string form is built inline by
        :func:`earthlens.usgs_water._helpers.query_kwargs` from this list.
        """
        return [self.space.west, self.space.south, self.space.east, self.space.north]

    def download(
        self,
        progress_bar: bool = True,
        aggregate: AggregationConfig | None = None,
    ) -> pd.DataFrame:
        """Fetch the selected service, write the table, and return it.

        Args:
            progress_bar: Accepted for signature parity with the other
                backends. USGS Water issues one bulk `dataretrieval`
                call per service rather than a per-item loop, so there
                is no progress bar to show — this is a no-op.
            aggregate: Must be `None`. USGS Water output is tabular, so
                there is no gridded reduction; the facade already
                rejects a non-`None` `aggregate=` for a `tabular`
                backend, and this is the belt-and-suspenders guard for
                direct callers. Use `service="statistics"` for a
                server-side temporal rollup instead.

        Returns:
            pd.DataFrame: The long-format observation table for the
                selected `service`.

        Raises:
            NotImplementedError: If `aggregate` is not `None` (tabular
                output has no gridded reduction; use
                `service="statistics"` instead).
            ValueError: If the selected `service` requires an explicit
                `sites=` (`peaks` / `ratings` / `statistics`) but none
                was supplied.
        """
        if aggregate is not None:
            raise NotImplementedError(
                "USGSWater.download(aggregate=...) is not supported: USGS "
                "water observations are tabular per-site rows, not gridded "
                "rasters, so there is no meaningful gridded reduction. Use "
                "service='statistics' for a server-side temporal rollup "
                "(daily/monthly/annual) instead."
            )
        # Each frame is already normalised to its service's schema (even
        # when empty), so concat all of them — preserving the right
        # columns for non-values services — rather than dropping empties.
        frames = self._api()
        df = (
            pd.concat(frames, ignore_index=True)
            if frames
            else _helpers.empty_canonical()
        )
        out_path = self._write_table(df)
        if len(df):
            logger.info(
                f"USGSWater {self._service}: {len(df)} row(s) written to {out_path}"
            )
        else:
            logger.warning(
                f"USGSWater {self._service}: no rows matched; wrote an empty "
                f"(schema-only) table to {out_path}"
            )
        return df

    def _api(self) -> list[pd.DataFrame]:
        """Compose `_search` and `_fetch` into the canonical C3 shape."""
        return self._api_via_search_fetch()

    def _search(self) -> list[RemoteProduct]:
        """Enumerate the products to fetch (one per request, pre-C8).

        The C3 search is a single product carrying the resolved
        parameter codes and any explicit `sites=`; the real bbox site
        discovery (one product per monitoring location) lands in C8.

        Returns:
            list[RemoteProduct]: One product whose `metadata` holds the
                resolved `codes` and the explicit `sites` (or `None`).

        Raises:
            ValueError: When the service requires an explicit `sites=`
                (no bbox filter is available for it) but none was given.
        """
        if self._service in _SITES_REQUIRED_SERVICES and not self._sites:
            raise ValueError(
                f"service={self._service!r} requires an explicit sites= "
                f"(no spatial bbox filter is available for it); pass "
                f"sites=[...] (e.g. sites='01646500')."
            )
        codes = [] if self._service in _SITE_KEYED_SERVICES else self._resolved_codes()
        return [
            RemoteProduct(
                id=self._service,
                metadata={"codes": codes, "sites": self._sites},
            )
        ]

    def _fetch(self, products: list[RemoteProduct]) -> list[pd.DataFrame]:
        """Query each product's service and normalise to the long schema.

        Widens the inherited `-> list[Path]` contract: a tabular
        backend returns in-memory long-format frames, not file paths
        (the write happens in :meth:`download` via :meth:`_write_table`).

        Args:
            products: The list returned by :meth:`_search`.

        Returns:
            list[pd.DataFrame]: One canonical long-schema frame per
                product, same order.
        """
        return [self._fetch_one(product) for product in products]

    def _fetch_one(self, product: RemoteProduct) -> pd.DataFrame:
        """Fetch one product's service frame and normalise it.

        Args:
            product: One :class:`RemoteProduct` from :meth:`_search`.

        Returns:
            pd.DataFrame: The canonical long-schema frame for this
                product (empty when the service returned nothing).
        """
        codes = product.metadata.get("codes", [])
        sites = product.metadata.get("sites")
        df, flavour = self._call_with_fallback(codes, sites)
        return _helpers.normalize(df, flavour, self._service, self._code_meta(codes))

    def _module(self, flavour: str):
        """Lazily import the `dataretrieval` submodule for a flavour.

        Args:
            flavour: `"waterdata"` (modern) or `"nwis"` (legacy).

        Returns:
            The imported `dataretrieval.waterdata` / `dataretrieval.nwis`
                module.

        Raises:
            ImportError: When `dataretrieval` is not installed (names
                the `earthlens[usgs-water]` extra).
        """
        _import_dataretrieval()
        if flavour == "waterdata":
            import dataretrieval.waterdata as mod
        else:
            import dataretrieval.nwis as mod
        return mod

    def _invoke(
        self, flavour: str, codes: list[str], sites: list[str] | None
    ) -> pd.DataFrame:
        """Call the resolved service function and return its raw frame.

        Args:
            flavour: `"waterdata"` or `"nwis"`.
            codes: Resolved parameter codes.
            sites: Explicit site numbers, or `None` for a bbox query.

        Returns:
            pd.DataFrame: The frame returned by the SDK call (the first
                element when the SDK returns a `(frame, metadata)`
                tuple).
        """
        module = self._module(flavour)
        function = getattr(module, _helpers.service_function(self._service, flavour))
        kwargs = _helpers.query_kwargs(
            service=self._service,
            flavour=flavour,
            codes=codes,
            sites=sites,
            bbox=self._bbox_list(),
            start=self.time.start_date.strftime("%Y-%m-%d"),
            end=self.time.end_date.strftime("%Y-%m-%d"),
            limit=self._limit,
            stat_type=self._stat_type,
        )
        result = function(**kwargs)
        return result[0] if isinstance(result, tuple) else result

    def _call_with_fallback(
        self, codes: list[str], sites: list[str] | None
    ) -> tuple[pd.DataFrame, str]:
        """Invoke the service, applying the modern→legacy fallbacks.

        Two fallbacks fold in here (both honouring `api=`): a service
        the modern endpoint can only query by site (e.g. instantaneous,
        which has no `bbox`) routes to legacy when only a bbox is given;
        and a modern HTTP 429 (anonymous rate-limit) under `api="auto"`
        retries on legacy.

        Args:
            codes: Resolved parameter codes.
            sites: Explicit site numbers, or `None` for a bbox query.

        Returns:
            tuple[pd.DataFrame, str]: The raw frame and the flavour
                (`"waterdata"` / `"nwis"`) that produced it.

        Raises:
            ValueError: When `api="waterdata"` is forced for a
                bbox-only query the modern endpoint cannot serve.
        """
        has_legacy = _helpers.service_function(self._service, "nwis") is not None
        use_legacy = self._api_flavour == "legacy"
        if use_legacy and not has_legacy:
            raise ValueError(
                f"service {self._service!r} has no legacy endpoint (it is "
                f"modern-only in dataretrieval); use api='auto' or "
                f"api='waterdata'."
            )

        bbox_only = not sites
        if (
            not use_legacy
            and bbox_only
            and not _helpers.modern_supports_bbox(self._service)
        ):
            if self._api_flavour == "waterdata" or not has_legacy:
                raise ValueError(
                    f"The modern endpoint cannot query service "
                    f"{self._service!r} by bbox (no bbox filter). Pass "
                    f"sites=[...] explicitly, or use api='auto'/'legacy'."
                )
            use_legacy = True  # api="auto": legacy supports bBox

        if use_legacy:
            return self._invoke("nwis", codes, sites), "nwis"
        try:
            return self._invoke("waterdata", codes, sites), "waterdata"
        except Exception as exc:  # noqa: BLE001 - re-raised unless a 429 fallback
            anonymous = self._auth is None or not self._auth.is_authenticated()
            if (
                self._api_flavour == "auto"
                and anonymous
                and _helpers.is_rate_limit_error(exc)
            ):
                if not has_legacy:
                    raise RuntimeError(
                        f"The modern USGS endpoint rate-limited this anonymous "
                        f"request (HTTP 429) and service {self._service!r} has "
                        f"no legacy fallback. Set API_USGS_PAT (or pass "
                        f"api_token=) to use the modern endpoint."
                    ) from exc
                self._warn_legacy_fallback()
                return self._invoke("nwis", codes, sites), "nwis"
            raise

    def _warn_legacy_fallback(self) -> None:
        """Log a one-time warning when falling back to the legacy endpoint."""
        if self._used_legacy_fallback:
            return
        self._used_legacy_fallback = True
        logger.warning(
            "USGS modern endpoint rate-limited this anonymous request "
            "(HTTP 429); falling back to the legacy waterservices.usgs.gov "
            "endpoint. Set API_USGS_PAT (or pass api_token=) to use the "
            "modern endpoint without throttling."
        )

    def _code_meta(self, codes: list[str]) -> dict[str, tuple[str, str]]:
        """Build a `{code: (name, units)}` map from the catalog.

        Args:
            codes: Resolved 5-digit parameter codes.

        Returns:
            dict[str, tuple[str, str]]: Friendly name + units per code
                that the catalog curates; uncurated codes map to
                `("", "")`.
        """
        by_code = {p.code: (p.name, p.units) for p in self._catalog.parameters.values()}
        return {code: by_code.get(code, ("", "")) for code in codes}

    def _write_table(self, df: pd.DataFrame) -> Path:
        """Write the long-format table to `root_dir` and return the path.

        Args:
            df: The canonical long-format frame.

        Returns:
            Path: The written CSV / Parquet file path.
        """
        codes = [] if self._service in _SITE_KEYED_SERVICES else self._resolved_codes()
        codes_part = "_".join(codes or ["all"])
        ext = "parquet" if self._output_format == "parquet" else "csv"
        out_path = self.root_dir / f"usgs_{self._service}_{codes_part}.{ext}"
        if self._output_format == "parquet":
            try:
                df.to_parquet(out_path, index=False)
            except ImportError as exc:  # pragma: no cover - depends on env
                raise ImportError(
                    "Writing Parquet requires 'pyarrow'. Install it (pip "
                    "install pyarrow) or use output_format='csv'."
                ) from exc
        else:
            df.to_csv(out_path, index=False)
        return out_path

__init__(start, end, variables, lat_lim, lon_lim, temporal_resolution='daily', path='', fmt='%Y-%m-%d', api_token=None, service='daily', sites=None, api='auto', output_format='csv', stat_type='daily', limit=None)

Initialise a USGS Water backend instance.

Parameters:

Name	Type	Description	Default
start	str	Inclusive start of the window, parsed with fmt.	required
end	str	Inclusive end of the window.	required
variables	list[str]	List of NWIS parameter codes or friendly names (["00060"], ["discharge", "gage_height"]), resolved to 5-digit codes via the catalog. An empty list defaults to discharge (["00060"]). Ignored by the site-keyed services (peaks, ratings).	required
lat_lim	list[float]	[lat_min, lat_max] bounding-box latitudes.	required
lon_lim	list[float]	[lon_min, lon_max] bounding-box longitudes.	required
temporal_resolution	str	Convenience alias mapped onto service when service is left at its default — "daily" keeps daily, a sub-daily value selects instantaneous. An explicit service= always wins.	'daily'
path	Path | str	Output directory for the written table.	''
fmt	str	strptime format for start / end.	'%Y-%m-%d'
api_token	str | None	Optional USGS Personal Access Token; falls back to the API_USGS_PAT env var, then anonymous access.	None
service	str	The NWIS / Water Data plane to query — one of :data:SERVICES. Defaults to "daily".	'daily'
sites	list[str] | str | None	Explicit USGS site number(s) to query, bypassing the bbox site discovery. Required for the services that have no spatial (bbox) filter — peaks, ratings, and statistics; a bbox-only request for one of those raises ValueError.	None
api	ApiFlavour	Endpoint selector — "auto" (default; modern with a 429 fallback to legacy), "waterdata" (force modern), or "legacy" (force the deprecated nwis endpoint).	'auto'
output_format	OutputFormat	On-disk format — "csv" (default) or "parquet".	'csv'
stat_type	str	For service="statistics" on the legacy endpoint (api="legacy"), the get_stats rollup period — "daily", "monthly", or "annual". Ignored by the modern endpoint, whose get_stats_date_range returns its own intervals over the start/end window.	'daily'
limit	int | None	Optional cap on the rows pulled per request (passed through to the modern endpoint's limit=). None means the SDK default.	None

Raises:

Type	Description
ValueError	When service, api, or output_format is not a recognised value.
TypeError	When variables is a mapping (this backend takes a flat list of parameter codes / names).

Source code in src/earthlens/usgs_water/backend.py

def __init__(
    self,
    start: str,
    end: str,
    variables: list[str],
    lat_lim: list[float],
    lon_lim: list[float],
    temporal_resolution: str = "daily",
    path: Path | str = "",
    fmt: str = "%Y-%m-%d",
    api_token: str | None = None,
    service: str = "daily",
    sites: list[str] | str | None = None,
    api: ApiFlavour = "auto",
    output_format: OutputFormat = "csv",
    stat_type: str = "daily",
    limit: int | None = None,
):
    """Initialise a USGS Water backend instance.

    Args:
        start: Inclusive start of the window, parsed with `fmt`.
        end: Inclusive end of the window.
        variables: List of NWIS parameter codes or friendly names
            (`["00060"]`, `["discharge", "gage_height"]`), resolved
            to 5-digit codes via the catalog. An empty list defaults
            to discharge (`["00060"]`). Ignored by the site-keyed
            services (`peaks`, `ratings`).
        lat_lim: `[lat_min, lat_max]` bounding-box latitudes.
        lon_lim: `[lon_min, lon_max]` bounding-box longitudes.
        temporal_resolution: Convenience alias mapped onto
            `service` when `service` is left at its default —
            `"daily"` keeps `daily`, a sub-daily value selects
            `instantaneous`. An explicit `service=` always wins.
        path: Output directory for the written table.
        fmt: `strptime` format for `start` / `end`.
        api_token: Optional USGS Personal Access Token; falls back
            to the `API_USGS_PAT` env var, then anonymous access.
        service: The NWIS / Water Data plane to query — one of
            :data:`SERVICES`. Defaults to `"daily"`.
        sites: Explicit USGS site number(s) to query, bypassing the
            bbox site discovery. **Required** for the services that
            have no spatial (bbox) filter — `peaks`, `ratings`, and
            `statistics`; a bbox-only request for one of those raises
            `ValueError`.
        api: Endpoint selector — `"auto"` (default; modern with a
            429 fallback to legacy), `"waterdata"` (force modern),
            or `"legacy"` (force the deprecated `nwis` endpoint).
        output_format: On-disk format — `"csv"` (default) or
            `"parquet"`.
        stat_type: For `service="statistics"` on the **legacy**
            endpoint (`api="legacy"`), the `get_stats` rollup period
            — `"daily"`, `"monthly"`, or `"annual"`. Ignored by the
            modern endpoint, whose `get_stats_date_range` returns its
            own intervals over the `start`/`end` window.
        limit: Optional cap on the rows pulled per request (passed
            through to the modern endpoint's `limit=`). `None`
            means the SDK default.

    Raises:
        ValueError: When `service`, `api`, or `output_format` is
            not a recognised value.
        TypeError: When `variables` is a mapping (this backend takes
            a flat list of parameter codes / names).
    """
    if isinstance(variables, dict):
        raise TypeError(
            "USGSWater `variables` must be a list of NWIS parameter "
            "codes or friendly names (e.g. ['00060', 'gage_height']), "
            "not a mapping. Query filters are explicit USGSWater(...) "
            "keyword arguments (service=, sites=, api=, ...)."
        )
    if service not in SERVICES:
        raise ValueError(
            f"service must be one of {list(SERVICES)}, got {service!r}."
        )
    if api not in API_FLAVOURS:
        raise ValueError(f"api must be one of {list(API_FLAVOURS)}, got {api!r}.")
    if output_format not in OUTPUT_FORMATS:
        raise ValueError(
            f"output_format must be one of {list(OUTPUT_FORMATS)}, "
            f"got {output_format!r}."
        )

    # `service` wins; otherwise honour the temporal_resolution alias —
    # but only for explicitly sub-daily tokens. Any other value
    # (e.g. "monthly", "yearly", "") leaves the default daily service
    # rather than silently selecting instantaneous.
    if service == "daily" and temporal_resolution.lower() in _SUBDAILY_RESOLUTIONS:
        service = "instantaneous"

    self._api_token = api_token
    self._service = service
    self._sites = [sites] if isinstance(sites, str) else sites
    self._api_flavour = api
    self._output_format: OutputFormat = output_format
    self._stat_type = stat_type
    self._limit = limit
    self._auth: UsgsWaterAuth | None = None
    self._catalog = Catalog()
    self._used_legacy_fallback = False
    super().__init__(
        start=start,
        end=end,
        variables=list(variables) or list(_DEFAULT_CODES),
        temporal_resolution=temporal_resolution,
        lat_lim=lat_lim,
        lon_lim=lon_lim,
        fmt=fmt,
        path=path,
    )

download(progress_bar=True, aggregate=None)

Fetch the selected service, write the table, and return it.

Parameters:

Name	Type	Description	Default
progress_bar	bool	Accepted for signature parity with the other backends. USGS Water issues one bulk dataretrieval call per service rather than a per-item loop, so there is no progress bar to show — this is a no-op.	True
aggregate	AggregationConfig | None	Must be None. USGS Water output is tabular, so there is no gridded reduction; the facade already rejects a non-None aggregate= for a tabular backend, and this is the belt-and-suspenders guard for direct callers. Use service="statistics" for a server-side temporal rollup instead.	None

Returns:

Type	Description
DataFrame	pd.DataFrame: The long-format observation table for the selected service.

Raises:

Type	Description
NotImplementedError	If aggregate is not None (tabular output has no gridded reduction; use service="statistics" instead).
ValueError	If the selected service requires an explicit sites= (peaks / ratings / statistics) but none was supplied.

Source code in src/earthlens/usgs_water/backend.py

def download(
    self,
    progress_bar: bool = True,
    aggregate: AggregationConfig | None = None,
) -> pd.DataFrame:
    """Fetch the selected service, write the table, and return it.

    Args:
        progress_bar: Accepted for signature parity with the other
            backends. USGS Water issues one bulk `dataretrieval`
            call per service rather than a per-item loop, so there
            is no progress bar to show — this is a no-op.
        aggregate: Must be `None`. USGS Water output is tabular, so
            there is no gridded reduction; the facade already
            rejects a non-`None` `aggregate=` for a `tabular`
            backend, and this is the belt-and-suspenders guard for
            direct callers. Use `service="statistics"` for a
            server-side temporal rollup instead.

    Returns:
        pd.DataFrame: The long-format observation table for the
            selected `service`.

    Raises:
        NotImplementedError: If `aggregate` is not `None` (tabular
            output has no gridded reduction; use
            `service="statistics"` instead).
        ValueError: If the selected `service` requires an explicit
            `sites=` (`peaks` / `ratings` / `statistics`) but none
            was supplied.
    """
    if aggregate is not None:
        raise NotImplementedError(
            "USGSWater.download(aggregate=...) is not supported: USGS "
            "water observations are tabular per-site rows, not gridded "
            "rasters, so there is no meaningful gridded reduction. Use "
            "service='statistics' for a server-side temporal rollup "
            "(daily/monthly/annual) instead."
        )
    # Each frame is already normalised to its service's schema (even
    # when empty), so concat all of them — preserving the right
    # columns for non-values services — rather than dropping empties.
    frames = self._api()
    df = (
        pd.concat(frames, ignore_index=True)
        if frames
        else _helpers.empty_canonical()
    )
    out_path = self._write_table(df)
    if len(df):
        logger.info(
            f"USGSWater {self._service}: {len(df)} row(s) written to {out_path}"
        )
    else:
        logger.warning(
            f"USGSWater {self._service}: no rows matched; wrote an empty "
            f"(schema-only) table to {out_path}"
        )
    return df

UsgsWaterAuth

Bases: AbstractAuth[UsgsWaterCredentials]

Resolve and hold the optional USGS Personal Access Token.

Implements the :class:earthlens.base.AbstractAuth contract for an optional-secret backend. Construction does not touch the environment; :meth:configure performs the resolution and is idempotent. After configure(), the token (or None for anonymous) is available via the :attr:token property, and — when a token was resolved — it is exported to the API_USGS_PAT environment variable so the dataretrieval SDK picks it up.

The class is a context manager (inherited from :class:AbstractAuth): with UsgsWaterAuth(creds) as auth: ... calls configure() on enter and the default no-op close() on exit.

Attributes:

Name	Type	Description
_creds		The :class:UsgsWaterCredentials passed at construction.

Examples:

• Resolve an explicit token:

  >>> from earthlens.usgs_water import UsgsWaterAuth, UsgsWaterCredentials
  >>> auth = UsgsWaterAuth(UsgsWaterCredentials(api_token="k"))
  >>> auth.is_authenticated()
  False
  >>> auth.configure()
  >>> auth.is_authenticated()
  True
  >>> auth.token
  'k'
  

• With no token, configure() succeeds and stays anonymous:

  >>> import os
  >>> from earthlens.usgs_water import UsgsWaterAuth, UsgsWaterCredentials
  >>> os.environ.pop("API_USGS_PAT", None) and None
  >>> auth = UsgsWaterAuth(UsgsWaterCredentials())
  >>> auth.configure()
  >>> auth.is_authenticated()
  False
  >>> auth.token is None
  True
  

Source code in src/earthlens/usgs_water/auth.py

class UsgsWaterAuth(AbstractAuth[UsgsWaterCredentials]):
    """Resolve and hold the optional USGS Personal Access Token.

    Implements the :class:`earthlens.base.AbstractAuth` contract for an
    **optional**-secret backend. Construction does not touch the
    environment; :meth:`configure` performs the resolution and is
    idempotent. After `configure()`, the token (or `None` for
    anonymous) is available via the :attr:`token` property, and — when
    a token was resolved — it is exported to the `API_USGS_PAT`
    environment variable so the `dataretrieval` SDK picks it up.

    The class is a context manager (inherited from
    :class:`AbstractAuth`): `with UsgsWaterAuth(creds) as auth: ...`
    calls `configure()` on enter and the default no-op `close()` on
    exit.

    Attributes:
        _creds: The :class:`UsgsWaterCredentials` passed at
            construction.

    Examples:
        - Resolve an explicit token:
            ```python
            >>> from earthlens.usgs_water import UsgsWaterAuth, UsgsWaterCredentials
            >>> auth = UsgsWaterAuth(UsgsWaterCredentials(api_token="k"))
            >>> auth.is_authenticated()
            False
            >>> auth.configure()
            >>> auth.is_authenticated()
            True
            >>> auth.token
            'k'

            ```
        - With no token, `configure()` succeeds and stays anonymous:
            ```python
            >>> import os
            >>> from earthlens.usgs_water import UsgsWaterAuth, UsgsWaterCredentials
            >>> os.environ.pop("API_USGS_PAT", None) and None
            >>> auth = UsgsWaterAuth(UsgsWaterCredentials())
            >>> auth.configure()
            >>> auth.is_authenticated()
            False
            >>> auth.token is None
            True

            ```
    """

    def __init__(self, credentials: UsgsWaterCredentials) -> None:
        """Store credentials; does not resolve the token yet.

        Args:
            credentials: The :class:`UsgsWaterCredentials` value object
                carrying the optional token.
        """
        super().__init__(credentials)
        self._configured = False
        self._token: str | None = None

    def configure(self) -> None:
        """Resolve the optional token and export it to the environment.

        Idempotent — short-circuits when :meth:`is_authenticated`
        already returns `True`. Resolves the token in this order: the
        explicit `api_token` on the credentials, then the
        `API_USGS_PAT` environment variable. When a token is found it
        is written back to `API_USGS_PAT` (the channel `dataretrieval`
        reads) and :meth:`is_authenticated` flips to `True`. When none
        is found this is a **no-op** — the request runs anonymously
        (rate-limited) and no error is raised.
        """
        if self.is_authenticated():
            return
        token = (
            self._creds.api_token.get_secret_value()
            if self._creds.api_token is not None
            else os.environ.get(_TOKEN_ENV_VAR)
        )
        if token:
            os.environ[_TOKEN_ENV_VAR] = token
            self._token = token
            self._configured = True

    def is_authenticated(self) -> bool:
        """Return `True` when a token was resolved (token-backed mode).

        Cheap predicate — does not call the network. `False` is a
        legitimate, non-error state meaning "anonymous access".

        Returns:
            bool: `True` after :meth:`configure` resolved a token,
                `False` when anonymous (no token).
        """
        return self._configured

    @property
    def token(self) -> str | None:
        """The resolved PAT, or `None` for anonymous access.

        Valid after :meth:`configure`; returns `None` both before
        configuration and when no token was available (anonymous).

        Returns:
            str | None: The USGS PAT string, or `None` when anonymous.
        """
        return self._token

token property

The resolved PAT, or None for anonymous access.

Valid after :meth:configure; returns None both before configuration and when no token was available (anonymous).

Returns:

Type	Description
str | None	str | None: The USGS PAT string, or None when anonymous.

__init__(credentials)

Store credentials; does not resolve the token yet.

Parameters:

Name	Type	Description	Default
credentials	UsgsWaterCredentials	The :class:UsgsWaterCredentials value object carrying the optional token.	required

Source code in src/earthlens/usgs_water/auth.py

def __init__(self, credentials: UsgsWaterCredentials) -> None:
    """Store credentials; does not resolve the token yet.

    Args:
        credentials: The :class:`UsgsWaterCredentials` value object
            carrying the optional token.
    """
    super().__init__(credentials)
    self._configured = False
    self._token: str | None = None

configure()

Resolve the optional token and export it to the environment.

Idempotent — short-circuits when :meth:is_authenticated already returns True. Resolves the token in this order: the explicit api_token on the credentials, then the API_USGS_PAT environment variable. When a token is found it is written back to API_USGS_PAT (the channel dataretrieval reads) and :meth:is_authenticated flips to True. When none is found this is a no-op — the request runs anonymously (rate-limited) and no error is raised.

Source code in src/earthlens/usgs_water/auth.py

def configure(self) -> None:
    """Resolve the optional token and export it to the environment.

    Idempotent — short-circuits when :meth:`is_authenticated`
    already returns `True`. Resolves the token in this order: the
    explicit `api_token` on the credentials, then the
    `API_USGS_PAT` environment variable. When a token is found it
    is written back to `API_USGS_PAT` (the channel `dataretrieval`
    reads) and :meth:`is_authenticated` flips to `True`. When none
    is found this is a **no-op** — the request runs anonymously
    (rate-limited) and no error is raised.
    """
    if self.is_authenticated():
        return
    token = (
        self._creds.api_token.get_secret_value()
        if self._creds.api_token is not None
        else os.environ.get(_TOKEN_ENV_VAR)
    )
    if token:
        os.environ[_TOKEN_ENV_VAR] = token
        self._token = token
        self._configured = True

is_authenticated()

Return True when a token was resolved (token-backed mode).

Cheap predicate — does not call the network. False is a legitimate, non-error state meaning "anonymous access".

Returns:

Name	Type	Description
bool	bool	True after :meth:configure resolved a token, False when anonymous (no token).

Source code in src/earthlens/usgs_water/auth.py

def is_authenticated(self) -> bool:
    """Return `True` when a token was resolved (token-backed mode).

    Cheap predicate — does not call the network. `False` is a
    legitimate, non-error state meaning "anonymous access".

    Returns:
        bool: `True` after :meth:`configure` resolved a token,
            `False` when anonymous (no token).
    """
    return self._configured

UsgsWaterCredentials

Bases: BaseModel

Frozen value object holding the optional USGS PAT.

The token is optional at construction time: None means "resolve from the API_USGS_PAT environment variable at :meth:UsgsWaterAuth.configure time, and if that is also unset, run anonymously".

Attributes:

Name	Type	Description
api_token	SecretStr | None	The USGS Personal Access Token, stored as a :class:pydantic.SecretStr so it is never echoed by repr(creds) or in logs. None defers resolution to the environment variable (and then to anonymous access).

Examples:

• Build from an explicit token; the secret is hidden in repr:

  >>> from earthlens.usgs_water import UsgsWaterCredentials
  >>> creds = UsgsWaterCredentials(api_token="topsecret")
  >>> creds.api_token.get_secret_value()
  'topsecret'
  >>> "topsecret" in repr(creds)
  False
  

• The token is optional — anonymous access is the default:

  >>> from earthlens.usgs_water import UsgsWaterCredentials
  >>> UsgsWaterCredentials().api_token is None
  True
  

Source code in src/earthlens/usgs_water/auth.py

class UsgsWaterCredentials(BaseModel):
    """Frozen value object holding the optional USGS PAT.

    The token is optional at construction time: `None` means "resolve
    from the `API_USGS_PAT` environment variable at
    :meth:`UsgsWaterAuth.configure` time, and if that is also unset,
    run anonymously".

    Attributes:
        api_token: The USGS Personal Access Token, stored as a
            :class:`pydantic.SecretStr` so it is never echoed by
            `repr(creds)` or in logs. `None` defers resolution to the
            environment variable (and then to anonymous access).

    Examples:
        - Build from an explicit token; the secret is hidden in `repr`:
            ```python
            >>> from earthlens.usgs_water import UsgsWaterCredentials
            >>> creds = UsgsWaterCredentials(api_token="topsecret")
            >>> creds.api_token.get_secret_value()
            'topsecret'
            >>> "topsecret" in repr(creds)
            False

            ```
        - The token is optional — anonymous access is the default:
            ```python
            >>> from earthlens.usgs_water import UsgsWaterCredentials
            >>> UsgsWaterCredentials().api_token is None
            True

            ```
    """

    model_config = ConfigDict(frozen=True)

    api_token: SecretStr | None = None

clear_catalog_cache()

Empty the module-level catalog parse cache.

Useful in tests that rewrite the catalog on disk and want to force a re-parse. Production callers do not need this — the cache key includes the file's st_mtime_ns, so any real edit invalidates the entry on its own.

Source code in src/earthlens/usgs_water/catalog.py

def clear_catalog_cache() -> None:
    """Empty the module-level catalog parse cache.

    Useful in tests that rewrite the catalog on disk and want to force
    a re-parse. Production callers do not need this — the cache key
    includes the file's `st_mtime_ns`, so any real edit invalidates the
    entry on its own.
    """
    _CATALOG_CACHE.clear()

earthlens.usgs_water.backend

Backend that fetches USGS water data from NWIS / the USGS Water Data API.

USGSWater(AbstractDataSource) wraps the official dataretrieval SDK to pull time-series and discrete water observations from the U.S. Geological Survey's National Water Information System — ~10,000 active stream gauges and many more groundwater / water-quality sites across the United States. A request is a bbox (or explicit sites=) + a time window + a list of NWIS parameter codes (["00060"] discharge, ["00065"] gage height, …); the backend returns a per-site time-series as a long-format :class:pandas.DataFrame, so OUTPUT_KIND = "tabular" and the :class:earthlens.earthlens.EarthLens facade rejects an aggregate= argument (use the server-side service="statistics" rollup instead).

The full NWIS / Water Data service surface is selectable via a service= keyword argument (default "daily"): daily, instantaneous, samples, statistics, gwlevels, field-measurements, peaks, ratings, and sites. The USGS is mid-migration from the legacy waterservices.usgs.gov endpoint (the dataretrieval.nwis module) to the modern api.waterdata.usgs.gov endpoint (the dataretrieval.waterdata module). The api= keyword selects which:

• "auto" (default) — try the modern endpoint, but because it rate-limits anonymous access aggressively (HTTP 429), transparently fall back to the legacy endpoint on a 429 when no token is set.
• "waterdata" — force the modern endpoint (a 429 surfaces as an error).
• "legacy" — force the legacy endpoint.

Authentication is an optional Personal Access Token (the API_USGS_PAT env var or the api_token= argument); anonymous access works at lower rate limits. See :class:earthlens.usgs_water.auth.

USGSWater

Bases: AbstractDataSource

USGS NWIS / Water Data backend (long-format tabular output).

Fetches per-site water observations for a bbox / explicit sites / date window / parameter-code list through the same download() shape every other earthlens backend uses, and returns a long-format :class:pandas.DataFrame (one row per observation). The query is a search/fetch split: :meth:_search enumerates the monitoring locations to pull, and :meth:_fetch pulls each service's observations and normalises them to one tidy long schema.

The service= argument selects the NWIS / Water Data plane (see :data:SERVICES); api= selects the modern / legacy endpoint with a 429 auto-fallback (see the module docstring). Authentication is an optional Personal Access Token.

Attributes:

Name	Type	Description
OUTPUT_KIND	OutputKind	"tabular" — the result is per-row site observations, so the facade rejects aggregate= with NotImplementedError.

Source code in src/earthlens/usgs_water/backend.py

class USGSWater(AbstractDataSource):
    """USGS NWIS / Water Data backend (long-format tabular output).

    Fetches per-site water observations for a bbox / explicit sites /
    date window / parameter-code list through the same `download()`
    shape every other earthlens backend uses, and returns a
    long-format :class:`pandas.DataFrame` (one row per observation).
    The query is a search/fetch split: :meth:`_search` enumerates the
    monitoring locations to pull, and :meth:`_fetch` pulls each
    service's observations and normalises them to one tidy long schema.

    The `service=` argument selects the NWIS / Water Data plane (see
    :data:`SERVICES`); `api=` selects the modern / legacy endpoint with
    a 429 auto-fallback (see the module docstring). Authentication is
    an optional Personal Access Token.

    Attributes:
        OUTPUT_KIND: `"tabular"` — the result is per-row site
            observations, so the facade rejects `aggregate=` with
            `NotImplementedError`.
    """

    OUTPUT_KIND: OutputKind = "tabular"

    def __init__(
        self,
        start: str,
        end: str,
        variables: list[str],
        lat_lim: list[float],
        lon_lim: list[float],
        temporal_resolution: str = "daily",
        path: Path | str = "",
        fmt: str = "%Y-%m-%d",
        api_token: str | None = None,
        service: str = "daily",
        sites: list[str] | str | None = None,
        api: ApiFlavour = "auto",
        output_format: OutputFormat = "csv",
        stat_type: str = "daily",
        limit: int | None = None,
    ):
        """Initialise a USGS Water backend instance.

        Args:
            start: Inclusive start of the window, parsed with `fmt`.
            end: Inclusive end of the window.
            variables: List of NWIS parameter codes or friendly names
                (`["00060"]`, `["discharge", "gage_height"]`), resolved
                to 5-digit codes via the catalog. An empty list defaults
                to discharge (`["00060"]`). Ignored by the site-keyed
                services (`peaks`, `ratings`).
            lat_lim: `[lat_min, lat_max]` bounding-box latitudes.
            lon_lim: `[lon_min, lon_max]` bounding-box longitudes.
            temporal_resolution: Convenience alias mapped onto
                `service` when `service` is left at its default —
                `"daily"` keeps `daily`, a sub-daily value selects
                `instantaneous`. An explicit `service=` always wins.
            path: Output directory for the written table.
            fmt: `strptime` format for `start` / `end`.
            api_token: Optional USGS Personal Access Token; falls back
                to the `API_USGS_PAT` env var, then anonymous access.
            service: The NWIS / Water Data plane to query — one of
                :data:`SERVICES`. Defaults to `"daily"`.
            sites: Explicit USGS site number(s) to query, bypassing the
                bbox site discovery. **Required** for the services that
                have no spatial (bbox) filter — `peaks`, `ratings`, and
                `statistics`; a bbox-only request for one of those raises
                `ValueError`.
            api: Endpoint selector — `"auto"` (default; modern with a
                429 fallback to legacy), `"waterdata"` (force modern),
                or `"legacy"` (force the deprecated `nwis` endpoint).
            output_format: On-disk format — `"csv"` (default) or
                `"parquet"`.
            stat_type: For `service="statistics"` on the **legacy**
                endpoint (`api="legacy"`), the `get_stats` rollup period
                — `"daily"`, `"monthly"`, or `"annual"`. Ignored by the
                modern endpoint, whose `get_stats_date_range` returns its
                own intervals over the `start`/`end` window.
            limit: Optional cap on the rows pulled per request (passed
                through to the modern endpoint's `limit=`). `None`
                means the SDK default.

        Raises:
            ValueError: When `service`, `api`, or `output_format` is
                not a recognised value.
            TypeError: When `variables` is a mapping (this backend takes
                a flat list of parameter codes / names).
        """
        if isinstance(variables, dict):
            raise TypeError(
                "USGSWater `variables` must be a list of NWIS parameter "
                "codes or friendly names (e.g. ['00060', 'gage_height']), "
                "not a mapping. Query filters are explicit USGSWater(...) "
                "keyword arguments (service=, sites=, api=, ...)."
            )
        if service not in SERVICES:
            raise ValueError(
                f"service must be one of {list(SERVICES)}, got {service!r}."
            )
        if api not in API_FLAVOURS:
            raise ValueError(f"api must be one of {list(API_FLAVOURS)}, got {api!r}.")
        if output_format not in OUTPUT_FORMATS:
            raise ValueError(
                f"output_format must be one of {list(OUTPUT_FORMATS)}, "
                f"got {output_format!r}."
            )

        # `service` wins; otherwise honour the temporal_resolution alias —
        # but only for explicitly sub-daily tokens. Any other value
        # (e.g. "monthly", "yearly", "") leaves the default daily service
        # rather than silently selecting instantaneous.
        if service == "daily" and temporal_resolution.lower() in _SUBDAILY_RESOLUTIONS:
            service = "instantaneous"

        self._api_token = api_token
        self._service = service
        self._sites = [sites] if isinstance(sites, str) else sites
        self._api_flavour = api
        self._output_format: OutputFormat = output_format
        self._stat_type = stat_type
        self._limit = limit
        self._auth: UsgsWaterAuth | None = None
        self._catalog = Catalog()
        self._used_legacy_fallback = False
        super().__init__(
            start=start,
            end=end,
            variables=list(variables) or list(_DEFAULT_CODES),
            temporal_resolution=temporal_resolution,
            lat_lim=lat_lim,
            lon_lim=lon_lim,
            fmt=fmt,
            path=path,
        )

    def _initialize(self):
        """Build :class:`UsgsWaterAuth` and resolve the optional token.

        Returns `None` (the SDK has no global client object; the token,
        when present, is exported to `API_USGS_PAT` by the auth).
        """
        self._auth = UsgsWaterAuth(UsgsWaterCredentials(api_token=self._api_token))
        self._auth.configure()
        return None

    def _create_grid(self, lat_lim: list, lon_lim: list) -> SpatialExtent:
        """Wrap the WGS84 bbox into a :class:`SpatialExtent` (no snapping).

        Args:
            lat_lim: `[lat_min, lat_max]` in degrees.
            lon_lim: `[lon_min, lon_max]` in degrees.

        Returns:
            SpatialExtent: Validated, frozen bbox.
        """
        return SpatialExtent.from_pairs(lat_lim=lat_lim, lon_lim=lon_lim)

    def _check_input_dates(
        self,
        start: str,
        end: str,
        temporal_resolution: str,
        fmt: str,
    ) -> TemporalExtent:
        """Parse the `[start, end]` window into a :class:`TemporalExtent`.

        The whole window is fetched per site in one call (NWIS takes a
        start/end range), so there is no per-date loop; `dates`
        collapses to the two endpoints.

        Args:
            start: Inclusive start date string.
            end: Inclusive end date string.
            temporal_resolution: Recorded as the resolution label.
            fmt: `strptime` format applied to `start` and `end`.

        Returns:
            TemporalExtent: Frozen model with the parsed endpoints.

        Raises:
            ValueError: If `start` parses to a date later than `end`.
        """
        start_dt = dt.datetime.strptime(start, fmt)
        end_dt = dt.datetime.strptime(end, fmt)
        return TemporalExtent(
            start_date=start_dt,
            end_date=end_dt,
            resolution=temporal_resolution,
            dates=pd.DatetimeIndex([start_dt, end_dt]),
        )

    def _resolved_codes(self) -> list[str]:
        """Resolve `variables` to 5-digit NWIS parameter codes, order-stable.

        Returns:
            list[str]: The resolved codes (de-duplicated, first-wins).

        Raises:
            ValueError: If a name is neither a known catalog entry nor a
                raw 5-digit code (with a did-you-mean hint).
        """
        codes: list[str] = []
        for name in self.vars:
            code = self._catalog.resolve(name)
            if code not in codes:
                codes.append(code)
        return codes

    def _bbox_list(self) -> list[float]:
        """Return the request bbox as modern `[west, south, east, north]`.

        The legacy `"west,south,east,north"` string form is built inline by
        :func:`earthlens.usgs_water._helpers.query_kwargs` from this list.
        """
        return [self.space.west, self.space.south, self.space.east, self.space.north]

    def download(
        self,
        progress_bar: bool = True,
        aggregate: AggregationConfig | None = None,
    ) -> pd.DataFrame:
        """Fetch the selected service, write the table, and return it.

        Args:
            progress_bar: Accepted for signature parity with the other
                backends. USGS Water issues one bulk `dataretrieval`
                call per service rather than a per-item loop, so there
                is no progress bar to show — this is a no-op.
            aggregate: Must be `None`. USGS Water output is tabular, so
                there is no gridded reduction; the facade already
                rejects a non-`None` `aggregate=` for a `tabular`
                backend, and this is the belt-and-suspenders guard for
                direct callers. Use `service="statistics"` for a
                server-side temporal rollup instead.

        Returns:
            pd.DataFrame: The long-format observation table for the
                selected `service`.

        Raises:
            NotImplementedError: If `aggregate` is not `None` (tabular
                output has no gridded reduction; use
                `service="statistics"` instead).
            ValueError: If the selected `service` requires an explicit
                `sites=` (`peaks` / `ratings` / `statistics`) but none
                was supplied.
        """
        if aggregate is not None:
            raise NotImplementedError(
                "USGSWater.download(aggregate=...) is not supported: USGS "
                "water observations are tabular per-site rows, not gridded "
                "rasters, so there is no meaningful gridded reduction. Use "
                "service='statistics' for a server-side temporal rollup "
                "(daily/monthly/annual) instead."
            )
        # Each frame is already normalised to its service's schema (even
        # when empty), so concat all of them — preserving the right
        # columns for non-values services — rather than dropping empties.
        frames = self._api()
        df = (
            pd.concat(frames, ignore_index=True)
            if frames
            else _helpers.empty_canonical()
        )
        out_path = self._write_table(df)
        if len(df):
            logger.info(
                f"USGSWater {self._service}: {len(df)} row(s) written to {out_path}"
            )
        else:
            logger.warning(
                f"USGSWater {self._service}: no rows matched; wrote an empty "
                f"(schema-only) table to {out_path}"
            )
        return df

    def _api(self) -> list[pd.DataFrame]:
        """Compose `_search` and `_fetch` into the canonical C3 shape."""
        return self._api_via_search_fetch()

    def _search(self) -> list[RemoteProduct]:
        """Enumerate the products to fetch (one per request, pre-C8).

        The C3 search is a single product carrying the resolved
        parameter codes and any explicit `sites=`; the real bbox site
        discovery (one product per monitoring location) lands in C8.

        Returns:
            list[RemoteProduct]: One product whose `metadata` holds the
                resolved `codes` and the explicit `sites` (or `None`).

        Raises:
            ValueError: When the service requires an explicit `sites=`
                (no bbox filter is available for it) but none was given.
        """
        if self._service in _SITES_REQUIRED_SERVICES and not self._sites:
            raise ValueError(
                f"service={self._service!r} requires an explicit sites= "
                f"(no spatial bbox filter is available for it); pass "
                f"sites=[...] (e.g. sites='01646500')."
            )
        codes = [] if self._service in _SITE_KEYED_SERVICES else self._resolved_codes()
        return [
            RemoteProduct(
                id=self._service,
                metadata={"codes": codes, "sites": self._sites},
            )
        ]

    def _fetch(self, products: list[RemoteProduct]) -> list[pd.DataFrame]:
        """Query each product's service and normalise to the long schema.

        Widens the inherited `-> list[Path]` contract: a tabular
        backend returns in-memory long-format frames, not file paths
        (the write happens in :meth:`download` via :meth:`_write_table`).

        Args:
            products: The list returned by :meth:`_search`.

        Returns:
            list[pd.DataFrame]: One canonical long-schema frame per
                product, same order.
        """
        return [self._fetch_one(product) for product in products]

    def _fetch_one(self, product: RemoteProduct) -> pd.DataFrame:
        """Fetch one product's service frame and normalise it.

        Args:
            product: One :class:`RemoteProduct` from :meth:`_search`.

        Returns:
            pd.DataFrame: The canonical long-schema frame for this
                product (empty when the service returned nothing).
        """
        codes = product.metadata.get("codes", [])
        sites = product.metadata.get("sites")
        df, flavour = self._call_with_fallback(codes, sites)
        return _helpers.normalize(df, flavour, self._service, self._code_meta(codes))

    def _module(self, flavour: str):
        """Lazily import the `dataretrieval` submodule for a flavour.

        Args:
            flavour: `"waterdata"` (modern) or `"nwis"` (legacy).

        Returns:
            The imported `dataretrieval.waterdata` / `dataretrieval.nwis`
                module.

        Raises:
            ImportError: When `dataretrieval` is not installed (names
                the `earthlens[usgs-water]` extra).
        """
        _import_dataretrieval()
        if flavour == "waterdata":
            import dataretrieval.waterdata as mod
        else:
            import dataretrieval.nwis as mod
        return mod

    def _invoke(
        self, flavour: str, codes: list[str], sites: list[str] | None
    ) -> pd.DataFrame:
        """Call the resolved service function and return its raw frame.

        Args:
            flavour: `"waterdata"` or `"nwis"`.
            codes: Resolved parameter codes.
            sites: Explicit site numbers, or `None` for a bbox query.

        Returns:
            pd.DataFrame: The frame returned by the SDK call (the first
                element when the SDK returns a `(frame, metadata)`
                tuple).
        """
        module = self._module(flavour)
        function = getattr(module, _helpers.service_function(self._service, flavour))
        kwargs = _helpers.query_kwargs(
            service=self._service,
            flavour=flavour,
            codes=codes,
            sites=sites,
            bbox=self._bbox_list(),
            start=self.time.start_date.strftime("%Y-%m-%d"),
            end=self.time.end_date.strftime("%Y-%m-%d"),
            limit=self._limit,
            stat_type=self._stat_type,
        )
        result = function(**kwargs)
        return result[0] if isinstance(result, tuple) else result

    def _call_with_fallback(
        self, codes: list[str], sites: list[str] | None
    ) -> tuple[pd.DataFrame, str]:
        """Invoke the service, applying the modern→legacy fallbacks.

        Two fallbacks fold in here (both honouring `api=`): a service
        the modern endpoint can only query by site (e.g. instantaneous,
        which has no `bbox`) routes to legacy when only a bbox is given;
        and a modern HTTP 429 (anonymous rate-limit) under `api="auto"`
        retries on legacy.

        Args:
            codes: Resolved parameter codes.
            sites: Explicit site numbers, or `None` for a bbox query.

        Returns:
            tuple[pd.DataFrame, str]: The raw frame and the flavour
                (`"waterdata"` / `"nwis"`) that produced it.

        Raises:
            ValueError: When `api="waterdata"` is forced for a
                bbox-only query the modern endpoint cannot serve.
        """
        has_legacy = _helpers.service_function(self._service, "nwis") is not None
        use_legacy = self._api_flavour == "legacy"
        if use_legacy and not has_legacy:
            raise ValueError(
                f"service {self._service!r} has no legacy endpoint (it is "
                f"modern-only in dataretrieval); use api='auto' or "
                f"api='waterdata'."
            )

        bbox_only = not sites
        if (
            not use_legacy
            and bbox_only
            and not _helpers.modern_supports_bbox(self._service)
        ):
            if self._api_flavour == "waterdata" or not has_legacy:
                raise ValueError(
                    f"The modern endpoint cannot query service "
                    f"{self._service!r} by bbox (no bbox filter). Pass "
                    f"sites=[...] explicitly, or use api='auto'/'legacy'."
                )
            use_legacy = True  # api="auto": legacy supports bBox

        if use_legacy:
            return self._invoke("nwis", codes, sites), "nwis"
        try:
            return self._invoke("waterdata", codes, sites), "waterdata"
        except Exception as exc:  # noqa: BLE001 - re-raised unless a 429 fallback
            anonymous = self._auth is None or not self._auth.is_authenticated()
            if (
                self._api_flavour == "auto"
                and anonymous
                and _helpers.is_rate_limit_error(exc)
            ):
                if not has_legacy:
                    raise RuntimeError(
                        f"The modern USGS endpoint rate-limited this anonymous "
                        f"request (HTTP 429) and service {self._service!r} has "
                        f"no legacy fallback. Set API_USGS_PAT (or pass "
                        f"api_token=) to use the modern endpoint."
                    ) from exc
                self._warn_legacy_fallback()
                return self._invoke("nwis", codes, sites), "nwis"
            raise

    def _warn_legacy_fallback(self) -> None:
        """Log a one-time warning when falling back to the legacy endpoint."""
        if self._used_legacy_fallback:
            return
        self._used_legacy_fallback = True
        logger.warning(
            "USGS modern endpoint rate-limited this anonymous request "
            "(HTTP 429); falling back to the legacy waterservices.usgs.gov "
            "endpoint. Set API_USGS_PAT (or pass api_token=) to use the "
            "modern endpoint without throttling."
        )

    def _code_meta(self, codes: list[str]) -> dict[str, tuple[str, str]]:
        """Build a `{code: (name, units)}` map from the catalog.

        Args:
            codes: Resolved 5-digit parameter codes.

        Returns:
            dict[str, tuple[str, str]]: Friendly name + units per code
                that the catalog curates; uncurated codes map to
                `("", "")`.
        """
        by_code = {p.code: (p.name, p.units) for p in self._catalog.parameters.values()}
        return {code: by_code.get(code, ("", "")) for code in codes}

    def _write_table(self, df: pd.DataFrame) -> Path:
        """Write the long-format table to `root_dir` and return the path.

        Args:
            df: The canonical long-format frame.

        Returns:
            Path: The written CSV / Parquet file path.
        """
        codes = [] if self._service in _SITE_KEYED_SERVICES else self._resolved_codes()
        codes_part = "_".join(codes or ["all"])
        ext = "parquet" if self._output_format == "parquet" else "csv"
        out_path = self.root_dir / f"usgs_{self._service}_{codes_part}.{ext}"
        if self._output_format == "parquet":
            try:
                df.to_parquet(out_path, index=False)
            except ImportError as exc:  # pragma: no cover - depends on env
                raise ImportError(
                    "Writing Parquet requires 'pyarrow'. Install it (pip "
                    "install pyarrow) or use output_format='csv'."
                ) from exc
        else:
            df.to_csv(out_path, index=False)
        return out_path

__init__(start, end, variables, lat_lim, lon_lim, temporal_resolution='daily', path='', fmt='%Y-%m-%d', api_token=None, service='daily', sites=None, api='auto', output_format='csv', stat_type='daily', limit=None)

Initialise a USGS Water backend instance.

Parameters:

Name	Type	Description	Default
start	str	Inclusive start of the window, parsed with fmt.	required
end	str	Inclusive end of the window.	required
variables	list[str]	List of NWIS parameter codes or friendly names (["00060"], ["discharge", "gage_height"]), resolved to 5-digit codes via the catalog. An empty list defaults to discharge (["00060"]). Ignored by the site-keyed services (peaks, ratings).	required
lat_lim	list[float]	[lat_min, lat_max] bounding-box latitudes.	required
lon_lim	list[float]	[lon_min, lon_max] bounding-box longitudes.	required
temporal_resolution	str	Convenience alias mapped onto service when service is left at its default — "daily" keeps daily, a sub-daily value selects instantaneous. An explicit service= always wins.	'daily'
path	Path | str	Output directory for the written table.	''
fmt	str	strptime format for start / end.	'%Y-%m-%d'
api_token	str | None	Optional USGS Personal Access Token; falls back to the API_USGS_PAT env var, then anonymous access.	None
service	str	The NWIS / Water Data plane to query — one of :data:SERVICES. Defaults to "daily".	'daily'
sites	list[str] | str | None	Explicit USGS site number(s) to query, bypassing the bbox site discovery. Required for the services that have no spatial (bbox) filter — peaks, ratings, and statistics; a bbox-only request for one of those raises ValueError.	None
api	ApiFlavour	Endpoint selector — "auto" (default; modern with a 429 fallback to legacy), "waterdata" (force modern), or "legacy" (force the deprecated nwis endpoint).	'auto'
output_format	OutputFormat	On-disk format — "csv" (default) or "parquet".	'csv'
stat_type	str	For service="statistics" on the legacy endpoint (api="legacy"), the get_stats rollup period — "daily", "monthly", or "annual". Ignored by the modern endpoint, whose get_stats_date_range returns its own intervals over the start/end window.	'daily'
limit	int | None	Optional cap on the rows pulled per request (passed through to the modern endpoint's limit=). None means the SDK default.	None

Raises:

Type	Description
ValueError	When service, api, or output_format is not a recognised value.
TypeError	When variables is a mapping (this backend takes a flat list of parameter codes / names).

Source code in src/earthlens/usgs_water/backend.py

def __init__(
    self,
    start: str,
    end: str,
    variables: list[str],
    lat_lim: list[float],
    lon_lim: list[float],
    temporal_resolution: str = "daily",
    path: Path | str = "",
    fmt: str = "%Y-%m-%d",
    api_token: str | None = None,
    service: str = "daily",
    sites: list[str] | str | None = None,
    api: ApiFlavour = "auto",
    output_format: OutputFormat = "csv",
    stat_type: str = "daily",
    limit: int | None = None,
):
    """Initialise a USGS Water backend instance.

    Args:
        start: Inclusive start of the window, parsed with `fmt`.
        end: Inclusive end of the window.
        variables: List of NWIS parameter codes or friendly names
            (`["00060"]`, `["discharge", "gage_height"]`), resolved
            to 5-digit codes via the catalog. An empty list defaults
            to discharge (`["00060"]`). Ignored by the site-keyed
            services (`peaks`, `ratings`).
        lat_lim: `[lat_min, lat_max]` bounding-box latitudes.
        lon_lim: `[lon_min, lon_max]` bounding-box longitudes.
        temporal_resolution: Convenience alias mapped onto
            `service` when `service` is left at its default —
            `"daily"` keeps `daily`, a sub-daily value selects
            `instantaneous`. An explicit `service=` always wins.
        path: Output directory for the written table.
        fmt: `strptime` format for `start` / `end`.
        api_token: Optional USGS Personal Access Token; falls back
            to the `API_USGS_PAT` env var, then anonymous access.
        service: The NWIS / Water Data plane to query — one of
            :data:`SERVICES`. Defaults to `"daily"`.
        sites: Explicit USGS site number(s) to query, bypassing the
            bbox site discovery. **Required** for the services that
            have no spatial (bbox) filter — `peaks`, `ratings`, and
            `statistics`; a bbox-only request for one of those raises
            `ValueError`.
        api: Endpoint selector — `"auto"` (default; modern with a
            429 fallback to legacy), `"waterdata"` (force modern),
            or `"legacy"` (force the deprecated `nwis` endpoint).
        output_format: On-disk format — `"csv"` (default) or
            `"parquet"`.
        stat_type: For `service="statistics"` on the **legacy**
            endpoint (`api="legacy"`), the `get_stats` rollup period
            — `"daily"`, `"monthly"`, or `"annual"`. Ignored by the
            modern endpoint, whose `get_stats_date_range` returns its
            own intervals over the `start`/`end` window.
        limit: Optional cap on the rows pulled per request (passed
            through to the modern endpoint's `limit=`). `None`
            means the SDK default.

    Raises:
        ValueError: When `service`, `api`, or `output_format` is
            not a recognised value.
        TypeError: When `variables` is a mapping (this backend takes
            a flat list of parameter codes / names).
    """
    if isinstance(variables, dict):
        raise TypeError(
            "USGSWater `variables` must be a list of NWIS parameter "
            "codes or friendly names (e.g. ['00060', 'gage_height']), "
            "not a mapping. Query filters are explicit USGSWater(...) "
            "keyword arguments (service=, sites=, api=, ...)."
        )
    if service not in SERVICES:
        raise ValueError(
            f"service must be one of {list(SERVICES)}, got {service!r}."
        )
    if api not in API_FLAVOURS:
        raise ValueError(f"api must be one of {list(API_FLAVOURS)}, got {api!r}.")
    if output_format not in OUTPUT_FORMATS:
        raise ValueError(
            f"output_format must be one of {list(OUTPUT_FORMATS)}, "
            f"got {output_format!r}."
        )

    # `service` wins; otherwise honour the temporal_resolution alias —
    # but only for explicitly sub-daily tokens. Any other value
    # (e.g. "monthly", "yearly", "") leaves the default daily service
    # rather than silently selecting instantaneous.
    if service == "daily" and temporal_resolution.lower() in _SUBDAILY_RESOLUTIONS:
        service = "instantaneous"

    self._api_token = api_token
    self._service = service
    self._sites = [sites] if isinstance(sites, str) else sites
    self._api_flavour = api
    self._output_format: OutputFormat = output_format
    self._stat_type = stat_type
    self._limit = limit
    self._auth: UsgsWaterAuth | None = None
    self._catalog = Catalog()
    self._used_legacy_fallback = False
    super().__init__(
        start=start,
        end=end,
        variables=list(variables) or list(_DEFAULT_CODES),
        temporal_resolution=temporal_resolution,
        lat_lim=lat_lim,
        lon_lim=lon_lim,
        fmt=fmt,
        path=path,
    )

download(progress_bar=True, aggregate=None)

Fetch the selected service, write the table, and return it.

Parameters:

Name	Type	Description	Default
progress_bar	bool	Accepted for signature parity with the other backends. USGS Water issues one bulk dataretrieval call per service rather than a per-item loop, so there is no progress bar to show — this is a no-op.	True
aggregate	AggregationConfig | None	Must be None. USGS Water output is tabular, so there is no gridded reduction; the facade already rejects a non-None aggregate= for a tabular backend, and this is the belt-and-suspenders guard for direct callers. Use service="statistics" for a server-side temporal rollup instead.	None

Returns:

Type	Description
DataFrame	pd.DataFrame: The long-format observation table for the selected service.

Raises:

Type	Description
NotImplementedError	If aggregate is not None (tabular output has no gridded reduction; use service="statistics" instead).
ValueError	If the selected service requires an explicit sites= (peaks / ratings / statistics) but none was supplied.

Source code in src/earthlens/usgs_water/backend.py

def download(
    self,
    progress_bar: bool = True,
    aggregate: AggregationConfig | None = None,
) -> pd.DataFrame:
    """Fetch the selected service, write the table, and return it.

    Args:
        progress_bar: Accepted for signature parity with the other
            backends. USGS Water issues one bulk `dataretrieval`
            call per service rather than a per-item loop, so there
            is no progress bar to show — this is a no-op.
        aggregate: Must be `None`. USGS Water output is tabular, so
            there is no gridded reduction; the facade already
            rejects a non-`None` `aggregate=` for a `tabular`
            backend, and this is the belt-and-suspenders guard for
            direct callers. Use `service="statistics"` for a
            server-side temporal rollup instead.

    Returns:
        pd.DataFrame: The long-format observation table for the
            selected `service`.

    Raises:
        NotImplementedError: If `aggregate` is not `None` (tabular
            output has no gridded reduction; use
            `service="statistics"` instead).
        ValueError: If the selected `service` requires an explicit
            `sites=` (`peaks` / `ratings` / `statistics`) but none
            was supplied.
    """
    if aggregate is not None:
        raise NotImplementedError(
            "USGSWater.download(aggregate=...) is not supported: USGS "
            "water observations are tabular per-site rows, not gridded "
            "rasters, so there is no meaningful gridded reduction. Use "
            "service='statistics' for a server-side temporal rollup "
            "(daily/monthly/annual) instead."
        )
    # Each frame is already normalised to its service's schema (even
    # when empty), so concat all of them — preserving the right
    # columns for non-values services — rather than dropping empties.
    frames = self._api()
    df = (
        pd.concat(frames, ignore_index=True)
        if frames
        else _helpers.empty_canonical()
    )
    out_path = self._write_table(df)
    if len(df):
        logger.info(
            f"USGSWater {self._service}: {len(df)} row(s) written to {out_path}"
        )
    else:
        logger.warning(
            f"USGSWater {self._service}: no rows matched; wrote an empty "
            f"(schema-only) table to {out_path}"
        )
    return df

earthlens.usgs_water.catalog

Parameter-code catalog for the USGS Water backend.

USGS NWIS filters its queries by 5-digit parameter codes (00060 discharge, 00065 gage height, 00010 water temperature, …), but users think in names. This module is the name-to-code bridge plus light metadata (units, display name, parameter group, and which services the code is valid on).

:class:Catalog is a thin :class:earthlens.base.AbstractCatalog subclass that loads the bundled usgs_water_data_catalog.yaml and exposes each row as a :class:Parameter. Resolve a single key with :meth:Catalog.resolve — which accepts either a friendly name ("discharge") or a raw 5-digit code ("00060", passed through unmapped), with a did-you-mean hint on an unknown name. The full ~25k USGS parameter-code table is not hand-curated here; the curated rows cover the common codes, and the refresh tool builds an informational available_parameters index from the live USGS reference table.

:data:CATALOG_PATH is the path to the bundled YAML.

Catalog

Bases: AbstractCatalog

Parameter-code catalog for the USGS Water backend.

Reads the bundled usgs_water_data_catalog.yaml (shipped as package data) and exposes its parameters: block as a map of :class:Parameter rows keyed by friendly name. Instantiate with no arguments (Catalog()). Resolve a name or raw code with :meth:resolve, or a single row with :meth:get_parameter.

Attributes:

Name	Type	Description
parameters	dict[str, Parameter]	Map from the friendly parameter name to its :class:Parameter row.

Examples:

• Resolve a friendly name and a raw code:

  >>> from earthlens.usgs_water import Catalog
  >>> cat = Catalog()
  >>> cat.resolve("discharge")
  '00060'
  >>> cat.resolve("00060")
  '00060'
  

• An unknown but close name raises with a did-you-mean hint:

  >>> from earthlens.usgs_water import Catalog
  >>> Catalog().resolve("dischrge")
  Traceback (most recent call last):
      ...
  ValueError: 'dischrge' is not in the USGS Water parameter catalog. Known parameters: [...]. Did you mean 'discharge'?
  

Source code in src/earthlens/usgs_water/catalog.py

class Catalog(AbstractCatalog):
    """Parameter-code catalog for the USGS Water backend.

    Reads the bundled `usgs_water_data_catalog.yaml` (shipped as
    package data) and exposes its `parameters:` block as a map of
    :class:`Parameter` rows keyed by friendly name. Instantiate with no
    arguments (`Catalog()`). Resolve a name or raw code with
    :meth:`resolve`, or a single row with :meth:`get_parameter`.

    Attributes:
        parameters: Map from the friendly parameter name to its
            :class:`Parameter` row.

    Examples:
        - Resolve a friendly name and a raw code:
            ```python
            >>> from earthlens.usgs_water import Catalog
            >>> cat = Catalog()
            >>> cat.resolve("discharge")
            '00060'
            >>> cat.resolve("00060")
            '00060'

            ```
        - An unknown but close name raises with a did-you-mean hint:
            ```python
            >>> from earthlens.usgs_water import Catalog
            >>> Catalog().resolve("dischrge")
            Traceback (most recent call last):
                ...
            ValueError: 'dischrge' is not in the USGS Water parameter catalog. Known parameters: [...]. Did you mean 'discharge'?

            ```
    """

    _catalog_kind: str = "USGS Water parameter catalog"
    _entry_noun: str = "parameters"

    #: The parameter rows live in the base :attr:`datasets` field so the
    #: inherited dict surface (`len`, `in`, `[]`, iteration) and
    #: :meth:`get_dataset`'s did-you-mean hint work unchanged. The field
    #: is narrowed here to :class:`Parameter` values; :attr:`parameters`
    #: is the domain-named read alias.
    datasets: dict[str, Parameter] = Field(default_factory=dict)

    @model_validator(mode="before")
    @classmethod
    def _accept_parameters_alias(cls, data: Any) -> Any:
        """Accept the legacy `parameters=` kwarg as an alias for `datasets`.

        Older callers (and tests) construct `Catalog(parameters={...})`.
        The rows now live in the base `datasets` field, so rewrite that
        key on the way in. An explicit `datasets=` always wins.

        Args:
            data: The raw model input (a mapping when constructed with
                keyword arguments).

        Returns:
            The input with `parameters` renamed to `datasets`, untouched
            otherwise.
        """
        if isinstance(data, dict) and "parameters" in data and "datasets" not in data:
            data = dict(data)
            data["datasets"] = data.pop("parameters")
        return data

    @property
    def parameters(self) -> dict[str, Parameter]:
        """The parameter map — alias for the base :attr:`datasets` field.

        Returns:
            dict[str, Parameter]: The same mapping stored in
                :attr:`datasets`.

        Examples:
            - The alias and the base field are the same object:
                ```python
                >>> from earthlens.usgs_water import Catalog
                >>> cat = Catalog()
                >>> cat.parameters is cat.datasets
                True
                >>> cat.parameters["discharge"].code
                '00060'

                ```
        """
        return self.datasets

    def model_post_init(self, __context: Any) -> None:
        """Auto-load the bundled catalog when no parameters were supplied.

        `Catalog()` with no args reads :data:`CATALOG_PATH`; passing
        `parameters=...` (or `datasets=...`) skips the disk read.

        Raises:
            ValueError: Propagated from :meth:`load` when the YAML is
                missing, empty, or has a malformed row.
        """
        if not self.datasets:
            self.datasets = Catalog.load().datasets
        super().model_post_init(__context)

    @classmethod
    def load(cls, catalog_path: Path | None = None) -> Catalog:
        """Read the USGS Water parameter catalog from disk.

        Args:
            catalog_path: Path to the catalog YAML. Defaults to the
                module-level :data:`CATALOG_PATH`.

        Returns:
            A fully-populated :class:`Catalog`.

        Raises:
            ValueError: If the file has no `parameters:` block, or a row
                fails :class:`Parameter` validation.
        """
        catalog_path = catalog_path if catalog_path is not None else CATALOG_PATH
        return cls(datasets=dict(_load_catalog_data(catalog_path)))

    def get_catalog(self) -> dict[str, Parameter]:
        """Return the parameter map (satisfies the abstract contract).

        Returns:
            dict[str, Parameter]: Same object as :attr:`datasets` /
                :attr:`parameters`.
        """
        return self.datasets

    @property
    def available_parameters(self) -> list[str]:
        """The sorted list of curated friendly parameter names.

        Returns:
            list[str]: Every curated catalog key, sorted.

        Examples:
            - List the curated names and check membership:
                ```python
                >>> from earthlens.usgs_water import Catalog
                >>> names = Catalog().available_parameters
                >>> "discharge" in names
                True
                >>> names == sorted(names)
                True

                ```
        """
        return sorted(self.datasets)

    def get_parameter(self, name: str) -> Parameter:
        """Resolve a friendly name to its :class:`Parameter` row.

        Thin wrapper over the inherited :meth:`get_dataset`, which raises
        a `ValueError` with a did-you-mean hint on an unknown name.

        Args:
            name: A friendly parameter name (`"discharge"`).

        Returns:
            Parameter: The matching catalog row.

        Raises:
            ValueError: If `name` is not a known parameter; the message
                names the catalog kind and, when a close match exists,
                adds a did-you-mean hint.

        Examples:
            - Resolve a row and read its fields:
                ```python
                >>> from earthlens.usgs_water import Catalog
                >>> p = Catalog().get_parameter("gage_height")
                >>> p.code
                '00065'
                >>> p.units
                'ft'

                ```
        """
        return self.get_dataset(name)

    def resolve(self, code_or_name: str) -> str:
        """Resolve a friendly name or a raw 5-digit code to a code.

        A raw 5-digit code (`"00060"`) passes through unmapped — the
        full USGS code table is far larger than the curated catalog, so
        any valid code is accepted directly. A non-code string is
        looked up as a friendly name via :meth:`get_parameter`.

        Args:
            code_or_name: A friendly name (`"discharge"`) or a raw
                5-digit NWIS code (`"00060"`).

        Returns:
            str: The 5-digit parameter code.

        Raises:
            ValueError: If `code_or_name` is neither a 5-digit code nor
                a known friendly name (with a did-you-mean hint).
        """
        if _CODE_RE.match(code_or_name):
            return code_or_name
        return self.get_parameter(code_or_name).code

available_parameters property

The sorted list of curated friendly parameter names.

Returns:

Type	Description
list[str]	list[str]: Every curated catalog key, sorted.

Examples:

• List the curated names and check membership:

  >>> from earthlens.usgs_water import Catalog
  >>> names = Catalog().available_parameters
  >>> "discharge" in names
  True
  >>> names == sorted(names)
  True
  

parameters property

The parameter map — alias for the base :attr:datasets field.

Returns:

Type	Description
dict[str, Parameter]	dict[str, Parameter]: The same mapping stored in :attr:datasets.

Examples:

• The alias and the base field are the same object:

  >>> from earthlens.usgs_water import Catalog
  >>> cat = Catalog()
  >>> cat.parameters is cat.datasets
  True
  >>> cat.parameters["discharge"].code
  '00060'
  

get_catalog()

Return the parameter map (satisfies the abstract contract).

Returns:

Type	Description
dict[str, Parameter]	dict[str, Parameter]: Same object as :attr:datasets / :attr:parameters.

Source code in src/earthlens/usgs_water/catalog.py

def get_catalog(self) -> dict[str, Parameter]:
    """Return the parameter map (satisfies the abstract contract).

    Returns:
        dict[str, Parameter]: Same object as :attr:`datasets` /
            :attr:`parameters`.
    """
    return self.datasets

get_parameter(name)

Resolve a friendly name to its :class:Parameter row.

Thin wrapper over the inherited :meth:get_dataset, which raises a ValueError with a did-you-mean hint on an unknown name.

Parameters:

Name	Type	Description	Default
name	str	A friendly parameter name ("discharge").	required

Returns:

Name	Type	Description
Parameter	Parameter	The matching catalog row.

Raises:

Type	Description
ValueError	If name is not a known parameter; the message names the catalog kind and, when a close match exists, adds a did-you-mean hint.

Examples:

• Resolve a row and read its fields:

  >>> from earthlens.usgs_water import Catalog
  >>> p = Catalog().get_parameter("gage_height")
  >>> p.code
  '00065'
  >>> p.units
  'ft'
  

Source code in src/earthlens/usgs_water/catalog.py

def get_parameter(self, name: str) -> Parameter:
    """Resolve a friendly name to its :class:`Parameter` row.

    Thin wrapper over the inherited :meth:`get_dataset`, which raises
    a `ValueError` with a did-you-mean hint on an unknown name.

    Args:
        name: A friendly parameter name (`"discharge"`).

    Returns:
        Parameter: The matching catalog row.

    Raises:
        ValueError: If `name` is not a known parameter; the message
            names the catalog kind and, when a close match exists,
            adds a did-you-mean hint.

    Examples:
        - Resolve a row and read its fields:
            ```python
            >>> from earthlens.usgs_water import Catalog
            >>> p = Catalog().get_parameter("gage_height")
            >>> p.code
            '00065'
            >>> p.units
            'ft'

            ```
    """
    return self.get_dataset(name)

load(catalog_path=None) classmethod

Read the USGS Water parameter catalog from disk.

Parameters:

Name	Type	Description	Default
catalog_path	Path | None	Path to the catalog YAML. Defaults to the module-level :data:CATALOG_PATH.	None

Returns:

Type	Description
Catalog	A fully-populated :class:Catalog.

Raises:

Type	Description
ValueError	If the file has no parameters: block, or a row fails :class:Parameter validation.

Source code in src/earthlens/usgs_water/catalog.py

@classmethod
def load(cls, catalog_path: Path | None = None) -> Catalog:
    """Read the USGS Water parameter catalog from disk.

    Args:
        catalog_path: Path to the catalog YAML. Defaults to the
            module-level :data:`CATALOG_PATH`.

    Returns:
        A fully-populated :class:`Catalog`.

    Raises:
        ValueError: If the file has no `parameters:` block, or a row
            fails :class:`Parameter` validation.
    """
    catalog_path = catalog_path if catalog_path is not None else CATALOG_PATH
    return cls(datasets=dict(_load_catalog_data(catalog_path)))

model_post_init(__context)

Auto-load the bundled catalog when no parameters were supplied.

Catalog() with no args reads :data:CATALOG_PATH; passing parameters=... (or datasets=...) skips the disk read.

Raises:

Type	Description
ValueError	Propagated from :meth:load when the YAML is missing, empty, or has a malformed row.

Source code in src/earthlens/usgs_water/catalog.py

def model_post_init(self, __context: Any) -> None:
    """Auto-load the bundled catalog when no parameters were supplied.

    `Catalog()` with no args reads :data:`CATALOG_PATH`; passing
    `parameters=...` (or `datasets=...`) skips the disk read.

    Raises:
        ValueError: Propagated from :meth:`load` when the YAML is
            missing, empty, or has a malformed row.
    """
    if not self.datasets:
        self.datasets = Catalog.load().datasets
    super().model_post_init(__context)

resolve(code_or_name)

Resolve a friendly name or a raw 5-digit code to a code.

A raw 5-digit code ("00060") passes through unmapped — the full USGS code table is far larger than the curated catalog, so any valid code is accepted directly. A non-code string is looked up as a friendly name via :meth:get_parameter.

Parameters:

Name	Type	Description	Default
code_or_name	str	A friendly name ("discharge") or a raw 5-digit NWIS code ("00060").	required

Returns:

Name	Type	Description
str	str	The 5-digit parameter code.

Raises:

Type	Description
ValueError	If code_or_name is neither a 5-digit code nor a known friendly name (with a did-you-mean hint).

Source code in src/earthlens/usgs_water/catalog.py

def resolve(self, code_or_name: str) -> str:
    """Resolve a friendly name or a raw 5-digit code to a code.

    A raw 5-digit code (`"00060"`) passes through unmapped — the
    full USGS code table is far larger than the curated catalog, so
    any valid code is accepted directly. A non-code string is
    looked up as a friendly name via :meth:`get_parameter`.

    Args:
        code_or_name: A friendly name (`"discharge"`) or a raw
            5-digit NWIS code (`"00060"`).

    Returns:
        str: The 5-digit parameter code.

    Raises:
        ValueError: If `code_or_name` is neither a 5-digit code nor
            a known friendly name (with a did-you-mean hint).
    """
    if _CODE_RE.match(code_or_name):
        return code_or_name
    return self.get_parameter(code_or_name).code

Parameter

Bases: BaseModel

One NWIS parameter code's catalog row.

The user-facing name is the parent key in :attr:Catalog.parameters and is also stored on the row as :attr:name so a resolved :class:Parameter is self-describing.

Attributes:

Name	Type	Description
code	str	The 5-digit NWIS parameter code ("00060").
name	str	Human-readable label ("Discharge").
units	str	The reporting units ("ft3/s", "degC").
group	ParameterGroup	Coarse USGS classification ("Physical", "Nutrients", …).
services	list[str]	The :data:earthlens.usgs_water.backend.SERVICES this code is typically available on (["daily", "instantaneous"]).

Examples:

• Build a row directly:

  >>> from earthlens.usgs_water import Parameter
  >>> p = Parameter(code="00060", name="Discharge", units="ft3/s")
  >>> p.code
  '00060'
  >>> p.group
  'Physical'
  

Source code in src/earthlens/usgs_water/catalog.py

class Parameter(BaseModel):
    """One NWIS parameter code's catalog row.

    The user-facing name is the parent key in
    :attr:`Catalog.parameters` and is also stored on the row as
    :attr:`name` so a resolved :class:`Parameter` is self-describing.

    Attributes:
        code: The 5-digit NWIS parameter code (`"00060"`).
        name: Human-readable label (`"Discharge"`).
        units: The reporting units (`"ft3/s"`, `"degC"`).
        group: Coarse USGS classification (`"Physical"`, `"Nutrients"`,
            …).
        services: The :data:`earthlens.usgs_water.backend.SERVICES`
            this code is typically available on (`["daily",
            "instantaneous"]`).

    Examples:
        - Build a row directly:
            ```python
            >>> from earthlens.usgs_water import Parameter
            >>> p = Parameter(code="00060", name="Discharge", units="ft3/s")
            >>> p.code
            '00060'
            >>> p.group
            'Physical'

            ```
    """

    model_config = ConfigDict(frozen=True, extra="forbid")

    code: str = Field(pattern=r"^\d{5}$")
    name: str = ""
    units: str = ""
    group: ParameterGroup = "Physical"
    services: list[str] = Field(default_factory=list)

clear_catalog_cache()

Empty the module-level catalog parse cache.

Useful in tests that rewrite the catalog on disk and want to force a re-parse. Production callers do not need this — the cache key includes the file's st_mtime_ns, so any real edit invalidates the entry on its own.

Source code in src/earthlens/usgs_water/catalog.py

def clear_catalog_cache() -> None:
    """Empty the module-level catalog parse cache.

    Useful in tests that rewrite the catalog on disk and want to force
    a re-parse. Production callers do not need this — the cache key
    includes the file's `st_mtime_ns`, so any real edit invalidates the
    entry on its own.
    """
    _CATALOG_CACHE.clear()

earthlens.usgs_water.auth

Optional Personal Access Token resolution for the USGS Water backend.

Hosts :class:UsgsWaterAuth, an :class:earthlens.base.AbstractAuth subclass that resolves an optional USGS Personal Access Token (PAT) from, in priority order, an explicit api_token= argument or the API_USGS_PAT environment variable. Unlike the OpenAQ backend (whose key is mandatory), USGS NWIS / Water Data works anonymously — a token only lifts the rate limit. So this auth never raises on a missing token: when none is found, the backend runs anonymously (the modern api.waterdata.usgs.gov endpoint then rate-limits aggressively, which the backend handles by falling back to the legacy endpoint).

The shape:

• :class:UsgsWaterCredentials is a frozen pydantic value object carrying the optional token as a :class:pydantic.SecretStr.
• :class:UsgsWaterAuth binds those credentials and, in :meth:UsgsWaterAuth.configure, resolves the token (explicit then env var) and — when one is present — exports it back to the API_USGS_PAT environment variable, which is the channel the dataretrieval SDK reads. configure() is a no-op when anonymous.
• :meth:UsgsWaterAuth.is_authenticated reports whether a token was resolved (i.e. whether the request is token-backed or anonymous); both are valid operating modes.

The resolved token is read back via the :attr:UsgsWaterAuth.token property (which returns None when anonymous).

UsgsWaterAuth

Bases: AbstractAuth[UsgsWaterCredentials]

Resolve and hold the optional USGS Personal Access Token.

Implements the :class:earthlens.base.AbstractAuth contract for an optional-secret backend. Construction does not touch the environment; :meth:configure performs the resolution and is idempotent. After configure(), the token (or None for anonymous) is available via the :attr:token property, and — when a token was resolved — it is exported to the API_USGS_PAT environment variable so the dataretrieval SDK picks it up.

The class is a context manager (inherited from :class:AbstractAuth): with UsgsWaterAuth(creds) as auth: ... calls configure() on enter and the default no-op close() on exit.

Attributes:

Name	Type	Description
_creds		The :class:UsgsWaterCredentials passed at construction.

Examples:

• Resolve an explicit token:

  >>> from earthlens.usgs_water import UsgsWaterAuth, UsgsWaterCredentials
  >>> auth = UsgsWaterAuth(UsgsWaterCredentials(api_token="k"))
  >>> auth.is_authenticated()
  False
  >>> auth.configure()
  >>> auth.is_authenticated()
  True
  >>> auth.token
  'k'
  

• With no token, configure() succeeds and stays anonymous:

  >>> import os
  >>> from earthlens.usgs_water import UsgsWaterAuth, UsgsWaterCredentials
  >>> os.environ.pop("API_USGS_PAT", None) and None
  >>> auth = UsgsWaterAuth(UsgsWaterCredentials())
  >>> auth.configure()
  >>> auth.is_authenticated()
  False
  >>> auth.token is None
  True
  

Source code in src/earthlens/usgs_water/auth.py

class UsgsWaterAuth(AbstractAuth[UsgsWaterCredentials]):
    """Resolve and hold the optional USGS Personal Access Token.

    Implements the :class:`earthlens.base.AbstractAuth` contract for an
    **optional**-secret backend. Construction does not touch the
    environment; :meth:`configure` performs the resolution and is
    idempotent. After `configure()`, the token (or `None` for
    anonymous) is available via the :attr:`token` property, and — when
    a token was resolved — it is exported to the `API_USGS_PAT`
    environment variable so the `dataretrieval` SDK picks it up.

    The class is a context manager (inherited from
    :class:`AbstractAuth`): `with UsgsWaterAuth(creds) as auth: ...`
    calls `configure()` on enter and the default no-op `close()` on
    exit.

    Attributes:
        _creds: The :class:`UsgsWaterCredentials` passed at
            construction.

    Examples:
        - Resolve an explicit token:
            ```python
            >>> from earthlens.usgs_water import UsgsWaterAuth, UsgsWaterCredentials
            >>> auth = UsgsWaterAuth(UsgsWaterCredentials(api_token="k"))
            >>> auth.is_authenticated()
            False
            >>> auth.configure()
            >>> auth.is_authenticated()
            True
            >>> auth.token
            'k'

            ```
        - With no token, `configure()` succeeds and stays anonymous:
            ```python
            >>> import os
            >>> from earthlens.usgs_water import UsgsWaterAuth, UsgsWaterCredentials
            >>> os.environ.pop("API_USGS_PAT", None) and None
            >>> auth = UsgsWaterAuth(UsgsWaterCredentials())
            >>> auth.configure()
            >>> auth.is_authenticated()
            False
            >>> auth.token is None
            True

            ```
    """

    def __init__(self, credentials: UsgsWaterCredentials) -> None:
        """Store credentials; does not resolve the token yet.

        Args:
            credentials: The :class:`UsgsWaterCredentials` value object
                carrying the optional token.
        """
        super().__init__(credentials)
        self._configured = False
        self._token: str | None = None

    def configure(self) -> None:
        """Resolve the optional token and export it to the environment.

        Idempotent — short-circuits when :meth:`is_authenticated`
        already returns `True`. Resolves the token in this order: the
        explicit `api_token` on the credentials, then the
        `API_USGS_PAT` environment variable. When a token is found it
        is written back to `API_USGS_PAT` (the channel `dataretrieval`
        reads) and :meth:`is_authenticated` flips to `True`. When none
        is found this is a **no-op** — the request runs anonymously
        (rate-limited) and no error is raised.
        """
        if self.is_authenticated():
            return
        token = (
            self._creds.api_token.get_secret_value()
            if self._creds.api_token is not None
            else os.environ.get(_TOKEN_ENV_VAR)
        )
        if token:
            os.environ[_TOKEN_ENV_VAR] = token
            self._token = token
            self._configured = True

    def is_authenticated(self) -> bool:
        """Return `True` when a token was resolved (token-backed mode).

        Cheap predicate — does not call the network. `False` is a
        legitimate, non-error state meaning "anonymous access".

        Returns:
            bool: `True` after :meth:`configure` resolved a token,
                `False` when anonymous (no token).
        """
        return self._configured

    @property
    def token(self) -> str | None:
        """The resolved PAT, or `None` for anonymous access.

        Valid after :meth:`configure`; returns `None` both before
        configuration and when no token was available (anonymous).

        Returns:
            str | None: The USGS PAT string, or `None` when anonymous.
        """
        return self._token

token property

The resolved PAT, or None for anonymous access.

Valid after :meth:configure; returns None both before configuration and when no token was available (anonymous).

Returns:

Type	Description
str | None	str | None: The USGS PAT string, or None when anonymous.

__init__(credentials)

Store credentials; does not resolve the token yet.

Parameters:

Name	Type	Description	Default
credentials	UsgsWaterCredentials	The :class:UsgsWaterCredentials value object carrying the optional token.	required

Source code in src/earthlens/usgs_water/auth.py

def __init__(self, credentials: UsgsWaterCredentials) -> None:
    """Store credentials; does not resolve the token yet.

    Args:
        credentials: The :class:`UsgsWaterCredentials` value object
            carrying the optional token.
    """
    super().__init__(credentials)
    self._configured = False
    self._token: str | None = None

configure()

Resolve the optional token and export it to the environment.

Idempotent — short-circuits when :meth:is_authenticated already returns True. Resolves the token in this order: the explicit api_token on the credentials, then the API_USGS_PAT environment variable. When a token is found it is written back to API_USGS_PAT (the channel dataretrieval reads) and :meth:is_authenticated flips to True. When none is found this is a no-op — the request runs anonymously (rate-limited) and no error is raised.

Source code in src/earthlens/usgs_water/auth.py

def configure(self) -> None:
    """Resolve the optional token and export it to the environment.

    Idempotent — short-circuits when :meth:`is_authenticated`
    already returns `True`. Resolves the token in this order: the
    explicit `api_token` on the credentials, then the
    `API_USGS_PAT` environment variable. When a token is found it
    is written back to `API_USGS_PAT` (the channel `dataretrieval`
    reads) and :meth:`is_authenticated` flips to `True`. When none
    is found this is a **no-op** — the request runs anonymously
    (rate-limited) and no error is raised.
    """
    if self.is_authenticated():
        return
    token = (
        self._creds.api_token.get_secret_value()
        if self._creds.api_token is not None
        else os.environ.get(_TOKEN_ENV_VAR)
    )
    if token:
        os.environ[_TOKEN_ENV_VAR] = token
        self._token = token
        self._configured = True

is_authenticated()

Return True when a token was resolved (token-backed mode).

Cheap predicate — does not call the network. False is a legitimate, non-error state meaning "anonymous access".

Returns:

Name	Type	Description
bool	bool	True after :meth:configure resolved a token, False when anonymous (no token).

Source code in src/earthlens/usgs_water/auth.py

def is_authenticated(self) -> bool:
    """Return `True` when a token was resolved (token-backed mode).

    Cheap predicate — does not call the network. `False` is a
    legitimate, non-error state meaning "anonymous access".

    Returns:
        bool: `True` after :meth:`configure` resolved a token,
            `False` when anonymous (no token).
    """
    return self._configured

UsgsWaterCredentials

Bases: BaseModel

Frozen value object holding the optional USGS PAT.

The token is optional at construction time: None means "resolve from the API_USGS_PAT environment variable at :meth:UsgsWaterAuth.configure time, and if that is also unset, run anonymously".

Attributes:

Name	Type	Description
api_token	SecretStr | None	The USGS Personal Access Token, stored as a :class:pydantic.SecretStr so it is never echoed by repr(creds) or in logs. None defers resolution to the environment variable (and then to anonymous access).

Examples:

• Build from an explicit token; the secret is hidden in repr:

  >>> from earthlens.usgs_water import UsgsWaterCredentials
  >>> creds = UsgsWaterCredentials(api_token="topsecret")
  >>> creds.api_token.get_secret_value()
  'topsecret'
  >>> "topsecret" in repr(creds)
  False
  

• The token is optional — anonymous access is the default:

  >>> from earthlens.usgs_water import UsgsWaterCredentials
  >>> UsgsWaterCredentials().api_token is None
  True
  

Source code in src/earthlens/usgs_water/auth.py

class UsgsWaterCredentials(BaseModel):
    """Frozen value object holding the optional USGS PAT.

    The token is optional at construction time: `None` means "resolve
    from the `API_USGS_PAT` environment variable at
    :meth:`UsgsWaterAuth.configure` time, and if that is also unset,
    run anonymously".

    Attributes:
        api_token: The USGS Personal Access Token, stored as a
            :class:`pydantic.SecretStr` so it is never echoed by
            `repr(creds)` or in logs. `None` defers resolution to the
            environment variable (and then to anonymous access).

    Examples:
        - Build from an explicit token; the secret is hidden in `repr`:
            ```python
            >>> from earthlens.usgs_water import UsgsWaterCredentials
            >>> creds = UsgsWaterCredentials(api_token="topsecret")
            >>> creds.api_token.get_secret_value()
            'topsecret'
            >>> "topsecret" in repr(creds)
            False

            ```
        - The token is optional — anonymous access is the default:
            ```python
            >>> from earthlens.usgs_water import UsgsWaterCredentials
            >>> UsgsWaterCredentials().api_token is None
            True

            ```
    """

    model_config = ConfigDict(frozen=True)

    api_token: SecretStr | None = None