Groundwater Chemistry API#

The groundwater chemistry API provides access to water quality measurements from monitoring stations, including chemical composition, pH, conductivity, and other parameters.

Overview#

The chemistry API is accessed through client.chemistry and provides methods to query chemical measurements from groundwater monitoring stations.

Quick Example:

from sgu_client import SGUClient

client = SGUClient()

# Get chemistry measurements
measurements = client.chemistry.get_measurements(
    bbox=[12.0, 55.0, 16.0, 58.0],
    datetime="2023-01-01/2024-01-01",
    limit=100
)

# Get stations with chemistry data
stations = client.chemistry.get_stations(limit=50)

# Convert to pandas DataFrame for analysis
df = measurements.to_dataframe()
print(df[['observation_date', 'parameter', 'value', 'unit']].head())

Chemistry Client#

The GroundwaterChemistryClient provides methods to query groundwater chemistry data.

Key Methods:

get_measurements() - Retrieve chemistry measurements
get_stations() - Retrieve stations with chemistry data
get_measurement_by_name() - Find measurements for a specific station

Filtering Options:

bbox: Spatial filter (bounding box)
datetime: Temporal filter (ISO 8601 datetime or interval)
filter: CQL filter expression for parameter-specific queries
limit: Maximum number of results to return

Common Parameters:

The chemistry API tracks various water quality parameters including:

pH levels
Electrical conductivity
Major ions (Ca, Mg, Na, K, Cl, SO4, HCO3)
Nutrients (NO3, NH4, PO4)
Trace elements
Dissolved gases

Groundwater chemistry client endpoints.

class sgu_client.client.chemistry.chemistry.GroundwaterChemistryClient(base_client)[source]#

Bases: object

Client for groundwater chemistry-related SGU API endpoints.

Parameters:: base_client (BaseClient)

BASE_PATH = 'collections'#

CHEMISTRY_BASE_URL = 'https://api.sgu.se/oppnadata/grundvattenkvalitet-analysresultat-provplatser/ogc/features/v1/'#

__init__(base_client)[source]#

Initialize groundwater chemistry client.

Parameters:: base_client (BaseClient) – Base HTTP client instance

get_sampling_sites(bbox=None, datetime=None, limit=None, filter_expr=None, sortby=None, **kwargs)[source]#

