"""Observed groundwater level client endpoints."""
from datetime import datetime
from typing import Any
from sgu_client.client.base import BaseClient
from sgu_client.client.collection import SGUCollection
from sgu_client.client.identifiers import IdentifierLookup, IdentifierValue
from sgu_client.models.observed import (
GroundwaterMeasurement,
GroundwaterMeasurementCollection,
GroundwaterStation,
GroundwaterStationCollection,
)
DATASET = "grundvattennivaer-observerade"
# Station identifier argument -> (API field, GroundwaterStationProperties attribute)
STATION_IDENTIFIERS: dict[str, tuple[str, str]] = {
"station_id": ("platsbeteckning", "station_id"),
"station_name": ("obsplatsnamn", "station_name"),
}
# The field measurements refer to their station by
MEASUREMENT_STATION_FIELD = "platsbeteckning"
def _stations_given(
station_id: IdentifierValue, station_name: IdentifierValue
) -> dict[str, IdentifierValue]:
"""A station lookup method's identifier arguments, for IdentifierLookup."""
return {"station_id": station_id, "station_name": station_name}
[docs]
class ObservedGroundwaterLevelClient:
"""Client for observed groundwater level-related SGU API endpoints."""
def __init__(self, base_client: BaseClient):
"""Initialize observed groundwater level client.
Args:
base_client: Base HTTP client instance
"""
self._client = base_client
self._stations = SGUCollection(
base_client,
dataset=DATASET,
name="stationer",
model=GroundwaterStationCollection,
label="Station",
)
self._measurements = SGUCollection(
base_client,
dataset=DATASET,
name="nivaer",
model=GroundwaterMeasurementCollection,
label="Measurement",
date_field="obsdatum",
)
self._station_lookup = IdentifierLookup(
self._stations, STATION_IDENTIFIERS, ref="station_id", noun="station"
)
[docs]
def get_stations(
self,
bbox: list[float] | None = None,
datetime: str | None = None,
limit: int | None = None,
filter_expr: str | None = None,
sortby: list[str] | None = None,
**kwargs: Any,
) -> GroundwaterStationCollection:
"""Get groundwater monitoring stations.
Args:
bbox: Bounding box as [min_lon, min_lat, max_lon, max_lat]
datetime: Date/time filter (RFC 3339 format or interval)
limit: Maximum number of features to return (automatically paginated if needed)
filter_expr: CQL filter expression
sortby: List of sort expressions (e.g., ['+name', '-date'])
**kwargs: Additional query parameters
Returns:
Typed collection of groundwater monitoring stations
Example:
>>> from sgu_client import SGUClient
>>> with SGUClient() as client:
... # get stations in southern Sweden
... stations = client.levels.observed.get_stations(
... bbox=[12.0, 55.0, 16.0, 58.0]
... )
... # filter stations by municipality using CQL
... stations = client.levels.observed.get_stations(
... filter_expr="kommun='Uppsala' AND akvifer='JS'"
... )
"""
return self._stations.query(
bbox=bbox,
datetime=datetime,
limit=limit,
filter_expr=filter_expr,
sortby=sortby,
**kwargs,
)
[docs]
def get_station(self, station_id: str) -> GroundwaterStation:
"""Get a specific groundwater monitoring station by ID. This endpoint is provided by the OGC API
but likely not used by any user.
Args:
station_id: Station identifier
Returns:
Typed groundwater monitoring station
Raises:
ValueError: If station not found or multiple stations returned
"""
return self._stations.get(station_id)
[docs]
def get_measurements(
self,
bbox: list[float] | None = None,
datetime: str | None = None,
limit: int | None = None,
filter_expr: str | None = None,
sortby: list[str] | None = None,
tmin: str | datetime | None = None,
tmax: str | datetime | None = None,
**kwargs: Any,
) -> GroundwaterMeasurementCollection:
"""Get groundwater level measurements.
This method is used internally by convenience functions like
``get_measurements_by_name()`` by constructing filter expressions. A user
may also do this, but readability is greater when using the built-in
convenience functions.
Args:
bbox: Bounding box as [min_lon, min_lat, max_lon, max_lat]
datetime: Raw OGC date/time filter; prefer ``tmin``/``tmax``
limit: Maximum number of features to return (automatically paginated if needed)
filter_expr: CQL filter expression
sortby: List of sort expressions (e.g., ['+date', '-value'])
tmin: Earliest observation date (ISO string or datetime object)
tmax: Latest observation date (ISO string or datetime object)
**kwargs: Additional query parameters
Returns:
Typed collection of groundwater level measurements
"""
return self._measurements.query(
bbox=bbox,
datetime=datetime,
limit=limit,
filter_expr=filter_expr,
sortby=sortby,
tmin=tmin,
tmax=tmax,
**kwargs,
)
[docs]
def get_measurement(self, measurement_id: str) -> GroundwaterMeasurement:
"""Get a specific groundwater level measurement by ID. This endpoint is provided by the OGGC API
but likely not used by any user.
Args:
measurement_id: Measurement identifier
Returns:
Typed groundwater level measurement
Raises:
ValueError: If measurement not found or multiple measurements returned
"""
return self._measurements.get(measurement_id)
[docs]
def get_station_by_name(
self,
station_id: str | None = None,
station_name: str | None = None,
**kwargs: Any,
) -> GroundwaterStation:
"""Convenience function to get a station by name ('station_id' or 'station_name').
Args:
station_id: Station identifier (maps to 'platsbeteckning' in API)
station_name: Station name (maps to 'obsplatsnamn' in API)
**kwargs: Additional query parameters (e.g., limit)
Returns:
Typed groundwater monitoring station
Raises:
SGUNotFoundError: If no station matches.
SGUAmbiguousError: If several stations match.
ValueError: If not exactly one of station_id or station_name is
provided.
Example:
>>> from sgu_client import SGUClient
>>> with SGUClient() as client:
... # find station by platsbeteckning (recommended)
... station = client.levels.observed.get_station_by_name(station_id="95_2")
... # find station by obsplatsnamn
... station = client.levels.observed.get_station_by_name(
... station_name="Lagga_2"
... )
... print(station.properties.aquifer_description)
"""
return self._station_lookup.find_one(
_stations_given(station_id, station_name), **kwargs
)
[docs]
def get_stations_by_names(
self,
station_id: list[str] | None = None,
station_name: list[str] | None = None,
**kwargs: Any,
) -> GroundwaterStationCollection:
"""Convenience function to get multiple stations by name ('station_id' or 'station_name').
Args:
station_id: List of station identifiers (maps to 'platsbeteckning' in API)
station_name: List of station names (maps to 'obsplatsnamn' in API)
**kwargs: Additional query parameters (e.g., limit)
Returns:
Typed collection of groundwater monitoring stations
Raises:
SGUNotFoundError: If any of the requested stations is not found.
ValueError: If not exactly one of station_id or station_name is
provided.
Example:
>>> from sgu_client import SGUClient
>>> with SGUClient() as client:
... # get multiple stations by their IDs
... stations = client.levels.observed.get_stations_by_names(
... station_id=["95_2", "101_1", "102_3"]
... )
... for station in stations.features:
... print(f"{station.properties.station_id}: {station.properties.aquifer_code}")
"""
return self._station_lookup.find_all(
_stations_given(station_id, station_name), **kwargs
)
[docs]
def get_measurements_by_name(
self,
station_id: str | None = None,
station_name: str | None = None,
tmin: str | datetime | None = None,
tmax: str | datetime | None = None,
limit: int | None = None,
**kwargs: Any,
) -> GroundwaterMeasurementCollection:
"""Get measurements for a specific station by name with optional time filtering.
Args:
station_id: Station identifier (maps to 'platsbeteckning' in API)
station_name: Station name (maps to 'obsplatsnamn' in API)
tmin: Start time (ISO string or datetime object)
tmax: End time (ISO string or datetime object)
limit: Maximum number of measurements to return
**kwargs: Additional query parameters
Returns:
Typed collection of groundwater level measurements
Raises:
SGUNotFoundError: If a station_name matches no station.
SGUAmbiguousError: If a station_name matches several stations.
ValueError: If not exactly one of station_id or station_name is
provided, or if the looked-up station has no station_id.
Example:
>>> from sgu_client import SGUClient
>>> from datetime import datetime, timezone
>>> with SGUClient() as client:
... # get all measurements for a station
... measurements = client.levels.observed.get_measurements_by_name(
... station_id="95_2",
... limit=100
... )
... # get measurements with time filtering
... measurements = client.levels.observed.get_measurements_by_name(
... station_id="95_2",
... tmin=datetime(2020, 1, 1, tzinfo=timezone.utc),
... tmax=datetime(2021, 1, 1, tzinfo=timezone.utc)
... )
... # using ISO string for time filtering
... measurements = client.levels.observed.get_measurements_by_name(
... station_id="95_2",
... tmin="2020-01-01T00:00:00Z",
... tmax="2021-01-01T00:00:00Z"
... )
"""
ref = self._station_lookup.ref(_stations_given(station_id, station_name))
return self._measurements.query(
where={MEASUREMENT_STATION_FIELD: ref},
tmin=tmin,
tmax=tmax,
limit=limit,
**kwargs,
)
[docs]
def get_measurements_by_names(
self,
station_id: list[str] | None = None,
station_name: list[str] | None = None,
tmin: str | datetime | None = None,
tmax: str | datetime | None = None,
limit: int | None = None,
**kwargs: Any,
) -> GroundwaterMeasurementCollection:
"""Get measurements for multiple stations by name with optional time filtering.
Args:
station_id: List of station identifiers (maps to 'platsbeteckning' in API)
station_name: List of station names (maps to 'obsplatsnamn' in API)
tmin: Start time (ISO string or datetime object)
tmax: End time (ISO string or datetime object)
limit: Maximum number of measurements to return
**kwargs: Additional query parameters
Returns:
Typed collection of groundwater level measurements
Raises:
SGUNotFoundError: If any station_name matches no station.
ValueError: If not exactly one of station_id or station_name is
provided, or if a looked-up station has no station_id.
Example:
>>> from sgu_client import SGUClient
>>> with SGUClient() as client:
... # get measurements for multiple stations
... measurements = client.levels.observed.get_measurements_by_names(
... station_id=["95_2", "101_1"],
... tmin="2020-01-01T00:00:00Z",
... tmax="2021-01-01T00:00:00Z",
... limit=1000
... )
... print(f"Found {len(measurements.features)} measurements")
"""
refs = self._station_lookup.refs(_stations_given(station_id, station_name))
return self._measurements.query(
where={MEASUREMENT_STATION_FIELD: refs},
tmin=tmin,
tmax=tmax,
limit=limit,
**kwargs,
)