...
 
Commits (2)
from . import toolbox, processing
from .wrapper import ekman, sverdrup
from . import toolbox, processing, plotting
from .processing import conversions, diagnostics
from .io.api import open_product, select, info, plot, report
__all__ = ["toolbox", "processing", "ekman", "sverdrup"]
__all__ = [
# modules
"toolbox",
"processing",
"plotting",
# functions
"open_product",
"info",
"select",
"conversions",
"diagnostics",
"plot",
"report",
]
"""
Data IO api
"""
# flake8: noqa
from .products import open_product, info, select
from ..plotting import plot
from .reports import report
import xarray as xr
from typing import Any, Dict
def open_product(path1, path2, **kwargs: Dict[str, Any]) -> Dict[str, xr.Dataset]:
raise NotImplementedError("Import of data through Intake is not yet implemented.")
def info(wndpr: Dict[str, xr.Dataset], *args: Any, **kwargs: Dict[str, Any]) -> None:
raise NotImplementedError("Info of wind products is not yet implemented.")
def select(
wndpr: Dict[str, xr.Dataset], *args: Any, **kwargs: Dict[str, Any]
) -> Dict[str, xr.Dataset]:
raise NotImplementedError(
"Selecting and slicing of dataset is not yet implemented."
)
import xarray as xr
from typing import Any, Dict
def report(wndpr: Dict[str, xr.Dataset], *args: Any, **kwargs: Dict[str, Any]) -> None:
raise NotImplementedError("Generation of reports is not yet implemented.")
import xarray as xr
from typing import Dict, Any
def plot(wndpr: Dict[str, xr.Dataset], *args: Any, **kwargs: Dict[str, Any]):
raise NotImplementedError("Plotting is not yet implemented.")
......@@ -2,7 +2,7 @@
import numpy as np
import xarray as xr
from typing import Callable, Union, Optional
from typing import Callable, Union, Optional, Dict, Any
class BulkFormula:
......@@ -600,3 +600,15 @@ def sverdrup_transport(X: xr.Dataset) -> xr.Dataset:
) / _Coriolis().derivative(lat)
return X
def conversions(
wndpr: Dict[str, xr.Dataset], *args: Any, **kwargs: Dict[str, Any]
) -> Dict[str, xr.Dataset]:
raise NotImplementedError("Conversions are not yet implemented.")
def diagnostics(
wndpr: Dict[str, xr.Dataset], *args: Any, **kwargs: Dict[str, Any]
) -> Dict[str, xr.Dataset]:
raise NotImplementedError("Diagnostics are not yet implemented.")
import pytest
from windeval.io import products
def test_open_product():
with pytest.raises(NotImplementedError):
products.open_product("path1", "path2", akwarg="kwarg")
def test_info(X):
with pytest.raises(NotImplementedError):
products.info({"ds": X})
def test_select(X):
with pytest.raises(NotImplementedError):
products.select({"ds": X})
import pytest
from windeval.io import reports
def test_report(X):
with pytest.raises(NotImplementedError):
reports.report({"ds": X})
import pytest
from windeval import plotting
def test_report(X):
with pytest.raises(NotImplementedError):
plotting.plot({"ds": X})
......@@ -182,3 +182,13 @@ def test_sverdrup_transport(X):
X.sverdrup_transport[0, 0, 0, 0].values, -62.40566775, rel_tol=1e-7
)
assert np.isnan(X.data_vars["sverdrup_transport"].values[0, 0, 0, 1])
def test_conversions(X):
with pytest.raises(NotImplementedError):
processing.conversions({"ds": X})
def test_diagnostics(X):
with pytest.raises(NotImplementedError):
processing.diagnostics({"ds": X})
import windeval
from windeval import wrapper
def test_ekman(X):
windeval.ekman(X, drag_coefficient="large_and_pond_1981", extend_ranges=True)
wrapper.ekman(X, drag_coefficient="large_and_pond_1981", extend_ranges=True)
def test_sverdrup(X):
windeval.sverdrup(X, drag_coefficient="large_and_yeager_2004")
wrapper.sverdrup(X, drag_coefficient="large_and_yeager_2004")
"""Wrapper."""
import xarray as xr
from typing import Optional
from . import processing as pr
from functools import singledispatch
from typing import Optional, Any, Dict, List
from . import processing
@singledispatch
def calculate(ds, *args, **kwargs):
raise NotImplementedError("Data type not supported.")
@calculate.register
def _(
ds: xr.Dataset, var: str, diag: str, *args: Any, **kwargs: Dict[str, Any]
) -> xr.DataArray:
return calculate(ds[var], diag, *args, **kwargs)
@calculate.register # type: ignore
def _(
da: xr.DataArray, diag: str, *args: Any, **kwargs: Dict[str, Any]
) -> xr.DataArray:
f = getattr(processing, diag, None)
if f is not None:
y = f(da, *args, **kwargs)
else:
y = getattr(da, diag)(*args, **kwargs)
return y
@calculate.register # type: ignore
def _(
d: dict, keys: List[str], diag: str, *args: Any, **kwargs: Dict[str, Any]
) -> xr.DataArray:
f = getattr(processing, diag)
return f(*[d[k] for k in keys], *args, **kwargs)
def ekman(
......@@ -40,8 +76,8 @@ def ekman(
continue
d[x] = locals()[x]
c = "northward" if component == "eastward" else "eastward"
getattr(pr, "surface_downward_" + c + "_stress")(X, **d)
getattr(pr, component + "_ekman_transport")(X)
getattr(processing, "surface_downward_" + c + "_stress")(X, **d)
getattr(processing, component + "_ekman_transport")(X)
return X
......@@ -77,8 +113,8 @@ def sverdrup(
if locals()[x] is None:
continue
d[x] = locals()[x]
pr.surface_downward_eastward_stress(X, **d)
pr.surface_downward_northward_stress(X, **d)
pr.sverdrup_transport(X)
processing.surface_downward_eastward_stress(X, **d)
processing.surface_downward_northward_stress(X, **d)
processing.sverdrup_transport(X)
return X