Get groundwater chemistry sampling sites. This method is used internally by convenience functions like `get_sampling_site_by_name() by constructing filter expressions. A user may also do this, but readability is greater when using the built-in convenience functions.

Parameters:

bbox (list[float] | None) – Bounding box as [min_lon, min_lat, max_lon, max_lat]
datetime (str | None) – Date/time filter (RFC 3339 format or interval)
limit (int | None) – Maximum number of features to return (automatically paginated if needed)
filter_expr (str | None) – CQL filter expression
sortby (list[str] | None) – List of sort expressions (e.g., [‘+name’, ‘-date’])
**kwargs (Any) – Additional query parameters

Return type:

SamplingSiteCollection

Returns:

Typed collection of groundwater chemistry sampling sites

get_sampling_site(site_id)[source]#

Get a specific groundwater chemistry sampling site by ID. This endpoint is provided by the OGC API but likely not used by any user.

Parameters:: site_id (str) – Site identifier
Return type:: SamplingSite
Returns:: Typed groundwater chemistry sampling site
Raises:: ValueError – If site not found or multiple sites returned

get_sampling_site_by_name(site_id=None, site_name=None, **kwargs)[source]#

Convenience function to get a sampling site by name (‘site_id’ or ‘site_name’).

Parameters:

site_id (str | None) – Site identifier (maps to ‘platsbeteckning’ in API)
site_name (str | None) – Site name (maps to ‘provplatsnamn’ in API)
**kwargs (Any) – Additional query parameters (e.g., limit)

Return type:

SamplingSite

Returns:

Typed groundwater chemistry sampling site

Raises:

ValueError – If neither parameter is provided, both are provided, or if multiple sites are found

Example

>>> from sgu_client import SGUClient
>>> client = SGUClient()
>>>
>>> # find site by id
>>> site = client.chemistry.get_sampling_site_by_name(site_id="10001_1")
>>>
>>> # find site by name
>>> site = client.chemistry.get_sampling_site_by_name(
...     site_name="Ringarum_1"
... )
>>> print(site.properties.municipality)

get_sampling_sites_by_names(site_id=None, site_name=None, **kwargs)[source]#

Convenience function to get multiple sampling sites by name (‘site_id’ or ‘site_name’).

Parameters:

site_id (list[str] | None) – List of site identifiers (maps to ‘platsbeteckning’ in API)
site_name (list[str] | None) – List of site names (maps to ‘provplatsnamn’ in API)
**kwargs (Any) – Additional query parameters (e.g., limit)

Return type:

SamplingSiteCollection

Returns:

Typed collection of groundwater chemistry sampling sites

Raises:

ValueError – If neither parameter is provided or both are provided

Example

>>> from sgu_client import SGUClient
>>> client = SGUClient()
>>>
>>> # get multiple sampling sites by their IDs
>>> sites = client.chemistry.get_sampling_sites_by_names(
...     site_id=["10001_1", "10002_1", "10003_1"]
... )
>>> for site in sites.features:
...     print(f"{site.properties.station_id}: {site.properties.municipality}")

get_analysis_results(bbox=None, datetime=None, limit=None, filter_expr=None, sortby=None, **kwargs)[source]#

Get groundwater chemistry analysis results. This method is used internally by convenience functions like `get_results_by_site() by constructing filter expressions. A user may also do this, but readability is greater when using the built-in convenience functions.

Parameters:

bbox (list[float] | None) – Bounding box as [min_lon, min_lat, max_lon, max_lat]
datetime (str | None) – Date/time filter (RFC 3339 format or interval)
limit (int | None) – Maximum number of features to return (automatically paginated if needed)
filter_expr (str | None) – CQL filter expression
sortby (list[str] | None) – List of sort expressions (e.g., [‘+date’, ‘-value’])
**kwargs (Any) – Additional query parameters

Return type:

AnalysisResultCollection

Returns:

Typed collection of groundwater chemistry analysis results

get_analysis_result(result_id)[source]#

Get a specific groundwater chemistry analysis result by ID. This endpoint is provided by the OGC API but likely not used by any user.

Parameters:: result_id (str) – Result identifier
Return type:: AnalysisResult
Returns:: Typed groundwater chemistry analysis result
Raises:: ValueError – If result not found or multiple results returned

get_results_by_site(site_id=None, site_name=None, tmin=None, tmax=None, limit=None, **kwargs)[source]#

Get analysis results for a specific site by name with optional time filtering.

Parameters:

site_id (str | None) – Site identifier (maps to ‘platsbeteckning’ in API)
site_name (str | None) – Site name (maps to ‘provplatsnamn’ in API)
tmin (str | datetime | None) – Start time (ISO string or datetime object)
tmax (str | datetime | None) – End time (ISO string or datetime object)
limit (int | None) – Maximum number of results to return
**kwargs (Any) – Additional query parameters

Return type:

AnalysisResultCollection

Returns:

Typed collection of groundwater chemistry analysis results

Raises:

ValueError – If neither or both name parameters are provided, or if site lookup fails

Example

>>> from sgu_client import SGUClient
>>> from datetime import datetime, timezone
>>> client = SGUClient()
>>>
>>> # get all results for a site
>>> results = client.chemistry.get_results_by_site(
...     site_id="10001_1",
...     limit=100
... )
>>>
>>> # get results with time filtering
>>> results = client.chemistry.get_results_by_site(
...     site_id="10001_1",
...     tmin=datetime(2020, 1, 1, tzinfo=timezone.utc),
...     tmax=datetime(2021, 1, 1, tzinfo=timezone.utc)
... )

