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 list of typed PI locations containing location identifiers, coordinates, names, and optional relations or attributes. Response envelope metadata such as PI document version is used for validation but not returned by this simplified public method.

Return type:

list[PiLocation]

Example

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

locations = client.get_locations()
first_location = 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 list of typed PI parameters containing metadata such as parameter IDs, units, parameter type, and optional attributes. Response envelope metadata such as PI document version is used for validation but not returned by this simplified public method.

Return type:

list[PiParameter]

Example

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

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

print(first_parameter.id)
print(first_parameter.unit)
get_timeseries(*, location_ids=None, parameter_ids=None, start_time=None, end_time=None, document_format='PI_NETCDF', **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.

  • document_format (str) – FEWS PI response format. Supported values are PI_JSON, PI_XML, PI_CSV and PI_NETCDF. Defaults to PI_NETCDF when omitted.

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

Returns:

A list of xarray.Dataset objects for PI_NETCDF responses, preserving the original NetCDF member layout and ZIP member order; a dictionary for PI_JSON; or a string for PI_XML and PI_CSV.

Return type:

list[Dataset] | dict[str, Any] | str

Example

Request time series as NetCDF. The ZIP payload returned by FEWS is unpacked automatically and returned as a list of xarray.Dataset objects.

from datetime import datetime, timezone

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

datasets = 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),
)

first_dataset = datasets[0]
print(first_dataset)

When FEWS writes multiple NetCDF files for a single request, each member is returned as a separate dataset so you can choose your own merge strategy.

datasets = 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),
)

merged = xr.merge(datasets, combine_attrs="override")
print(merged)

Request raw PI JSON explicitly.

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),
    document_format="PI_JSON",
)

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

By default, get_timeseries() requests PI_NETCDF and returns a list[xarray.Dataset], preserving the original NetCDF member layout returned by FEWS.

PI JSON responses are returned as raw dictionaries. Use PI_NETCDF when you want the wrapper to return one or more xarray.Dataset objects.

post_timeseries(*, pi_time_series_xml_content=None, pi_time_series_json_content=None, filter_id=None, convert_datum=None)[source]#

Write PI time series data to the FEWS web services using POST.

The FEWS POST /timeseries endpoint writes PI time series that belong to time series sets configured in the default filter or in the filter identified by filter_id. Provide the PI XML or PI JSON content to be written through the dedicated content arguments.

Parameters:

  • pi_time_series_xml_content (str | None) – Optional PI XML payload to write.

  • pi_time_series_json_content (str | None) – Optional PI JSON payload to write.

  • filter_id (str | None) – Optional FEWS filter identifier restricting which time series sets may be written.

  • convert_datum (bool | None) – Optional FEWS convert-datum flag.

Returns:

A PI diagnostic XML string describing the import result.

Return type:

str

Example

Post PI XML content.

from pathlib import Path

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

xml_payload = Path("tests/test_data/post_timeseries.xml").read_text(
    encoding="utf-8"
)

diag_xml = client.post_timeseries(
    pi_time_series_xml_content=xml_payload,
    filter_id="MEAS",
)

print(diag_xml)

Post PI JSON content.

from pathlib import Path

json_payload = Path("tests/test_data/post_timeseries.json").read_text(
    encoding="utf-8"
)

diag_xml = client.post_timeseries(
    pi_time_series_json_content=json_payload,
)

print(diag_xml)
get_filters(filter_id=None, *, document_format='PI_JSON', document_version=None)[source]#

Get filters from the FEWS web services.

Retrieves filters that are subfilters of the default filter. An existing subfilter ID can be specified to narrow the results.

Parameters:

  • filter_id (str | None) – Optional FEWS filter identifier. When provided, only subfilters of this filter are returned.

  • document_format (str | None) – Response format supported by the FEWS filters endpoint. Defaults to PI_JSON.

  • document_version (str | None) – Optional PI document version.

Returns:

A list of typed PI filters for PI_JSON by default, or a string when a text-based format such as PI_XML is requested. Response envelope metadata such as PI document version is used for validation but not returned by this simplified public method.

Return type:

list[PiFilter] | str

Example

Retrieve all available filters.

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

filters = client.get_filters()
print(filters[0].id)

Retrieve subfilters of a specific filter.

filters = client.get_filters(filter_id="MEAS")
print(list(filters))
post_runtask(*, workflow_id, start_time=None, end_time=None, time_zero=None, cold_state_id=None, scenario_id=None, user_id=None, description=None, run_option=None, run_locally_and_promote_to_server=None, pi_parameters_xml_content=None)[source]#

Run a one-off FEWS task for a workflow and return the task ID.

This wraps FEWS POST /runtask using application/x-www-form-urlencoded request encoding. FEWS returns a plain-text taskId that can be used to track the task status.

