"""Declarative, reproducible realization of wind farms.
:class:`~oowl.Sampler` describes a scenario. :meth:`~oowl.Sampler.sample` and
:meth:`~oowl.Sampler.stream` draw from it. Any part of a description may be a
specification such as an OOWL object, a class, a constructor recipe, a
:class:`~oowl.sampling.Choice`, a callable taking one ``Generator``, or a
frozen ``scipy.stats`` distribution.
"""
from __future__ import annotations
import inspect
import itertools
import typing
from collections.abc import Iterator, Mapping, Sequence
from copy import copy
from dataclasses import replace
from functools import lru_cache, partial
from typing import Callable, NamedTuple
import numpy as np
import shapely
from shapely import affinity
from ._api import _attach_components, _build, _enumerates_members
from ._cluster import Cluster
from ._links import (
Link,
Linked,
chosen,
is_choice,
linked_value,
number_links,
per_farm_linked,
scenario_dials,
)
from ._random import as_rng, derive_seed, resolve_seed, spawn_seeds
from ._repr import _ArgumentRepr
from ._sampling_specs import (
Choice,
Literal,
_json_safe,
as_real,
attempt_choice_seed,
describe_scalar,
draw_scalar,
is_distribution,
is_number,
is_numeric_pair,
is_recipe,
is_whole,
nested_choice_seed,
)
from ._validation import _validate_integer
from .boundaries.base import Boundary, _cells, _GeometryBoundary, _project_inside
from .layouts.base import (
Farm,
LayoutAlgorithm,
_closest_pair,
_loneliest,
_mean_nearest,
_placed_for,
)
from .metrics import aspect_ratio
from .resources import Dimitrov, VonMises, WindResource
__all__ = ["Choice", "Link", "Literal", "RotorDiameters", "Sampler"]
@lru_cache(maxsize=1)
def _versions_cached() -> tuple[tuple[str, str], ...]:
import scipy
from ._version import __version__
return (
("oowl", __version__),
("numpy", np.__version__),
("scipy", scipy.__version__),
("shapely", shapely.__version__),
)
def _versions() -> dict[str, str]:
"""The versions of OOWL and of the libraries its numbers depend on.
A farm records them, as the same seed reproduces a farm only with the
same code. ``pixi install --frozen`` restores the environment the lock
file pins.
"""
return dict(_versions_cached())
_CONTEXT_KEYS = frozenset(
{
"boundary",
"layout",
"n_turbines",
"area",
"area_per_turbine",
"power_density",
"minimum_spacing",
"mean_spacing",
"maximum_spacing",
"resource",
"turbines",
"surface",
}
)
#: The keys a farm configuration may set, its density as one mapping.
_CONFIGURATION_KEYS = (
_CONTEXT_KEYS
- {
"area_per_turbine",
"power_density",
"minimum_spacing",
"mean_spacing",
"maximum_spacing",
}
) | {"density"}
#: The independent random streams of one farm draw, in spawn order. Spawned
#: children do not depend on how many are spawned, so a stage appended later
#: leaves the earlier streams unchanged. ``area`` also draws
#: ``area_per_turbine`` and ``power_density``, and ``spacing`` every spacing.
_FARM_STAGES = (
"n_turbines",
"area",
"boundary",
"layout",
"resource",
"turbines",
"surface",
"build",
"spacing",
)
#: The quantities a sampler takes for a farm's density, at most one given.
_DENSITY_KEYS = (
"area_per_turbine",
"power_density",
"minimum_spacing",
"mean_spacing",
"maximum_spacing",
)
#: The densities a layout holds as distances, which may be in rotor diameters.
_SPACING_KEYS = ("minimum_spacing", "mean_spacing", "maximum_spacing")
#: The densities a layout carries itself, which stand in for a sampler's.
_LAYOUT_DENSITIES = (
"area_per_turbine",
"minimum_spacing",
"mean_spacing",
"maximum_spacing",
)
#: Components a cluster or region shares, so a member cannot override them.
_CLUSTER_SCOPED = frozenset({"resource", "surface"})
def _copied_spec(value):
"""Copy the dict, list, tuple and array containers of a specification.
Every other leaf, such as a callable, Link, Choice, boundary or
distribution, is kept by identity so shared Links stay shared.
"""
kind = type(value)
if kind is dict:
return {key: _copied_spec(item) for key, item in value.items()}
if kind is list or kind is tuple:
return kind(_copied_spec(item) for item in value)
if kind is np.ndarray:
return value.copy()
return value
def _start_index(value) -> int:
"""Validate the first scenario index of a draw."""
if not is_number(value):
raise TypeError(f"start must be an integer, not {type(value).__name__}")
if not is_whole(value) or value < 0:
raise ValueError("start must be a non-negative integer")
return int(value)
def _resolved_record(value, record, seed: int):
"""Return a JSON-safe description of a resolved component."""
result = dict(record or {})
result["seed"] = int(seed)
metadata = getattr(value, "metadata", None)
if callable(metadata):
metadata = metadata()
if isinstance(metadata, Mapping):
result["metadata"] = _json_safe(metadata)
elif isinstance(value, Sequence) and not isinstance(value, (str, bytes)):
result["items"] = [_resolved_record(item, {}, seed) for item in value]
return _json_safe(result)
def _validate_nested_spec(value, name: str = "value") -> None:
"""Reject malformed recipes before any farm is drawn."""
if isinstance(value, Literal):
return
if isinstance(value, Choice):
for index, item in enumerate(value.values):
_validate_nested_spec(item, f"{name}[{index}]")
return
if is_recipe(value):
constructor, kwargs = value
constructors = _recipe_constructors(constructor, f"{name} constructor")
for candidate in constructors:
_validate_recipe_keywords(candidate, kwargs, name)
for key, item in kwargs.items():
_validate_nested_spec(item, f"{name}.{key}")
return
if isinstance(value, tuple) and len(value) == 2 and isinstance(value[1], Mapping):
raise TypeError(f"{name} must have a callable or Choice constructor")
if isinstance(value, (list, tuple)):
for index, item in enumerate(value):
_validate_nested_spec(item, f"{name}[{index}]")
return
if isinstance(value, Mapping):
for key, item in value.items():
if key in _SCALAR_BOUNDS:
_validate_scalar_spec(item, f"{name}.{key}", _SCALAR_BOUNDS[key])
_validate_nested_spec(item, f"{name}.{key}")
def _component_types(name: str) -> tuple[type, ...]:
"""The types a sampler component of ``name`` resolves to."""
from .surfaces import Surface, SurfaceModel
from .turbines import Turbine
return {
"boundary": (Boundary,),
"domain": (Boundary,),
"layout": (LayoutAlgorithm,),
"resource": (WindResource, Dimitrov),
"turbines": (Turbine,),
"surface": (Surface, SurfaceModel),
}[name]
def _validate_component(value, kind: str, label: str | None = None) -> None:
"""Reject a fixed component of the wrong type before any farm is drawn.
``kind`` names the component and ``label`` how errors name it. Fixed
values and literals, also among a choice's options or a sequence of
turbines, are checked. What a recipe, class or callable builds is known
only when it is drawn, and checked then.
"""
name = label or kind
expected = _component_types(kind)
if value is None:
return
if isinstance(value, Literal):
_check_type(value.value, expected, f"{name} is")
return
if isinstance(value, Choice):
for item in value.values:
_validate_component(item, kind, label)
return
if (
is_recipe(value)
or isinstance(value, (type, Mapping, Linked))
or is_choice(value)
or (callable(value) and not isinstance(value, expected))
):
return
# Clusters take one boundary or layout per member, and turbines one per
# position.
if isinstance(value, (list, tuple)):
for item in value:
_validate_component(item, kind, label)
return
if isinstance(value, Cluster) and Boundary in expected:
raise TypeError(
f"{name} must be a Boundary; a Cluster is already realized farms, so "
"pass the boundaries you want developed instead"
)
_check_type(value, expected, f"{name} is")
def _recipe_constructors(spec, name: str) -> tuple[Callable, ...]:
"""Return every constructor a recipe head may select."""
if isinstance(spec, Choice):
result: list[Callable] = []
for index, item in enumerate(spec.values):
result.extend(_recipe_constructors(item, f"{name}[{index}]"))
return tuple(result)
if not callable(spec):
raise TypeError(f"{name} must resolve to a callable")
return (spec,)
def _validate_recipe_keywords(constructor, kwargs: Mapping, name: str) -> None:
"""Reject recipe arguments that one selectable constructor does not accept."""
unsupported = sorted(str(key) for key in kwargs if not _accepts(constructor, key))
if unsupported:
label = getattr(constructor, "__name__", type(constructor).__name__)
raise TypeError(f"{name} arguments {unsupported} are not accepted by {label}")
class _Bounds(NamedTuple):
"""The values one scalar field can take when drawn."""
integer: bool = False
low: float | None = None
low_inclusive: bool = True
high: float | None = None
_POSITIVE = _Bounds(low=0.0, low_inclusive=False)
_COUNT = _Bounds(integer=True, low=1)
_SCALAR_BOUNDS = {
"n_turbines": _COUNT,
"area": _POSITIVE,
"area_per_turbine": _POSITIVE,
"power_density": _POSITIVE,
"minimum_spacing": _POSITIVE,
"mean_spacing": _POSITIVE,
"maximum_spacing": _POSITIVE,
"n_sites": _COUNT,
"n_farms": _COUNT,
"gap": _Bounds(low=0.0),
}
[docs]
class RotorDiameters:
"""A spacing given in rotor diameters, as ``RotorDiameters(7)`` for 7D.
A sampler turns it into meters with each scenario's turbines, the largest
rotor diameter among them, so a farm of a drawn turbine keeps its spacing
in diameters. ``diameters`` is any scalar specification, such as a number
or a ``(low, high)`` range.
"""
def __init__(self, diameters) -> None:
_check_scalar_spec(diameters, "RotorDiameters", _POSITIVE)
self.diameters = diameters
def __repr__(self) -> str:
return f"RotorDiameters({self.diameters!r})"
def _validate_scalar_spec(value, name: str, bounds: _Bounds) -> None:
"""Check a constant, both ends of a range or every choice against ``bounds``.
``None`` leaves the field unset. Callables and distributions are checked
when drawn. A spacing may also be :class:`RotorDiameters` of such a value.
"""
if value is None:
return
if isinstance(value, RotorDiameters):
if name not in _SPACING_KEYS:
raise TypeError(f"{name} cannot be given in rotor diameters")
value = value.diameters
_check_scalar_spec(value, name, bounds)
def _check_scalar_spec(value, name: str, bounds: _Bounds) -> None:
"""Check one set scalar specification, as :func:`_validate_scalar_spec`."""
integer, low, low_inclusive, high = bounds
if isinstance(value, Linked):
# A linked range may run either way, so check its ends in order.
value = (
value.spec if isinstance(value.spec, Choice) else tuple(sorted(value.spec))
)
if isinstance(value, Choice):
for position, item in enumerate(value.values):
_check_scalar_spec(item, f"{name}[{position}]", bounds)
return
noun = "whole number" if integer else "number"
if isinstance(value, Literal):
# A Literal is not drawn, so it holds the value itself.
if not is_number(value.value):
raise TypeError(
f"{name} Literal must hold a {noun}, got {type(value.value).__name__}"
)
value = value.value
if is_distribution(value) or (callable(value) and not isinstance(value, type)):
return
valid = is_number(value) or is_numeric_pair(value)
if not valid:
raise TypeError(
f"{name} must be a {noun}, a ({noun}, {noun}) range, a scipy.stats "
f"distribution, f(rng) -> {noun}, a Literal {noun}, or a Choice of "
f"those; got {type(value).__name__}"
)
if is_numeric_pair(value) and value[0] > value[1]:
raise ValueError(f"{name} range must have low <= high")
def within(number) -> bool:
if integer and not is_whole(number):
return False
number = float(number)
return (
np.isfinite(number)
and (low is None or (number >= low if low_inclusive else number > low))
and (high is None or number <= high)
)
if all(
within(number) for number in (value if is_numeric_pair(value) else (value,))
):
return
if integer and low == 1 and high is None:
requirement = "a positive integer"
elif low is not None and high is not None:
requirement = f"in {'[' if low_inclusive else '('}{low:g}, {high:g}]"
else:
requirement = f"finite and {'>=' if low_inclusive else '>'} {low:g}"
label = f"{name} range ends" if is_numeric_pair(value) else name
verb = "must each be" if is_numeric_pair(value) else "must be"
raise ValueError(f"{label} {verb} {requirement}")
#: Consecutive scenarios ``on_error="skip"`` leaves out before it raises, as
#: a description nothing meets would otherwise stream forever. One scenario in
#: a hundred passing still gives up only once in 23,000 runs.
_SKIP_GIVE_UP = 1_000
#: The farm measures a sampler can require to lie in a range, with the values
#: each end may take. None of them sizes the farm.
_REQUIREMENT_BOUNDS = {
"spacing": _POSITIVE,
"n_turbines": _COUNT,
"capacity": _POSITIVE,
"coverage": _Bounds(low=0.0, high=1.0),
"aspect_ratio": _Bounds(low=1.0),
}
def _requirement_range(limits: tuple, farm: Farm, name: str) -> list:
"""A required range in the farm's units, rotor diameters turned into meters."""
return [
end.diameters * _largest_rotor(farm.turbines, f"requirements['{name}']")
if isinstance(end, RotorDiameters)
else end
for end in limits
]
def _check_requirements(farm: Farm, ranges: Mapping) -> None:
"""Raise ``ValueError`` for the first of ``ranges`` that ``farm`` misses."""
where = f"{farm.n_turbines} turbines" + (
"" if farm.boundary is None else f" in {farm.boundary.area / 1e6:.4g} km2"
)
for name, (low, high) in ranges.items():
missed = _missed_requirement(farm, name, low, high, where)
if missed is not None:
raise ValueError(missed)
def _missed_requirement(farm: Farm, name: str, low, high, where: str) -> str | None:
"""How ``farm`` misses the range ``name`` requires, or ``None``."""
count = farm.n_turbines
unit, scale = "", 1.0
if name == "spacing":
# Below two turbines there is no neighbor to measure.
if count < 2:
return None
nearest = farm.nearest_neighbor_distances()
lowest, highest = float(nearest.min()), float(nearest.max())
below = f"come {lowest:.4g} m apart, closer"
above = f"leave one {highest:.4g} m from its nearest neighbor, farther"
unit = " m"
elif name == "n_turbines":
lowest = highest = count
below, above = "are fewer", "are more"
elif name == "capacity":
lowest = highest = farm.capacity()
if not np.isfinite(lowest):
raise ValueError("requirements['capacity'] needs turbines to measure")
below = f"install {lowest / 1e6:.4g} MW, less"
above = f"install {highest / 1e6:.4g} MW, more"
unit, scale = " MW", 1e6
elif name == "coverage":
lowest = highest = farm.coverage()
if not np.isfinite(lowest):
raise ValueError("requirements['coverage'] needs a boundary to measure")
below = f"span {lowest:.3g} of their site, less"
above = f"span {highest:.3g} of their site, more"
else:
if count < 2:
return None
# Turbines in one line have no spread across it, as long as can be.
ratio = aspect_ratio(farm)
lowest = highest = np.inf if np.isnan(ratio) else ratio
spread = f"spread {lowest:.3g} times as far along their long axis as across it"
below, above = f"{spread}, less", f"{spread}, more"
if low is not None and lowest < low * (1.0 - 1e-9):
return (
f"{where} {below} than the required {name} of at least "
f"{low / scale:.4g}{unit}"
)
if high is not None and highest > high * (1.0 + 1e-9):
return (
f"{where} {above} than the required {name} of at most "
f"{high / scale:.4g}{unit}"
)
return None
def _refuse_unmeasurable(requirements, *, turbines, boundary, declared) -> None:
"""Refuse a requirement the farms could never be measured for.
``declared`` farms, from a configuration or a factory, may bring their own
turbines and boundary, so they are checked as drawn.
"""
if declared:
return
for name, ends in requirements.items():
needs = None
if turbines is None and (
name == "capacity" or any(isinstance(end, RotorDiameters) for end in ends)
):
needs = "turbines"
elif boundary is None and name == "coverage":
needs = "a boundary"
if needs is not None:
raise ValueError(
f"requirements['{name}'] is measured on the farm's "
f"{needs.split()[-1]}, so give {needs} too"
)
def _fixed_count(n_turbines, layout, configuration, *, cluster) -> bool:
"""Whether every farm is given a fixed turbine count before drawing.
A built layout shares its count among a cluster's members by area, so
only a single farm keeps it whole.
"""
count = None
if n_turbines is not None:
count = n_turbines
elif isinstance(configuration, Mapping):
count = configuration.get("n_turbines")
elif is_recipe(layout):
count = layout[1].get("n_turbines")
elif isinstance(layout, LayoutAlgorithm) and not cluster:
count = layout.n_turbines
return is_number(count)
def _floor_advice(density: str) -> str:
"""How to keep a minimum spacing a fixed count and area leave nothing to size."""
if "minimum_spacing" not in density:
return ""
return (
", or keep the spacing as a floor the farm is checked against with "
"Sampler(requirements={'spacing': (minimum, None)})"
)
def _validated_density(value, label: str) -> dict:
"""The one density in mapping ``value`` as ``{name: spec}``, ``{}`` for ``None``.
The density is one slot, so its name says which measure fills it, and
its value is any specification. A mapping of several is refused, as the
densities would size the farm twice.
"""
if value is None:
return {}
if not isinstance(value, Mapping):
raise TypeError(
f"{label} must be a mapping such as {{'minimum_spacing': 800.0}}, not "
f"{type(value).__name__}. A Choice of configurations varies it per scenario"
)
unknown = sorted(str(name) for name in value if name not in _DENSITY_KEYS)
if unknown:
raise ValueError(
f"unknown {label} {unknown}, which may be {', '.join(_DENSITY_KEYS)}"
)
if len(value) != 1:
raise ValueError(
f"{label} takes one of {', '.join(_DENSITY_KEYS)}, got {len(value)}"
)
name, spec = next(iter(value.items()))
return {name: _copied_spec(_literal_number(spec))}
def _validated_requirements(value) -> dict:
"""The checked ``requirements`` mapping, or ``{}`` for ``None``.
Each requirement is a fixed ``(low, high)`` range a farm measure must lie
in, ``None`` leaving an end open. It is never drawn, as a drawn range would
make whether a scenario counts random.
"""
if value is None:
return {}
if not isinstance(value, Mapping):
raise TypeError(f"requirements must be a mapping, not {type(value).__name__}")
unknown = sorted(str(name) for name in value if name not in _REQUIREMENT_BOUNDS)
if unknown:
raise ValueError(
f"unknown requirements {unknown}, which may be "
f"{', '.join(_REQUIREMENT_BOUNDS)}"
)
checked: dict = {}
for name, item in value.items():
label = f"requirements['{name}']"
if not isinstance(item, tuple) or len(item) != 2:
raise TypeError(
f"{label} must be a (low, high) range, None leaving an end open, "
f"as in (300e6, None), got {type(item).__name__}"
)
if item[0] is None and item[1] is None:
raise ValueError(f"{label} leaves both ends open, so it checks nothing")
low, high = (
_requirement_end(name, end, side)
for end, side in zip(item, ("low", "high"))
)
if low is not None and high is not None and type(low) is type(high):
# Ends in one unit compare before drawing, mixed ones per farm.
first, last = (
(low.diameters, high.diameters)
if isinstance(low, RotorDiameters)
else (low, high)
)
if first > last:
raise ValueError(
f"{label} runs from {first:g} down to {last:g}, so no farm meets it"
)
checked[name] = (low, high)
return checked
def _requirement_end(name: str, end, side: str):
"""One checked end of required range ``name``, ``None`` when it is open."""
if end is None:
return None
label = f"requirements['{name}'] {side} end"
diameters = isinstance(end, RotorDiameters)
if diameters and name != "spacing":
raise TypeError(f"{label} cannot be given in rotor diameters")
number = _literal_number(end.diameters if diameters else end)
if not is_number(number):
kinds = "a number or RotorDiameters of one" if name == "spacing" else "a number"
raise TypeError(
f"{label} must be {kinds} or None, as a requirement is a fixed range "
f"rather than drawn, got {type(number).__name__}"
)
bounds = _REQUIREMENT_BOUNDS[name]
_check_scalar_spec(number, label, _POSITIVE if diameters else bounds)
number = int(number) if bounds.integer else float(number)
return RotorDiameters(number) if diameters else number
def _literal_number(value):
"""The number a :class:`Literal` holds, any other value as given."""
if isinstance(value, Literal) and is_number(value.value):
return value.value
return value
def _validate_configuration_spec(
value, forbidden: frozenset[str] = frozenset()
) -> None:
"""Validate every farm configuration branch known before drawing."""
if isinstance(value, Choice):
for item in value.values:
_validate_configuration_spec(item, forbidden)
return
if not isinstance(value, Mapping):
return
unknown = set(value) - _CONFIGURATION_KEYS
if unknown:
raise ValueError(f"unknown configuration keys: {sorted(unknown)}")
if "density" in value:
_validated_density(value["density"], "configuration density")
disallowed = sorted(set(value) & forbidden)
if disallowed:
raise ValueError(
f"farm configuration cannot override cluster-scoped {disallowed}"
)
for key, item in value.items():
if key in ("boundary", "layout", "resource", "turbines", "surface"):
_validate_component(item, key, f"configuration {key}")
def _draw_count(
spec,
rng: np.random.Generator,
name: str,
*,
index: int = 0,
choice_seed: int | None = None,
attempt: int = 0,
) -> int:
"""Draw a positive whole number, as :func:`draw_scalar` draws a real one."""
if is_choice(spec):
with chosen(spec, index, attempt_choice_seed(spec, choice_seed, attempt)) as (
_,
_,
selected,
):
return _draw_count(
selected,
rng,
name,
index=index,
choice_seed=nested_choice_seed(choice_seed),
attempt=attempt,
)
if isinstance(spec, Literal):
# A Literal is not drawn, so it holds the count itself.
if not is_number(spec.value):
raise TypeError(
f"{name} Literal must hold a whole number, "
f"got {type(spec.value).__name__}"
)
spec = spec.value
if isinstance(spec, Linked):
if not spec.integer:
raise TypeError(f"{name} must be linked to a whole-number range")
value = linked_value(spec, index)
elif is_distribution(spec):
value = spec.rvs(random_state=rng)
elif callable(spec) and not isinstance(spec, type):
value = spec(rng)
elif is_whole(spec):
value = spec
elif is_numeric_pair(spec) and all(is_whole(item) for item in spec):
low, high = (int(item) for item in spec)
if low > high:
raise ValueError(f"{name} range must have low <= high")
value = rng.integers(low, high + 1)
elif is_number(spec) or is_numeric_pair(spec):
raise ValueError(f"{name} must be a whole number or a whole-number range")
else:
raise TypeError(
f"{name} must be an integer, an integer range, a scipy.stats "
"distribution, or f(rng) -> number"
)
value = as_real(value, name)
if not np.isfinite(value):
raise ValueError(f"{name} must be finite")
# A callable or distribution may draw a float, rounded to the nearest count.
value = int(round(value))
if value < 1:
raise ValueError(f"{name} must be >= 1")
return value
def _inspected_parameters(constructor) -> tuple[frozenset[str], bool] | None:
"""The parameter names of ``constructor`` and whether it takes any keyword.
``None`` when its signature cannot be inspected.
"""
try:
parameters = inspect.signature(constructor).parameters
except (TypeError, ValueError):
return None
return frozenset(parameters), any(
parameter.kind is inspect.Parameter.VAR_KEYWORD
for parameter in parameters.values()
)
_cached_parameters = lru_cache(maxsize=256)(_inspected_parameters)
def _parameters(constructor) -> tuple[frozenset[str], bool] | None:
""":func:`_inspected_parameters`, inspected once per hashable constructor."""
try:
hash(constructor)
except TypeError:
return _inspected_parameters(constructor)
return _cached_parameters(constructor)
def _accepts(constructor, name: str) -> bool:
"""Whether ``constructor`` takes keyword ``name``, assumed when unknowable."""
parameters = _parameters(constructor)
return parameters is None or name in parameters[0] or parameters[1]
def _has_explicit_parameter(constructor, name: str) -> bool:
"""Whether ``constructor`` names parameter ``name`` in its signature."""
parameters = _parameters(constructor)
return parameters is not None and name in parameters[0]
def _accepts_one_argument(function) -> bool:
"""Whether ``function`` can be called with one positional argument.
A signature that cannot be inspected is assumed to accept it.
"""
try:
inspect.signature(function).bind(None)
except ValueError:
return True
except TypeError:
return False
return True
def _kept_inside(geometry, positions: np.ndarray) -> np.ndarray:
"""Project turbines that rounding left just outside ``geometry`` back in."""
if shapely.intersects_xy(geometry, positions[:, 0], positions[:, 1]).all():
return positions
inner = geometry.buffer(-1e-7)
return positions if inner.is_empty else _project_inside(geometry, inner, positions)
def _turbine_list(turbines, name: str) -> list:
"""The turbines of one farm as a list, or ``ValueError`` naming ``name``."""
from .turbines import Turbine
if turbines is None:
raise ValueError(f"{name} needs turbines to resolve, so give turbines too")
return [turbines] if isinstance(turbines, Turbine) else list(turbines)
def _largest_rotor(turbines, name: str) -> float:
"""The largest rotor diameter in meters among a farm's ``turbines``."""
return max(
float(turbine.rotor_diameter) for turbine in _turbine_list(turbines, name)
)
def _rated_power(turbines) -> float:
"""The rated power in watts shared by a farm's ``turbines``.
A power density turns into a count through one rated power, and turbines
given one per position already fix the count.
"""
powers = {
float(turbine.rated_power)
for turbine in _turbine_list(turbines, "power_density")
}
if len(powers) != 1:
raise ValueError(
"power_density needs one turbine rating to turn watts into a count"
)
return powers.pop()
#: Relative widening of a farm scaled to a minimum spacing, far beyond the
#: rounding of its coordinates and far below any tolerance it is held to.
_SCALED_MINIMUM_MARGIN = 1e-12
def _bare_layout(layout: LayoutAlgorithm, build_seed) -> LayoutAlgorithm:
"""``layout`` without its modifiers, drawing the turbines a removal drops.
It is seeded as a farm built from ``build_seed`` seeds the layout, so the
modifiers run after it from the seed they would have run from inside it.
Run inside the sizing, PositionNoise brought a pair 0.14 of the spacing
together, and the farm was grown seven times wider to part them.
"""
bare = copy(layout)
if bare.seed is None:
bare.seed = derive_seed(resolve_seed(build_seed), "placement")
bare.modifiers = ()
if layout.n_turbines is not None:
bare.n_turbines = _placed_for(layout.modifiers, layout.n_turbines)
return bare
def _scaled_to_spacing(farm: Farm, spacing: float, measure) -> tuple[Farm, float]:
"""Scale a drawn farm so its ``measure`` of spacing sits exactly at ``spacing``.
``measure`` is the layout's closest pair or mean nearest-neighbor
distance, which both scale with the farm. Boundary, positions and guide
share one similarity transform about the boundary centroid. Returns the
scaled farm and the factor.
"""
if farm.boundary is None:
raise ValueError("spacing requires a boundary to scale")
closest = measure(farm)
if not np.isfinite(closest):
raise ValueError("spacing needs at least two turbines to separate")
factor = spacing / closest
if measure is _closest_pair:
# A minimum is a guarantee, so rounding in the scaled positions must
# not bring the closest pair a hair below it.
factor *= 1.0 + _SCALED_MINIMUM_MARGIN
elif measure is _loneliest:
# A maximum is one too, so the loneliest turbine ends a hair inside it.
factor *= 1.0 - _SCALED_MINIMUM_MARGIN
geometry = farm.boundary.geometry
origin = geometry.centroid
offset = np.array(
[origin.x - factor * origin.x, origin.y - factor * origin.y], dtype=float
)
def scaled(shape):
return affinity.scale(shape, xfact=factor, yfact=factor, origin=origin)
site = scaled(geometry)
positions = _kept_inside(site, farm.positions * factor + offset)
guide = None if farm.guide is None else farm.guide._transformed(factor, offset)
result = replace(
farm,
positions=positions,
# A cluster still places a member it has not been told where to put.
# Partition cells scale with the site, so the farm keeps its parts.
boundary=_GeometryBoundary(
site,
positioned=getattr(farm.boundary, "_positioned", True),
cells=_cells(farm.boundary, scaled),
),
guide=guide,
metrics={},
)
return result, factor
#: Draws allowed for a spacing to settle on a boundary with modifiers in meters.
_MAX_SPACING_DRAWS = 32
#: Share of a spacing within which a redrawn farm settles on it, a minimum
#: from above. Layouts fit their patterns to a millionth, so a search to a
#: billionth spent its last draws on their rounding, three of six for a
#: String of 376 turbines at 3.5 s a draw.
_SPACING_SETTLED = 1e-6
#: A bracket across which the closest pair changes this many times faster
#: than it scales with the extent holds a jump of the layout, not a root. A
#: smooth layout changes this fast only on a site its modifiers shrink to a
#: twentieth of its width.
_SPACING_JUMP_STEEPNESS = 20.0
#: Most a draw may grow the side of a site redrawn toward a spacing. A
#: Contour resting on a plateau gave a secant of rounding error, and the next
#: draw was 1.6 billion km2.
_SPACING_STEP_LIMIT = 8.0
#: Width, as a share of the site's side, to which a bracket across a jump of
#: the layout is halved before the farm just above it is kept.
_SPACING_JUMP_WIDTH = 1e-3
#: Width, as a share of the site's side, below which a bracket still
#: unsettled holds the layout's noise, not a root. A smooth layout changes its
#: spacing by about this share across it, a hundred times the settling share.
#: A mixture with uniform cells jittered by a meter as its site grew, and ten
#: draws at 1.3 s each went to the jitter.
_SPACING_NOISE_WIDTH = 1e-4
def _hole_area(boundary: Boundary) -> float:
"""The area the holes of ``boundary`` cut from its outlines."""
geometry = boundary.geometry
parts = shapely.get_parts(geometry)
polygons = parts[shapely.get_type_id(parts) == shapely.GeometryType.POLYGON]
outlines = shapely.polygons(shapely.get_exterior_ring(polygons))
return max(float(shapely.area(outlines).sum()) - geometry.area, 0.0)
def _drawn_to_spacing(
build, spacing: float, area: float, measure=_closest_pair
) -> Farm:
"""Redraw a farm at the extent whose ``measure`` of spacing lands on ``spacing``.
``measure`` is the layout's closest pair or mean nearest-neighbor distance.
A modifier given in meters, such as a setback, does not scale with its
boundary, so a drawn farm cannot be scaled to the spacing. The closest
pair grows about linearly with the side of the site's outline, holes
filled in. A hole in meters keeps its area as the site grows, so the
first step scales that outline to the spacing and keeps the hole area.
Secant steps on the side then bracket the spacing in a few draws, and the
Illinois method closes the bracket, bisecting where a step halves neither
the bracket nor the miss. A layout can also rearrange as its site changes
and jump over the spacing. The bracket then closes on the jump and the
farm just above the spacing is kept, so no pair is closer than requested,
as it is when the draws run out inside a bracket. Raises ``ValueError``
when no draw within the allowed number brackets the spacing.
"""
failure: ValueError | None = None
def drawn(side: float) -> tuple[Farm | None, float]:
"""The farm at ``side`` and its miss, ``None`` below every spacing if
the boundary cannot be built that small."""
nonlocal failure
try:
farm = build(at_area=side**2)
except ValueError as error:
# Seven channels 713 m apart could not shrink to the 9.6 km2 a
# first draw asked for, though the spacing lay on a larger site.
failure = error
return None, -spacing
closest = measure(farm)
if not np.isfinite(closest):
raise ValueError("spacing needs at least two turbines to separate")
return farm, closest - spacing
side = float(np.sqrt(area))
farm, residual = drawn(side)
previous = None
# The bracket ends as ``(side, residual, farm)``, with the Illinois
# weight halving the residual of an end kept twice running.
below: tuple[float, float, Farm | None] | None = None
above: tuple[float, float, Farm] | None = None
weights = {"below": 1.0, "above": 1.0}
kept = None
# Bracket widths and misses after each bracketed draw, to catch a stall.
widths: list[float] = []
misses: list[float] = []
# A minimum is held from above, a maximum from below, a mean from either
# side, so a bracket keeps its farm on the side the spacing holds.
ceiling = measure is _loneliest
floor = 0.0 if measure is _closest_pair else -_SPACING_SETTLED * spacing
top = 0.0 if ceiling else _SPACING_SETTLED * spacing
def held(below, above) -> Farm:
if not ceiling:
return above[2]
if below[2] is None:
raise ValueError(
f"no draw of the layout left every turbine within {spacing:.4g} m "
"of a neighbor"
) from failure
return below[2]
for _ in range(_MAX_SPACING_DRAWS):
if farm is not None and floor <= residual <= top:
return farm
if residual < 0.0 or farm is None:
below, weights["below"] = (side, residual, farm), 1.0
else:
above, weights["above"] = (side, residual, farm), 1.0
if below is None or above is None:
if farm is None:
# The boundary could not be built this small, so grow it.
step = 2.0 * side
elif previous is None:
# A layout scaling with its outline reaches the spacing in one
# step, as the holes keep their area.
holes = 0.0 if farm.boundary is None else _hole_area(farm.boundary)
outline = (side**2 + holes) * (spacing / (residual + spacing)) ** 2
step = float(np.sqrt(max(outline - holes, 0.25 * side**2)))
else:
slope = (residual - previous[1]) / (side - previous[0])
# A pair that closed as the site grew, or a layout resting on
# a plateau, gives a secant whose step runs off, so the slope
# is at least half that of a layout scaling with its site.
slope = max(slope, 0.5 * (residual + spacing) / side)
step = max(side - residual / slope, 0.5 * side)
# A step that left the miss on the same side and not halved
# rests on a plateau, which a zone in meters can pin, so the
# next moves at least twice as far rather than creeping along.
stalled = abs(residual) > 0.5 * abs(previous[1])
if stalled and (residual < 0.0) == (previous[1] < 0.0):
least = 2.0 * abs(side - previous[0])
step = side + np.sign(step - side) * max(abs(step - side), least)
previous = (side, residual)
side = float(min(max(step, 0.5 * side), _SPACING_STEP_LIMIT * side))
else:
steepness = (above[1] - below[1]) / (above[0] - below[0])
width = abs(above[0] - below[0])
if width <= _SPACING_NOISE_WIDTH * above[0]:
# The layout varies with its site's size below this width, as
# a uniform part redraws its points, so the farm just above
# the spacing is kept.
return held(below, above)
if steepness > _SPACING_JUMP_STEEPNESS * spacing / side:
# A jump of the layout lies in the bracket, which is halved
# onto it, and the farm just above it is kept.
if width <= _SPACING_JUMP_WIDTH * above[0]:
return held(below, above)
side = 0.5 * (below[0] + above[0])
elif (
widths
and abs(above[0] - below[0]) > 0.5 * widths[-1]
and abs(residual) > 0.5 * misses[-1]
):
# A step that halved neither the bracket nor the miss crept
# along a flat end, as false position does beside a jump, so
# bisect.
side = 0.5 * (below[0] + above[0])
else:
now = "above" if residual < 0.0 else "below"
if now == kept:
weights[now] *= 0.5
kept = now
low, high = below[1] * weights["below"], above[1] * weights["above"]
side = (below[0] * high - above[0] * low) / (high - low)
widths.append(abs(above[0] - below[0]))
misses.append(abs(residual))
farm, residual = drawn(side)
# The last draw joins the bracket, whose end on the held side is kept.
if farm is not None and residual >= 0.0:
if above is None or residual < above[1]:
above = (side, residual, farm)
elif farm is not None and (below is None or residual > below[1]):
below = (side, residual, farm)
elif below is None:
below = (side, residual, farm)
if below is not None and above is not None:
return held(below, above)
raise ValueError(
f"the layout's spacing did not settle on {spacing:.4g} m in "
f"{_MAX_SPACING_DRAWS} draws, it was last {residual + spacing:.4g} m"
) from failure
def _selection(
spec, seed: int, index: int, context: Mapping, path: tuple
) -> tuple[int, int | None]:
"""The index and seed a choice selects with, the seed ``None`` if it has its own.
A balanced schedule spans the population, so it derives from the sampler
root and the choice's place in the description, not from the draw seed.
A component shared by a cluster balances over ``_balanced_index``, the
cluster's scenario index. A linked choice reads its dial instead.
"""
if isinstance(spec, Linked):
return index, None
if spec.selection != "balanced":
return index, seed if spec.seed is None else None
index = context.get("_balanced_index", index)
if spec.seed is not None:
return index, None
return index, derive_seed(context.get("_choice_seed", seed), *path)
#: Arguments whose pair is a point or a vector rather than a range to draw.
_PAIR_ARGUMENTS = frozenset({"center", "gradient"})
#: Arguments that take points, whose pairs are coordinates rather than ranges
#: when they are given as numbers, as a Polygon's corners, a partition's sites
#: and a Polyline's vertices are.
_POINT_ARGUMENTS = frozenset({"exterior", "holes", "sites", "coordinates"})
def _is_points(value) -> bool:
"""Whether ``value`` holds numbers as points, ``(N, 2)`` or a list of such."""
if isinstance(value, (str, bytes, Mapping)) or is_numeric_pair(value):
return False
try:
array = np.asarray(value, dtype=float)
except (TypeError, ValueError):
# Rings of different lengths, as a polygon's holes, are a list of them.
return (
isinstance(value, (list, tuple))
and bool(value)
and all(_is_points(item) for item in value)
)
return array.ndim >= 2 and array.shape[-1] == 2
def _resolve_argument(
name,
value,
seed: int,
index: int,
context: Mapping,
path: tuple,
):
"""Resolve one constructor argument from its specification."""
if isinstance(value, Literal):
return value.value
if name in _POINT_ARGUMENTS and _is_points(value):
# Each corner of a list of them was drawn as a range, and failed.
return value
if is_choice(value):
with chosen(value, *_selection(value, seed, index, context, path)) as (
_,
_,
selected,
):
return _resolve_argument(
name,
selected,
derive_seed(seed, "selected"),
index,
context,
(*path, "selected"),
)
if isinstance(value, Linked):
return linked_value(value, index)
if is_recipe(value):
return _resolve(value, None, seed, index, context, field=name, path=path)[0]
if is_distribution(value):
return value.rvs(random_state=as_rng(seed))
if callable(value) and not isinstance(value, type):
return value(as_rng(seed))
if isinstance(value, type):
return _construct(value, {}, None, seed, index, context, path)
if (
(str(name).endswith("_range") or name in _PAIR_ARGUMENTS)
and isinstance(value, tuple)
and len(value) == 2
):
# A range, point or vector argument takes its pair as given, and a pair
# of ranges draws each end, as in elevation_range=((0, 10), (90, 110)).
if is_numeric_pair(value):
return value
return tuple(
_resolve_argument(
f"{name} end {position}",
item,
derive_seed(seed, "item", position),
index,
context,
(*path, position),
)
for position, item in enumerate(value)
)
if is_numeric_pair(value):
low, high = value
if low > high:
raise ValueError(f"{name} range must have low <= high")
if all(is_whole(item) for item in value):
return int(as_rng(seed).integers(int(low), int(high) + 1))
return float(as_rng(seed).uniform(float(low), float(high)))
if isinstance(value, Mapping):
return {
key: _resolve_argument(
key,
item,
derive_seed(seed, "key", str(key)),
index,
context,
(*path, str(key)),
)
for key, item in value.items()
}
if isinstance(value, list):
return [
_resolve_argument(
name,
item,
derive_seed(seed, "item", item_index),
index,
context,
(*path, item_index),
)
for item_index, item in enumerate(value)
]
return _scenario_seeded(value, seed)
def _retry_seed(seed, attempt: int) -> int:
"""The seed retry ``attempt`` of a scenario draws a seeded layout from."""
return derive_seed(resolve_seed(seed), "attempt", attempt)
def _scenario_seeded(value, seed: int):
"""Redraw an unseeded wind resource or direction model from ``seed``.
It drew fresh entropy when built, so without this two runs of one sampler
seed would differ.
"""
if isinstance(value, (WindResource, VonMises)) and not value._pinned_seed:
return value._reseeded(seed)
return value
def _expected_types(expected) -> tuple[type, ...]:
"""The types a resolved component may have, empty for any."""
if expected is None:
return ()
return (expected,) if isinstance(expected, type) else tuple(expected)
def _check_type(value, expected_types: tuple[type, ...], subject: str) -> None:
"""Raise a ``TypeError`` naming ``subject`` unless ``value`` is expected."""
if expected_types and not isinstance(value, expected_types):
names = ", ".join(item.__name__ for item in expected_types)
raise TypeError(f"{subject} {type(value).__name__}; expected {names}")
def _construct(
constructor,
kwargs: Mapping,
expected,
seed: int,
index: int,
context: Mapping,
path: tuple,
):
"""Call ``constructor`` with its resolved arguments and the draw's values.
A boundary takes the drawn ``area`` and a layout the drawn ``n_turbines``
unless the recipe sets them, and an explicit ``seed`` parameter takes the
draw seed.
"""
resolved = {
key: _resolve_argument(
key,
value,
derive_seed(seed, "argument", str(key)),
index,
context,
(*path, str(key)),
)
for key, value in kwargs.items()
}
expected_types = _expected_types(expected)
for name, kind in (("area", Boundary), ("n_turbines", LayoutAlgorithm)):
if (
name not in resolved
and context.get(name) is not None
and any(issubclass(item, kind) for item in expected_types)
and _accepts(constructor, name)
):
resolved[name] = context[name]
if "seed" not in resolved and _has_explicit_parameter(constructor, "seed"):
resolved["seed"] = seed
elif (
resolved.get("seed") is not None
and context.get("_attempt")
and any(issubclass(item, LayoutAlgorithm) for item in expected_types)
):
# A layout recipe given its own seed draws each retry from one derived
# from it, as a built layout does.
resolved["seed"] = _retry_seed(resolved["seed"], context["_attempt"])
try:
result = constructor(**resolved)
except TypeError as exc:
label = getattr(constructor, "__name__", type(constructor).__name__)
raise TypeError(f"could not construct {label}: {exc}") from exc
_check_type(result, expected_types, "constructor returned")
return result
def _resolve_constructor(
spec, seed: int, index: int, context: Mapping, path: tuple, field: str
):
"""Resolve a recipe head without constructing the selected class."""
if is_choice(spec):
with chosen(spec, *_selection(spec, seed, index, context, path)) as (
position,
selection,
selected,
):
constructor, record = _resolve_constructor(
selected,
derive_seed(seed, "selected"),
index,
context,
(*path, "selected"),
field,
)
return constructor, {
"index": position,
"selection": selection,
"selected": record,
}
if not callable(spec):
raise TypeError(f"{field} recipe constructor must resolve to a callable")
return spec, {"constructor": getattr(spec, "__name__", type(spec).__name__)}
def _resolve(
spec,
expected,
seed: int,
index: int,
context: Mapping | None = None,
field: str = "component",
path: tuple = (),
):
"""Resolve one component specification and record how it was resolved.
Returns the value and a JSON-compatible record. ``expected`` is a type or
tuple of types the value must have, or ``None`` for any.
"""
context = context or {}
path = path or (field,)
expected_types = _expected_types(expected)
if isinstance(spec, Literal):
_check_type(spec.value, expected_types, f"{field} is")
return spec.value, {"literal": True}
if is_choice(spec):
with chosen(spec, *_selection(spec, seed, index, context, path)) as (
position,
selection,
selected,
):
result, record = _resolve(
selected,
expected,
spawn_seeds(seed, 2)[1],
index,
context,
field=field,
path=(*path, "selected"),
)
return result, {"index": position, "selection": selection, "selected": record}
if is_recipe(spec):
constructor, record = _resolve_constructor(
spec[0],
derive_seed(seed, "constructor"),
index,
context,
(*path, "constructor"),
field,
)
result = _construct(constructor, spec[1], expected, seed, index, context, path)
return result, record
if isinstance(spec, type):
result = _construct(spec, {}, expected, seed, index, context, path)
return result, {"constructor": spec.__name__}
if callable(spec) and not isinstance(spec, expected_types):
try:
result = spec(as_rng(seed))
except TypeError as exc:
raise TypeError(
"component factory must accept one numpy.random.Generator"
) from exc
_check_type(result, expected_types, "component factory returned")
return result, {"factory": getattr(spec, "__name__", type(spec).__name__)}
if isinstance(spec, Mapping):
if context.get("_raw_mapping"):
return dict(spec), {"mapping": True}
return _resolve_argument(field, spec, seed, index, context, path), {
"mapping": True
}
if isinstance(spec, list) or (
isinstance(spec, tuple) and not is_numeric_pair(spec)
):
items = [
_resolve(
item,
None,
derive_seed(seed, "item", item_index),
index,
context,
field=field,
path=(*path, item_index),
)[0]
for item_index, item in enumerate(spec)
]
return (items if isinstance(spec, list) else tuple(items)), {"sequence": True}
# Copy a layout so the seed and count set below stay off the caller's.
result = copy(spec) if isinstance(spec, LayoutAlgorithm) else spec
if isinstance(result, Cluster) and Boundary in expected_types:
raise TypeError(
"boundary must be a Boundary; a Cluster is already realized "
"farms, so pass the boundaries you want developed instead"
)
_check_type(result, expected_types, f"{field} is")
if isinstance(result, LayoutAlgorithm):
if context.get("n_turbines") is not None:
result.n_turbines = context["n_turbines"]
# An explicitly seeded layout keeps its seed, and draws each retry
# from one derived from it, so retries replay and still differ.
if getattr(result, "seed", None) is None:
result.seed = seed
elif context.get("_attempt"):
result.seed = _retry_seed(result.seed, context["_attempt"])
return _scenario_seeded(result, seed), {"object": type(result).__name__}
def _move_farm(farm: Farm, boundary: Boundary, translation) -> Farm:
"""Move a realized farm rigidly onto its arranged cluster boundary."""
offset = np.asarray(translation, dtype=float)
farm.positions = _kept_inside(boundary.geometry, farm.positions + offset)
farm.boundary = boundary
if farm.guide is not None:
farm.guide = farm.guide._transformed(1.0, offset)
farm.metrics.clear()
farm.metadata["boundary"] = boundary.metadata()
return farm
[docs]
class Sampler(_ArgumentRepr):
"""A reproducible description of wind farm scenarios to draw from.
Building the sampler checks every fixed value, range and component type,
and the arguments of every recipe, so most mistakes raise before anything
is drawn. What a recipe, class or callable returns is known only when it
runs, so it is checked as each scenario is drawn. Any component may be a fixed
value or a specification, meaning a ``(low, high)`` range, a frozen
``scipy.stats`` distribution such as ``stats.loguniform(0.03, 1.0)``, a
callable taking one ``Generator``, a :class:`~oowl.sampling.Choice`, a
class, or a ``(constructor, kwargs)`` recipe. An argument whose name ends
in ``_range``, and the points and vectors ``center`` and ``gradient``,
take their pair as given, and a pair of ranges draws each end or
coordinate. The points of ``exterior``, ``holes``, ``sites`` and
``coordinates``, given as numbers, are taken as given too. A distribution
is drawn once per scenario, so wrap it in
:class:`~oowl.sampling.Literal` for an argument that takes the
distribution itself. ``Literal`` passes any value through unresolved. A
range or choice read from a :class:`~oowl.sampling.Link` moves with the
other values on that link.
``boundary`` and ``layout`` describe one farm. ``configuration`` is a
mapping specification that varies boundary, layout, count, area and
density together. It is resolved once per farm, or once per member of a
cluster or region.
``n_turbines`` is the turbine count and ``area`` the farm's area in square
meters. ``density`` is one mapping naming the farm's density,
``{"area_per_turbine": square meters}``, ``{"power_density": watts per
square meter}``, which needs turbines of one rated power, ``{"minimum_spacing":
meters}``, the closest pair, ``{"mean_spacing": meters}``, the mean
nearest-neighbor distance, or ``{"maximum_spacing": meters}``, the
loneliest turbine's nearest-neighbor distance. Its value is any
specification, and a spacing
may also be :class:`~oowl.sampling.RotorDiameters` of the farm's largest
rotor. Give at most two of count, area and density and the third follows.
A boundary given its own area fixes the area as ``area`` does. A layout's
own count or density stands in for the sampler's.
``requirements`` only checks each farm built and sizes nothing. Each key
names a farm measure and maps to the fixed ``(low, high)`` range it must
lie in, ``None`` leaving an end open. ``"spacing"`` holds every turbine's
distance to its nearest neighbor in meters, an end also given as
:class:`~oowl.sampling.RotorDiameters`, so its low end keeps pairs apart
even in a farm of fixed count and area, and its high end leaves no turbine
alone. ``"n_turbines"`` holds a count an area and a density derive,
``"capacity"`` the installed capacity in watts, ``"coverage"`` the share of
the boundary inside the convex hull of the turbines and ``"aspect_ratio"``
how many times as far the turbine positions spread along their long axis
as across it. A farm that misses one raises a ``ValueError``, which
``on_error`` handles, and each records the ranges it met in
``metadata["sample"]["requirements"]``.
``resource``, ``turbines`` and ``surface`` attach to every scenario. In a
cluster turbines are drawn per farm, one resource is shared by the region
and one surface covers the whole scenario.
By default a scenario is one :class:`~oowl.Farm`. Without ``boundary``,
``layout`` and ``configuration`` it is the drawn ``resource`` or
``turbines`` alone, the same one a farm scenario of that index carries.
``n_farms`` makes it a :class:`~oowl.Cluster` of adjacent farms with
clearance ``gap`` in meters. ``domain`` makes it a region scattered with
sites described by ``boundary``, added at least ``gap`` apart until
``n_sites`` are placed. A site that finds no room is replaced by another, and 12 in a row
that find none end the fill. Each farm records its scenario in
``metadata["sample"]["index"]`` and, in a region, its site in
``metadata["group"]``.
``factory`` is an alternative ``f(rng) -> Farm`` that excludes the
declarative fields. It draws every random value from ``rng``, seeds
included, as ``L.Grid(n, seed=rng)``, since an unseeded component inside
it draws a new seed each call. ``on_error="resample"`` draws a scenario that
raises a ``ValueError`` again, up to ``max_attempts`` attempts in all, redrawing its random values
and random choices, while balanced and seeded choices keep theirs. A
layout given its own seed draws each retry from a seed derived from it
and the attempt, so the retries replay, and the farm records the seed it
was drawn from. ``on_error="skip"`` leaves out a scenario that raises a
``ValueError``, so ``sample(n)`` and ``stream(n)`` hold the scenarios of
indices ``start`` to ``start + n - 1`` that could be built, each recording its
index, and ``itertools.islice(sampler.stream(), n)`` gives ``n`` of them.
After 1,000 scenarios in a row are skipped it raises a ``ValueError``, as
a description nothing meets would otherwise stream forever.
``seed`` is the root seed, and a generator or ``None`` is resolved to an
integer once.
"""
def __init__(
self,
boundary=None,
layout=None,
resource=None,
*,
configuration=None,
n_turbines=None,
area=None,
density=None,
turbines=None,
surface=None,
domain=None,
gap=0.0,
n_farms=None,
n_sites=None,
requirements=None,
factory: Callable[[np.random.Generator], Farm] | None = None,
on_error: typing.Literal["raise", "resample", "skip"] = "raise",
max_attempts: int = 10,
seed: int | np.random.Generator | None = None,
):
# Later edits to a caller's dicts, lists or arrays must not reach draws.
boundary, layout, resource, configuration = map(
_copied_spec, (boundary, layout, resource, configuration)
)
# A Literal sizing value is the number it holds, as nothing is drawn.
n_turbines, area = map(_literal_number, (n_turbines, area))
gap, n_farms, n_sites = map(_literal_number, (gap, n_farms, n_sites))
n_turbines, area = map(_copied_spec, (n_turbines, area))
density = _validated_density(density, "density")
densities = {name: density.get(name) for name in _DENSITY_KEYS}
turbines, surface, domain, gap, n_farms, n_sites = map(
_copied_spec, (turbines, surface, domain, gap, n_farms, n_sites)
)
if factory is not None and any(
item is not None
for item in (
boundary,
layout,
configuration,
n_turbines,
area,
*densities.values(),
resource,
turbines,
surface,
domain,
)
):
raise ValueError("factory is mutually exclusive with declarative fields")
# Without a farm, a scenario is one drawn component.
self._component: str | None = None
if (
factory is None
and boundary is None
and layout is None
and configuration is None
):
farm_fields = [
name
for name, value in (
("n_turbines", n_turbines),
("area", area),
*densities.items(),
("surface", surface),
("domain", domain),
("n_farms", n_farms),
("n_sites", n_sites),
)
if value is not None
]
if gap != 0.0:
farm_fields.append("gap")
if requirements is not None:
farm_fields.append("requirements")
if farm_fields:
raise ValueError(
f"{', '.join(farm_fields)} describe a farm, so give a boundary "
"and layout too"
)
given = [
name
for name, value in (("resource", resource), ("turbines", turbines))
if value is not None
]
if len(given) != 1:
raise ValueError(
"without a farm a Sampler draws one component, a resource "
"or turbines"
)
self._component = given[0]
elif factory is None and layout is None and configuration is None:
raise ValueError("layout or configuration is required")
# Domain, farm-count and member-list sizing apply only to plain farm
# draws, as each member of a cluster is sized on its own.
members = _enumerates_members(boundary)
if domain is not None or n_farms is not None or members:
ignored = sorted(
name
for name, value in (
("area", area),
*densities.items(),
("n_turbines", n_turbines),
)
if value is not None
)
if ignored:
unit = "a region" if domain is not None else "a cluster"
names = ", ".join(ignored[:-1])
subject = f"{names} and {ignored[-1]}" if names else ignored[0]
verb = "sizes" if len(ignored) == 1 else "size"
if set(ignored) & set(_DENSITY_KEYS):
advice = (
"Put the specification in a farm configuration instead, "
"as in configuration={'boundary': OffshorePolygon, "
"'layout': PoissonDisc, 'n_turbines': 40, "
"'density': {'minimum_spacing': 600.0}}, which is drawn "
"once per member"
)
else:
advice = (
"Put the specification in the boundary or layout instead, "
"as in boundary=(OffshorePolygon, {'area': (8e6, 20e6)}), "
"which is drawn once per member"
)
if members:
advice = (
"A boundary sequence lists each member, so size the "
"members themselves, as in [Circle(area=1e6), "
"Circle(area=2e6)] with layouts [Grid(5), Grid(7)]"
)
elif n_turbines is not None and domain is None:
# Derive cluster count from the layout when provided.
advice = (
"A built layout carries the cluster total, as in "
"layout=L.Grid(100) with n_farms=5, and a layout "
"recipe carries each farm's own count, as in "
"layout=(L.Grid, {'n_turbines': 100})"
)
raise ValueError(
f"{subject} {verb} a single farm and cannot size {unit}. {advice}"
)
if domain is not None:
if configuration is None and (boundary is None or layout is None):
raise ValueError(
"a domain requires boundary and layout or a farm configuration"
)
if n_farms is not None:
# A domain controls site count, not farm count.
raise TypeError(
"a domain caps sites, so pass n_sites rather than n_farms"
)
elif n_sites is not None:
raise TypeError("n_sites applies to a domain; without one pass n_farms")
if (
n_farms is not None
and configuration is None
and (boundary is None or layout is None)
):
raise ValueError(
"a cluster requires boundary and layout or a farm configuration"
)
if factory is not None and (
not callable(factory) or not _accepts_one_argument(factory)
):
raise TypeError("factory must be callable as f(rng) -> Farm")
given_densities = [
name for name, value in densities.items() if value is not None
]
# Turbine count, area and density cannot all be independently fixed.
sizing = [
name
for name, value in (("n_turbines", n_turbines), ("area", area))
if value is not None
] + given_densities
if len(sizing) == 3:
raise ValueError(
f"{sizing[0]}, {sizing[1]} and {sizing[2]} over-constrain the "
f"same geometry. Give any two and the third follows"
+ _floor_advice(sizing[2])
)
requirements = _validated_requirements(requirements)
_refuse_unmeasurable(
requirements,
turbines=turbines,
boundary=boundary,
declared=configuration is not None or factory is not None,
)
if "n_turbines" in requirements and (
_fixed_count(
n_turbines,
layout,
configuration,
cluster=domain is not None or n_farms is not None or members,
)
):
raise ValueError(
"every farm is given its turbine count, so a requirement on it "
"always or never holds. Bound the count where an area and a "
"density derive it"
)
if on_error not in ("raise", "resample", "skip"):
raise ValueError("on_error must be 'raise', 'resample' or 'skip'")
max_attempts = _validate_integer("max_attempts", max_attempts)
for name, value in (
("boundary", boundary),
("layout", layout),
("configuration", configuration),
("resource", resource),
("turbines", turbines),
("surface", surface),
("domain", domain),
):
_validate_nested_spec(value, name)
if name != "configuration":
_validate_component(value, name)
cluster = domain is not None or n_farms is not None
_validate_configuration_spec(
configuration, _CLUSTER_SCOPED if cluster else frozenset()
)
for name, value in (
("n_turbines", n_turbines),
("area", area),
*densities.items(),
("n_sites", n_sites),
("n_farms", n_farms),
("gap", gap),
):
_validate_scalar_spec(value, name, _SCALAR_BOUNDS[name])
self.boundary = boundary
self.layout = layout
self.configuration = configuration
self.n_turbines = n_turbines
self.area = area
self.density = density or None
self._densities = densities
self.resource = resource
self.turbines = turbines
self.surface = surface
self.domain = domain
self.gap = gap
self.requirements = requirements
self.n_farms = n_farms
self.n_sites = n_sites
self.factory = factory
self.on_error = on_error
self.max_attempts = max_attempts
# Resolve the root seed once so draws are reproducible.
self.seed = resolve_seed(seed)
# Links are numbered in order of first appearance, so dials replay.
self._links = number_links(
configuration,
boundary,
layout,
resource,
n_turbines,
area,
*densities.values(),
turbines,
surface,
domain,
gap,
n_farms,
n_sites,
)
if self._draws_clusters():
# One value serves every farm, so it has no farm to read a dial for.
for name in ("resource", "surface", "domain", "n_farms", "n_sites"):
linked = per_farm_linked(getattr(self, name))
if linked is not None:
raise ValueError(
f"{name} is shared by every farm of a cluster, so it "
f"cannot read link {self._links[linked.link]}, which "
f"turns per farm, in {linked!r}. Read it from a "
"Link(per='scenario') instead"
)
def _configuration(
self, index: int, seed: int, sampler_seed: int
) -> tuple[dict, dict]:
"""Resolve the farm configuration of ``index`` and its record."""
if self.configuration is None:
return {}, {}
selected, record = _resolve(
self.configuration,
Mapping,
seed,
index,
{"_choice_seed": sampler_seed, "_raw_mapping": True},
)
if not isinstance(selected, Mapping):
raise TypeError("configuration choices must resolve to mappings")
unknown = set(selected) - _CONFIGURATION_KEYS
if unknown:
raise ValueError(f"unknown configuration keys: {sorted(unknown)}")
return dict(selected), record
def _retried(self, unit: str, index: int, root_seed: int, draw: Callable):
"""Call ``draw(seed, errors)``, retrying with fresh seeds when resampling.
``errors`` lists the failures of earlier attempts. The last failure is
raised as a ``ValueError`` naming ``unit`` and ``index``.
"""
attempts = self.max_attempts if self.on_error == "resample" else 1
errors: list[str] = []
for attempt, seed in enumerate(spawn_seeds(root_seed, attempts), start=1):
try:
drawn = draw(seed, errors)
self._held_to_requirements(drawn)
return drawn
except ValueError as exc:
if attempt == attempts:
raise ValueError(
f"could not sample {unit} {index} after {attempt} "
f"attempt(s): {exc}"
) from exc
errors.append(f"{type(exc).__name__}: {exc}")
raise AssertionError("unreachable")
def _held_to_requirements(self, unit) -> None:
"""Raise ``ValueError`` when ``unit`` misses a requirement, else record them.
A requirement only checks the drawn farm or cluster, so whatever fails
is drawn again, skipped or raised as ``on_error`` says.
"""
if not self.requirements or not isinstance(unit, (Farm, Cluster)):
return
farms = list(unit) if isinstance(unit, Cluster) else [unit]
for farm in farms:
ranges = {
name: _requirement_range(ends, farm, name)
for name, ends in self.requirements.items()
}
_check_requirements(farm, ranges)
farm.metadata["sample"]["requirements"] = ranges
def _sample_one(
self,
index: int,
root_seed: int,
sampler_seed: int,
cluster_scoped: frozenset[str] = frozenset(),
choice_seed: int | None = None,
) -> Farm:
"""Draw farm ``index``, recording any link dials it turned."""
def draw(seed: int, errors: list[str]) -> Farm:
with scenario_dials(self._links, index, seed, sampler_seed) as dials:
farm = typing.cast(
Farm,
self._draw_farm(
index, seed, errors, sampler_seed, cluster_scoped, choice_seed
),
)
if dials.turned:
farm.metadata["sample"]["links"] = dials.record()
return farm
return self._retried("farm", index, root_seed, draw)
def _preview_boundary(
self,
index: int,
root_seed: int,
sampler_seed: int,
cluster_scoped: frozenset[str],
choice_seed: int,
) -> Boundary | None:
"""The boundary farm ``index`` would be built in, without building it.
It is the first attempt's boundary, from the same seeds and dials as
:meth:`_sample_one`. ``None`` when the farm's boundary depends on its
layout, a count and a spacing sizing it, or when the first attempt
cannot resolve one, so the caller draws the farm instead.
"""
if self.factory is not None:
return None
# A first child does not depend on how many are spawned.
(seed,) = spawn_seeds(root_seed, 1)
try:
with scenario_dials(self._links, index, seed, sampler_seed):
preview = self._draw_farm(
index, seed, [], sampler_seed, cluster_scoped, choice_seed, True
)
except ValueError:
return None
return preview if isinstance(preview, Boundary) else None
def _draw_farm(
self,
index: int,
seed: int,
errors: list[str],
sampler_seed: int,
cluster_scoped: frozenset[str],
choice_seed: int | None,
preview: bool = False,
) -> Farm | Boundary | None:
"""Draw one farm from ``seed`` and record how in ``metadata["sample"]``.
A cluster member leaves the ``cluster_scoped`` components to the
cluster and selects balanced choices from the cluster's
``choice_seed``. ``preview`` returns the boundary the farm is built
in before building, or ``None`` when building sizes it.
"""
if self.factory is not None:
farm = self.factory(as_rng(seed))
if not isinstance(farm, Farm):
raise TypeError("factory must return a Farm")
farm.metadata["sample"] = _json_safe(
{
"index": index,
"seed": seed,
"root": int(sampler_seed),
"factory": getattr(
self.factory, "__name__", type(self.factory).__name__
),
"attempts": len(errors) + 1,
"errors": errors,
"versions": _versions(),
}
)
return farm
choice_seed = sampler_seed if choice_seed is None else choice_seed
config, config_record = self._configuration(index, seed, choice_seed)
overridden = sorted(set(config) & cluster_scoped)
if overridden:
raise ValueError(
f"farm configuration cannot override cluster-scoped {overridden}"
)
values = {
name: getattr(self, name)
for name in _CONTEXT_KEYS
if name not in _DENSITY_KEYS
}
values.update(self._densities)
if "density" in config:
# A configuration's density replaces the sampler's whole.
values.update(dict.fromkeys(_DENSITY_KEYS))
values.update(
_validated_density(config["density"], "configuration density")
)
values.update({key: item for key, item in config.items() if key != "density"})
for name in cluster_scoped:
values[name] = None
stage = dict(zip(_FARM_STAGES, spawn_seeds(seed, len(_FARM_STAGES))))
def scalar(name: str, draw: Callable = draw_scalar, stream: str = ""):
"""Draw one sizing value from its stage stream, or ``None``.
Each field chooses from its own seed, so choices on two fields do
not pick the same position together. A retry reselects a random
choice, as the attempt seed redraws a range.
"""
spec = values[name]
if spec is None:
return None
rng = as_rng(stage[stream or name])
field_seed = derive_seed(choice_seed, name) if is_choice(spec) else None
return draw(
spec,
rng,
name,
index=index,
choice_seed=field_seed,
attempt=len(errors),
)
def component(name: str, expected, context: Mapping | None = None):
"""Resolve one optional component from its stage seed, with its record."""
if values[name] is None:
return None, None
return _resolve(
values[name],
expected,
stage[name],
index,
{**(context or {}), "_choice_seed": choice_seed},
field=name,
)
def density_value(name: str, turbines) -> float:
"""Draw one density, a spacing in rotor diameters turned into meters."""
spec = values[name]
stream = "spacing" if name in _SPACING_KEYS else "area"
if isinstance(spec, RotorDiameters):
rng = as_rng(stage[stream])
inner = spec.diameters
field_seed = (
derive_seed(choice_seed, name) if is_choice(inner) else None
)
diameters = draw_scalar(
inner,
rng,
name,
index=index,
choice_seed=field_seed,
attempt=len(errors),
)
return diameters * _largest_rotor(turbines, name)
return scalar(name, stream=stream)
count = scalar("n_turbines", _draw_count)
area = scalar("area")
boundary, boundary_record = component("boundary", Boundary, {"area": area})
if boundary is not None and not isinstance(boundary, Boundary):
raise TypeError("boundary must resolve to a Boundary or None")
layout, layout_record = _resolve(
values["layout"],
LayoutAlgorithm,
stage["layout"],
index,
{
"area": area,
"n_turbines": count,
"_choice_seed": choice_seed,
"_attempt": len(errors),
},
field="layout",
)
if not isinstance(layout, LayoutAlgorithm):
raise TypeError("layout must resolve to a LayoutAlgorithm")
if count is None:
count = layout.n_turbines
# Power density and rotor diameters turn into a count and meters with
# the farm's turbines, so those resolve first.
resolved: dict = {}
records: dict = {}
value, record = component("turbines", None)
resolved["turbines"] = value
records["turbines"] = (
None
if record is None
else _resolved_record(value, record, stage["turbines"])
)
# A layout's own density stands in for the sampler's, as its count does.
given = [name for name in _DENSITY_KEYS if values[name] is not None]
if len(given) > 1:
raise ValueError(
f"{' and '.join(given)} each set the farm's density, so give one "
"of them"
)
own = {
name: getattr(layout, name)
for name in _LAYOUT_DENSITIES
if getattr(layout, name) is not None
}
for name in _LAYOUT_DENSITIES:
setattr(layout, name, None)
densities: dict[str, float] = {}
density_name: str | None = None
source: str | None = None
if given:
density_name = source = given[0]
densities[density_name] = density_value(density_name, resolved["turbines"])
elif own:
density_name, own_value = next(iter(own.items()))
source = f"the layout's {density_name}"
densities[density_name] = float(own_value)
# A boundary given its area fixes the area as a drawn area does, so a
# count and a density over-constrain it.
fixed_area = area is not None or (boundary is not None and boundary._sized)
if count is not None and fixed_area and density_name is not None:
counted = (
"n_turbines"
if values["n_turbines"] is not None
else "the layout's n_turbines"
)
sized = "area" if area is not None else "the boundary's area"
raise ValueError(
f"{counted}, {sized} and {source} over-constrain the same "
"geometry. Give any two and the third follows"
+ _floor_advice(source or "")
)
if count is None and density_name is None:
raise ValueError(
"n_turbines is required unless the selected layout has a count or "
"a density, or a density sets the count"
)
if boundary is None and area is not None:
raise ValueError("area cannot be used by a standalone layout")
spacing_name = spacing = None
if density_name in ("area_per_turbine", "power_density"):
if boundary is None:
raise ValueError(f"{density_name} requires a boundary")
per_turbine = densities[density_name]
if density_name == "power_density":
per_turbine = _rated_power(resolved["turbines"]) / per_turbine
if count is not None:
area = per_turbine * count
elif area is not None:
count = max(1, round(area / per_turbine))
else:
# The boundary's own area sets the count.
layout.area_per_turbine = per_turbine
elif density_name is not None:
spacing_name, spacing = density_name, densities[density_name]
if count is None or boundary is None:
# The layout fills the boundary to it, or sizes itself by it.
setattr(layout, spacing_name, spacing)
if count is not None:
layout.n_turbines = count
if boundary is not None and area is not None:
# Resize a copy so draws do not mutate a caller-owned boundary.
boundary = boundary._resized(area)
if preview:
from ._api import _farm_boundary
# A count and a spacing size the boundary by drawing the farm.
sized = spacing is not None and count is not None
if boundary is None or sized:
return None
return _farm_boundary(boundary, stage["build"])
from .surfaces import Surface, SurfaceModel
for name, expected, context in (
("resource", (WindResource, Dimitrov), None),
("surface", (Surface, SurfaceModel), {"area": area}),
):
value, record = component(name, expected, context)
resolved[name] = value
if record is not None:
record = _resolved_record(value, record, stage[name])
records[name] = record
def build(at_area=None, drawn=layout, attached=True):
"""Draw one farm of ``drawn``, optionally at a trial extent."""
target = boundary
# The boundary already resized to ``area`` keeps its built shape.
if at_area is not None and at_area != area and boundary is not None:
target = boundary._resized(at_area)
return _build(
target if target is not None else drawn,
None if target is None else drawn,
resource=resolved["resource"] if attached else None,
seed=stage["build"],
turbines=resolved["turbines"] if attached else None,
surface=resolved["surface"] if attached else None,
)
scaled_by = None
if boundary is not None and count is not None and spacing is not None:
minimum = spacing_name == "minimum_spacing"
measure = {
"minimum_spacing": _closest_pair,
"mean_spacing": _mean_nearest,
"maximum_spacing": _loneliest,
}[typing.cast(str, spacing_name)]
at_area = count * spacing**2
# The layout is sized to the spacing without its modifiers, which
# then run once, as after a fill, those that move turbines held
# to a minimum spacing. The components attach last, so a surface
# covers the final extent and turbines the final count.
bare = _bare_layout(layout, stage["build"])
draw = partial(build, drawn=bare, attached=False)
root = resolve_seed(bare.seed)
if boundary._in_meters:
from ._api import _farm_boundary
# Every draw reseeds an unseeded boundary alike, so it is drawn
# once here, and each draw sizes its base rather than searching
# for an area the spacing does not need.
fixed = _farm_boundary(boundary, stage["build"])._uncached_copy()
fixed._pinned_seed = True
def draw_at_base(at_area):
return _build(
fixed._at_base_area(at_area), bare, seed=stage["build"]
)
farm = _drawn_to_spacing(draw_at_base, spacing, at_area, measure)
farm = layout._modified(farm, root, spacing if minimum else None)
else:
# Draw once at about the right scale, run the modifiers held to
# the closest pair there, then scale the farm to the spacing,
# its record staying in the frame the modifiers read it in.
farm = draw(at_area=at_area)
held = measure(farm) if minimum else np.nan
farm = layout._modified(farm, root, held if np.isfinite(held) else None)
farm, scaled_by = _scaled_to_spacing(farm, spacing, measure)
farm.metadata["requested_n"] = count
_attach_components(
farm, resolved["resource"], resolved["turbines"], resolved["surface"]
)
else:
farm = build()
if records["surface"] is not None:
# A generated surface draws its seed when the farm is built.
realized_surface = farm.metadata.get("surface", {})
if isinstance(realized_surface, Mapping) and "seed" in realized_surface:
records["surface"]["seed"] = realized_surface["seed"]
farm.metadata["sample"] = _json_safe(
{
"index": index,
"seed": seed,
"root": int(sampler_seed),
"attempts": len(errors) + 1,
"configuration": config_record,
"boundary": boundary_record,
"layout": {**(layout_record or {}), "algorithm": type(layout).__name__},
"n_turbines": int(farm.n_turbines),
# The realized boundary, not the initial proposal.
"area": None if farm.boundary is None else float(farm.boundary.area),
**{name: densities.get(name) for name in _DENSITY_KEYS},
"versions": _versions(),
"scaled_by": None if scaled_by is None else float(scaled_by),
"closest_pair": _closest_pair(farm)
if len(farm.positions) > 1
else None,
"resource": records["resource"],
"turbines": records["turbines"],
"surface": records["surface"],
"factory": False,
"errors": errors,
}
)
return farm
def _units(self, first: int, count: int | None) -> Iterator:
"""Yield drawn units from index ``first``, endlessly for ``count=None``."""
root = resolve_seed(self.seed)
draw: Callable[[int, int, int], object]
if self._component is not None:
draw = self._sample_component
elif self._draws_clusters():
draw = self._cluster_one
else:
draw = self._sample_one
indices = (
itertools.count(first) if count is None else range(first, first + count)
)
skipped = 0
for index in indices:
# A unit, a farm or a whole cluster, is seeded by its index alone.
try:
unit = draw(index, derive_seed(root, "unit", index), root)
except ValueError as exc:
# A scenario that cannot be built is left out, and the others
# keep their indices, until so many in a row fail that the
# description likely cannot be met at all.
if self.on_error != "skip":
raise
skipped += 1
if skipped >= _SKIP_GIVE_UP:
raise ValueError(
f"skipped {skipped} scenarios in a row, up to index "
f"{index}, so the description is likely unmeetable. "
f"The last failed with {exc}"
) from exc
continue
skipped = 0
yield unit
def _sample_component(self, index: int, root_seed: int, sampler_seed: int):
"""Draw the one component of a farm-free sampler, as a farm draw would."""
name = str(self._component)
expected = (WindResource, Dimitrov) if name == "resource" else None
def draw(seed: int, errors: list[str]):
stream = spawn_seeds(seed, len(_FARM_STAGES))[_FARM_STAGES.index(name)]
with scenario_dials(self._links, index, seed, sampler_seed):
return _resolve(
getattr(self, name),
expected,
stream,
index,
{"_choice_seed": sampler_seed},
field=name,
)[0]
return self._retried(name, index, root_seed, draw)
def _configured_cluster(
self,
seed: int,
sampler_seed: int,
domain: Boundary | None,
n_sites: int | None,
n_farms: int | None,
index: int,
) -> Cluster:
"""Draw complete farm configurations, then place the realized farms.
``index`` is the scenario index, over which the shared resource and
surface balance their choices.
"""
from ._api import _attach_components, _cluster_surface, _region_resource
from ._cluster import _arrange_boundaries, _scatter_sites
cached: dict[int, Farm] = {}
def draw(member_index: int) -> Farm:
if member_index not in cached:
farm = self._sample_one(
member_index,
derive_seed(seed, "member", member_index),
sampler_seed,
cluster_scoped=_CLUSTER_SCOPED,
choice_seed=seed,
)
if farm.boundary is None:
raise ValueError(
"a farm configuration used in a cluster must define a boundary"
)
cached[member_index] = farm
return cached[member_index]
if domain is None:
if n_farms is None:
raise ValueError("a configured cluster requires n_farms")
source_farms = [draw(member_index) for member_index in range(n_farms)]
boundaries, placement = _arrange_boundaries(
tuple(farm.boundary for farm in source_farms),
self.gap,
derive_seed(seed, "placement"),
)
farms = [
_move_farm(farm, boundary, record["translation"])
for farm, boundary, record in zip(source_farms, boundaries, placement)
]
boundary_records = [
{
"index": member_index,
"boundary": type(farm.boundary).__name__,
"seed": int(derive_seed(seed, "member", member_index)),
}
for member_index, farm in enumerate(farms)
]
else:
previews: dict[int, Boundary] = {}
def site_boundary(member_index: int, previewed: bool) -> Boundary:
source = None
if previewed:
# A site the scatter rejects is never built, so only its
# boundary is drawn until it is placed.
source = self._preview_boundary(
member_index,
derive_seed(seed, "member", member_index),
sampler_seed,
_CLUSTER_SCOPED,
seed,
)
if source is not None:
previews[member_index] = source
if source is None:
source = draw(member_index).boundary
if source is None:
raise ValueError(
"a configured region member must define a boundary"
)
# The cells come along, so the site splits as its farm's parts.
return _GeometryBoundary(
source.geometry,
positioned=getattr(source, "_positioned", True),
cells=_cells(source),
)
def scattered(previewed: bool):
return _scatter_sites(
lambda member_index: site_boundary(member_index, previewed),
domain,
self.gap,
n_sites,
derive_seed(seed, "sites"),
)
boundaries, _, placement = scattered(True)
# A retry, or a layout that sizes its boundary, can build a farm in
# another boundary than its preview, which is then scattered again.
for record in placement:
site = int(record["site"])
built = draw(site).boundary
preview = previews.get(site)
if preview is not None and (
built is None
or not built.geometry.equals_exact(preview.geometry, 0.0)
):
boundaries, _, placement = scattered(False)
break
farms = []
boundary_records = []
cursor = 0
for record in placement:
site = int(record["site"])
parts = draw(site).parts
for part, moved in zip(parts, boundaries[cursor : cursor + len(parts)]):
if part.boundary is None:
raise ValueError(
"a configured region member must define a boundary"
)
# Each part moves with its site, centroid onto centroid.
translation = np.subtract(
moved.geometry.centroid.coords[0],
part.boundary.geometry.centroid.coords[0],
)
farm = _move_farm(part, moved, translation)
farm.metadata["group"] = int(record["group"])
farms.append(farm)
boundary_records.append(
{
"index": len(farms) - 1,
"boundary": type(moved).__name__,
"seed": int(derive_seed(seed, "member", site)),
}
)
cursor += len(parts)
region_resource = None
resource_record = None
if self.resource is not None:
region_resource, resource_record = _region_resource(
self.resource, seed, (index, sampler_seed)
)
cluster_surface = None
surface_record = None
if self.surface is not None:
cluster_surface, surface_record = _cluster_surface(
self.surface,
seed,
tuple(farm.boundary for farm in farms),
domain,
(index, sampler_seed),
)
for farm in farms:
_attach_components(
farm,
region_resource,
farm.turbines,
cluster_surface,
)
counts = [farm.n_turbines for farm in farms]
return Cluster(
farms,
metadata={
"kind": "generated_cluster",
"seed": int(seed),
"requested_n": sum(counts),
"achieved_n": sum(counts),
"counts": counts,
"partition": None,
"boundaries": boundary_records,
"placement": {
"method": "scattered" if domain is not None else "configured",
"gap": describe_scalar(self.gap),
"members": placement,
**(
{
"domain": domain.geometry.wkt,
"domain_area": float(domain.area),
"coverage": (
sum(record["area"] for record in placement)
/ float(domain.area)
),
"n_sites": len(placement),
}
if domain is not None
else {}
),
},
"resource": resource_record,
"turbines": {"scope": "member"},
"surface": surface_record,
},
)
def _cluster_one(self, index: int, root_seed: int, sampler_seed: int) -> Cluster:
"""Draw cluster or region ``index``, recording any link dials it turned."""
def draw(seed: int, errors: list[str]) -> Cluster:
with scenario_dials(self._links, index, seed, sampler_seed) as dials:
return self._draw_cluster(index, seed, errors, sampler_seed, dials)
return self._retried("cluster", index, root_seed, draw)
def _draw_cluster(
self, index: int, seed: int, errors: list[str], sampler_seed: int, dials
) -> Cluster:
"""Draw one cluster or region from ``seed`` and record how on each farm."""
domain_seed, count_seed, build_seed = spawn_seeds(seed, 3)
domain = None
domain_record: dict | None = None
if self.domain is not None:
domain, domain_record = _resolve(
self.domain,
Boundary,
domain_seed,
index,
{"_choice_seed": sampler_seed},
field="domain",
)
if not isinstance(domain, Boundary):
raise TypeError("domain must resolve to a Boundary")
# An unseeded random domain is redrawn per scenario, as a farm
# boundary is, so two runs of one sampler seed match.
if domain._is_random and not domain._pinned_seed:
domain = domain._reseeded(domain_seed)
def count(name: str) -> int | None:
"""Draw the site or farm count, which share one stream."""
spec = getattr(self, name)
if spec is None:
return None
return _draw_count(
spec,
as_rng(count_seed),
name,
index=index,
choice_seed=sampler_seed,
attempt=len(errors),
)
n_sites = count("n_sites")
n_farms = count("n_farms")
if self.configuration is not None:
cluster = self._configured_cluster(
build_seed, sampler_seed, domain, n_sites, n_farms, index
)
else:
boundary, layout = self.boundary, self.layout
if boundary is None or layout is None:
raise ValueError("a cluster requires boundary and layout")
cluster = _build(
boundary,
layout,
n_farms=1 if n_farms is None else n_farms,
domain=domain,
gap=self.gap,
n_sites=n_sites,
resource=self.resource,
turbines=self.turbines,
surface=self.surface,
seed=build_seed,
scenario=(index, sampler_seed),
)
record: dict = {
"index": index,
"seed": seed,
"root": int(sampler_seed),
"attempts": len(errors) + 1,
"errors": errors,
"versions": _versions(),
}
if dials.turned:
record["links"] = dials.record()
if domain is not None:
record["domain"] = {
**(domain_record or {}),
"boundary": type(domain).__name__,
"area": float(domain.area),
"seed": int(domain_seed),
}
record["n_sites"] = n_sites
else:
record["n_farms"] = n_farms
# A cluster of one stays a Cluster, so a population never mixes types.
if isinstance(cluster, Farm):
cluster = Cluster((cluster,), metadata={"kind": "generated_cluster"})
for farm in cluster:
farm.metadata["sample"] = _json_safe(
{
**record,
"layout": {
"algorithm": farm.metadata.get("layout", {}).get("type")
},
}
)
return cluster
def _draws_clusters(self) -> bool:
"""Whether the drawn unit is a cluster rather than one farm."""
return (
self.domain is not None
or self.n_farms is not None
or _enumerates_members(self.boundary)
)
[docs]
def sample(self, n: int = 1, *, start: int = 0) -> tuple:
"""Draw ``n`` scenarios starting at index ``start`` and return a tuple.
Each scenario is a :class:`~oowl.Farm`, or a :class:`~oowl.Cluster`
when the sampler has ``n_farms``, a ``domain`` or one boundary per
member. A sampler without a boundary and layout draws its
``resource``, a :class:`~oowl.resources.WindResource` or
:class:`~oowl.resources.Dimitrov`, or else its ``turbines``, one
:class:`~oowl.turbines.Turbine` or a tuple of them. Scenario ``i`` is
the same however many are drawn, so repeated calls give the same
result and ``start`` shards a population.
"""
return tuple(self._units(_start_index(start), _validate_integer("n", n)))
[docs]
def stream(self, n: int | None = None, *, start: int = 0) -> Iterator:
"""Iterate over the scenarios :meth:`sample` would return, one at a time.
``n`` of ``None`` streams indefinitely from index ``start``. ``n`` and
``start`` are checked when the stream is created.
"""
return self._units(
_start_index(start), None if n is None else _validate_integer("n", n)
)
