"""Neutral base classes for layout algorithms and generated results."""
from __future__ import annotations
from abc import ABC, abstractmethod
from collections.abc import Mapping, Sequence
from copy import copy
from dataclasses import dataclass, field
from functools import partial
from pathlib import Path
from typing import TYPE_CHECKING, Any, Self
import numpy as np
from .._random import as_rng, derive_seed, pin_seed, resolve_seed
from .._repr import _ArgumentRepr
from .._validation import _validate_integer, _validate_positive
if TYPE_CHECKING: # pragma: no cover
from collections.abc import Iterable
import matplotlib.axes
from numpy.typing import ArrayLike
from ..boundaries.base import Boundary
from ..modifiers import LayoutModifier
from ..resources import Dimitrov, WindResource
from ..strings.base import StringGuide
from ..surfaces import Surface
from ..turbines import Turbine
from .fitters import Fitter
__all__ = ["Farm", "LayoutAlgorithm"]
def _allocate_counts(n_turbines: int, weights) -> np.ndarray:
"""Split a count by weight with deterministic largest remainder."""
weights = np.asarray(weights, dtype=float)
if weights.ndim != 1 or not len(weights):
raise ValueError("weights must be a non-empty one-dimensional sequence")
if not np.all(np.isfinite(weights)) or np.any(weights < 0):
raise ValueError("weights must be finite and non-negative")
if weights.sum() <= 0:
raise ValueError("at least one weight must be positive")
quotas = int(n_turbines) * weights / weights.sum()
counts = np.floor(quotas).astype(int)
remainder = int(n_turbines) - int(counts.sum())
fractions = quotas - counts
order = sorted(range(len(weights)), key=lambda index: (-fractions[index], index))
for index in order[:remainder]:
counts[index] += 1
return counts
def _nearest_neighbor_distances(points: np.ndarray) -> np.ndarray:
"""Distance from each of at least two ``(N, 2)`` points to its nearest other."""
from scipy.spatial import cKDTree # ty: ignore[unresolved-import]
return cKDTree(points).query(points, k=2)[0][:, 1]
def _median_spacing(points: np.ndarray) -> float | None:
"""Median positive nearest-neighbor distance, or ``None`` if there is none."""
if len(points) < 2:
return None
nearest = _nearest_neighbor_distances(points)
usable = nearest[np.isfinite(nearest) & (nearest > 0)]
return float(np.median(usable)) if len(usable) else None
def _closest_pair(farm: Farm) -> float:
"""Distance between the two nearest turbines, or ``inf`` if there is one."""
from ..metrics import nearest_neighbor_distances
if len(farm.positions) < 2:
return float("inf")
return float(np.nanmin(nearest_neighbor_distances(farm)))
def _mean_nearest(farm: Farm) -> float:
"""Mean nearest-neighbor distance, or ``nan`` below two turbines."""
return farm.mean_spacing()
def _loneliest(farm: Farm) -> float:
"""The loneliest turbine's nearest-neighbor distance, ``nan`` below two."""
return farm.maximum_spacing()
#: The quantities that size a layout inside a boundary, at most one given.
_SIZING = (
"n_turbines",
"area_per_turbine",
"minimum_spacing",
"mean_spacing",
"maximum_spacing",
)
#: The sizing quantities a layout holds by searching or filling.
_SPACINGS = ("minimum_spacing", "mean_spacing", "maximum_spacing")
#: The densities a layout takes as ``density={name: value}``, one at most.
_DENSITIES = ("area_per_turbine", "minimum_spacing", "mean_spacing", "maximum_spacing")
def _optional_positive(name: str, value) -> float | None:
"""``value`` as a positive finite float, or ``None``."""
return None if value is None else _validate_positive(name, value)
def _layout_density(density: "Mapping[str, Any] | None") -> dict[str, float]:
"""A layout's checked ``density`` as ``{name: value}``, empty for ``None``."""
if density is None:
return {}
if not isinstance(density, Mapping):
raise TypeError(
"density must be a mapping such as {'minimum_spacing': 800.0}, not "
f"{type(density).__name__}"
)
if "power_density" in density:
raise ValueError(
"a layout has no turbines to turn a power_density into a count, so "
"give it to a Sampler with turbines"
)
unknown = sorted(str(name) for name in density if name not in _DENSITIES)
if unknown:
raise ValueError(
f"unknown density {unknown}, which may be {', '.join(_DENSITIES)}"
)
if len(density) != 1:
raise ValueError(
f"density takes one of {', '.join(_DENSITIES)}, got {len(density)}"
)
name, value = next(iter(density.items()))
return {name: _validate_positive(f"density['{name}']", value)}
class _NoFitError(ValueError):
"""No two turbines of a layout fit its boundary at the spacing asked."""
def _placed_for(modifiers, n_kept: int) -> int:
"""Turbines to place for ``n_kept`` to remain after ``modifiers`` run in order."""
for modifier in reversed(modifiers):
n_kept = modifier._placed_for(n_kept)
return n_kept
def _no_fit_error(spacing: float, area: float, name: str) -> _NoFitError:
return _NoFitError(
f"no {name} of two or more turbines fits {area / 1e6:.3g} km2 "
f"with {spacing:.3g} m spacing"
)
#: Most turbines a fill may test, as many as a candidate fitter's default
#: ``candidate_cap``, so a spacing far too small for its boundary is refused
#: at once rather than filled for minutes.
_MAX_FILL_TURBINES = 100_000
def _too_many_error(spacing: float, count: int, holder: str) -> ValueError:
return ValueError(
f"a spacing of {spacing:.4g} m is too small for this boundary, "
f"whose {holder} would test about {count:,} turbines"
)
#: How to size a farm's boundary, the advice of :func:`_unsized_error`.
_SIZE_A_FARM = (
"Give it area=... in square meters, or let a Sampler size it with area, a "
"density, or n_turbines with a spacing"
)
def _unsized_error(boundary, advice: str = _SIZE_A_FARM) -> ValueError:
"""The error for laying a farm out on a boundary nothing gave a size."""
return ValueError(
f"{type(boundary).__name__} has no area, so it is a unit-size shape. {advice}"
)
#: Draws that refine a count by the prediction a mean distance gives.
_MEAN_REFINEMENTS = 6
def _refined_count(built, measure, spacing: float, count: int) -> int:
"""The count whose ``measure`` a mean distance predicts to hold ``spacing``.
A mean nearest-neighbor distance falls about as ``1 / sqrt(n)``, so each
draw predicts the count ``n * (measure / spacing) ** 2``, held within a
factor of two of the last. A lattice's mean jumps where its rows change,
and the prediction then swung across the jump until the draws ran out,
so the most counts known to hold the spacing and the fewest known not to
bracket it, and a prediction past either ends the refinement at that
edge, for the steps of :func:`_most_turbines_for_spacing` to settle.
Returns the last count when the prediction repeats, the draws fail or
:data:`_MEAN_REFINEMENTS` run out.
"""
holds, fails = 0, _MAX_FILL_TURBINES + 1
for _ in range(_MEAN_REFINEMENTS):
if count > _MAX_FILL_TURBINES:
break
farm = built(count)
value = np.nan if farm is None else measure(farm)
if not np.isfinite(value):
break
if value >= spacing:
holds = max(holds, count)
else:
fails = min(fails, count)
predicted = round(count * (value / spacing) ** 2)
predicted = int(min(max(predicted, max(2, count // 2)), 2 * count))
if predicted == count:
break
if predicted <= holds:
return holds
if predicted >= fails:
return max(holds, fails - 1)
count = predicted
return count
def _most_turbines_for_spacing(
build,
spacing: float,
area: float,
estimate: int | None = None,
name="layout",
measure=None,
) -> Farm:
"""Draw the most turbines whose ``measure`` is at least ``spacing``.
``measure`` maps a farm to a distance, its closest pair unless given.
``build(count)`` draws ``count`` turbines in a boundary of ``area``. The
count starts at ``estimate``, or else from one trial draw, since spacing
scales as ``sqrt(area / n)``. It is stepped by one, as most estimates miss
by a few, and after four steps by doubling ones until a count flips
between fitting and not, the two then bisected. An estimate far off so
costs a few draws, not one per count. A step down removes at most an
eighth of the count, since the closest pair of a layout such as Delaunay
need not grow as the count falls, and a longer step could pass over every
count that fits. A count the layout cannot build at all does not fit
either, and the last such error is kept as the cause when no count fits.
A mean distance falls smoothly with the count, so the draws first follow
its prediction and the steps only confirm the count found.
"""
if measure is None:
measure = _closest_pair
failure: ValueError | None = None
# Every count is drawn from one seed, so a count drawn twice is the same.
draws: dict[int, Farm | None] = {}
def built(count: int) -> Farm | None:
nonlocal failure
if count not in draws:
try:
draws[count] = build(count)
except ValueError as error:
failure = error
draws[count] = None
return draws[count]
if estimate is None:
trial_n = 16
trial = built(trial_n)
if trial is None:
# Too many to build, so step up from the fewest instead.
estimate = 2
else:
closest = measure(trial)
if not np.isfinite(closest):
raise ValueError("spacing needs at least two turbines to separate")
estimate = int(trial_n * (closest / spacing) ** 2)
count = max(2, int(estimate))
if measure is not _closest_pair:
count = _refined_count(built, measure, spacing, count)
if count > _MAX_FILL_TURBINES:
raise _too_many_error(spacing, count, "layout")
def honoured(count: int) -> Farm | None:
if count > _MAX_FILL_TURBINES:
return None
farm = built(count)
return farm if farm is not None and measure(farm) >= spacing else None
def steps():
yield from (1, 1, 1, 1)
step = 2
while True:
yield step
step *= 2
# ``low`` fits and ``high`` does not, once both are found.
first = honoured(count)
if first is not None:
low, farm = count, first
for step in steps():
more = honoured(low + step)
if more is None:
high = low + step
break
low, farm = low + step, more
else:
high = count
for step in steps():
low = max(2, high - min(step, max(1, high // 8)))
found = honoured(low)
if found is not None:
farm = found
break
if low == 2:
raise _no_fit_error(spacing, area, name) from failure
high = low
while high - low > 1:
middle = (low + high) // 2
more = honoured(middle)
if more is None:
high = middle
else:
low, farm = middle, more
return farm
def _closest_distance(points: np.ndarray) -> float:
"""Distance between the two nearest of ``points``, or ``inf`` below two."""
if len(points) < 2:
return float("inf")
return float(_nearest_neighbor_distances(points).min())
def _fewest_turbines_for_ceiling(
build, spacing: float, area: float, name="layout"
) -> Farm:
"""Draw the fewest turbines, at least two, that leave none farther than ``spacing``.
The loneliest turbine's nearest neighbor shrinks about as ``1 / sqrt(n)``
as the count grows, so the count starts where a lattice of ``area /
spacing ** 2`` cells would sit and each draw predicts the count that
meets the ceiling from the distance it measured, a little past it so the
next draw lands on the other side. A count that holds the ceiling and
the one drawn before it, which did not, are then bisected. A step rises
at most fourfold and falls at most a quarter, so a poor prediction does
not pass over every count that holds. A count the layout cannot build
holds nothing.
"""
draws: dict[int, tuple[Farm | None, float]] = {}
def measured(count: int) -> tuple[Farm | None, float]:
if count > _MAX_FILL_TURBINES:
raise _too_many_error(spacing, count, "layout")
if count not in draws:
try:
farm = build(count)
except ValueError:
draws[count] = (None, np.inf)
else:
loneliest = _loneliest(farm)
draws[count] = (farm, loneliest if np.isfinite(loneliest) else np.inf)
return draws[count]
def holding(count: int) -> Farm | None:
farm, loneliest = measured(count)
return farm if loneliest <= spacing else None
count = max(2, round(area / spacing**2))
# ``low`` does not hold and ``high`` does, once both are found.
while True:
found, loneliest = measured(count)
if found is not None and loneliest <= spacing:
high, farm = count, found
if count == 2:
return found
target = count * (loneliest / spacing) ** 2 * 0.97
following = min(
count - 1, max(int(np.floor(target)), count - count // 4, 2)
)
if holding(following) is None:
low = following
break
else:
low = count
target = (
count * 2.0
if np.isinf(loneliest)
else count * (loneliest / spacing) ** 2
)
following = max(count + 1, min(int(np.ceil(target * 1.03)), 4 * count))
farm = holding(following)
if farm is not None:
high = following
break
count = following
while high - low > 1:
middle = (low + high) // 2
found = holding(middle)
if found is None:
low = middle
else:
high, farm = middle, found
if farm is None:
raise _no_fit_error(spacing, area, name)
return farm
def _kept_apart(positions: np.ndarray, spacing: float) -> np.ndarray:
"""``positions`` without each turbine closer than ``spacing`` to an earlier one."""
return positions[_kept_apart_mask(positions, spacing)]
def _kept_apart_mask(positions: np.ndarray, spacing: float) -> np.ndarray:
"""Which of ``positions`` no earlier kept one lies closer than ``spacing`` to."""
from scipy.spatial import cKDTree # ty: ignore[unresolved-import]
kept = np.ones(len(positions), dtype=bool)
if len(positions) < 2:
return kept
# Pairs at most just under ``spacing`` apart are the ones closer than it.
pairs = cKDTree(positions).query_pairs(
np.nextafter(spacing, 0.0), output_type="ndarray"
)
if not len(pairs):
return kept
earlier = pairs.min(axis=1)
later = pairs.max(axis=1)
order = np.argsort(later, kind="stable")
# Taken by their later point, every earlier point is settled when reached.
flags = [True] * len(positions)
for first, second in zip(earlier[order].tolist(), later[order].tolist()):
if flags[first]:
flags[second] = False
return np.asarray(flags)
def _apart_from(points, references, spacing):
"""``points`` no closer than ``spacing`` to any of ``references``."""
from scipy.spatial import cKDTree # ty: ignore[unresolved-import]
if not len(points) or not len(references):
return points
distances, _ = cKDTree(references).query(points)
return points[distances >= spacing]
#: Share of a piece that clearing it of its neighbors may remove. A piece that
#: a full clearance would cut by more is cleared by less, the most that
#: removes at most this share.
_MAX_CLEARED_SHARE = 0.5
#: Steps that shrink a clearance toward the share it may remove.
_CLEARANCE_STEPS = 8
def _lattice_clearance(piece, count: int) -> float:
"""Half the spacing of ``count`` turbines on an equilateral lattice over ``piece``.
Two neighbors sharing an edge each give up that much of their own, so
their turbines stay about a spacing apart across it.
"""
return 0.5 * float(np.sqrt(2.0 * piece.area / (np.sqrt(3.0) * count)))
def _within_share(piece, cleared, clearance: float, cleared_at):
"""The shape cleared by the largest clearance up to ``clearance`` within the share.
``cleared`` is ``piece`` cleared by ``clearance``, and ``cleared_at(c)``
clears it by ``c``. Clearing may remove at most :data:`_MAX_CLEARED_SHARE`
of the piece. Along a shared edge the area removed grows about in
proportion to the clearance, so each step scales the clearance by the
share over the area removed, a little below it. Returns the shape and the
clearance, or ``piece`` and ``0.0`` without one.
"""
allowed = _MAX_CLEARED_SHARE * piece.area
for step in range(_CLEARANCE_STEPS + 1):
removed = piece.area - (0.0 if cleared.is_empty else cleared.area)
if not cleared.is_empty and removed <= allowed:
return cleared, clearance
if not clearance or step == _CLEARANCE_STEPS:
break
clearance *= 0.95 * allowed / removed
cleared = cleared_at(clearance)
return piece, 0.0
def _cleared(pieces, index: int, clearance: float):
"""Piece ``index`` without what lies within ``clearance`` of the others.
Returns the shape and the clearance, less than ``clearance`` where that
would remove more than half the piece, and ``0.0`` with the piece itself
when it has no neighbor that close.
"""
import shapely
piece = pieces[index]
neighbors = [
other
for position, other in enumerate(pieces)
if position != index and piece.distance(other) < clearance
]
if not neighbors:
return piece, 0.0
others = shapely.union_all(neighbors)
def cleared_at(distance):
return piece.difference(others.buffer(distance))
return _within_share(piece, cleared_at(clearance), clearance, cleared_at)
def _laid_out_clear(piece, cleared, clearance: float, build):
"""``build`` laid out on ``cleared``, a part of ``piece``, and the clearance.
``build(boundary)`` returns a farm inside the boundary it is given. A
piece too tight for its turbines once cleared is laid out whole, with
clearance ``0.0``, and an error laying it out whole propagates.
"""
from ..boundaries.base import _GeometryBoundary
if clearance:
try:
return build(_GeometryBoundary(cleared)), clearance
except (TypeError, ValueError):
pass
return build(_GeometryBoundary(piece)), 0.0
#: Distance, as a share of the farm's span, within which a turbine in a
#: rounding crack between the parts of a boundary joins the nearest part.
_PIECE_CRACK = 1e-9
def _grouped_by_piece(pieces, positions: np.ndarray) -> list[np.ndarray]:
"""Indices of ``positions`` in each of ``pieces``, a turbine in its first piece.
Partition cells cut with no gap meet only to rounding, so a turbine on a
shared edge, as a Sunflower's center was, can fall in a crack 2e-13 m
wide. Such a turbine joins its nearest piece. Raises ``ValueError`` for
one farther than rounding from every piece.
"""
import shapely
remaining = np.ones(len(positions), dtype=bool)
groups = []
for piece in pieces:
inside = remaining & shapely.intersects_xy(
piece, positions[:, 0], positions[:, 1]
)
groups.append(np.flatnonzero(inside))
remaining &= ~inside
if remaining.any():
stray = np.flatnonzero(remaining)
distances = shapely.distance(
np.asarray(pieces, dtype=object)[:, None],
shapely.points(positions[stray])[None, :],
)
bounds = shapely.total_bounds(np.asarray(pieces, dtype=object))
span = max(float(np.hypot(bounds[2] - bounds[0], bounds[3] - bounds[1])), 1.0)
if distances.min(axis=0).max() > _PIECE_CRACK * span:
raise ValueError("a filled turbine lies outside every part of the boundary")
nearest = distances.argmin(axis=0)
for index, piece_index in zip(stray, nearest):
groups[piece_index] = np.sort(np.append(groups[piece_index], index))
return groups
def _spread_selection(candidates, count: int, references, rng) -> np.ndarray:
"""Greedily pick ``count`` candidates, each farthest from all placed points.
``references`` are fixed points the picks keep away from. Without any, the
first pick is a random candidate.
"""
if count > len(candidates):
raise ValueError(f"cannot pick {count} of {len(candidates)} candidates")
selected = []
first = None
if not len(references):
first = int(rng.integers(len(candidates)))
selected.append(candidates[first])
references = candidates[first].reshape(1, 2)
# The squares of the norm's distances, whose square root is monotone, so
# the root of the least is the least root, without the norm's large
# intermediate of every coordinate difference.
dx = candidates[:, None, 0] - references[None, :, 0]
dy = candidates[:, None, 1] - references[None, :, 1]
nearest = np.sqrt((dx * dx + dy * dy).min(axis=1))
# A picked candidate is marked rather than removed, which would copy the
# arrays at every pick.
if first is not None:
nearest[first] = -np.inf
x = np.ascontiguousarray(candidates[:, 0])
y = np.ascontiguousarray(candidates[:, 1])
dx = np.empty_like(x)
dy = np.empty_like(y)
box = float(np.ptp(x) * np.ptp(y)) if len(x) else 0.0
rows = tree = None
check = np.inf
while len(selected) < count:
choice = int(np.argmax(nearest))
selected.append(candidates[choice])
reach = float(nearest[choice])
if (
rows is None
and len(x) > _SPREAD_ROWS_FROM
and reach <= check
and np.pi * reach * reach * len(x) <= _SPREAD_BALL * box
):
rows, targets, distances, tree = _neighbor_rows(x, y, reach, tree)
# Candidates that crowd part of their box hold more, so they are
# counted again once the distance has shrunk by a fifth.
check = 0.8 * reach
if rows is None:
# Each pick only lowers the distances by its own, so update them in
# place of measuring every candidate against every placed point.
np.subtract(x, x[choice], out=dx)
np.subtract(y, y[choice], out=dy)
np.multiply(dx, dx, out=dx)
np.multiply(dy, dy, out=dy)
np.add(dx, dy, out=dx)
np.sqrt(dx, out=dx)
np.minimum(nearest, dx, out=nearest)
else:
# A pick lies farthest from the placed points, so it brings no
# candidate farther from it than that any closer to them, and the
# farthest distance only shrinks. So the pairs within it at the
# first such pick hold every distance a later pick can lower, and
# their row in place of a full pass gives the same picks.
near = targets[rows[choice] : rows[choice + 1]]
nearest[near] = np.minimum(
nearest[near], distances[rows[choice] : rows[choice + 1]]
)
nearest[choice] = -np.inf
return np.asarray(selected)
#: Candidates a pick's farthest distance holds around it on average, below
#: which :func:`_spread_selection` lists each candidate's neighbors once
#: rather than measure every candidate at every pick.
_SPREAD_BALL = 64.0
#: Candidates above which listing the neighbors pays. Below, a full pass is
#: quicker than building the lists, twice as quick for 400 candidates.
_SPREAD_ROWS_FROM = 1_000
def _neighbor_rows(x, y, reach: float, tree=None):
"""Each candidate's neighbors within ``reach`` and their distances.
Rows of a sparse adjacency, so candidate ``i`` has neighbors
``targets[rows[i]:rows[i + 1]]``, or ``None`` where a sample of the
candidates holds more than :data:`_SPREAD_BALL` on average. A distance is
measured as a full pass of :func:`_spread_selection` measures it, and the
tree's own rounding is covered by a margin. The tree comes back for the
next call.
"""
from scipy.spatial import cKDTree # ty: ignore[unresolved-import]
points = np.column_stack([x, y])
if tree is None:
tree = cKDTree(points)
radius = reach * (1.0 + 1e-9)
sample = points[:: max(1, len(points) // 64)]
if tree.query_ball_point(sample, radius, return_length=True).mean() > _SPREAD_BALL:
return None, None, None, tree
pairs = tree.query_pairs(radius, output_type="ndarray")
sources = np.concatenate([pairs[:, 0], pairs[:, 1]])
targets = np.concatenate([pairs[:, 1], pairs[:, 0]])
# A row's order does not matter, and a stable sort of 16-bit keys is a
# radix sort, several times faster than sorting the pairs' own indices.
keys = sources.astype(np.uint16) if len(x) <= 65_535 else sources
order = np.argsort(keys, kind="stable")
sources, targets = sources[order], targets[order]
dx = x[targets] - x[sources]
dy = y[targets] - y[sources]
distances = np.sqrt(dx * dx + dy * dy)
rows = np.concatenate([[0], np.cumsum(np.bincount(sources, minlength=len(x)))])
return rows, targets, distances, tree
def _inside(boundary, positions: np.ndarray) -> np.ndarray:
"""``positions`` with any a rounding error outside ``boundary`` moved inside.
The cells of a partition without gaps do not share the exact noding of
their outline, so a turbine on a cell's outer edge can test just outside.
"""
from ..boundaries.base import _inset_interior, _project_inside
if not len(positions):
return positions
outline = boundary.geometry
return _project_inside(outline, _inset_interior(outline), positions)
def _combine_guides(guides):
"""One string guide holding every logical string of ``guides``, or ``None``."""
guides = [guide for guide in guides if guide is not None]
if not guides:
return None
from ..strings.base import _GeometryString
return _GeometryString(
tuple(logical_string for guide in guides for logical_string in guide._strings)
)
def _type_label(value) -> str:
"""Public name of an attached object's type, for the farm repr."""
from ..resources import _WindIOResource
if isinstance(value, _WindIOResource):
return "windIO resource"
return type(value).__name__
def _turbines_for(turbines, start: int, stop: int):
"""Turbines for one component's position slice.
A slice holding a single type collapses to that one turbine.
"""
if not isinstance(turbines, tuple):
return turbines
sliced = turbines[start:stop]
if sliced and len({id(item) for item in sliced}) == 1:
return sliced[0]
return sliced
def _boundary_pieces(boundary, count: int) -> tuple:
"""One boundary per recorded component, in the order generation filled them.
Generation records one component per partition cell of the boundary, or
per polygon without cells, so a different count means the farm's
metadata and boundary no longer describe the same farm, and
``ValueError`` is raised.
"""
from ..boundaries.base import _farm_pieces, _GeometryBoundary
if boundary is None:
return (None,) * count
pieces = _farm_pieces(boundary)
if len(pieces) != count:
raise ValueError(
f"farm metadata records {count} components but its boundary has "
f"{len(pieces)} parts"
)
return tuple(_GeometryBoundary(piece) for piece in pieces)
def _same_geometry(first, second) -> bool:
"""Whether two boundaries or guides, or ``None``, are alike in type and shape."""
if first is None or second is None:
return first is second
return type(first) is type(second) and bool(
first.geometry.equals_exact(second.geometry, 0.0)
)
[docs]
@dataclass(init=False, repr=False, eq=False)
class Farm:
"""One wind farm with turbine positions and their associated context.
``positions`` has shape ``(N, 2)`` in meters. ``turbines`` is one shared
:class:`~oowl.turbines.Turbine` or one per position. ``metadata`` records
how the farm was generated, and ``metrics`` caches :meth:`summary`.
"""
positions: np.ndarray
boundary: "Boundary | None"
metadata: dict = field(default_factory=dict)
guide: "StringGuide | None" = None
resource: "WindResource | Dimitrov | None" = None
turbines: "Turbine | Sequence[Turbine] | None" = None
metrics: dict = field(default_factory=dict, compare=False, repr=False)
surface: "Surface | None" = None
def __init__(
self,
positions: ArrayLike,
boundary: "Boundary | None",
*,
metadata: dict | None = None,
guide: "StringGuide | None" = None,
resource: "WindResource | Dimitrov | None" = None,
turbines: "Turbine | Sequence[Turbine] | None" = None,
metrics: dict | None = None,
surface: "Surface | None" = None,
) -> None:
# Copy, so later edits to the caller's array leave the farm unchanged.
points = np.array(positions, dtype=float)
if points.size == 0:
points = np.empty((0, 2), dtype=float)
if points.ndim != 2 or points.shape[1] != 2:
raise ValueError("positions must have shape (N, 2)")
if not np.all(np.isfinite(points)):
raise ValueError("positions must contain only finite values")
# Positions come first, so the turbine count is checked against them.
self.positions = points
self.boundary = boundary
if metadata is not None and not isinstance(metadata, dict):
raise TypeError("metadata must be a dictionary")
# Copy, so farms built from one template dict keep their own records.
self.metadata = {} if metadata is None else dict(metadata)
self.guide = guide
self.resource = resource
self.turbines = turbines
self.metrics = {} if metrics is None else metrics
self.surface = surface
def __setattr__(self, name, value):
if name == "boundary" and value is not None:
from ..boundaries.base import Boundary
if not isinstance(value, Boundary):
raise TypeError(
f"boundary must be a Boundary or None, not {type(value).__name__}"
)
if name == "resource" and value is not None:
if not hasattr(value, "_to_windio_energy_resource"):
raise TypeError("resource must be a WindResource, Dimitrov or None")
if name == "turbines" and value is not None:
from ..turbines import Turbine
if not isinstance(value, Turbine):
if isinstance(value, (str, bytes)):
raise TypeError(
"turbines must be a Turbine or a sequence of Turbine "
f"objects, not {type(value).__name__}"
)
try:
value = tuple(value)
except TypeError as exc:
raise TypeError(
"turbines must be a Turbine or a sequence of Turbine objects"
) from exc
# The type first, so a wrong kind of value is not reported as
# a wrong count.
if not all(isinstance(item, Turbine) for item in value):
raise TypeError("turbines must contain only Turbine objects")
positions = self.__dict__.get("positions")
if positions is not None and len(value) != len(positions):
raise ValueError(
"number of turbines must match the number of positions"
)
if name == "surface" and value is not None:
from ..surfaces import Surface
if not isinstance(value, Surface):
raise TypeError("surface must be a Terrain, Bathymetry, or None")
object.__setattr__(self, name, value)
def __eq__(self, other: object) -> bool:
"""Whether ``other`` is a farm with the same positions and context.
The boundary and guide compare by type and exact geometry, the other
parts by their own ``==``. The cached ``metrics`` are not compared.
"""
if other.__class__ is not self.__class__:
return NotImplemented
return (
bool(np.array_equal(self.positions, other.positions))
and _same_geometry(self.boundary, other.boundary)
and _same_geometry(self.guide, other.guide)
and all(
getattr(self, name) == getattr(other, name)
for name in ("metadata", "resource", "turbines", "surface")
)
)
def __repr__(self) -> str:
# Key quantities only, not every position and metadata entry.
parts = [f"n_turbines={self.n_turbines}"]
if self.boundary is not None:
parts.append(f"area={self.boundary.area:.6g}")
record = self.metadata.get("sample")
if isinstance(record, dict):
layout = record.get("layout")
if isinstance(layout, dict) and layout.get("algorithm"):
parts.append(f"layout={layout['algorithm']}")
parts.append(f"seed={record.get('root')}, index={record.get('index')}")
for name in ("resource", "surface"):
value = getattr(self, name)
if value is not None:
parts.append(f"{name}={_type_label(value)}")
if self.turbines is not None:
if isinstance(self.turbines, tuple):
kinds = len({id(turbine) for turbine in self.turbines})
parts.append(f"turbines={kinds} type{'s' if kinds != 1 else ''}")
else:
parts.append(f"turbines={type(self.turbines).__name__}")
return f"Farm({', '.join(parts)})"
@property
def n_turbines(self) -> int:
"""Number of turbine positions in this farm."""
return int(len(self.positions))
@property
def parts(self) -> "tuple[Farm, ...]":
"""This farm's recorded pieces, each as a farm of its own.
Pieces come from a disjoint boundary, a partition or a region. On a
partitioned boundary ``farm.parts[i]`` is the farm in
``farm.boundary.parts[i]``. A farm of one piece returns ``(self,)``. A
boundary whose part count differs from the recorded ``components``
raises ``ValueError``.
"""
components = self.metadata.get("components")
if not components:
return (self,)
pieces = _boundary_pieces(self.boundary, len(components))
farms = []
for piece, record in zip(pieces, components):
start, stop = int(record["start"]), int(record["stop"])
farms.append(
Farm(
positions=self.positions[start:stop],
boundary=piece,
metadata={**record, "kind": "part"},
resource=self.resource,
turbines=_turbines_for(self.turbines, start, stop),
surface=self.surface,
)
)
return tuple(farms)
[docs]
def summary(
self, metrics: Iterable[str] | None = None, refresh: bool = False
) -> dict:
"""Overview of this farm as an ordered ``name -> value`` dict.
``metrics=None`` gives the turbine count, the mean nearest-neighbor
distance and its coefficient of variation, the density and the area.
Otherwise pass names from :data:`oowl.metrics.SCALAR_METRICS`. The result
is cached on ``self.metrics`` until the positions or the boundary
change. Pass ``refresh=True`` to recompute.
"""
from ..metrics import _metric_names
from ..metrics import summary as _summary
requested = _metric_names(metrics)
# The positions are a public array, so a farm edited in place gave the
# summary of its old positions.
measured = self.__dict__.get("_measured")
if (
not refresh
and self.metrics
and tuple(self.metrics) == requested
and measured is not None
and measured[1] is self.boundary
and np.array_equal(measured[0], self.positions)
):
return dict(self.metrics)
self.metrics = _summary(self, requested)
self._measured = (self.positions.copy(), self.boundary)
return dict(self.metrics)
@property
def x(self) -> np.ndarray:
"""X coordinates of all turbines in meters."""
return self.positions[:, 0]
@property
def y(self) -> np.ndarray:
"""Y coordinates of all turbines in meters."""
return self.positions[:, 1]
@property
def bounds(self) -> tuple[float, float, float, float]:
"""Turbine-position bounding box ``(min_x, min_y, max_x, max_y)`` in meters."""
if not len(self.positions):
raise ValueError("empty farm has no turbine bounds")
return (
float(np.min(self.x)),
float(np.min(self.y)),
float(np.max(self.x)),
float(np.max(self.y)),
)
@property
def centroid(self) -> tuple[float, float]:
"""Mean turbine position ``(x, y)`` in meters."""
if not len(self.positions):
raise ValueError("empty farm has no turbine centroid")
center = np.mean(self.positions, axis=0)
return (float(center[0]), float(center[1]))
@property
def extent(self) -> tuple[float, float]:
"""Turbine-position extent ``(width, height)`` in meters."""
minx, miny, maxx, maxy = self.bounds
return (maxx - minx, maxy - miny)
@property
def coordinates(self) -> np.ndarray:
"""Return turbine coordinates as ``(x, y, z)`` rows in meters.
``z`` is the terrain elevation when the surface is terrain, else zero.
"""
z = np.zeros(self.n_turbines, dtype=float)
if self.surface is not None and self.surface.kind == "terrain":
z = self.surface.at(self.positions)
return np.column_stack((self.positions, z))
[docs]
def nearest_neighbor_distances(self) -> np.ndarray:
"""Distance in meters from each turbine to its nearest neighbor.
Shape ``(N,)``, filled with ``nan`` when the farm has fewer than two
turbines, as in :func:`oowl.metrics.nearest_neighbor_distances`.
"""
if len(self.positions) < 2:
return np.full(len(self.positions), np.nan)
return _nearest_neighbor_distances(self.positions)
[docs]
def minimum_spacing(self) -> float:
"""Closest pair in meters, ``nan`` below two turbines.
What a layout's ``minimum_spacing`` holds, so a farm sized by one
reaches at least it. A sampled farm records it in
``metadata["sample"]["closest_pair"]``.
"""
distances = self.nearest_neighbor_distances()
if len(distances) < 2:
return float("nan")
return float(np.min(distances))
[docs]
def mean_spacing(self) -> float:
"""Mean nearest-neighbor spacing in meters, ``nan`` below two turbines."""
distances = self.nearest_neighbor_distances()
if len(distances) < 2:
return float("nan")
return float(np.mean(distances))
[docs]
def maximum_spacing(self) -> float:
"""Nearest-neighbor spacing of the loneliest turbine in meters.
The largest distance from a turbine to its nearest neighbor, ``nan``
below two turbines, which the top end of a sampler's
``requirements={"spacing": (None, maximum)}`` bounds.
"""
distances = self.nearest_neighbor_distances()
if len(distances) < 2:
return float("nan")
return float(np.max(distances))
[docs]
def capacity(self) -> float:
"""Installed capacity in watts, the rated powers of the turbines summed.
``nan`` for a farm without turbines.
"""
from ..turbines import Turbine
if self.turbines is None:
return float("nan")
if isinstance(self.turbines, Turbine):
return float(self.turbines.rated_power) * self.n_turbines
return float(sum(turbine.rated_power for turbine in self.turbines))
[docs]
def coverage(self) -> float:
"""Share of the boundary's area inside the convex hull of the turbines.
Near 1 when the turbines span their whole site and small when they
bunch in one part of it. Turbines that enclose no area, fewer than
three or in one line, cover none. ``nan`` for a farm without a boundary.
"""
if self.boundary is None:
return float("nan")
import shapely
site = self.boundary.geometry
hull = shapely.MultiPoint(self.positions).convex_hull
return float(hull.intersection(site).area / site.area)
[docs]
def save(self, path: str | Path, **kwargs: Any) -> Path:
"""Write this farm to *path* as a windIO document.
Wraps :func:`oowl.save` and takes the same keyword arguments (``name``,
``crs``, ``energy_resource``, ``document``). ``document`` is
``"wind_energy_system"`` (default), ``"wind_farm"`` (the default for a
farm without a boundary) or ``"site"``. Returns the written path.
"""
from .._windio import save
return save(self, path, **kwargs)
[docs]
def plot(
self,
ax: "matplotlib.axes.Axes | None" = None,
**kwargs: Any,
) -> "matplotlib.axes.Axes":
"""Plot the farm and return its Matplotlib axes.
Extra keyword arguments go to Matplotlib's ``scatter`` for the turbines,
such as ``s`` and ``color``. Turbines default to the first color of
the style's property cycle. ``facecolor``, ``edgecolor``,
``alpha`` and ``linewidth`` style the boundary. ``regions`` shades
exclusion zones, and ``regions=False`` punches them out.
"""
from ..plotting import _plot_farm
return _plot_farm(self, ax=ax, **kwargs)
[docs]
class LayoutAlgorithm(_ArgumentRepr, ABC):
"""Base class for one turbine-placement method applied to one farm.
A boundary fixes the farm's area, so a layout inside one takes one more
quantity to size the farm, ``n_turbines`` or a ``density`` mapping that
names one of these.
- ``n_turbines``, the exact count of turbines.
- ``{"area_per_turbine": square meters}``, the boundary's area per
turbine, which rounds to the nearest count of at least one.
- ``{"minimum_spacing": meters}``, the most turbines that fit with no pair
closer than it. :class:`~oowl.layouts.Uniform` and
:class:`~oowl.layouts.Sobol` hold no minimum, so a random farm is sized
by its single closest pair and varies widely between seeds.
- ``{"mean_spacing": meters}``, the most turbines whose mean
nearest-neighbor distance, :meth:`~oowl.Farm.mean_spacing`, is at least
it over the whole farm, the turbines shared over its parts by area.
- ``{"maximum_spacing": meters}``, the fewest turbines that leave none
farther than it from its nearest neighbor,
:meth:`~oowl.Farm.maximum_spacing`, over the whole farm.
A count and a density over-constrain the boundary's area, so a farm of
fixed count is held to a spacing by a sampler's ``requirements`` instead.
A spacing that fits fewer than two turbines gives a farm of the one that
fits, recorded as ``metadata["fits_one_turbine"]``. A layout with neither
is a template for :class:`~oowl.layouts.Mixture` or a sampler, which then
supplies one. ``seed`` replays the placement and every unseeded modifier
added with :meth:`modify`.
"""
def __init__(
self,
n_turbines: int | None = None,
*,
density: dict[str, float] | None = None,
fitter: Fitter | None = None,
seed: int | np.random.Generator | None = None,
):
if n_turbines is not None:
n_turbines = _validate_integer("n_turbines", n_turbines)
self.n_turbines = n_turbines
given = _layout_density(density)
self.area_per_turbine = given.get("area_per_turbine")
self.minimum_spacing = given.get("minimum_spacing")
self.mean_spacing = given.get("mean_spacing")
self.maximum_spacing = given.get("maximum_spacing")
self.fitter = fitter
self.seed = pin_seed(seed)
self.modifiers: tuple[LayoutModifier, ...] = ()
@property
def density(self) -> "dict[str, float] | None":
"""The density sizing this layout as ``{name: value}``, or ``None``."""
given = {
name: getattr(self, name)
for name in _DENSITIES
if getattr(self, name) is not None
}
return given or None
def _sizing(self) -> str | None:
"""The one quantity sizing this layout, or ``None`` for a template.
Raises ``ValueError`` when several are set.
"""
given = [name for name in _SIZING if getattr(self, name) is not None]
if len(given) > 1:
floor = (
", or keep the spacing as a floor the farm is checked against "
"with Sampler(requirements={'spacing': (minimum, None)})"
if "minimum_spacing" in given
else ""
)
raise ValueError(
f"{' and '.join(given)} each size a layout, so give one of them" + floor
)
return given[0] if given else None
@property
def _required_n(self) -> int:
"""``n_turbines`` once a count is known to be set."""
if self.n_turbines is None:
raise ValueError("n_turbines is required for generation")
return self.n_turbines
[docs]
def generate(self, boundary: "Boundary | None" = None) -> Farm:
"""Generate this layout, optionally fitted to ``boundary``.
A multipart boundary is filled part by part, each part receiving its
share of turbines by area. The parts are the partition cells of a
partitioned boundary, in its ``parts`` order, else the polygons of its
shape. ``n_turbines`` counts the turbines left after the modifiers, so
the layout places the ones a removal drops on top. A spacing is
recorded in the farm's metadata under its own name. With
``minimum_spacing`` a layout that fills directly fills the whole
boundary at once, holding its modifiers to the spacing too. Others
search over counts, each part on its own, later parts giving way where
they meet earlier ones closer than the spacing. ``mean_spacing`` searches
over counts of the whole farm, shared over its parts by area. Raises
``TypeError`` when ``boundary`` is not a boundary, and ``ValueError``
when no quantity or several size the layout, when the boundary was
never given an area, or when it cannot place its turbines.
"""
from ..boundaries.base import Boundary
if boundary is not None and not isinstance(boundary, Boundary):
raise TypeError(
f"boundary must be a Boundary or None, not {type(boundary).__name__}"
)
if boundary is not None and not boundary._sized:
raise _unsized_error(boundary)
# A boundary fixes the area, so one more quantity sizes the farm.
sizing = self._sizing() if boundary is not None else None
if boundary is not None and sizing in _SPACINGS:
return self._filled_to(boundary, sizing, getattr(self, sizing))
if boundary is not None and self.area_per_turbine is not None:
area = float(boundary.area)
count = max(1, round(area / self.area_per_turbine))
farm = self._with_count(count).generate(boundary)
farm.metadata["area_per_turbine"] = self.area_per_turbine
return farm
if self.n_turbines is None:
if boundary is None:
raise ValueError("a layout without a boundary needs n_turbines")
raise ValueError(
"n_turbines or a density is required for direct generation, "
"layouts without one can be used inside Mixture or a Sampler"
)
root_seed = resolve_seed(self.seed)
core_seed = derive_seed(root_seed, "placement")
# ``n_turbines`` is the count after the modifiers, so the turbines
# they remove are drawn on top of it.
drawn = copy(self)
drawn.n_turbines = _placed_for(self.modifiers, self.n_turbines)
if boundary is None:
result = drawn._generate_standalone(as_rng(core_seed), root_seed)
else:
from ..boundaries.base import _farm_pieces
pieces = _farm_pieces(boundary)
if len(pieces) > 1:
result = drawn._generate_multipart(
boundary, pieces, root_seed, core_seed
)
else:
result = drawn._generate_single(boundary, as_rng(core_seed), root_seed)
farm = self._modified(result, root_seed)
if farm.n_turbines != self.n_turbines:
raise ValueError(
f"the modifiers of this {self._layout_name} left {farm.n_turbines} "
f"turbines, not the n_turbines={self.n_turbines} it was given"
)
farm.metadata["requested_n"] = self.n_turbines
return farm
def _filled_to(self, boundary: "Boundary", sizing: str, spacing: float) -> Farm:
"""The farm holding ``boundary`` to ``spacing``, a minimum, mean or maximum.
Where fewer than two turbines fit, the farm holds the one that does.
"""
try:
if sizing == "minimum_spacing":
farm = self._spaced(boundary, spacing)
elif sizing == "maximum_spacing":
farm = self._fewest(boundary, spacing)
else:
farm = self._fill(boundary, spacing, mean=True)
except _NoFitError:
root_seed = resolve_seed(self.seed)
single = self._with_count(1)
single.modifiers = ()
single.seed = root_seed
minimum = spacing if sizing == "minimum_spacing" else None
farm = self._modified(single.generate(boundary), root_seed, minimum)
farm.metadata["fits_one_turbine"] = True
farm.metadata[sizing] = spacing
return farm
def _fewest(self, boundary: "Boundary", spacing: float) -> Farm:
"""The fewest turbines in ``boundary`` that leave none farther than ``spacing``.
Searches exact counts of the whole farm without the modifiers, which
then run once on the farm found, held to no spacing, as for a mean.
"""
pinned = copy(self)
pinned.seed = resolve_seed(self.seed)
bare = copy(pinned)
bare.modifiers = ()
farm = _fewest_turbines_for_ceiling(
lambda count: bare._with_count(count).generate(boundary),
spacing,
float(boundary.area),
self._layout_name,
)
return self._modified(farm, pinned.seed)
def _modified(
self, result: Farm, root_seed: int, minimum_spacing: float | None = None
) -> Farm:
"""``result`` through this layout's modifiers, each from its own seed.
With ``minimum_spacing``, the farm of a minimum-spacing fill, each
modifier keeps its pairs that far apart, and ``ValueError`` names one
that does not. A mean spacing is a density, which no modifier is held to.
"""
for index, modifier in enumerate(self.modifiers):
effective_seed = (
modifier.seed
if modifier.seed is not None
else derive_seed(root_seed, "modifiers", index)
)
if minimum_spacing is not None:
modifier = modifier._held_apart(minimum_spacing)
result = modifier.apply(result, as_rng(effective_seed), effective_seed)
# The margin covers rounding in the distances a modifier measured.
if minimum_spacing is not None and _closest_pair(result) < (
minimum_spacing * (1.0 - 1e-9)
):
raise ValueError(
f"{type(modifier).__name__} brought turbines closer than"
f" minimum_spacing={minimum_spacing:g} m"
)
return result
@property
def _layout_name(self) -> str:
"""How errors name this layout, ``"Grid layout"`` for a Grid."""
return f"{type(self).__name__} layout"
#: Whether :meth:`_fill_positions` places turbines at a spacing directly.
_fills_directly = False
def _fill_positions(
self, boundary: "Boundary", spacing: float, rng: np.random.Generator
) -> tuple[np.ndarray, dict]:
"""Turbines filling the whole of ``boundary`` no closer than ``spacing``.
Returns the positions, in any order, and the fit record. A layout that
sets :attr:`_fills_directly` implements it.
"""
raise NotImplementedError
def _spaced(self, boundary: "Boundary", spacing: float) -> Farm:
"""The farm filling ``boundary`` with no pair closer than ``spacing``.
A layout that fills directly does so once over the whole boundary, so
pairs across its parts keep the spacing too, and searches exact counts
where that fill holds fewer than two turbines. Otherwise exact counts
are searched. The modifiers run once on the filled farm.
"""
if not self._fills_directly:
return self._fill(boundary, spacing)
root_seed = resolve_seed(self.seed)
rng = as_rng(derive_seed(root_seed, "placement"))
positions, fit = self._fill_positions(boundary, spacing, rng)
# A part's edge on the farm's outline does not share its exact noding,
# so a turbine filled on it can lie a rounding error outside the farm.
# Rounding must not leave a pair closer than asked either.
positions = _kept_apart(
_inside(boundary, np.asarray(positions, dtype=float).reshape(-1, 2)),
spacing,
)
if len(positions) < 2:
# A fill grows from one scale, start or center and can miss where
# the exact counts fit, their pattern placed to suit each count.
return self._fill(boundary, spacing)
return self._modified(
self._filled_farm(boundary, positions, fit, root_seed), root_seed, spacing
)
def _filled_farm(self, boundary, positions, fit, root_seed: int) -> Farm:
"""The farm of filled ``positions``, one component per part of ``boundary``."""
from ..boundaries.base import _farm_pieces
record = {
"algorithm": type(self).__name__,
"seed": int(root_seed),
"requested_n": None,
}
pieces = _farm_pieces(boundary)
if len(pieces) == 1:
metadata = {
"kind": "single",
**record,
"achieved_n": int(len(positions)),
"fit": fit,
"modifiers": [],
}
return Farm(positions, boundary, metadata=metadata)
groups = _grouped_by_piece(pieces, positions)
components = []
start = 0
for index, (piece, group) in enumerate(zip(pieces, groups)):
components.append(
{
"component_index": index,
"area": float(piece.area),
"n_turbines": int(len(group)),
"start": start,
"stop": start + len(group),
"seed": int(root_seed),
"layout": {"kind": "single", **record, "fit": fit},
}
)
start += len(group)
ordered = positions[np.concatenate(groups)]
metadata = {
"kind": "multipart",
**record,
"achieved_n": int(len(ordered)),
"fit": fit,
"components": components,
"modifiers": [],
}
return Farm(ordered, boundary, metadata=metadata)
def _fill(self, boundary: "Boundary", spacing: float, mean: bool = False) -> Farm:
"""The most turbines in ``boundary`` held to ``spacing``.
``mean`` holds their mean nearest-neighbor distance to it, else their
closest pair. This searches over exact counts without the modifiers,
every try and the first estimate drawn from one seed, and runs the
modifiers once on the farm found, as after a direct fill. For a closest
pair each part of a multipart boundary is searched on its own. A mean
is searched over the whole farm, its count shared over the parts by
area, so neighbors across the parts count toward it.
"""
from ..boundaries.base import _farm_pieces
pinned = copy(self)
pinned.seed = resolve_seed(self.seed)
bare = copy(pinned)
bare.modifiers = ()
pieces = _farm_pieces(boundary)
if len(pieces) > 1 and not mean:
farm = bare._filled_by_part(boundary, pieces, spacing)
else:
farm = _most_turbines_for_spacing(
lambda count: bare._with_count(count).generate(boundary),
spacing,
float(boundary.area),
bare._fill_estimate(boundary, spacing, mean),
self._layout_name,
_mean_nearest if mean else _closest_pair,
)
return self._modified(farm, pinned.seed, None if mean else spacing)
def _filled_by_part(self, boundary, pieces, spacing: float) -> Farm:
"""Each of ``pieces`` searched on its own, later ones giving way to earlier.
A count shared over parts by area puts turbines either side of a
shared edge, so over a partition without gaps nearly every count
failed. A part too small for two turbines holds none.
"""
from ..boundaries.base import _GeometryBoundary
root_seed = resolve_seed(self.seed)
placed = np.empty((0, 2))
guides = []
components = []
for index, piece in enumerate(pieces):
part = _GeometryBoundary(piece)
child = copy(self)
child.seed = derive_seed(root_seed, "part", index)
try:
farm = _most_turbines_for_spacing(
lambda count: child._with_count(count).generate(part),
spacing,
float(piece.area),
child._fill_estimate(part, spacing),
self._layout_name,
)
except ValueError:
farm = None
positions = (
np.empty((0, 2)) if farm is None else _inside(boundary, farm.positions)
)
positions = _apart_from(positions, placed, spacing)
if farm is not None:
guides.append(farm.guide)
components.append(
{
"component_index": index,
"area": float(piece.area),
"n_turbines": int(len(positions)),
"start": len(placed),
"stop": len(placed) + len(positions),
"seed": int(child.seed),
"given_way": 0
if farm is None
else farm.n_turbines - len(positions),
"layout": None if farm is None else farm.metadata,
}
)
placed = np.vstack([placed, positions])
if len(placed) < 2:
raise _no_fit_error(spacing, float(boundary.area), self._layout_name)
return Farm(
placed,
boundary,
metadata={
"kind": "multipart",
"algorithm": type(self).__name__,
"seed": root_seed,
"requested_n": None,
"achieved_n": int(len(placed)),
"components": components,
"modifiers": [],
},
guide=_combine_guides(guides),
)
def _with_count(self, count: int) -> Self:
"""This layout drawing exactly ``count`` turbines instead of filling."""
layout = copy(self)
for name in _SIZING:
setattr(layout, name, None)
layout.n_turbines = count
return layout
def _fill_estimate(
self, boundary: "Boundary", spacing: float, mean: bool = False
) -> int | None:
"""A first count for the search, or ``None`` to estimate one by a trial draw.
``mean`` asks for a mean nearest-neighbor ``spacing``, else a minimum.
"""
return None
@abstractmethod
def _generate_single(self, boundary, rng, seed: int) -> Farm:
"""Generate one exact-count layout inside one polygonal farm."""
def _generate_standalone(self, rng, seed: int) -> Farm:
raise ValueError(f"{type(self).__name__} requires a boundary for generation")
def _single_result(self, positions, boundary, seed, fit, guide=None) -> Farm:
positions = np.asarray(positions, dtype=float).reshape(-1, 2)
if len(positions) != self.n_turbines:
raise ValueError(
f"generated {len(positions)} turbines; expected {self.n_turbines}"
)
return Farm(
positions,
boundary,
metadata={
"kind": "single",
"algorithm": type(self).__name__,
"seed": int(seed),
"requested_n": self.n_turbines,
"achieved_n": int(len(positions)),
"fit": fit,
"modifiers": [],
},
guide=guide,
)
def _allocate_parts(self, areas: np.ndarray) -> tuple[np.ndarray, list[dict]]:
"""Turbines per part, and any extra record each part's layout needs."""
counts = _allocate_counts(self._required_n, areas)
return counts, [{} for _ in counts]
def _part_layout(self, count: int, record: dict) -> Self:
"""The unmodified copy of this layout that fills one part."""
child = copy(self)
child.n_turbines = count
child.modifiers = ()
return child
def _multipart_metadata(self) -> dict:
"""Farm-level entries a subclass adds to the multipart record."""
return {}
def _place_parts(self, pieces, child_seeds) -> list[tuple[int, dict, Farm | None]]:
"""Fill each piece with its area share, as ``(count, record, farm)``.
A piece of several polygons, such as a cell split by a setback, is
filled as a multipart farm of its own. Each piece is laid out clear of
the others by half its spacing, or by the most that leaves it half its
area, recorded as ``clearance``, so the cells of a partition do not
place turbines either side of the edge they share.
"""
areas = np.asarray([piece.area for piece in pieces], dtype=float)
counts, records = self._allocate_parts(areas)
parts = []
for index, (piece, count, record, child_seed) in enumerate(
zip(pieces, counts, records, child_seeds)
):
count = int(count)
result = None
record = {**record, "clearance": 0.0}
if count:
child = self._part_layout(count, record)
cleared, clearance = _cleared(
pieces, index, _lattice_clearance(piece, count)
)
try:
# A cell too tight for its turbines once cleared is laid
# out whole, as the record shows.
result, clearance = _laid_out_clear(
piece,
cleared,
clearance,
partial(child._laid_out_in, seed=child_seed),
)
except (TypeError, ValueError) as exc:
raise type(exc)(
f"Could not generate polygon component {index} "
f"with {count} turbines: {exc}"
) from exc
record["clearance"] = float(clearance)
parts.append((count, record, result))
return parts
def _laid_out_in(self, site: "Boundary", seed: int) -> Farm:
"""This exact-count layout in ``site``, part by part if it has several."""
from ..boundaries.base import _farm_pieces
polygons = _farm_pieces(site)
if len(polygons) > 1:
return self._generate_multipart(site, polygons, seed, seed)
return self._generate_single(site, as_rng(seed), seed)
def _generate_multipart(
self, boundary, pieces, root_seed: int, child_root: int
) -> Farm:
"""Fill ``pieces`` of ``boundary`` in order, one component each."""
polygons = list(pieces)
child_seeds = [
derive_seed(child_root, "part", index) for index in range(len(polygons))
]
parts = self._place_parts(polygons, child_seeds)
positions = []
guides = []
components = []
start = 0
for index, (polygon, child_seed, (count, record, result)) in enumerate(
zip(polygons, child_seeds, parts)
):
stop = start + count
if result is not None:
positions.append(result.positions)
guides.append(result.guide)
components.append(
{
"component_index": index,
"area": float(polygon.area),
"n_turbines": count,
**record,
"start": start,
"stop": stop,
"seed": int(child_seed),
"layout": None if result is None else result.metadata,
}
)
start = stop
combined = np.concatenate(positions) if positions else np.empty((0, 2))
combined = _inside(boundary, combined)
return Farm(
combined,
boundary,
metadata={
"kind": "multipart",
"algorithm": type(self).__name__,
"seed": int(root_seed),
"requested_n": self.n_turbines,
"achieved_n": int(len(combined)),
**self._multipart_metadata(),
"components": components,
"modifiers": [],
},
guide=_combine_guides(guides),
)
[docs]
def modify(self, *modifiers: "LayoutModifier") -> Self:
"""Return a copy with post-generation modifiers appended in order."""
from ..modifiers import LayoutModifier
for modifier in modifiers:
if not isinstance(modifier, LayoutModifier):
raise TypeError("layout modifiers must be LayoutModifier objects")
layout = copy(self)
layout.modifiers = self.modifiers + tuple(modifiers)
return layout
def __repr__(self) -> str:
text = super().__repr__()
if self.modifiers:
text += f".modify({', '.join(map(repr, self.modifiers))})"
return text