get_results_by_sites(site_id=None, site_name=None, tmin=None, tmax=None, limit=None, **kwargs)[source]#

Get analysis results for multiple sites by name with optional time filtering.

Parameters:

site_id (list[str] | None) – List of station identifiers (maps to ‘platsbeteckning’ in API)
site_name (list[str] | None) – List of site names (maps to ‘provplatsnamn’ in API)
tmin (str | datetime | None) – Start time (ISO string or datetime object)
tmax (str | datetime | None) – End time (ISO string or datetime object)
limit (int | None) – Maximum number of results to return
**kwargs (Any) – Additional query parameters

Return type:

AnalysisResultCollection

Returns:

Typed collection of groundwater chemistry analysis results

Raises:

ValueError – If neither or both name parameters are provided, or if site lookup fails

Example

>>> from sgu_client import SGUClient
>>> client = SGUClient()
>>>
>>> # get results for multiple sites
>>> results = client.chemistry.get_results_by_sites(
...     station_id=["10001_1", "10002_1"],
...     tmin="2020-01-01T00:00:00Z",
...     tmax="2021-01-01T00:00:00Z",
...     limit=1000
... )
>>> print(f"Found {len(results.features)} analysis results")

get_results_by_parameter(parameter, site_id=None, tmin=None, tmax=None, limit=None, **kwargs)[source]#

Get analysis results filtered by chemical parameter (e.g., pH, nitrate).

This convenience method makes it easy to retrieve all measurements for a specific chemical parameter, optionally filtered by site and time range.

Parameters:

parameter (str) – Parameter short name (e.g., ‘PH’, ‘NITRATE’, ‘KLORID’)
site_id (str | list[str] | None) – Optional station identifier(s) to filter by
tmin (str | datetime | None) – Start time (ISO string or datetime object)
tmax (str | datetime | None) – End time (ISO string or datetime object)
limit (int | None) – Maximum number of results to return
**kwargs (Any) – Additional query parameters

Return type:

AnalysisResultCollection

Returns:

Typed collection of groundwater chemistry analysis results

Example

>>> from sgu_client import SGUClient
>>> client = SGUClient()
>>>
>>> # get all pH measurements
>>> results = client.chemistry.get_results_by_parameter(
...     parameter="PH",
...     limit=1000
... )
>>>
>>> # get pH measurements for a specific site
>>> results = client.chemistry.get_results_by_parameter(
...     parameter="PH",
...     site_id="10001_1",
...     tmin="2020-01-01",
...     tmax="2021-01-01"
... )
>>>
>>> # ... or multiple sites
>>> results = client.chemistry.get_results_by_parameter(
...     parameter="PH",
...     site_id=["10001_1", "10002_1"]
... )

Chemistry Data Models#

The chemistry data uses typed Pydantic models for measurements and station metadata.

Pydantic models for groundwater chemistry data from SGU API.

class sgu_client.models.chemistry.SamplingSiteProperties(**data)[source]#

Bases: SGUBaseModel

Properties for a groundwater chemistry sampling site.

Parameters:: data (Any)

station_id: str | None#

site_name: str | None#

national_site_id: int | None#

eu_station_code: str | None#

eu_groundwater_body: str | None#

site_type_code: str | None#

site_type_description: str | None#

site_category_code: str | None#

site_category_description: str | None#

north_coordinate: float | None#

east_coordinate: float | None#

positioning_method_code: str | None#

positioning_method_description: str | None#

position_quality_code: str | None#

position_quality_description: str | None#

county_code: str | None#

county: str | None#

municipality_code: str | None#

municipality: str | None#

region_code: str | None#

region_description: str | None#

water_district_code: str | None#

water_district_description: str | None#

reference_level: float | None#

elevation_system: str | None#

well_depth: float | None#

well_depth_qualifier: str | None#

filter_depth_top: float | None#

filter_depth_bottom: float | None#

