FewsWebServiceClient#

class fews_py_wrapper.fews_webservices.FewsWebServiceClient(base_url, authenticate=False, token=None, verify_ssl=True)[source]#

Client for interacting with FEWS web services.

Parameters:

  • base_url (str)

  • authenticate (bool)

  • token (str | None)

  • verify_ssl (bool)

client: Client | AuthenticatedClient#
authenticate(token, verify_ssl)[source]#

Authenticate with the FEWS web services.

Parameters:

  • token (str)

  • verify_ssl (bool)

Return type:

None

get_locations()[source]#

Get locations from the FEWS web services as a typed PI model.

Returns:

A validated PI locations response containing location identifiers, coordinates, names, and optional relations or attributes.

Return type:

PiLocationsResponse

Example

client = FewsWebServiceClient(
    base_url="https://example.com/FewsWebServices/rest"
)

locations = client.get_locations()
first_location = locations.locations[0]

print(first_location.location_id)
print(first_location.lat, first_location.lon)
get_parameters()[source]#

Get parameters from the FEWS web services as a typed PI model.

Returns:

A validated PI parameters response containing parameter metadata such as parameter IDs, units, parameter type, and optional attributes.

Return type:

PiParametersResponse

Example

client = FewsWebServiceClient(
    base_url="https://example.com/FewsWebServices/rest"
)

parameters = client.get_parameters()
first_parameter = parameters.parameters[0]

print(first_parameter.id)
print(first_parameter.unit)
get_timeseries(*, location_ids=None, parameter_ids=None, start_time=None, end_time=None, to_xarray=False, document_format='PI_JSON', **kwargs)[source]#

Get time series data from the FEWS web services.

Parameters:

  • location_ids (list[str] | None) – One or more FEWS location identifiers.

  • parameter_ids (list[str] | None) – One or more FEWS parameter identifiers.

  • start_time (datetime | None) – Inclusive start timestamp. Must be timezone-aware.

  • end_time (datetime | None) – Inclusive end timestamp. Must be timezone-aware.

  • to_xarray (bool) – If True, convert the PI JSON response into an xarray.Dataset.

  • document_format (str | None) – FEWS response format. PI_JSON is currently the supported option.

  • **kwargs (Any) – Additional endpoint arguments accepted by the underlying FEWS time series endpoint.

Returns:

Either the raw PI JSON response as a dictionary, or an xarray.Dataset when to_xarray=True.

Return type:

Dataset | dict[str, Any]

Example

Request raw PI JSON for a single parameter and location.

from datetime import datetime, timezone

client = FewsWebServiceClient(
    base_url="https://example.com/FewsWebServices/rest"
)

response = client.get_timeseries(
    location_ids=["Amanzimtoti_River_level"],
    parameter_ids=["H.obs"],
    start_time=datetime(2025, 3, 14, 10, 0, tzinfo=timezone.utc),
    end_time=datetime(2025, 3, 15, 0, 0, tzinfo=timezone.utc),
)

print(response["timeSeries"][0]["header"]["parameterId"])

Request the same data and convert it to xarray for analysis.

dataset = client.get_timeseries(
    location_ids=["Amanzimtoti_River_level"],
    parameter_ids=["H.obs"],
    start_time=datetime(2025, 3, 14, 10, 0, tzinfo=timezone.utc),
    end_time=datetime(2025, 3, 15, 0, 0, tzinfo=timezone.utc),
    to_xarray=True,
)

print(dataset)
get_taskruns(workflow_id, task_ids=None)[source]#

Get the status of a task run in the FEWS web services.

Parameters:

  • workflow_id (str)

  • task_ids (list[str] | str | None)

Return type:

dict[str, Any]

execute_workflow(*args, **kwargs)[source]#

Execute a workflow in the FEWS web services.

Parameters:

  • args (Any)

  • kwargs (Any)

Return type:

None

execute_whatif_scenario(what_if_template_id=None, single_run_what_if=None, name=None, document_format=None, document_version=None)[source]#

Execute a what-if scenario in the FEWS web services.

Parameters:

  • what_if_template_id (str | None)

  • single_run_what_if (str | None)

  • name (str | None)

  • document_format (str | None)

  • document_version (str | None)

Return type:

dict[str, Any]

get_workflows()[source]#
Return type:

dict[str, Any]

endpoint_arguments(endpoint)[source]#

Get the arguments for a specific FEWS web service endpoint.

Parameters:

endpoint (str) – The name of the endpoint, options: “timeseries”, “taskruns”, “whatif_scenarios”, “workflows”.

Returns:

The argument names for the specified endpoint.

Return type:

list[str]