veriflow.datasources.fewswebservice#

Module for reading from and writing to a fews webservice.

Classes

FewsWebservice(config)

For downloading data using the Delft-FEWS PI Webservice.

FewsWebserviceConfig(*, import_adapter, ...)

A fews webservice input config element.

FewsWebserviceAuthConfig([_case_sensitive, ...])

Get url, username and password safely from environment variables.
class veriflow.datasources.fewswebservice.FewsWebservice(config)[source]#

For downloading data using the Delft-FEWS PI Webservice.

Parameters:

config (FewsWebserviceConfig)

kind: str = 'fewswebservice'#
config_class#

alias of FewsWebserviceConfig

supported_data_types: ClassVar[set[DataType]] = {DataType.observed_historical, DataType.simulated_forecast_ensemble, DataType.simulated_forecast_probabilistic, DataType.simulated_forecast_single, DataType.simulated_historical}#
datetime_format = '%Y-%m-%dT%H:%M:%SZ'#
timeout = 30#
config: FewsWebserviceConfig#
property data_type: DataType#

Whether the instance represents sim or obs data.

dataset: Dataset#
property configured_stations: set[str]#

Return the internal station identifiers configured for this datasource.

This is needed for standardization of station identifiers across sources.

property cache: ZarrCache | None#

Return the cache instance if caching is enabled, otherwise None.

property configured_variables: set[str]#

Return the internal variable identifiers configured for this datasource.

This is needed for standardization of variable identifiers across sources.

property configured_variables_internal: set[str] | None#

Return the standardized internal variable identifiers configured for this datasource.

This standardized format is needed for caching across different datasources. If your datasource implementation does not have any configurable variables, return None.

create_cache_request(cached_dataset)#

Get the cache request based on the cached dataset and the config.

Parameters:

cached_dataset (Dataset)

Return type:

CacheRequest

fetch_validate_filter_cache(*, clear_cache=False)#

High-level wrapper to fetch, validate, filter and apply id mapping to the dataset.

Parameters:

clear_cache (bool)

Return type:

Self

filter_dataset(dataset)#

Filter the dataset on lead times and times outside the verification period.

Parameters:

dataset (Dataset)

Return type:

Dataset

classmethod from_config(raw_config)#

Initialize class from config dict.

Parameters:

raw_config (dict[str, Any])

Return type:

Self

static get_cached_data(cached_dataset, datasource)#

Get the dataset from the cache based on the datasource configuration.

Parameters:
Return type:

Dataset

get_data()#

Get data and make use of cache if configured.

Return type:

Self

validate_fetched_data()#

Validate that the dataset is consistent with the config.

Return type:

None

static write_netcdf_response_to_dir(response, write_dir, unique_prefix=None)[source]#

Write the NetCDF file(s) from a webservice response to a directory.

The Delft-FEWS Webservice may return either a zip archive containing one or more NetCDF files (one per requested parameter), or a single, unzipped NetCDF (.nc) file. Both cases are handled here.

Optional parameter unique_prefix is used only when using the leadTime parameter in the request. In this case, the Delft-FEWS webservice response does not contain any meta data on the exact leadTime that was used. That’s why we need to store it in the filename, so that we can internally assign it later as a proper coordinate on the internal xr.DataArray.

Parameters:

  • response (Response)

  • write_dir (Path)

  • unique_prefix (str | None)

Return type:

Path

fetch_data()[source]#

Retrieve :py::class`~xarray.Dataset` from Delft-FEWS Webservice.

Return type:

Self

class veriflow.datasources.fewswebservice.FewsWebserviceConfig(*, import_adapter, source, data_type, general, id_mapping=None, auth_config=<factory>, location_ids, parameter_ids, module_instance_id, ensemble_id=None, qualifier_ids=None, export_id_map=None, webservice_version, archive_kind=ArchiveKind.open_archive, forecast_retrieval_method=ForecastRetrievalMethod.retrieve_all_forecast_data, max_workers_in_thread_pool=2, **extra_data)[source]#

