wps_parameters¶

wps_parameters is a process that runs the parameters module of a VIC streamflow routing model, called RVIC. To get started, first instatiate the client. Here, the client will try to connect to a remote osprey instance using the url parameter.¶

In [1]:
from birdy import WPSClient
import os
from datetime import datetime
from wps_tools.testing import get_target_url
from netCDF4 import Dataset
from rvic.core.config import read_config
from importlib.resources import files

# Ensure we are in the working directory with access to the data
while os.path.basename(os.getcwd()) != "osprey":
    os.chdir('../')
In [2]:
# NBVAL_IGNORE_OUTPUT
url = get_target_url("osprey")
print(f"Using osprey on {url}")

Using osprey on https://marble-dev01.pcic.uvic.ca/twitcher/ows/proxy/osprey/wps
In [3]:
osprey = WPSClient(url)

Help for individual processes can be diplayed using the ? command (ex. bird.process?).¶

In [4]:
# NBVAL_IGNORE_OUTPUT
osprey.parameters?

Signature:
osprey.parameters(
    case_id,
    grid_id,
    pour_points_csv=None,
    uh_box_csv=None,
    routing=None,
    domain=None,
    loglevel='INFO',
    np=1,
    version=True,
    params_config_file=None,
    params_config_dict=None,
)
Docstring:
Develop impulse response functions using inputs from a configuration file or dictionary

Parameters
----------
loglevel : {'CRITICAL', 'ERROR', 'WARNING', 'INFO', 'DEBUG', 'NOTSET'}string
    Logging level
np : integer
    Number of processors used to run job
version : boolean
    Return RVIC version string
case_id : string
    Case ID for the RVIC process
grid_id : string
    Routing domain grid shortname
pour_points_csv : ComplexData:mimetype:`text/plain`, :mimetype:`text/csv`
    Pour Points File content; A comma separated file of outlets to route to [lons, lats] Use open(filename).read() for local files and a URL for remote files.
uh_box_csv : ComplexData:mimetype:`text/plain`, :mimetype:`text/csv`
    UH Box File content. Use open(filename).read() for local files and a URL for remote files. This defines the unit hydrograph to rout flow to the edge of each grid cell.
routing : ComplexData:mimetype:`application/x-netcdf`, :mimetype:`application/x-ogc-dods`
    Path to routing inputs netCDF.
domain : ComplexData:mimetype:`application/x-netcdf`, :mimetype:`application/x-ogc-dods`
    Path to CESM complaint domain file
params_config_file : ComplexData:mimetype:`text/cfg`
    Path to input configuration file for Parameters process
params_config_dict : string
    Dictionary containing input configuration for Parameters process

Returns
-------
output : ComplexData:mimetype:`application/x-netcdf`
    Output Netcdf File
File:      ~/python-sprint/py-repos/osprey/</home/quintins/.cache/pypoetry/virtualenvs/osprey-ZXncEVFc-py3.11/lib/python3.11/site-packages/birdy/client/base.py-2>
Type:      method

We can use the docstring to ensure we provide the appropriate parameters.¶

In [5]:
# run parameters only with required inputs
output_basic = osprey.parameters(
    case_id = "sample",
    grid_id = "COLUMBIA",
    pour_points_csv = "https://marble-dev01.pcic.uvic.ca/twitcher/ows/proxy/thredds/fileServer/datasets/storage/data/projects/comp_support/climate_explorer_data_prep/hydro/sample_data/set4/sample_pour.txt",
    uh_box_csv = (files("tests") / "data/samples/uhbox.csv").read_text(),
    routing = "https://marble-dev01.pcic.uvic.ca/twitcher/ows/proxy/thredds/dodsC/datasets/storage/data/projects/comp_support/climate_explorer_data_prep/hydro/sample_data/set4/sample_flow_parameters.nc",
    domain = str((files("tests") / "data/samples/sample_routing_domain.nc").resolve()),
)
# Use asobj=True to access the data from the output file
basic_data = output_basic.get(asobj=True)[0]
In [6]:
# run parameters with local configuration file
output_config_file = osprey.parameters(
    case_id = "sample",
    grid_id = "COLUMBIA",
    pour_points_csv = "https://marble-dev01.pcic.uvic.ca/twitcher/ows/proxy/thredds/fileServer/datasets/storage/data/projects/comp_support/climate_explorer_data_prep/hydro/sample_data/set4/sample_pour.txt",
    uh_box_csv = (files("tests") / "data/samples/uhbox.csv").read_text(),
    routing = "https://marble-dev01.pcic.uvic.ca/twitcher/ows/proxy/thredds/dodsC/datasets/storage/data/projects/comp_support/climate_explorer_data_prep/hydro/sample_data/set4/sample_flow_parameters.nc",
    domain = str((files("tests") / "data/samples/sample_routing_domain.nc").resolve()),
    params_config_file = str((files("tests") / "data/configs/parameters.cfg").resolve()),
)
config_file_data = output_config_file.get(asobj=True)[0]
In [7]:
# run parameters with configuration dictionary
output_config_dict = osprey.parameters(
    case_id = "sample",
    grid_id = "COLUMBIA",
    pour_points_csv = "https://marble-dev01.pcic.uvic.ca/twitcher/ows/proxy/thredds/fileServer/datasets/storage/data/projects/comp_support/climate_explorer_data_prep/hydro/sample_data/set4/sample_pour.txt",
    uh_box_csv = (files("tests") / "data/samples/uhbox.csv").read_text(),
    routing = "https://marble-dev01.pcic.uvic.ca/twitcher/ows/proxy/thredds/dodsC/datasets/storage/data/projects/comp_support/climate_explorer_data_prep/hydro/sample_data/set4/sample_flow_parameters.nc",
    domain = str((files("tests") / "data/samples/sample_routing_domain.nc").resolve()),
    params_config_dict = {
        "OPTIONS": {
            "LOG_LEVEL": "CRITICAL",
        },
    }
)
config_dict_data = output_config_dict.get(asobj=True)[0]

Once the process has completed we can extract the results and ensure it is what we expected.¶

In [8]:
input_config = read_config("tests/data/configs/parameters.cfg")

# Expected output metadata determined from the input configuration file
expected = {
    "title": "RVIC parameter file",
    "year": datetime.now().strftime("%Y"),
    "month": datetime.now().strftime("%b"),
    "day": datetime.now().strftime("%d"),
    "RvicFdrFile": "sample_flow_parameters.nc",
}
In [9]:
def test_parameters_output(output_data, expected):
    # Metadata of the output of RVIC parameters module
    metadata = {
        "title": output_data.title,
        "year": output_data.history.split()[5],
        "month": output_data.history.split()[2],
        "day": output_data.history.split()[3].zfill(2),
        "RvicFdrFile": output_data.RvicFdrFile,
    }

    assert metadata == expected
In [10]:
test_parameters_output(basic_data, expected)
test_parameters_output(config_file_data, expected)
test_parameters_output(config_dict_data, expected)