# (c) 2026 DTU Wind Energy
"""Weibull wind climate module
When measuring over a long period the frequency of occurence of wind speed usually follows a
`Weibull distribution <https://en.wikipedia.org/wiki/Weibull_distribution>`_. It is therefore common
practice in the wind energy industry to use the Weibull *A* and *k*
parameters to denote the wind resource at a certain location.
Because there can be large differences in the wind climate when the wind is
coming from different wind directions, the Weibull distributions are usually specified
per sector.
A valid Weibull wind climate therefore has a dimension ``sector`` and the variables
``A``, ``k`` and ``wdfreq``. Also it must have a valid spatial structure. This module contains
functions that operate on and create weibull wind climates.
"""
__all__ = [
"validate_wwc",
"is_wwc",
"create_wwc",
"read_wwc",
"read_mfwwc",
"wwc_to_file",
"wwc_to_bwc",
"weibull_combined",
]
import logging
import re
from pathlib import Path
import numpy as np
import pandas as pd
import xarray as xr
from scipy.special import gamma
from windkit.xarray_structures._schema import (
Check,
DataVar,
DatasetSchema,
make_is_valid,
make_validator,
validate,
validate_input,
)
from windkit.xarray_structures.empty import (
_copy_chunks,
_native_batch,
_native_frame,
_native_store,
)
from windkit.xarray_structures.metadata import (
_BWC_ATTRS,
_WEIB_ATTRS,
_update_history,
_update_var_attrs,
)
from windkit.xarray_structures.sector import create_sector_coords
from windkit.xarray_structures.wsbin import create_wsbin_coords
from windkit.spatial import (
_raster,
count_spatial_points,
is_cuboid,
set_crs,
to_raster,
)
from windkit.utils import _infer_file_format
from windkit.weibull import (
fit_weibull_k_sumlogm,
weibull_cdf,
weibull_freq_gt_mean,
weibull_moment,
)
from windkit.wind_climate._validation import _validate_wdfreq_sums_to_one
SUPPORTED_WWC_FILE_FORMATS_READ = ["rsf", "wrg", "pwc", "nc", "yml", "yaml"]
SUPPORTED_WWC_FILE_FORMATS_WRITE = ["rsf", "wrg", "yml", "yaml"]
VAR_WEIBULL_A = "A"
VAR_WEIBULL_k = "k"
VAR_WDFREQ = "wdfreq"
DIM_SECTOR = "sector"
DIM_WSBIN = "wsbin"
def _validate_values_range(wwc):
"""Helper function to validate data variables are within range"""
response_list = []
if "A" in wwc.data_vars and any(
xr.where(wwc.A.sum(dim="sector") <= 0.0, True, False).values.flatten()
):
response_list.append("Sum of A values not positive.")
if "k" in wwc.data_vars and any(
xr.where(wwc.k == 0.0, True, False).values.flatten()
):
response_list.append("At least one of the k values is zero.")
return response_list
_WWC_SCHEMA = DatasetSchema(
name="wwc",
data_vars={
"A": DataVar(dims=("sector",)),
"k": DataVar(dims=("sector",)),
"wdfreq": DataVar(dims=("sector",)),
},
dims=("sector",),
coords=("sector", "sector_ceil", "sector_floor"),
extra_checks=(
Check(_validate_values_range),
Check(_validate_wdfreq_sums_to_one),
),
)
validate_wwc = make_validator(_WWC_SCHEMA)
is_wwc = make_is_valid(_WWC_SCHEMA)
[docs]
def create_wwc(output_locs, n_sectors=12, not_empty=True, seed=9876538, **kwargs):
"""Create empty weibull wind climate dataset.
If not_empty=True,the data variables are filled with meaninful random numbers, e.g.
the values from A are generated from a uniform function between 5
and 10 and the values for k from a uniform function between 1.5 and 2.5.
Parameters
----------
output_locs : xarray.Dataset
Output geospatial information
n_sectors : int
Number of sectors, defaults to 12.
not_empty : bool
If true, the empty dataset is filled with random
meaningful data. Defaults to True.
seed : int
Seed for the random data, defaults to 9876538.
kwargs : dict
Additional arguments.
Returns
-------
ds : xarray.Dataset
Weibull wind climate dataset either empty or filled with
random numbers.
Notes
-----
``A``, ``k`` and ``wdfreq`` are stored float32, C-contiguous, with
``sector`` trailing the spatial batch dimensions — the reverse of
``aep_points``' declared ``A(n_sectors, npoints)`` — so PyWAsP's compiled
boundary can take a Fortran-ordered view of this buffer instead of
converting a copy (windkit#807). This is not a supported API contract:
``create_wwc`` is a test-data utility, and neither its dtype nor its
dimension order is guaranteed to a caller.
"""
dims, shape = _native_batch(output_locs)
n_pt = int(np.prod(shape, dtype=int))
if not_empty:
rng = np.random.default_rng(seed)
k = rng.uniform(1.5, 2.5, [n_sectors, n_pt])
A = rng.uniform(5, 10, [n_sectors, n_pt])
wdfreq = rng.dirichlet(np.ones(n_sectors), n_pt).T
else:
k = A = wdfreq = np.full((n_sectors, n_pt), np.nan)
ds = _native_frame(output_locs).assign_coords(
create_sector_coords(n_sectors).coords
)
for name, values in (("A", A), ("k", k), ("wdfreq", wdfreq)):
out_dims, array = _native_store(values, dims, shape, core_dims=("sector",))
ds[name] = xr.DataArray(
array, dims=out_dims, attrs={"height_dim_policy": "optional"}
)
ds = _update_var_attrs(_copy_chunks(output_locs, ds), _WEIB_ATTRS)
return _update_history(ds)
def _rsf_data_to_dataset(data, crs, to_cuboid=False, use_production=False):
"""Convert a dict from :func:`windkit.io.wasp.read_rsf` to a weibull
wind climate dataset.
Parameters
----------
data : dict
Parsed resource-file data as returned by
:func:`windkit.io.wasp.read_rsf`.
crs : int, dict, str or CRS
Value to create CRS object or an existing CRS object
to_cuboid: boolean
If true, the dataset will be converted to the cuboid spatial
structure (dimensions south_north, west_east, height).
use_production: bool
If True, the column with power in the file is interpreted as power production,
i.e. stored as 'gross_aep' in the dataset. If False, it is stored as 'power_density'.
If the values in the file are power production, they are originally stored in Wh/y,
but they are saved with units GWh/y in the dataset. Defaults to False.
Returns
-------
wwc: xarray.Dataset
Weibull wind climate dataset.
"""
if crs is None:
raise ValueError("crs must be specified")
coords = data["coords"]
n_sectors = data["A"].shape[1]
power_var = "gross_aep" if use_production else "power_density"
wwc = xr.Dataset(
data_vars={
"site_elev": ("point", data["site_elev"]),
"A_combined": ("point", data["A_combined"]),
"k_combined": ("point", data["k_combined"]),
power_var: ("point", data["power"]),
"k": (("sector", "point"), data["k"].T),
"A": (("sector", "point"), data["A"].T),
"wdfreq": (("sector", "point"), data["wdfreq"].T),
},
coords={
"name": ("point", np.asarray(data["name"], dtype=object)),
"west_east": ("point", coords["west_east"]),
"south_north": ("point", coords["south_north"]),
"height": ("point", coords["height"]),
},
)
wwc["wdfreq"] = wwc["wdfreq"] / wwc["wdfreq"].sum(dim="sector", skipna=False)
if use_production:
wwc["gross_aep"] = wwc["gross_aep"] / 1e9 # Wh/y -> GWh/y
wwc = set_crs(wwc, crs)
wdcenters = create_sector_coords(n_sectors)
wwc = wwc.assign_coords(**wdcenters.coords)
can_be_raster = _raster._can_be_raster(coords["west_east"], coords["south_north"])
n_spatial = count_spatial_points(wwc)
if (not can_be_raster) and to_cuboid and (n_spatial > 1):
logging.warning("Data cannot be converted to raster, returning point.")
if can_be_raster and to_cuboid or n_spatial == 1 and to_cuboid:
wwc = to_raster(wwc, ignore_raster_check=True)
wwc = _update_var_attrs(wwc, _WEIB_ATTRS)
return _update_history(wwc)
def _read_rsffile(rsffile, crs=None, *, to_cuboid=False, use_production=False):
"""Reads .rsf file into a weibull wind climate dataset.
Thin wrapper around :func:`windkit.io.wasp.read_rsf` that converts
the raw dict into an xarray Dataset.
Parameters
----------
rsffile : str, pathlib.Path or file-like
Path to .rsf file
crs : int, dict, str or CRS
Value to create CRS object or an existing CRS object
to_cuboid: boolean
If true, the dataset will be converted to the cuboid spatial
structure (dimensions south_north, west_east, height).
use_production: bool
If True, the column with power in the file is interpreted as power production,
i.e. stored as 'gross_aep' in the dataset. If False, it is stored as 'power_density'.
If the values in the file are power production, they are originally stored in Wh/y,
but they are saved with units GWh/y in the dataset. Defaults to False.
Returns
-------
wwc: xarray.Dataset
Weibull wind climate dataset.
"""
from windkit.io.wasp import read_rsf
return _rsf_data_to_dataset(
read_rsf(rsffile), crs, to_cuboid=to_cuboid, use_production=use_production
)
def _read_wrgfile(wrgfile, crs=None, *, to_cuboid=True, use_production=False):
"""Reads .wrg file into a weibull wind climate dataset.
Thin wrapper around :func:`windkit.io.wasp.read_rsf` that converts
the raw dict into an xarray Dataset.
Parameters
----------
wrgfile : str, pathlib.Path or file-like
Path to .wrg file
crs : int, dict, str or CRS
Value to create CRS object or an existing CRS object
to_cuboid: boolean
If true, the dataset will be converted to the cuboid spatial
structure (dimensions south_north, west_east, height).
use_production: bool
If True, the column with power in the file is interpreted as power production,
i.e. stored as 'gross_aep' in the dataset. If False, it is stored as 'power_density'.
If the values in the file are power production, they are originally stored in Wh/y,
but they are saved with units GWh/y in the dataset. Defaults to False.
Returns
-------
wwc: xarray.Dataset
Weibull wind climate dataset.
"""
from windkit.io.wasp import read_rsf
return _rsf_data_to_dataset(
read_rsf(wrgfile), crs, to_cuboid=to_cuboid, use_production=use_production
)
def _read_pwcfile(pwcfile, spatial_dataset):
"""Reads .pwc predicted wind climate file from WAsP in XML format.
The .pwc file does not include spatial information, so it must
be passed as a parameter.
Parameters
----------
pwcfile: str,pathlib.Path, io.StringIO
Path to .pwc file
spatial_dataset: xarray.Dataset
xarray dataset with the spatial info
"""
ds = pd.read_xml(
pwcfile, names=["index", "sector", "sector_width", "wdfreq", "A", "k"]
)
ds = ds.set_index("sector")[["A", "k", "wdfreq"]].to_xarray()
ds = ds.assign_coords(create_sector_coords(ds.sector.size).coords)
ds = ds.expand_dims(spatial_dataset.sizes)
ds = ds.assign_coords(spatial_dataset.coords)
wwc = _update_var_attrs(ds, _WEIB_ATTRS)
return _update_history(wwc)
def _wwc_to_rsf_data(wwc, use_production=False):
"""Convert a weibull wind climate dataset to the plain-array dict
consumed by :func:`windkit.io.wasp.write_rsf` /
:func:`windkit.io.wasp.format_rsf`.
Missing ``A_combined``/``k_combined`` and ``power_density`` variables
are derived from the sector-wise Weibull parameters.
Parameters
----------
wwc : xarray.Dataset
Weibull wind climate xarray dataset.
use_production : bool
If True, the ``gross_aep`` variable (converted from GWh/y to the
Wh/y stored in the file) is used as the power column instead of
``power_density``.
Returns
-------
dict
Keyword arguments for :func:`windkit.io.wasp.format_rsf`.
Raises
------
ValueError
If the dataset has unsupported dimensions, or ``gross_aep`` is
missing while ``use_production=True``.
"""
approved_dims = [
"sector",
"point",
"stacked_point",
"south_north",
"west_east",
"height",
]
if any([dim not in approved_dims for dim in wwc.dims]):
raise ValueError(
f"Unsupported dimensions in wwc: {wwc.dims}. "
f"Supported dimensions: {approved_dims}"
)
if use_production and "gross_aep" not in wwc.data_vars:
raise ValueError(
"The gross_aep variable is required to write a resource file with "
"use_production=True."
)
var_power = "gross_aep" if use_production else "power_density"
wwc_cp = wwc.copy()
if "A_combined" not in wwc_cp.data_vars:
wwc_cp[["A_combined", "k_combined"]] = weibull_combined(wwc_cp)
if not use_production and "power_density" not in wwc_cp.data_vars:
try:
air_dens = wwc_cp["air_density"]
except KeyError:
air_dens = 1.225
wwc_cp["power_density"] = _mean_power_density(
wwc_cp, air_density=air_dens, bysector=False
)
wwc_cp = wwc_cp.drop_vars(
["sector", "sector_ceil", "sector_floor", "crs"], errors="ignore"
)
n_sectors = wwc_cp.sizes["sector"]
# Flatten to one dataframe row per site, preserving the row order the
# format has always been written in
df = wwc_cp[["site_elev", "A_combined", "k_combined", var_power]].to_dataframe()
sec_cols = {
var: [str((var, sec)) for sec in range(n_sectors)]
for var in ["wdfreq", "A", "k"]
}
df_sec = wwc_cp[["wdfreq", "A", "k"]].to_dataframe()
df_sec = df_sec.pivot_table(
index=df.index.names, columns="sector", values=["wdfreq", "A", "k"]
)
df_total = (
pd.concat([df, df_sec], axis=1)
.reset_index()
.drop(["point", "stacked_point"], axis=1, errors="ignore")
)
df_total.columns = [str(x) for x in df_total.columns]
power = df_total[var_power].to_numpy(dtype=float)
if use_production:
power = power * 1e9 # GWh/y -> Wh/y
return {
"A": df_total[sec_cols["A"]].to_numpy(),
"k": df_total[sec_cols["k"]].to_numpy(),
"wdfreq": df_total[sec_cols["wdfreq"]].to_numpy(),
"A_combined": df_total["A_combined"].to_numpy(dtype=float),
"k_combined": df_total["k_combined"].to_numpy(dtype=float),
"power": power,
"site_elev": df_total["site_elev"].to_numpy(dtype=float),
"coords": {
"west_east": df_total["west_east"].to_numpy(dtype=float),
"south_north": df_total["south_north"].to_numpy(dtype=float),
"height": df_total["height"].to_numpy(dtype=float),
},
"name": df_total["name"].to_numpy() if "name" in df_total.columns else None,
}
def _to_rsffile(wwc, rsffile, wrg_header=False, use_production=False):
"""Write weibull wind climate dataset to .rsf file.
Thin wrapper around :func:`windkit.io.wasp.write_rsf`.
Parameters
----------
wwc : xarray.Dataset
Weibull wind climate xarray dataset.
rsffile : str
Path to .rsf file
wrg_header : bool
If True, the WRG grid header will be added to the file.
Requires a cuboid dataset.
use_production : bool
If True, the "gross_aep" variable will be used instead of
"power_density".
"""
from windkit.io.wasp import write_rsf
header = None
if wrg_header:
if not is_cuboid(wwc):
raise ValueError("WWC must be a 'cuboid' to add WRG header!")
nx, ny = _raster._shape(wwc)
header = {
"nx": nx,
"ny": ny,
"xmin": wwc.west_east.values.min(),
"ymin": wwc.south_north.values.min(),
"cell_size": _raster._spacing(wwc),
}
data = _wwc_to_rsf_data(wwc, use_production=use_production)
write_rsf(
rsffile,
**data,
header=header,
power_decimals=0 if use_production else 4,
)
def _to_wrgfile(wwc, wrgfile, use_production=False):
"""Write weibull wind climate dataset to .wrg file.
Thin wrapper around :func:`windkit.io.wasp.write_rsf`.
Parameters
----------
wwc : xarray.Dataset
Weibull wind climate xarray dataset.
wrgfile : str
Path to .wrg file
use_production : bool
If True, the "gross_aep" variable will be used instead of
"power_density".
"""
return _to_rsffile(wwc, wrgfile, wrg_header=True, use_production=use_production)
def _read_grdfiles(grdfiles, *, regex_pattern=None, regex_var_order=None):
"""Reads a .grd file into a weibull wind climate dataset.
Parameters
----------
grdfiles: str or list
path of .grd file or list of .grd files.
regex_pattern: re str
Filename regex pattern to extract height, sector, and variable name.
Defaults to None.
regex_var_order: list or tuple
Order of 'height', 'sector', and 'var' in regex_pattern. Defaults to None.
Returns
-------
wwc: xarray.Dataset
Weibull wind climate dataset.
"""
def _rename_var(var):
"""
Function to rename WAsP variable names to short hand name
"""
_rename = {
"Flow inclination": "flow_inclination",
"Mean speed": "wspd",
"Meso roughness": "z0meso",
"Obstacles speed": "obstacle_speedups",
"Orographic speed": "orographic_speedups",
"Orographic turn": "orographic_turnings",
"Power density": "power_density",
"RIX": "rix",
"Roughness changes": "nrch",
"Roughness speed": "roughness_speedups",
"Sector frequency": "wdfreq",
"Turbulence intensity": "turbulence_intensity",
"Weibull-A": "A",
"Weibull-k": "k",
"Elevation": "site_elev",
}
return _rename[var]
def _read_grd_data(filename):
def _parse_line_floats(f):
return [float(i) for i in f.readline().strip().split()]
def _parse_line_ints(f):
return [int(i) for i in f.readline().strip().split()]
with open(filename, "rb") as f:
_ = f.readline().strip().decode() # file_id
nx, ny = _parse_line_ints(f)
xl, xu = _parse_line_floats(f)
yl, yu = _parse_line_floats(f)
zl, zu = _parse_line_floats(f)
values = np.genfromtxt(f)
xarr = np.linspace(xl, xu, nx)
yarr = np.linspace(yl, yu, ny)
# note that the indexing of WAsP grd file is 'xy' type, i.e.,
# values.shape == (xarr.shape[0], yarr.shape[0])
# we need to transpose values to match the 'ij' indexing
values = values.T
return xarr, yarr, values
def _parse_grdfile(grdfile, regex_pattern=None, regex_var_order=None):
match = re.findall(regex_pattern, grdfile.name)[0]
meta = {k: v for k, v in zip(regex_var_order, match)}
meta["var"] = _rename_var(meta["var"])
xarr, yarr, values = _read_grd_data(grdfile)
dims = ["west_east", "south_north", "height"]
coords = {
"height": [float(meta["height"])],
"x": (("west_east",), xarr),
"y": (("south_north",), yarr),
"west_east": (("west_east",), xarr),
"south_north": (("south_north",), yarr),
}
values = values[..., np.newaxis]
if not meta["sector"].lower() == "all":
dims += ["sector"]
coords["sector"] = [int(meta["sector"])]
values = values[..., np.newaxis]
else:
if meta["var"] != "site_elev":
meta["var"] = meta["var"] + "_combined"
da = xr.DataArray(values, dims=dims, coords=coords, name=meta["var"])
if da.name == "site_elev":
da = da.isel(height=0, drop=True)
return da
if not isinstance(grdfiles, list):
grdfiles = [grdfiles]
grdfiles = list(Path(f) for f in grdfiles)
if regex_pattern is None:
regex_pattern = r"Sector (\w+|\d+) \s+ Height (\d+)m \s+ ([a-zA-Z0-9- ]+)"
if regex_var_order is None:
regex_var_order = ("sector", "height", "var")
wwc = xr.merge(
[
_parse_grdfile(
grdfile, regex_pattern=regex_pattern, regex_var_order=regex_var_order
)
for grdfile in grdfiles
],
compat="no_conflicts",
join="outer",
)
ds = _update_var_attrs(wwc, _WEIB_ATTRS)
return _update_history(ds)
def _read_windiofile(windiofile, crs=None, level="plant", **kwargs):
"""Read a windIO YAML file into a weibull wind climate dataset.
Parameters
----------
windiofile : str or pathlib.Path
Path to windIO YAML file.
crs : int, dict, str or CRS, None, optional
Coordinate reference system. Required if not present in windIO file.
level : str
The windIO hierarchy level of the input file: 'plant', 'site',
'energy_resource', or 'wind_resource'. Defaults to 'plant'.
**kwargs
Additional keyword arguments (unused, for API consistency).
Returns
-------
wwc : xarray.Dataset
Weibull wind climate dataset.
Raises
------
ValueError
If CRS is not provided and not present in windIO file.
ValueError
If level is not valid.
"""
from windkit.io.windio.translate import wwc_from_windio
from windkit.io.windio.utils import _read_windio_climate
return _read_windio_climate(windiofile, wwc_from_windio, crs=crs, level=level)
[docs]
def read_wwc(filename, *, file_format="infer", **kwargs):
"""Read a weibull wind climate dataset from a file.
Parameters
----------
filename : str or pathlib.Path
Path to the file to read the dataset from.
file_format : str
File format of the file. If "infer", the file format is inferred from the file extension.
Supported file formats are "rsf", "wrg", "grd", "nc", and "yaml" (windIO). Defaults to "infer".
**kwargs
Additional keyword arguments passed to the read function.
For windIO format (yaml), ``crs`` is required if not present in the file. Also, ``level``
can be specified to indicate the windIO hierarchy level of the input file:
'plant', 'site', 'energy_resource', or 'wind_resource'. Defaults to 'plant'.
Returns
-------
xarray.Dataset
Weibull wind climate dataset.
Examples
--------
>>> import windkit as wk
>>> # Read from .rsf file
>>> wwc = wk.read_wwc("resource.rsf", crs="EPSG:32632")
>>> # Read from .wrg file
>>> wwc = wk.read_wwc("resource.wrg", crs="EPSG:32632")
>>> # Read from windIO YAML format
>>> wwc = wk.read_wwc("plant.yaml", crs="EPSG:32632")
>>> # Read from a standalone wind_resource file
>>> wwc = wk.read_wwc("wind_resource.yaml", crs="EPSG:32632")
"""
if file_format == "infer":
file_format = _infer_file_format(filename)
if file_format == "rsf":
ds = _read_rsffile(filename, **kwargs)
elif file_format == "wrg":
ds = _read_wrgfile(filename, **kwargs)
elif file_format == "grd":
ds = _read_grdfiles(filename, **kwargs)
elif file_format == "nc":
ds = xr.open_dataset(filename, **kwargs)
elif file_format in ["yml", "yaml"]:
ds = _read_windiofile(filename, **kwargs)
else:
raise ValueError(f"Unsupported file format: {file_format}")
validate(ds, _WWC_SCHEMA, run_extra_checks=False)
ds = _update_var_attrs(ds, _WEIB_ATTRS)
return _update_history(ds)
[docs]
def read_mfwwc(files, file_format="infer", **kwargs):
"""Read multiple weibull wind climate datasets from files.
Parameters
----------
files : list of str or pathlib.Path
Paths to the files to read the datasets from.
file_format : str
File format of the files. If "infer", the file format is inferred from the file extension.
The only supported file format is "grd". Defaults to "infer".
**kwargs
Additional keyword arguments passed to the read function
Returns
-------
xarray.Dataset
Weibull wind climate dataset.
"""
if file_format == "infer":
file_format = _infer_file_format(files[0])
if file_format == "grd":
ds = _read_grdfiles(files, **kwargs)
else:
raise ValueError(f"Unsupported file format: {file_format}")
validate(ds, _WWC_SCHEMA, run_extra_checks=False)
ds = _update_var_attrs(ds, _WEIB_ATTRS)
return _update_history(ds)
def _to_windiofile(wwc, windiofile, level="plant", **kwargs):
"""Write a weibull wind climate dataset to windIO YAML format.
Parameters
----------
wwc : xarray.Dataset
Weibull wind climate dataset.
windiofile : str or pathlib.Path
Path to output windIO YAML file.
level : str
Export level: 'plant', 'site', 'energy_resource', or 'wind_resource'.
Defaults to 'plant'.
**kwargs
Additional keyword arguments (unused, for API consistency).
"""
from windkit.io.windio.translate import wwc_to_windio
from windkit.io.windio.utils import _write_windio_climate
_write_windio_climate(wwc, windiofile, wwc_to_windio, level=level)
[docs]
@validate_input(_WWC_SCHEMA)
def wwc_to_file(wwc, filename, *, file_format="infer", **kwargs):
"""Write a weibull wind climate dataset to a file.
Parameters
----------
wwc : xarray.Dataset
Weibull wind climate dataset.
filename : str or pathlib.Path
Path to the file to write the dataset to.
file_format : str
File format of the file. If "infer", the file format is inferred from the file extension.
Supported file formats are "rsf", "wrg", "nc", and "yaml" (windIO). Defaults to "infer".
**kwargs
Additional keyword arguments passed to the write function
For windIO format, the level can be specified via the 'level' keyword argument
(e.g. level='plant', level='site', level='energy_resource', or level='wind_resource').
Examples
--------
>>> import windkit as wk
>>> # Write to .rsf file
>>> wk.wwc_to_file(wwc, "output.rsf")
>>> # Write to .wrg file
>>> wk.wwc_to_file(wwc, "output.wrg")
>>> # Write to windIO YAML format
>>> wk.wwc_to_file(wwc, "output.yaml")
"""
if file_format == "infer":
file_format = _infer_file_format(filename)
if file_format == "rsf":
_to_rsffile(wwc, filename, **kwargs)
elif file_format == "wrg":
_to_wrgfile(wwc, filename, **kwargs)
elif file_format in ["yml", "yaml"]:
_to_windiofile(wwc, filename, **kwargs)
else:
raise ValueError(f"Unsupported file format: {file_format}")
[docs]
@validate_input(_WWC_SCHEMA)
def weibull_combined(wwc, atol=0.000001):
"""Return the all sector A & k.
This is known as the combined weibull A and k in the
WAsP GUI. For more information, see here:
https://wasp.dtu.dk/support/frequently-asked-questions/wasp-faq/emergent-and-combined-all-sector-weibull-distributions
Using the combined weibull A and k are calculated
using first and third moment conservation rules.
Parameters
----------
wwc: xarray.Dataset
Weibull Wind Climate dataset.
Returns
-------
tuple of xr.DataArray
All sector A & k DataArrays
"""
sum1 = (wwc["wdfreq"] * weibull_moment(wwc["A"], wwc["k"], 1)).sum(
dim="sector", skipna=False
)
sum3 = (wwc["wdfreq"] * weibull_moment(wwc["A"], wwc["k"], 3)).sum(
dim="sector", skipna=False
)
sum1 = sum1 / wwc["wdfreq"].sum(dim="sector", skipna=False)
sum3 = sum3 / wwc["wdfreq"].sum(dim="sector", skipna=False)
sum_logm = np.log(sum3) / 3.0 - np.log(sum1)
k_combined = fit_weibull_k_sumlogm(
sum_logm, order_m_first=1, order_m_higher=3, atol=atol
)
A_combined = sum1 / gamma(1.0 + 1.0 / k_combined)
return A_combined, k_combined
def _mean_ws_moment(wwc, moment=1, *, bysector=False):
"""Calculate the mean wind speed from a weibull wind climate dataset.
Parameters
----------
wwc: xarray.Dataset
Weibull Wind Climate dataset.
moment: int
Moment to calculate. Defaults to 1.
bysector: bool
Return results by sector or as an all-sector value. Defaults to False.
Returns
-------
xarray.DataArray
DataArray with the mean wind speed.
"""
result = weibull_moment(wwc[VAR_WEIBULL_A], wwc[VAR_WEIBULL_k], moment)
if not bysector:
result = (result * wwc[VAR_WDFREQ]).sum(dim=DIM_SECTOR, skipna=False)
return result
def _mean_wind_speed(wwc, *, bysector=False):
"""Calculate the mean wind speed from a weibull wind climate dataset.
Parameters
----------
wwc: xarray.Dataset
Weibull Wind Climate dataset.
bysector: bool
Return results by sector or as an all-sector value. Defaults to False.
Returns
-------
xarray.DataArray
DataArray with the mean wind speed.
"""
return _mean_ws_moment(wwc, moment=1, bysector=bysector)
def _mean_power_density(wwc, *, air_density=1.225, bysector=False):
"""Calculate the power density
Parameters
----------
wwc: xarray.Dataset
Weibull wind climate dataset.
air_density : float
Air density.
bysector: bool
Return sectorwise mean wind speed if True. defaults to False.
Returns
pd : xarray.DataArray
Data array with power density.
"""
return 0.5 * air_density * _mean_ws_moment(wwc, moment=3, bysector=bysector)
[docs]
@validate_input(_WWC_SCHEMA)
def wwc_to_bwc(wwc, ws_bins):
"""Creates object from directional A's and k's.
Parameters
----------
wwc: xarray.Dataset
Weibull wind climate xr.Dataset object
ws_bins: np.array
Wind speed bin edges
Returns
-------
bwc : xarray.Dataset
binned wind climate from a Weibull distribution.
"""
wwc = wwc.copy() # make a copy to avoid modifications leaving scope
ws_bins = xr.DataArray(ws_bins, dims=("wsbin",))
cdfs = weibull_cdf(wwc.A, wwc.k, ws_bins)
ws_freq = cdfs.isel(wsbin=slice(1, None)) - cdfs.isel(wsbin=slice(None, -1))
ws_freq = ws_freq / ws_freq.sum(dim="wsbin")
ws_freq = ws_freq.fillna(0.0)
bwc = wwc[
[v for v in wwc.data_vars if v not in ["A", "k"]]
] # pass through other variables
bwc["wsfreq"] = ws_freq
wscenters = create_wsbin_coords(ws_bins)
bwc = bwc.assign_coords(
{
**wscenters.coords,
}
)
bwc = _update_var_attrs(bwc, _BWC_ATTRS)
return _update_history(bwc)
def _ws_cdf(wwc, *, bysector=False, ws_bins=None):
"""Calculate the wind speed CDF at the wind speed bin ceilings.
Parameters
----------
wwc: xarray.Dataset
Weibull wind climate dataset.
bysector: bool
Return the sectorwise CDF if True, else the emergent all-sector CDF,
the ``wdfreq``-weighted sum of the sectorwise CDFs. Defaults to False.
ws_bins: int or array_like, optional
Wind speed bins as accepted by :func:`windkit.create_wsbin_coords`.
Defaults to its default bins.
Returns
-------
xarray.DataArray
Exact Weibull CDF at the ``wsceil`` of each ``wsbin``.
"""
wsbins = create_wsbin_coords() if ws_bins is None else create_wsbin_coords(ws_bins)
cdf = weibull_cdf(wwc[VAR_WEIBULL_A], wwc[VAR_WEIBULL_k], wsbins["wsceil"])
cdf = cdf.transpose(..., DIM_WSBIN)
if bysector:
return cdf
return (wwc[VAR_WDFREQ] * cdf).sum(dim=DIM_SECTOR, skipna=False)
def _ws_freq_gt_mean(wwc, *, bysector=False):
"""Calculate the frequency of wind speeds greater than the mean wind speed.
Parameters
----------
wwc: xarray.Dataset
Weibull wind climate dataset.
bysector: bool
Return the sectorwise frequency if True, else the frequency of the
emergent all-sector distribution above its mean. Defaults to False.
Returns
-------
xarray.DataArray
Frequency of wind speeds greater than the mean wind speed.
"""
A, k = wwc[VAR_WEIBULL_A], wwc[VAR_WEIBULL_k]
if bysector:
return weibull_freq_gt_mean(A, k)
mean = _mean_ws_moment(wwc, moment=1, bysector=False)
cdf_at_mean = (wwc[VAR_WDFREQ] * weibull_cdf(A, k, mean)).sum(
dim=DIM_SECTOR, skipna=False
)
return 1.0 - cdf_at_mean