A fews webservice input config element.

Parameters:

  • import_adapter (Literal[DataSourceKind.FEWSWEBSERVICE])

  • source (Annotated[str, StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=^[A-Za-z][A-Za-z0-9_]*$, ascii_only=None), MinLen(min_length=1)])

  • data_type (DataType)

  • general (Annotated[GeneralInfoConfig, SkipJsonSchema()])

  • id_mapping (Annotated[IdMappingConfig, SkipJsonSchema()] | None)

  • auth_config (FewsWebserviceAuthConfig)

  • location_ids (Annotated[list[str], MinLen(min_length=1)])

  • parameter_ids (Annotated[list[str], MinLen(min_length=1)])

  • module_instance_id (Annotated[str, MinLen(min_length=1)])

  • ensemble_id (Annotated[str, FieldInfo(annotation=NoneType, required=True, metadata=[MinLen(min_length=1)])] | None)

  • qualifier_ids (Annotated[list[str], FieldInfo(annotation=NoneType, required=True, metadata=[MinLen(min_length=1)])] | None)

  • export_id_map (Annotated[str, FieldInfo(annotation=NoneType, required=True, metadata=[MinLen(min_length=1)])] | None)

  • webservice_version (Annotated[str, StringConstraints(strip_whitespace=None, to_upper=None, to_lower=None, strict=None, min_length=None, max_length=None, pattern=^\d{4}\.(0[1-2])$, ascii_only=None)])

  • archive_kind (ArchiveKind)

  • forecast_retrieval_method (ForecastRetrievalMethod)

  • max_workers_in_thread_pool (int)

  • extra_data (Any)

import_adapter: Literal[DataSourceKind.FEWSWEBSERVICE]#
auth_config: FewsWebserviceAuthConfig#
location_ids: Annotated[list[str], FieldInfo(annotation=NoneType, required=True, metadata=[MinLen(min_length=1)])]#
parameter_ids: Annotated[list[str], FieldInfo(annotation=NoneType, required=True, metadata=[MinLen(min_length=1)])]#
module_instance_id: Annotated[str, FieldInfo(annotation=NoneType, required=True, metadata=[MinLen(min_length=1)])]#
ensemble_id: Annotated[str, FieldInfo(annotation=NoneType, required=True, metadata=[MinLen(min_length=1)])] | None#
qualifier_ids: Annotated[list[str], FieldInfo(annotation=NoneType, required=True, metadata=[MinLen(min_length=1)])] | None#
export_id_map: Annotated[str, FieldInfo(annotation=NoneType, required=True, metadata=[MinLen(min_length=1)])] | None#
webservice_version: 02")]#
archive_kind: Annotated[ArchiveKind, FieldInfo(annotation=NoneType, required=True, description='Archive kind. Defaults to a Delft-FEWS Open Archive, which is the Delft-FEWS standard.')]#
forecast_retrieval_method: Annotated[ForecastRetrievalMethod, FieldInfo(annotation=NoneType, required=True, description='Since Delft-FEWS 2025.01, the Delft-FEWS Webservice canretrieve forecasts for specific lead times (lead times). This avoid having to retrieve all forecast data outside of the configured lead times (lead times) for the verification pipeline. If not provided, the method will be automatically determined based on the configured webservice version.')]#
max_workers_in_thread_pool: Annotated[int, FieldInfo(annotation=NoneType, required=True, description='This datasource asynchronously retrieves data from the Delft-FEWS webservice. Define here the maximum workers it can use. Use 5-10 for gentle load on the server-side and keep below 30 to avoid instability and minimize the risk of internal server errors.')]#
property webservice_supports_lead_time_in_get_timeseries: bool#

Wether or not the leadTime parameter is supported.

This determines the forecast retrieval method.

validate_forecast_retrieval_method()[source]#

