Python wrappers for the Deltares FEWS PI REST Web Service

get_filters(url, filter_id=None, document_format='PI_JSON', verify=False, logger=LOGGER)

Get FEWS qualifiers as a pandas DataFrame

Parameters:

Name	Type	Description	Default
url	str	url Delft-FEWS PI REST WebService.	required
E.g.	http	//localhost:8080/FewsWebServices/rest/fewspiservice/v1/filters	required
filter_id	str	the FEWS id of the filter to pass as request parameter	None
document_format	str	request document format to return. Defaults to PI_JSON.	'PI_JSON'
verify	bool	passed to requests.get verify parameter.	False
logger	Logger	Logger to pass logging to. By	LOGGER

Returns:

Name	Type	Description
df	DataFrame	Pandas dataframe with index "id" and columns
	List[dict]	"name" and "group_id".

Source code in src\fewspy\wrappers\get_filters.py

def get_filters(
    url: str,
    filter_id: str = None,
    document_format: str = "PI_JSON",
    verify: bool = False,
    logger=LOGGER,
) -> List[dict]:
    """
    Get FEWS qualifiers as a pandas DataFrame

    Args:
        url (str): url Delft-FEWS PI REST WebService.
        E.g. http://localhost:8080/FewsWebServices/rest/fewspiservice/v1/filters
        filter_id (str): the FEWS id of the filter to pass as request parameter
        document_format (str): request document format to return. Defaults to PI_JSON.
        verify (bool, optional): passed to requests.get verify parameter.
        Defaults to False.
        logger (logging.Logger, optional): Logger to pass logging to. By
        default, a logger will ge created.

    Returns:
        df (pandas.DataFrame): Pandas dataframe with index "id" and columns
        "name" and "group_id".

    """

    # do the request
    timer = Timer(logger)
    parameters = parameters_to_fews(locals())
    response = requests.get(url, parameters, verify=verify)
    timer.report("Filters request")

    # parse the response
    result = []
    if response.status_code == 200:
        if "filters" in response.json().keys():
            result = response.json()["filters"]
        timer.report("Filters parsed")
    else:
        logger.error(f"FEWS Server responds {response.text}")

    return result

get_locations(url, filter_id=None, document_format='PI_JSON', attributes=[], verify=False, logger=LOGGER, remove_duplicates=False)

Get FEWS qualifiers as a pandas DataFrame

Parameters:

Name	Type	Description	Default
url	str	url Delft-FEWS PI REST WebService.	required
E.g.	http	//localhost:8080/FewsWebServices/rest/fewspiservice/v1/qualifiers	required
filter_id	str	the FEWS id of the filter to pass as request parameter	None
document_format	str	request document format to return. Defaults to PI_JSON.	'PI_JSON'
attributes	list	if not emtpy, the location attributes to include as columns in the pandas DataFrame.	[]
verify	bool	passed to requests.get verify parameter.	False
logger	Logger	Logger to pass logging to. By default, a logger will ge created.	LOGGER

Returns:

Name	Type	Description
df	DataFrame	Pandas dataframe with index "id" and columns
	DataFrame	"name" and "group_id".

Source code in src\fewspy\wrappers\get_locations.py

