# (c) 2022 DTU Wind Energy
"""Calculate meteorological variables from weibull or binned wind climate
This module contains functions that are used to calculate downstream variables
from a binned or weibull wind climate. This includes wind speed, power density, and
probabilities. It also is used to calculate the total weibull fit.
There is a convenience function add_met_fields which adds the
wind speeds and power densities to an existing wind
climate object.
Sector aggregation
WAsP typically works on sector-wise data, but there are times that we want to know
about the all-sector wind distribution, for example when calculating the mean wind
speed or power density for a location. There are typically two approaches for
calculating these all-sector wind distributions. The best option is to use the
**emergent** calculations, which allows for a non-Weibull shaped distribution in
the combined calculations. The other alternative is the **combined** calculations,
which refit a Weibull distribution to the combined data, this can be reasonably accurate
for many locations, but in areas with varied distibutions can lead to significant errors
due to the multi-modal nature of the aggregated distribution. The advantage of using
the combined approach is that you can get Weibull A & k values that can be stored and
used as an alternative to keeping all of the sector values.
"""
__all__ = ["add_met_fields"]
from enum import Enum
import xarray as xr
import windkit as wk
from windkit.spatial import _stack_then_unstack
from windkit.wind_climate.weibull_wind_climate import (
weibull_combined,
is_wwc,
_validate_wwc_wrapper_factory,
)
from windkit.xarray_structures.metadata import (
_MET_ATTRS,
_update_var_attrs,
_update_history,
)
from pywasp.core import Rvea0326
from pywasp.wasp.air_density import get_air_density
from pywasp._errors import PywaspError
_FORT_WTOT = Rvea0326.calc_points.wtot_points
_FORT_MEAN_WS = Rvea0326.calc_points.mean_ws_points
_FORT_TOTAL_PD = Rvea0326.calc_points.total_pd_points
_FORT_WSPROB = Rvea0326.calc_points.wsprob_points
EMERGENT_FIELDS = [
"wspd",
"power_density",
]
COMBINED_FIELDS = [
"A_combined",
"k_combined",
"wspd_combined",
"power_density_combined",
]
SECTOR_FIELDS = [
"A",
"k",
"wspd_sector",
"power_density_sector",
]
ALL_FIELDS = EMERGENT_FIELDS + COMBINED_FIELDS + SECTOR_FIELDS
class _CALC_TYPE(Enum):
EMERGENT = 0
SECTORWISE = 1
COMBINED = 2
@_validate_wwc_wrapper_factory(run_extra_checks=False)
@_stack_then_unstack
def _weibull_combined(wwc):
"""Return the all sector A & k
This is know as the combined weibull A and k in the
WAsP GUI. For more information, see here:
https://www.wasp.dk/support/faq#general__emergent-and-combined-weibull-all-sector-distributions
Using the combined weibull A and k are calculated
using first and third moment conservation rules.
Parameters
----------
wwc: xarray.Dataset
Weibull Wind Climate object
Returns
-------
xarray.Dataset
Dataset containing all sector A & k variables
"""
A, k = weibull_combined(wwc)
return xr.Dataset(dict(A_combined=A, k_combined=k))
@_stack_then_unstack
def _probabilities(wwc, speed_bins, bysector=True):
"""Get the probabilities for all speed bins
.. warning::
This function is experimental and its signature may change.
Parameters
----------
wwc: xarray.Dataset
Weibull Wind Climate or Binned Wind Climate Object
speed_bins: xr.DataArray
Array containing the wind speed bins, most often from a bwc
bysector: bool
Should results be returned by sector?
Returns
-------
xarray.Dataset
DataArray with the probabilities
"""
prob, _ = xr.apply_ufunc(
_FORT_WSPROB,
wwc.A,
wwc.k,
wwc.wdfreq,
speed_bins,
bysector,
input_core_dims=3 * [["sector", "point"]] + [["wsbin"]] + [list()],
output_core_dims=[["wsbin", "sector", "point"], ["point"]],
)
if not bysector:
prob = prob.isel(sector=0, drop=True)
return prob.to_dataset(name="probabilities")
def _get_air_density_dataset(wco, air_density=None):
"""Return xr.Dataset of air density"""
if isinstance(air_density, xr.DataArray):
pass
elif air_density is None:
try:
air_density = wco["air_density"]
except KeyError:
air_density = get_air_density(wco)
else:
air_density = xr.full_like(wco.wdfreq.isel(sector=0, drop=True), air_density)
return air_density.to_dataset(name="air_density")
[docs]
def add_met_fields(wco, fields="emergent", air_density=None):
"""Add additional fields to a weibull wind climate object
This function can adds post-processed variables to a
wind climate with sectorwise A's and k's.
.. warning::
This function is experimental and its signature may change.
Parameters
----------
wco: xarray.Dataset
Wind Climate object (Weibull or binned)
fields: str or list
Either a string = ["all", "emergent", "combined", "sector"] or a list of possible field values.
A variable containing a list of included values for each string can be found by
appending_fields to the string name listed here.
air_density: xr.DataArray or float
xr.DataArray with air densities with the same dimensions as wco. Default is None,
where it will first check if variable air_density is present in the wco and if not, add it.
One can also set a float to use, which will be broadcasted to dimensions of wco.
Returns
-------
xarray.Dataset
Updated dataset with additional fields
"""
# Check right fields flag
if fields not in ["all", "emergent", "combined", "sector"]:
raise PywaspError(
f"Unknow flag {fields}, options are 'all','emergent','combined','sector'."
)
if fields == "all":
fields_to_check = ALL_FIELDS
elif fields == "emergent":
fields_to_check = EMERGENT_FIELDS
elif fields == "combined":
fields_to_check = COMBINED_FIELDS
elif fields == "sector":
fields_to_check = SECTOR_FIELDS
# Returns a dataset with one variable air_dens
air_density = _get_air_density_dataset(wco, air_density).air_density
wco["air_density"] = air_density
for field in fields_to_check:
# Only need to check one of A_combined and k_combined
if field == "A_combined":
if is_wwc(wco) and field not in wco:
weib_comb = _weibull_combined(wco)
wco["A_combined"] = weib_comb["A_combined"]
wco["k_combined"] = weib_comb["k_combined"]
elif field == "wspd_sector":
wco["wspd_sector"] = wk.mean_wind_speed(wco, bysector=True)
elif field == "wspd_combined":
if is_wwc(wco) and field not in wco:
wco["wspd_combined"] = wk.weibull.weibull_moment(
wco["A_combined"], wco["k_combined"], 1.0
)
elif field == "wspd":
wco["wspd"] = wk.mean_wind_speed(wco, bysector=False)
elif field == "power_density_sector":
wco["power_density_sector"] = wk.mean_power_density(
wco, bysector=True, air_density=air_density
)
elif field == "power_density_combined":
if is_wwc(wco) and field not in wco:
wco["power_density_combined"] = (
0.5
* air_density
* wk.weibull.weibull_moment(
wco["A_combined"], wco["k_combined"], 3.0
)
)
elif field == "power_density":
wco["power_density"] = wk.mean_power_density(
wco, bysector=False, air_density=air_density
)
wco = _update_var_attrs(wco, _MET_ATTRS)
return _update_history(wco)