Validate that the configures retrieval method is compatible with the webservice.

Return type:

FewsWebserviceConfig

property lead_times: LeadTimes | None#
property verification_period: TimePeriod#
property verification_period_on_frt: TimePeriod#
property verification_period_on_time: TimePeriod#
source#
data_type#
general#
id_mapping#
class veriflow.datasources.fewswebservice.FewsWebserviceAuthConfig(_case_sensitive=None, _nested_model_default_partial_update=None, _env_prefix=None, _env_prefix_target=None, _env_file=PosixPath('.'), _env_file_encoding=None, _env_ignore_empty=None, _env_nested_delimiter=None, _env_nested_max_split=None, _env_parse_none_str=None, _env_parse_enums=None, _cli_prog_name=None, _cli_parse_args=None, _cli_settings_source=None, _cli_parse_none_str=None, _cli_hide_none_type=None, _cli_avoid_json=None, _cli_enforce_required=None, _cli_use_class_docs_for_groups=None, _cli_exit_on_error=None, _cli_prefix=None, _cli_flag_prefix_char=None, _cli_implicit_flags=None, _cli_ignore_unknown_args=None, _cli_kebab_case=None, _cli_shortcuts=None, _secrets_dir=None, _build_sources=None, *, url, username, password)[source]#

Get url, username and password safely from environment variables.

This config class inherits from pydantic_settings.BaseSettings, that will try to infer field values from environment variables.

Environment variables:

  • FEWSWEBSERVICE_URL: URL of the FEWS webservice (required).

  • FEWSWEBSERVICE_USERNAME: Username for the FEWS webservice (set to “” if not required).

  • FEWSWEBSERVICE_PASSWORD: Password for the FEWS webservice (set to “” if not required).

see: https://docs.pydantic.dev/latest/concepts/pydantic_settings/#usage

Notes

For local development, an easy and recommended way to set environment variables is to use a .env file in the project root with the required environment variables, and they will be automatically loaded when you instantiate this config class. For example:

from dotenv import load_dotenv
load_dotenv()  # Load environment variables from .env file

# You can check that the variables are loaded correctly
import os
print(os.getenv("FEWSWEBSERVICE_URL"))
Parameters:

  • _case_sensitive (bool | None)

  • _nested_model_default_partial_update (bool | None)

  • _env_prefix (str | None)

  • _env_prefix_target (EnvPrefixTarget | None)

  • _env_file (DotenvType | None)

  • _env_file_encoding (str | None)

  • _env_ignore_empty (bool | None)

  • _env_nested_delimiter (str | None)

  • _env_nested_max_split (int | None)

  • _env_parse_none_str (str | None)

  • _env_parse_enums (bool | None)

  • _cli_prog_name (str | None)

  • _cli_parse_args (bool | list[str] | tuple[str, ...] | None)

  • _cli_settings_source (CliSettingsSource[Any] | None)

  • _cli_parse_none_str (str | None)

  • _cli_hide_none_type (bool | None)

  • _cli_avoid_json (bool | None)

  • _cli_enforce_required (bool | None)

  • _cli_use_class_docs_for_groups (bool | None)

  • _cli_exit_on_error (bool | None)

  • _cli_prefix (str | None)

  • _cli_flag_prefix_char (str | None)

  • _cli_implicit_flags (bool | Literal['dual', 'toggle'] | None)

  • _cli_ignore_unknown_args (bool | None)

  • _cli_kebab_case (bool | Literal['all', 'no_enums'] | None)

  • _cli_shortcuts (Mapping[str, str | list[str]] | None)

  • _secrets_dir (PathType | None)

  • _build_sources (tuple[tuple[PydanticBaseSettingsSource, ...], dict[str, Any]] | None)

  • url (AnyUrl)

  • username (SecretStr)

  • password (SecretStr)

url: AnyUrl#
username: SecretStr#
password: SecretStr#
On this page
Show Source