def get_locations(
    url: str,
    filter_id: str = None,
    document_format: str = "PI_JSON",
    attributes: list = [],
    verify: bool = False,
    logger=LOGGER,
    remove_duplicates: bool = False
) -> pd.DataFrame:
    """
    Get FEWS qualifiers as a pandas DataFrame

    Args:
        url (str): url Delft-FEWS PI REST WebService.
        E.g. http://localhost:8080/FewsWebServices/rest/fewspiservice/v1/qualifiers
        filter_id (str): the FEWS id of the filter to pass as request parameter
        document_format (str): request document format to return. Defaults to PI_JSON.
        attributes (list): if not emtpy, the location attributes to include as columns in the pandas DataFrame.
        verify (bool, optional): passed to requests.get verify parameter.
        Defaults to False.
        logger (logging.Logger, optional): Logger to pass logging to. By default, a logger will ge created.

    Returns:
        df (pandas.DataFrame): Pandas dataframe with index "id" and columns
        "name" and "group_id".

    """

    # do the request
    timer = Timer(logger)
    parameters = parameters_to_fews(locals())
    response = requests.get(url, parameters, verify=verify)
    timer.report("Locations request")

    # parse the response
    if response.status_code == 200:
        # convert to gdf and snake_case
        gdf = gpd.GeoDataFrame(response.json()["locations"])
        gdf.columns = [camel_to_snake_case(i) for i in gdf.columns]

        # remove duplicates
        if remove_duplicates:
            gdf.drop_duplicates(subset="location_id", inplace=True, ignore_index=True)

        # set index
        gdf.set_index("location_id", inplace=True)

        # handle geometry and crs
        gdf["geometry"] = xy_array_to_point(gdf[["x", "y"]].values)
        gdf.crs = geo_datum_to_crs(response.json()["geoDatum"])

        # handle attributes
        if attributes:
            gdf.loc[:, attributes] = attributes_to_array(
                gdf["attributes"].values, attributes
            )
        gdf.drop(columns=["attributes"], inplace=True)

        timer.report("Locations parsed")

    else:
        logger.error(f"FEWS Server responds {response.text}")
        gdf = gpd.GeoDataFrame()

    return gdf

get_parameters(url, filter_id=None, document_format='PI_JSON', verify=False, logger=LOGGER)

Get FEWS qualifiers as a pandas DataFrame

Parameters:

Name	Type	Description	Default
url	str	url Delft-FEWS PI REST WebService.	required
E.g.	http	//localhost:8080/FewsWebServices/rest/fewspiservice/v1/qualifiers	required
filter_id	str	the FEWS id of the filter to pass as request parameter	None
document_format	str	request document format to return. Defaults to PI_JSON.	'PI_JSON'
verify	bool	passed to requests.get verify parameter.	False
logger	Logger	Logger to pass logging to. By	LOGGER

Returns:

Name	Type	Description
df	DataFrame	Pandas dataframe with index "id" and columns
	List[dict]	"name" and "group_id".

Source code in src\fewspy\wrappers\get_parameters.py

def get_parameters(
    url: str,
    filter_id: str = None,
    document_format: str = "PI_JSON",
    verify: bool = False,
    logger=LOGGER,
) -> List[dict]:
    """
    Get FEWS qualifiers as a pandas DataFrame

    Args:
        url (str): url Delft-FEWS PI REST WebService.
        E.g. http://localhost:8080/FewsWebServices/rest/fewspiservice/v1/qualifiers
        filter_id (str): the FEWS id of the filter to pass as request parameter
        document_format (str): request document format to return. Defaults to PI_JSON.
        verify (bool, optional): passed to requests.get verify parameter.
        Defaults to False.
        logger (logging.Logger, optional): Logger to pass logging to. By
        default, a logger will ge created.

    Returns:
        df (pandas.DataFrame): Pandas dataframe with index "id" and columns
        "name" and "group_id".

    """

    # do the request
    timer = Timer(logger)
    parameters = parameters_to_fews(locals())
    response = requests.get(url, parameters, verify=verify)
    timer.report("Parameters request")

    # parse the response
    df = pd.DataFrame(columns=COLUMNS)
    if response.status_code == 200:
        if "timeSeriesParameters" in response.json().keys():
            df = pd.DataFrame(response.json()["timeSeriesParameters"])
            df.columns = [camel_to_snake_case(i) for i in df.columns]
            df["uses_datum"] = df["uses_datum"] == "true"
            timer.report("Parameters parsed")
    else:
        logger.error(f"FEWS Server responds {response.text}")

    df.set_index("id", inplace=True)

    return df

get_qualifiers(url, verify=False, logger=LOGGER)

