Source code for pywasp.wasp.wind_climate

# (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.
"""

from enum import Enum
import xarray as xr

from windkit.spatial import stack_then_unstack
from windkit.wind_climate import (
    mean_windspeed as _mean_windspeed,
    power_density as _power_dens,
)
from windkit.weibull_wind_climate import weibull_combined, _is_wwc, wwc_validate_wrapper
from windkit.metadata import _MET_ATTRS, update_var_attrs, update_history
from ..core import Rvea0326
from .air_density import get_air_density
from .._errors import PywaspError

__all__ = ["add_met_fields", "probabilities"]


_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


@wwc_validate_wrapper
@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))


[docs] @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_dens=None): """Return xr.Dataset of air density""" if isinstance(air_dens, xr.DataArray): pass elif air_dens is None: try: air_dens = wco["air_density"] except KeyError: air_dens = get_air_density(wco) else: air_dens = xr.full_like(wco.wdfreq.isel(sector=0, drop=True), air_dens) return air_dens.to_dataset(name="air_dens")
[docs] def add_met_fields(wco, fields="emergent", air_dens=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_dens: 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_dens = _get_air_density_dataset(wco, air_dens).air_dens wco["air_density"] = air_dens 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"] = _mean_windspeed(wco, bysector=True, emergent=False) elif field == "wspd_combined": wco["wspd_combined"] = _mean_windspeed(wco, bysector=False, emergent=False) elif field == "wspd": wco["wspd"] = _mean_windspeed(wco, bysector=False, emergent=True) elif field == "power_density_sector": wco["power_density_sector"] = _power_dens( wco, bysector=True, emergent=False, air_density=air_dens ) elif field == "power_density_combined": wco["power_density_combined"] = _power_dens( wco, bysector=False, emergent=False, air_density=air_dens ) elif field == "power_density": wco["power_density"] = _power_dens( wco, bysector=False, emergent=True, air_density=air_dens ) wco = update_var_attrs(wco, _MET_ATTRS) return update_history(wco)