Parameters:

  • workflow_id (str) – Required FEWS workflow identifier.

  • start_time (datetime | None) – Optional workflow start time. Must be timezone-aware.

  • end_time (datetime | None) – Optional workflow end time. Must be timezone-aware.

  • time_zero (datetime | None) – Optional forecast time zero. Must be timezone-aware.

  • cold_state_id (str | None) – Optional FEWS cold-state identifier.

  • scenario_id (str | None) – Optional FEWS scenario identifier.

  • user_id (str | None) – Optional FEWS user identifier.

  • description (str | None) – Optional task description stored by FEWS.

  • run_option (str | None) – Optional FEWS run option. Supported values are all, allmostrecentonly, and alloneatatime.

  • run_locally_and_promote_to_server (bool | None) – Optional FEWS execution flag.

  • pi_parameters_xml_content (str | None) – Optional PI model parameters XML content.

Returns:

The FEWS task identifier returned by POST /runtask.

Return type:

str

Example

from datetime import datetime, timezone

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

task_id = client.post_runtask(
    workflow_id="ImportObscape",
    start_time=datetime(2025, 3, 18, 15, 0, tzinfo=timezone.utc),
    end_time=datetime(2025, 3, 18, 16, 0, tzinfo=timezone.utc),
    description="Run ImportObscape once from the wrapper",
    run_option="all",
)

print(task_id)
get_taskruns(*, workflow_id, topology_node_id=None, forecast_count=None, task_run_ids=None, scenario_id=None, mc_id=None, start_forecast_time=None, end_forecast_time=None, start_dispatch_time=None, end_dispatch_time=None, task_run_status_ids=None, only_forecasts=None, task_run_count=None, only_current=None, document_format='PI_JSON', document_version=None)[source]#

Get task runs for a FEWS workflow.

Retrieves task runs from FEWS GET /taskruns for the specified workflow_id, optionally filtered by identifiers, status, forecast time, or dispatch time.

FEWS returns only forecast task runs by default. As a result, non-forecast workflows can legitimately produce an empty task_runs list unless you pass only_forecasts=False.

Parameters:

  • workflow_id (str) – Required FEWS workflow identifier.

  • topology_node_id (str | None) – Optional FEWS topology-node identifier.

  • forecast_count (int | str | None) – Optional forecast-count filter accepted by FEWS.

  • task_run_ids (list[str] | None) – Optional FEWS task-run IDs to filter by.

  • scenario_id (str | None) – Optional FEWS scenario identifier.

  • mc_id (str | None) – Optional FEWS MC identifier.

  • start_forecast_time (datetime | None) – Optional inclusive forecast-time lower bound. Must be timezone-aware.

  • end_forecast_time (datetime | None) – Optional inclusive forecast-time upper bound. Must be timezone-aware.

  • start_dispatch_time (datetime | None) – Optional inclusive dispatch-time lower bound. Must be timezone-aware.

  • end_dispatch_time (datetime | None) – Optional inclusive dispatch-time upper bound. Must be timezone-aware.

  • task_run_status_ids (list[str] | None) – Optional FEWS task-run status identifiers.

  • only_forecasts (bool | None) – Optional FEWS forecast-only filter. When omitted, FEWS may default to returning only forecast task runs.

  • task_run_count (int | str | None) – Optional maximum number of returned task runs.

  • only_current (bool | None) – Optional FEWS current-only filter.

  • document_format (str | None) – Response format. Defaults to PI_JSON.

  • document_version (str | None) – Optional PI document version.

Returns:

A list of typed task-run descriptors for PI_JSON by default, or a string when a text-based format such as PI_XML is requested.

Return type:

list[PiTaskRun] | str

Example

Retrieve the latest forecast task runs for a workflow.

taskruns = client.get_taskruns(
    workflow_id="ImportObscape",
    task_run_count=10,
)

for task_run in taskruns:
    print(task_run.id, task_run.status, task_run.dispatch_time)

Retrieve task runs for a non-forecast workflow.

taskruns = client.get_taskruns(
    workflow_id="ftpClientConfig",
    only_forecasts=False,
    task_run_count=10,
)

print(taskruns)

Retrieve the raw PI XML response.

taskruns_xml = client.get_taskruns(
    workflow_id="ImportObscape",
    document_format="PI_XML",
)
print(taskruns_xml)
get_taskrunstatus(*, task_id, max_wait_millis=None, document_format='PI_JSON', document_version=None)[source]#

Get the status of a FEWS task run.

Retrieves the status of a FEWS task run from GET /taskrunstatus. The current OpenAPI specification exposes only PI_JSON for this endpoint, and the wrapper returns a typed status response.

Possible FEWS status codes are I (invalid), P (pending), T (terminated), R (running), F (failed), C (completed fully successful), D (completed partly successful), A (approved), and B (approved partly successful).

Parameters:

  • task_id (str) – Required FEWS task identifier returned by post_runtask.

  • max_wait_millis (int | str | None) – Optional FEWS long-poll timeout in milliseconds.

  • document_format (str | None) – Response format. The current specification supports PI_JSON.

  • document_version (str | None) – Optional PI document version.

Returns:

A validated typed task-run-status response.

Return type:

PiTaskRunStatusResponse

Example

task_id = client.post_runtask(workflow_id="ImportObscape")