filter_depth_qualifier: str | None#

aquifer_code: str | None#

aquifer_description: str | None#

soil_genesis_code: str | None#

soil_genesis_description: str | None#

rock_type_code: str | None#

rock_type_description: str | None#

established_date: str | None#

decommissioned_date: str | None#

sample_count: int | None#

program_affiliation: str | None#

national_monitoring: str | None#

regional_monitoring: str | None#

local_monitoring: str | None#

symbol: str | None#

analyses_csv_url: str | None#

analyses_json_url: str | None#

property established_datetime: datetime | None#: Parse establishment date as datetime object.

property decommissioned_datetime: datetime | None#: Parse decommissioning date as datetime object.

model_config: ClassVar[ConfigDict] = {'extra': 'allow', 'use_enum_values': True, 'validate_assignment': True, 'validate_by_alias': True, 'validate_by_name': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class sgu_client.models.chemistry.SamplingSite(**data)[source]#

Bases: SGUBaseModel

A groundwater chemistry sampling site (GeoJSON Feature).

Parameters:: data (Any)

type: Literal['Feature']#

id: str#

geometry: Point | MultiPoint | LineString | Polygon | MultiPolygon | None#

properties: SamplingSiteProperties#

model_config: ClassVar[ConfigDict] = {'extra': 'allow', 'use_enum_values': True, 'validate_assignment': True, 'validate_by_alias': True, 'validate_by_name': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class sgu_client.models.chemistry.AnalysisResultProperties(**data)[source]#

Bases: SGUBaseModel

Properties for a groundwater chemistry analysis result.

Parameters:: data (Any)

station_id: str | None#

national_site_id: int | None#

county_code: str | None#

sample_id: str | None#

sample_type: str | None#

delivery_id: str | None#

program_name: str | None#

program_id: str | None#

monitoring_manual: str | None#

sampling_date: str | None#

submission_date: str | None#

parameter_name: str | None#

parameter_short_name: str | None#

parameter_sequence_number: int | None#

water_preparation: str | None#

sample_preparation: str | None#

laboratory: str | None#

method: str | None#

reporting_limit: float | None#

detection_limit: float | None#

measurement_value_annotation: str | None#

measurement_value: float | None#

measurement_value_span: str | None#

measurement_value_text: str | None#

unit: str | None#

measurement_uncertainty: str | None#

last_updated: str | None#

row_number: int | None#

property sampling_datetime: datetime | None#: Parse sampling date as datetime object.

property submission_datetime: datetime | None#: Parse submission date as datetime object.

property last_updated_datetime: datetime | None#: Parse last update as datetime object.

model_config: ClassVar[ConfigDict] = {'extra': 'allow', 'use_enum_values': True, 'validate_assignment': True, 'validate_by_alias': True, 'validate_by_name': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class sgu_client.models.chemistry.AnalysisResult(**data)[source]#

Bases: SGUBaseModel

A groundwater chemistry analysis result (GeoJSON Feature).

Parameters:: data (Any)

type: Literal['Feature']#

id: str#

geometry: Point | MultiPoint | LineString | Polygon | MultiPolygon | None#

properties: AnalysisResultProperties#

model_config: ClassVar[ConfigDict] = {'extra': 'allow', 'use_enum_values': True, 'validate_assignment': True, 'validate_by_alias': True, 'validate_by_name': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class sgu_client.models.chemistry.SamplingSiteCollection(**data)[source]#

Bases: SGUResponse

Collection of groundwater chemistry sampling sites (GeoJSON FeatureCollection).

Parameters:: data (Any)

type: Literal['FeatureCollection']#

features: list[SamplingSite]#

totalFeatures: int | None#

numberMatched: int | None#

numberReturned: int | None#

timeStamp: str | None#

links: list[Link] | None#

crs: CRS | None#

to_dataframe(**kwargs) → pd.DataFrame#

Convert to pandas DataFrame with flattened sampling site properties.

Returns:: DataFrame containing sampling site data with parsed datetime columns.

Examples

>>> from sgu_client import SGUClient
>>> client = SGUClient()
>>>
>>> sites = client.chemistry.get_sampling_sites(limit=10)
>>> df = sites.to_dataframe()
>>>
>>> # dataFrame includes site properties with datetime parsing
>>> print(df[['station_id', 'site_name', 'municipality', 'established_date', 'sample_count']].head())
>>> # established_date and decommissioned_date are parsed as datetime objects

model_config: ClassVar[ConfigDict] = {'extra': 'allow', 'use_enum_values': True, 'validate_assignment': True, 'validate_by_alias': True, 'validate_by_name': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class sgu_client.models.chemistry.AnalysisResultCollection(**data)[source]#

Bases: SGUResponse

Collection of groundwater chemistry analysis results (GeoJSON FeatureCollection).

Parameters:: data (Any)

type: Literal['FeatureCollection']#

features: list[AnalysisResult]#

totalFeatures: int | None#

numberMatched: int | None#

numberReturned: int | None#

timeStamp: str | None#

links: list[Link] | None#

crs: CRS | None#

to_dataframe(**kwargs) → pd.DataFrame#

Convert to pandas DataFrame with analysis result data.

Parameters:: sort_by_date – Whether to sort the DataFrame by sampling date.
Returns:: DataFrame containing analysis result data.

Examples

>>> from sgu_client import SGUClient
>>> client = SGUClient()
>>>
>>> results = client.chemistry.get_results_by_site(site_id="10001_1", limit=100)
>>> df = results.to_dataframe()
>>>
>>> # dataFrame includes chemical analysis results with multiple datetime columns
>>> print(df[['sampling_date', 'parameter_short_name', 'measurement_value', 'unit']].head())
>>> # sampling_date, submission_date, and last_update are all parsed as datetime objects

to_series(**kwargs) → pd.Series#

Convert to pandas Series with analysis result data.

Parameters:

index – Column name to use as index. If None, sampling_date is used.
data – Column name to use as data. If None, measurement_value is used.
sort_by_date – Whether to sort the data by sampling date before creating the Series.

Returns:

Series containing analysis result data.

Examples

>>> from sgu_client import SGUClient
>>> client = SGUClient()
>>> results = client.chemistry.get_results_by_site(site_id="10001_1", limit=100)
>>>
>>> # create time series with default columns (sampling_date, measurement_value)
>>> series = results.to_series()
>>> print(series.head())
>>>
>>> # use custom columns - e.g., parameter names as data
>>> series_params = results.to_series(
...     index="sampling_date",
...     data="parameter_short_name"
... )

pivot_by_parameter(**kwargs) → pd.DataFrame#

Pivot analysis results by parameter for easier time series analysis.

This creates a wide-format DataFrame where each chemical parameter becomes a column, making it easy to analyze multiple parameters over time.

Parameters:

values – Column to use for values (default: ‘measurement_value’)
index – Column to use as index (default: ‘sampling_date’)
columns – Column to pivot into columns (default: ‘parameter_short_name’)
aggfunc – Aggregation function if there are duplicate index/column pairs (default: ‘mean’). Can be ‘mean’, ‘median’, ‘first’, ‘last’, etc.

Returns:

Pivoted DataFrame with parameters as columns.

Example

>>> from sgu_client import SGUClient()
>>> client = SGUClient()
>>>
>>> results = client.chemistry.get_results_by_site(site_id="10001_1")
>>> df_pivot = results.pivot_by_parameter()
>>> # now df_pivot has columns like 'PH', 'NITRATE', 'CHLORIDE', etc.

model_config: ClassVar[ConfigDict] = {'extra': 'allow', 'use_enum_values': True, 'validate_assignment': True, 'validate_by_alias': True, 'validate_by_name': True}#: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Groundwater Chemistry API#

Overview#

Chemistry Client#

Chemistry Data Models#

See Also#

This Page