Get FEWS qualifiers as Pandas DataFrame

Parameters:

Name	Type	Description	Default
url	str	url Delft-FEWS PI REST WebService.	required
E.g.	http	//localhost:8080/FewsWebServices/rest/fewspiservice/v1/qualifiers	required
verify	bool	passed to requests.get verify parameter.	False
logger	Logger	Logger to pass logging to. By	LOGGER

Returns:

Name	Type	Description
df	DataFrame	Pandas dataframe with index "id" and columns
	DataFrame	"name" and "group_id".

Source code in src\fewspy\wrappers\get_qualifiers.py

def get_qualifiers(url: str, verify: bool = False, logger=LOGGER) -> pd.DataFrame:
    """
    Get FEWS qualifiers as Pandas DataFrame

    Args:
        url (str): url Delft-FEWS PI REST WebService.
        E.g. http://localhost:8080/FewsWebServices/rest/fewspiservice/v1/qualifiers
        verify (bool, optional): passed to requests.get verify parameter.
        Defaults to False.
        logger (logging.Logger, optional): Logger to pass logging to. By
        default, a new logger will ge created.

    Returns:
        df (pandas.DataFrame): Pandas dataframe with index "id" and columns
        "name" and "group_id".

    """

    # do the request
    timer = Timer(logger)
    response = requests.get(url, verify=False)
    timer.report("Qualifiers request")

    # parse the response
    if response.status_code == 200:
        tree = ElementTree.fromstring(response.content)
        qualifiers_tree = [i for i in tree.iter(tag=f"{NS}qualifier")]
        qualifiers_tuple = (_element_to_tuple(i) for i in qualifiers_tree)
        df = pd.DataFrame(qualifiers_tuple, columns=COLUMNS)
        timer.report("Qualifiers parsed")
    else:
        logger.error(f"FEWS Server responds {response.text}")
        df = pd.DataFrame(columns=COLUMNS)
    df.set_index("id", inplace=True)

    return df

get_time_series(url, filter_id, location_ids=None, parameter_ids=None, qualifier_ids=None, start_time=None, end_time=None, thinning=None, only_headers=False, omit_missing=True, show_statistics=False, document_format='PI_JSON', verify=False, logger=LOGGER)

Get FEWS qualifiers as a pandas DataFrame

Parameters:

Name	Type	Description	Default
url	str	url Delft-FEWS PI REST WebService.	required
E.g.	http	//localhost:8080/FewsWebServices/rest/fewspiservice/v1/qualifiers	required
filter_id	str	the FEWS id of the filter to pass as request parameter	required
location_ids	list	list with FEWS location ids to extract timeseries from. Defaults to None.	None
parameter_ids	list	list with FEWS parameter ids to extract timeseries from. Defaults to None.	None
qualifier_ids	list	list with FEWS qualifier ids to extract timeseries from. Defaults to None.	None
start_time	datetime	datetime-object with start datetime to use in request. Defaults to None.	None
end_time	datetime	datetime-object with end datetime to use in request. Defaults to None.	None
thinning	int	integer value for thinning parameter to use in request. Defaults to None.	None
only_headers	bool	if True, only headers will be returned. Defaults to False.	False
omit_missing	bool	if True, no missings values will be returned. Defaults to True.	True
show_statistics	bool	if True, time series statistics will be included in header. Defaults to False.	False
document_format	str	request document format to return. Defaults to PI_JSON.	'PI_JSON'
verify	bool	passed to requests.get verify parameter.	False
logger	Logger	Logger to pass logging to. By	LOGGER

Returns:

Name	Type	Description
df	DataFrame	Pandas dataframe with index "id" and columns
	DataFrame	"name" and "group_id".

Source code in src\fewspy\wrappers\get_time_series.py