status = client.get_taskrunstatus(
    task_id=task_id,
    max_wait_millis=1000,
)

print(status.code, status.description, status.task_run_id)
get_whatiftemplates(*, what_if_template_id=None, document_format='PI_JSON', document_version=None)[source]#

Get FEWS what-if templates.

Retrieves the configured what-if templates from GET /whatiftemplates. The current OpenAPI specification exposes PI_JSON for this endpoint, and the wrapper returns a typed templates response.

Parameters:

  • what_if_template_id (str | None) – Optional FEWS what-if template identifier used to narrow the response to a single template.

  • document_format (str | None) – Response format. The current specification supports PI_JSON.

  • document_version (str | None) – Optional PI document version.

Returns:

A list of typed what-if-template descriptors.

Return type:

list[PiWhatIfTemplate]

Example

Retrieve all what-if templates.

templates = client.get_whatiftemplates()

for template in templates:
    print(template.id, template.name)

Retrieve one specific what-if template by ID.

templates = client.get_whatiftemplates(
    what_if_template_id="sfincs_palmiet_scenario_map",
)

print(templates[0].properties)
get_whatifscenarios(*, what_if_template_id=None, what_if_scenario_id=None, workflow_id=None, document_format='PI_JSON', document_version=None)[source]#

Get FEWS what-if scenarios.

Retrieves the configured what-if scenarios from GET /whatifscenarios. The current OpenAPI specification exposes PI_JSON for this endpoint, and the wrapper returns a typed scenarios response.

Parameters:

  • what_if_template_id (str | None) – Optional FEWS what-if template identifier used to return scenarios belonging to one template.

  • what_if_scenario_id (str | None) – Optional FEWS what-if scenario identifier used to return one specific scenario descriptor.

  • workflow_id (str | None) – Optional FEWS workflow identifier used to return scenarios linked to one workflow.

  • document_format (str | None) – Response format. The current specification supports PI_JSON.

  • document_version (str | None) – Optional PI document version.

Returns:

A list of typed what-if-scenario descriptors.

Return type:

list[PiWhatIfScenarioDescriptor]

Example

Retrieve all what-if scenarios.

scenarios = client.get_whatifscenarios()

for scenario in scenarios:
    print(scenario.id, scenario.name)

Retrieve one specific what-if scenario by ID.

scenarios = client.get_whatifscenarios(
    what_if_scenario_id="SA107:2",
)

print(scenarios[0].what_if_template_id)
post_whatifscenarios(*, what_if_template_id=None, single_run_what_if=None, name=None, document_format='PI_JSON', document_version=None)[source]#

Create a FEWS what-if scenario.

Wraps POST /whatifscenarios using the parameters currently exposed by the generated FEWS OpenAPI client. The current specification exposes query parameters such as what_if_template_id, single_run_what_if, and name, and returns a JSON descriptor of the created scenario.

Parameters:

  • what_if_template_id (str | None) – Optional FEWS what-if template identifier.

  • single_run_what_if (bool | None) – Optional FEWS single-run what-if flag.

  • name (str | None) – Optional FEWS what-if scenario name.

  • document_format (str | None) – Response format. The current specification supports PI_JSON.

  • document_version (str | None) – Optional PI document version.

Returns:

A validated typed descriptor for the created what-if scenario.

Return type:

PiWhatIfScenarioDescriptor

Example

scenario = client.post_whatifscenarios(
    what_if_template_id="sfincs_palmiet_scenario_map",
    name="Wrapper what-if scenario",
    single_run_what_if=False,
)

print(scenario.id, scenario.name, scenario.what_if_template_id)

Note

The current generated FEWS OpenAPI client does not expose a request body for setting scenario properties on POST /whatifscenarios. This wrapper therefore forwards only the parameters defined by that client.

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

Backward-compatible alias for post_runtask().

Parameters:

  • args (Any)

  • kwargs (Any)

Return type:

str

get_workflows(*, document_format='PI_JSON', document_version=None)[source]#

Get available FEWS workflows.

Retrieves the default workflow XML files exposed by the FEWS /workflows endpoint.

Parameters:

  • document_format (str | None) – Response format supported by the FEWS workflows endpoint. Defaults to PI_JSON.

  • document_version (str | None) – Optional PI document version.

Returns:

A list of typed workflow descriptors for PI_JSON by default, or a string when a text-based format such as PI_XML is requested.

Return type:

list[PiWorkflow] | str

Example

Retrieve the available workflows as PI JSON.

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

workflows = client.get_workflows()
print(workflows[0].id)

Retrieve the raw PI XML response.

workflows_xml = client.get_workflows(document_format="PI_XML")
print(workflows_xml)
endpoint_arguments(endpoint)[source]#

Get the arguments for a specific FEWS web service endpoint.

Parameters:

endpoint (str) – The name of the endpoint, options: timeseries, post_timeseries, post_runtask, taskruns, taskrunstatus, whatiftemplates, whatifscenarios, post_whatifscenarios, filters, and workflows.

Returns:

The argument names for the specified endpoint.

Return type:

list[str]