def get_time_series(
    url: str,
    filter_id: str,
    location_ids: Union[str, List[str]] = None,
    parameter_ids: Union[str, List[str]] = None,
    qualifier_ids: Union[str, List[str]] = None,
    start_time: datetime = None,
    end_time: datetime = None,
    thinning: int = None,
    only_headers: bool = False,
    omit_missing: bool = True,
    show_statistics: bool = False,
    document_format: str = "PI_JSON",
    verify: bool = False,
    logger=LOGGER,
) -> pd.DataFrame:
    """
    Get FEWS qualifiers as a pandas DataFrame

    Args:
        url (str): url Delft-FEWS PI REST WebService.
        E.g. http://localhost:8080/FewsWebServices/rest/fewspiservice/v1/qualifiers
        filter_id (str): the FEWS id of the filter to pass as request parameter
        location_ids (list): list with FEWS location ids to extract timeseries from. Defaults to None.
        parameter_ids (list): list with FEWS parameter ids to extract timeseries from. Defaults to None.
        qualifier_ids (list): list with FEWS qualifier ids to extract timeseries from. Defaults to None.
        start_time (datetime.datetime): datetime-object with start datetime to use in request. Defaults to None.
        end_time (datetime.datetime): datetime-object with end datetime to use in request. Defaults to None.
        thinning (int): integer value for thinning parameter to use in request. Defaults to None.
        only_headers (bool): if True, only headers will be returned. Defaults to False.
        omit_missing (bool): if True, no missings values will be returned. Defaults to True.
        show_statistics (bool): if True, time series statistics will be included in header. Defaults to False.
        document_format (str): request document format to return. Defaults to PI_JSON.
        verify (bool, optional): passed to requests.get verify parameter.
        Defaults to False.
        logger (logging.Logger, optional): Logger to pass logging to. By
        default, a logger will ge created.

    Returns:
        df (pandas.DataFrame): Pandas dataframe with index "id" and columns
        "name" and "group_id".

    """
    report_string = _ts_or_headers(only_headers)

    # do the request
    timer = Timer(logger)
    parameters = parameters_to_fews(locals())
    response = requests.get(url, parameters, verify=verify)
    timer.report(report_string.format(status="request"))

    # parse the response
    if response.ok:
        pi_time_series = response.json()
        logger.debug(response.url)
        time_series_set = TimeSeriesSet.from_pi_time_series(pi_time_series)
        timer.report(report_string.format(status="parsed"))
        if time_series_set.empty:
            logger.debug(f"FEWS WebService request passing empty set: {response.url}")
    else:
        logger.error(f"FEWS WebService request {response.url} responds {response.text}")
        time_series_set = TimeSeriesSet()

    return time_series_set

get_time_series_async(url, filter_id, location_ids=None, parameter_ids=None, qualifier_ids=None, start_time=None, end_time=None, thinning=None, document_format='PI_JSON', omit_missing=True, verify=False, logger=LOGGER)

Parameters:

Name	Type	Description	Default
url	str	url Delft-FEWS PI REST WebService.	required
E.g.	http	//localhost:8080/FewsWebServices/rest/fewspiservice/v1/qualifiers	required
filter_id	str	the FEWS id of the filter to pass as request parameter	required
location_ids	list	list with FEWS location ids to extract timeseries from. Defaults to None.	None
parameter_ids	list	list with FEWS parameter ids to extract timeseries from. Defaults to None.	None
qualifier_ids	list	list with FEWS qualifier ids to extract timeseries from. Defaults to None.	None
start_time	datetime	datetime-object with start datetime to use in request. Defaults to None.	None
end_time	datetime	datetime-object with end datetime to use in request. Defaults to None.	None
thinning	int	integer value for thinning parameter to use in request. Defaults to None.	None
document_format	str	request document format to return. Defaults to PI_JSON.	'PI_JSON'
omit_missing	bool	if True, no missings values will be returned. Defaults to True	True
verify	bool	passed to requests.get verify parameter.	False
logger	Logger	Logger to pass logging to. By	LOGGER

Returns:

Name	Type	Description
df	DataFrame	Pandas dataframe with index "id" and columns
	DataFrame	"name" and "group_id".

Source code in src\fewspy\wrappers\get_time_series_async.py

def get_time_series_async(
    url: str,
    filter_id: str,
    location_ids: Union[str, List[str]] = None,
    parameter_ids: Union[str, List[str]] = None,
    qualifier_ids: Union[str, List[str]] = None,
    start_time: datetime = None,
    end_time: datetime = None,
    thinning: int = None,
    document_format: str = "PI_JSON",
    omit_missing: bool = True,
    verify: bool = False,
    logger=LOGGER,
) -> pd.DataFrame:
    """

    Args:
        url (str): url Delft-FEWS PI REST WebService.
        E.g. http://localhost:8080/FewsWebServices/rest/fewspiservice/v1/qualifiers
        filter_id (str): the FEWS id of the filter to pass as request parameter
        location_ids (list): list with FEWS location ids to extract timeseries from. Defaults to None.
        parameter_ids (list): list with FEWS parameter ids to extract timeseries from. Defaults to None.
        qualifier_ids (list): list with FEWS qualifier ids to extract timeseries from. Defaults to None.
        start_time (datetime.datetime): datetime-object with start datetime to use in request. Defaults to None.
        end_time (datetime.datetime): datetime-object with end datetime to use in request. Defaults to None.
        thinning (int): integer value for thinning parameter to use in request. Defaults to None.
        document_format (str): request document format to return. Defaults to PI_JSON.
        omit_missing (bool): if True, no missings values will be returned. Defaults to True
        verify (bool, optional): passed to requests.get verify parameter.
        Defaults to False.
        logger (logging.Logger, optional): Logger to pass logging to. By
        default, a logger will ge created.

    Returns:
        df (pandas.DataFrame): Pandas dataframe with index "id" and columns
        "name" and "group_id".

    """
    parameters = parameters_to_fews(locals(), bool_to_string=True)

    def _get_loop():
        try:
            loop = asyncio.get_event_loop()
        except RuntimeError:
            loop = asyncio.new_event_loop()
            asyncio.set_event_loop(loop)
        finally:
            loop.set_debug(True)
            return loop

    async def get_timeseries_async(location_id, parameter_id, qualifier_id, session):
        """Get timerseries using FEWS (asynchronously)"""
        parameters["locationIds"] = [location_id]
        parameters["parameterIds"] = [parameter_id]
        if qualifier_id is not None:
            parameters["qualifierIds"] = qualifier_id
        try:
            response = await session.request(
                method="GET", url=url, params=parameters, verify_ssl=verify
            )
            response.raise_for_status()
        except Exception as err:
            logger.error(
                f"An error ocurred: {err} while executing url {url} with parameters {parameters}"
            )
            response = None
        response_json = await response.json()
        return response_json

    async def run_program(location_id, parameter_id, qualifier_id, session):
        """Wrapper for running program in an asynchronous manner"""
        try:
            response = await get_timeseries_async(
                location_id, parameter_id, qualifier_id, session
            )
        except Exception as err:
            logger.error(f"Exception occured: {err}")
            response = None
            pass
        return response

    async def asynciee():
        async with aiohttp.ClientSession(loop=loop) as session:
            args = [
                (location_id, parameter_id)
                for location_id in location_ids
                for parameter_id in parameter_ids
            ]
            if qualifier_ids is None:
                args = [(*i, None) for i in args]
            else:
                args = [
                    (*i, qualifier_id) for i in args for qualifier_id in qualifier_ids
                ]
            fetch_all = [run_program(*i, session) for i in args]
            result_async = await asyncio.gather(*fetch_all)
            return result_async

    if __name__ == "fewspy.wrappers.get_time_series_async":
        loop = _get_loop()
        result_async = loop.run_until_complete(asynciee())
        time_series_set = __result_async_to_time_series_set(result_async)
    return time_series_set