"""Spatial compositions of layout algorithms."""
from __future__ import annotations
from copy import copy
from typing import TYPE_CHECKING
import numpy as np
from .._random import (
as_rng,
derive_seed,
farthest_point_sample,
resolve_seed,
spawn_seeds,
)
from .._validation import _validate_real_array
from .base import (
_SIZING,
Farm,
LayoutAlgorithm,
_allocate_counts,
_combine_guides,
_inside,
_kept_apart_mask,
_laid_out_clear,
_lattice_clearance,
_no_fit_error,
_NoFitError,
_within_share,
)
from .uniform import _uniform_points
if TYPE_CHECKING: # pragma: no cover
from collections.abc import Sequence
from ..boundaries.base import Boundary
__all__ = ["Mixture"]
#: Uniform points drawn per cell for the farthest-point sampling of its site.
#: More let the spread sites reach further into the corners of the farm.
_SITE_SAMPLES = 64
#: Share of its own lattice spacing that a cell keeps between its turbines
#: and those of the cells laid out before it.
_SEAM_SEPARATION = 0.8
#: Share of its own lattice spacing that a cell's domain reaches past its
#: seams into the cells laid out before it, to fill the strip their
#: templates leave short of the seam.
_SEAM_REACH = 1.0
#: Share of its own lattice spacing that a cell leaves along its seams with
#: the cells laid out after it, which they keep their separation from less
#: far. Over 240 farms it lowered the largest empty circle from 1.67 to 1.48
#: spacings and moved the other measures by under 2 %.
_SEAM_YIELD = 0.2
#: Segments per quarter circle of the disks cut around placed turbines.
_DISK_SEGMENTS = 8
def _validate_layouts(layouts) -> tuple[LayoutAlgorithm, ...]:
"""At least two layout templates sized by none of the sizing quantities."""
layouts = tuple(layouts)
if len(layouts) < 2:
raise ValueError("Mixture requires at least two layout templates")
for index, layout in enumerate(layouts):
if not isinstance(layout, LayoutAlgorithm):
raise TypeError(
f"layout {index} must be a LayoutAlgorithm, not {type(layout).__name__}"
)
for name in _SIZING:
if getattr(layout, name) is not None:
omitted = name if name == "n_turbines" else "its density"
raise ValueError(
f"layout {index} must omit {omitted} when used as a template"
)
return layouts
def _validate_weights(values, count):
"""Weights normalized to sum to one, equal when ``values`` is ``None``.
A zero weight marks an unused source. At least one must be positive.
"""
if values is None:
weights = np.ones(count, dtype=float)
else:
weights = _validate_real_array("weights", values)
if weights.shape != (count,):
raise ValueError("weights must match the number of layouts")
if not np.all(np.isfinite(weights)) or np.any(weights < 0):
raise ValueError("weights must be finite and >= 0")
if not np.any(weights > 0):
raise ValueError("weights must include at least one positive value")
# Scaling by the largest first keeps the sum finite for any finite weights.
weights = weights / weights.max()
return tuple(float(value) for value in weights / weights.sum())
def _component_source_counts(component_counts, source_counts):
"""Allocate one integer matrix while preserving both exact margins."""
remaining = np.asarray(source_counts, dtype=int).copy()
matrix = np.zeros((len(component_counts), len(source_counts)), dtype=int)
nonempty = np.flatnonzero(np.asarray(component_counts) > 0)
for position, component_index in enumerate(nonempty):
count = int(component_counts[component_index])
if position == len(nonempty) - 1:
allocated = remaining.copy()
else:
allocated = _allocate_counts(count, remaining)
matrix[component_index] = allocated
remaining -= allocated
return matrix
def _source_cells(geometry, weights, rng):
"""Power cells of ``geometry`` with areas in proportion to ``weights``.
Farthest-point sampling over uniform points inside ``geometry`` spreads
one site per cell to the far reaches of the farm, and power weights fitted
to those sites give each cell its share of the area, as
:class:`~oowl.boundaries.partition.PowerDiagram` fits them. Cells grown
from sites far apart are compact, mostly cut from the farm by straight
seams. As there, the sites are drawn again when a fit fails, and the cells
are snapped to one fine grid, so their shared edges match and clearing a
cell of its neighbors does not lose one in their union. Returns the cells
in the order of ``weights`` and a JSON-compatible record of them.
"""
import shapely
from ..boundaries.partition import (
_MAX_SITE_FIT_ATTEMPTS,
_fit_power_cells,
_partition_precision,
)
target = np.asarray(weights, dtype=float)
target = target / target.sum()
area = float(geometry.area)
for attempt in range(1, _MAX_SITE_FIT_ATTEMPTS + 1):
points = _uniform_points(geometry, _SITE_SAMPLES * len(target), rng)
sites = farthest_point_sample(points, len(target), rng)
try:
cells, power_weights, fit = _fit_power_cells(geometry, sites, target)
except ValueError:
if attempt == _MAX_SITE_FIT_ATTEMPTS:
raise
continue
precision = _partition_precision(geometry)
cells = [shapely.set_precision(cell, precision) for cell in cells]
areas = [float(cell.area) for cell in cells]
return cells, {
"sites": sites.tolist(),
"power_weights_m2": power_weights.tolist(),
"areas": areas,
"target_area_fractions": target.tolist(),
"area_fractions": [value / area for value in areas],
"site_attempts": attempt,
"fit": fit,
}
def _reaching(cells, reach: float):
"""The last of ``cells`` and the parts of the others within ``reach`` of it.
The cells tile a farm, so the shape stays inside it.
"""
import shapely
cell = cells[-1]
if len(cells) == 1:
return cell
domain = cell.buffer(reach).intersection(shapely.union_all(cells))
parts = shapely.get_parts(domain)
# Edges the cells share may leave lines and points in the intersection.
return shapely.union_all(
parts[shapely.get_type_id(parts) == shapely.GeometryType.POLYGON]
)
def _around(cell, domain, placed: np.ndarray, radius: float):
"""``domain`` without the disks of ``radius`` around the ``placed`` turbines.
``domain`` holds ``cell``, and only the parts of the shape that meet the
cell are kept, so the turbines do not stray into gaps among those placed.
Returns the shape and ``radius``, or ``domain`` and ``0.0`` when no
placed turbine lies that close. Each disk is a polygon whose edges touch
its circle, so no point of the shape lies closer than ``radius`` to one.
"""
import shapely
if not len(placed):
return domain, 0.0
points = shapely.points(placed)
near = shapely.distance(domain, points) < radius
if not near.any():
return domain, 0.0
disks = shapely.buffer(
points[near],
radius / np.cos(np.pi / (4 * _DISK_SEGMENTS)),
quad_segs=_DISK_SEGMENTS,
)
parts = shapely.get_parts(domain.difference(shapely.union_all(disks)))
return shapely.union_all(parts[shapely.intersects(parts, cell)]), radius
def _filled(layout: LayoutAlgorithm, site: Boundary) -> Farm:
"""``layout`` filled into ``site`` at its spacing, one turbine if two do not fit."""
try:
return layout.generate(site)
except _NoFitError:
return layout._with_count(1).generate(site)
def _inside_by_cell(boundary, positions) -> list[np.ndarray]:
"""Each array of ``positions``, any a rounding error outside ``boundary`` moved in.
A cell's edge on the outline of the farm does not share its exact noding.
"""
moved = _inside(boundary, np.concatenate(positions))
return np.split(moved, np.cumsum([len(values) for values in positions])[:-1])
[docs]
class Mixture(LayoutAlgorithm):
"""Combine layout methods in weighted regions of one farm.
Every template in ``layouts`` must omit ``n_turbines`` and ``density``. The farm
is split into one compact cell per source, a power diagram whose cell
areas follow ``weights`` exactly, equal by default, and each template is
laid out in its own cell. The weights also fix each source's exact
turbine count, and a source allotted no turbines takes no cell.
The cells are laid out in turn. A later cell keeps 0.8 of its own spacing
from the turbines already placed and reaches one spacing past its seams,
so it fills the strip an earlier cell leaves short of a seam, and an
earlier cell leaves 0.2 of its spacing free along the later ones.
Turbines either side of a seam stay at least 0.8 of the smaller spacing
apart. A cell keeps at least half its area clear, and one its template
cannot fill once cleared is laid out whole, as ``laid_out_whole`` in its
cell record shows, and only its turbines may come closer.
With ``density={"minimum_spacing": meters}`` each cell is filled at it, a
later one no closer to the turbines already placed than the spacing. A
cell too small for two turbines at the spacing takes one, and one left no
room takes none.
A template given its own seed keeps it, and the others draw from seeds
spawned from the mixture's. ``metadata["origin"]`` gives each turbine's
source, ``parts`` each source's layout and seed and ``cells`` the power
sites and weights, the cell areas, the clearance each cell kept and
whether it was laid out whole.
"""
def __init__(
self,
layouts: Sequence[LayoutAlgorithm],
n_turbines: int | None = None,
*,
density: dict[str, float] | None = None,
weights: Sequence[float] | None = None,
seed: int | np.random.Generator | None = None,
):
self.layouts = _validate_layouts(layouts)
self.weights = _validate_weights(weights, len(self.layouts))
self._fixed_counts = None
super().__init__(
n_turbines,
density=density,
seed=seed,
)
def _source_counts(self) -> np.ndarray:
"""Each source's turbines, a part's share of them inside a multipart farm."""
if self._fixed_counts is not None:
return np.asarray(self._fixed_counts, dtype=int)
return _allocate_counts(self._required_n, self.weights)
def _source_seeds(self, seed: int) -> list[int]:
"""Each source's seed, its own if it was given one, else spawned from ``seed``."""
spawned = spawn_seeds(seed, len(self.layouts))
return [
int(spawned[source])
if template.seed is None
else resolve_seed(template.seed)
for source, template in enumerate(self.layouts)
]
def _mixture(self, seed: int, source_seeds, requested_n, farms, positions, cells):
"""The positions, metadata and guide of one mixture of its sources' farms.
``source_seeds`` are the seeds the sources drew with. ``farms`` and
``positions`` map each source given a cell to its farm and the
positions kept of it, which a fill records as ``given_way``.
"""
ordered = []
origin = []
parts = []
start = 0
for source, template in enumerate(self.layouts):
farm = farms.get(source)
values = positions.get(source, np.empty((0, 2)))
stop = start + len(values)
part = {
"index": source,
"algorithm": type(template).__name__,
"weight": self.weights[source],
"n_turbines": int(len(values)),
"start": start,
"stop": stop,
"seed": source_seeds[source],
"layout": None if farm is None else farm.metadata,
}
if self.minimum_spacing is not None:
part["given_way"] = 0 if farm is None else farm.n_turbines - len(values)
parts.append(part)
ordered.append(values)
origin.extend([source] * len(values))
start = stop
metadata = {
"kind": "mixture",
"algorithm": "Mixture",
"seed": int(seed),
"requested_n": requested_n,
"achieved_n": start,
"weights": list(self.weights),
"counts": [part["n_turbines"] for part in parts],
"origin": origin,
"cells": cells,
"parts": parts,
"modifiers": [],
}
guide = _combine_guides([farm.guide for farm in farms.values()])
return np.concatenate(ordered), metadata, guide
def _generate_single(self, boundary, rng, seed):
"""One layout of each source in its cell, the cells laid out in order.
A later cell reaches ``_SEAM_REACH`` of its lattice spacing into the
earlier ones and keeps ``_SEAM_SEPARATION`` of it from their
turbines, so its template fills in up to them, also in the strip an
earlier template leaves short of the seam. An earlier cell leaves
``_SEAM_YIELD`` of its spacing free along the later cells.
"""
counts = self._source_counts()
if len(counts) != len(self.layouts) or int(counts.sum()) != self._required_n:
raise ValueError("mixture source counts do not match n_turbines")
seeds = self._source_seeds(seed)
active = np.flatnonzero(counts > 0)
cells, record = _source_cells(
boundary.geometry, np.asarray(self.weights)[active], rng
)
farms = {}
clearances = []
whole = []
placed = np.empty((0, 2))
for index, (cell, source) in enumerate(zip(cells, active.tolist())):
count = int(counts[source])
layout = copy(self.layouts[source])
layout.n_turbines = count
layout.seed = seeds[source]
spacing = 2.0 * _lattice_clearance(cell, count)
domain = _reaching(cells[: index + 1], _SEAM_REACH * spacing)
if index + 1 < len(cells):
import shapely
# The cells after it take the strip along their seams.
later = shapely.union_all(cells[index + 1 :])
domain = domain.difference(later.buffer(_SEAM_YIELD * spacing))
radius = _SEAM_SEPARATION * spacing
region, wanted = _around(cell, domain, placed, radius)
region, clearance = _within_share(
cell,
region,
wanted,
lambda distance: _around(cell, domain, placed, distance)[0],
)
try:
farm, clearance = _laid_out_clear(
cell, region, clearance, layout.generate
)
except (TypeError, ValueError) as exc:
raise type(exc)(
f"Could not lay out mixture source {source} "
f"with {count} turbines in its cell: {exc}"
) from exc
farms[source] = farm
clearances.append(float(clearance))
# Turbines lay near it, but clearing it of them took too much.
whole.append(bool(wanted) and not clearance)
placed = np.concatenate([placed, farm.positions])
moved = _inside_by_cell(boundary, [farm.positions for farm in farms.values()])
cells_record = {
"sources": active.tolist(),
**record,
"clearances": clearances,
"laid_out_whole": whole,
}
combined, metadata, guide = self._mixture(
seed,
seeds,
self.n_turbines,
farms,
dict(zip(farms, moved)),
cells_record,
)
return Farm(combined, boundary, metadata=metadata, guide=guide)
def _spaced(self, boundary: Boundary, spacing: float) -> Farm:
"""Every cell filled with its template at a minimum ``spacing``, in order.
Each part of a multipart boundary is split into cells of its own. A
later cell is filled around the turbines already placed, at least the
spacing from them, and reaches ``_SEAM_REACH`` of the spacing into
the earlier cells of its part, so seams and parts keep it and stay
dense on both sides. A turbine that rounding leaves closer gives way.
The modifiers run once on the filled farm.
"""
from ..boundaries.base import _farm_pieces, _GeometryBoundary
root_seed = resolve_seed(self.seed)
pieces = _farm_pieces(boundary)
if len(pieces) == 1:
seeds = [root_seed]
rngs = [as_rng(derive_seed(root_seed, "placement"))]
else:
seeds = [
derive_seed(root_seed, "part", index) for index in range(len(pieces))
]
rngs = [as_rng(seed) for seed in seeds]
active = np.flatnonzero(np.asarray(self.weights) > 0).tolist()
weights = np.asarray(self.weights)[active]
partitions = [
_source_cells(piece, weights, rng) for piece, rng in zip(pieces, rngs)
]
cells = [cell for piece_cells, _ in partitions for cell in piece_cells]
part_seeds = [self._source_seeds(seed) for seed in seeds]
farms = []
clearances = []
placed = np.empty((0, 2))
for index, cell in enumerate(cells):
part, slot = divmod(index, len(active))
source = active[slot]
layout = copy(self.layouts[source])
layout.minimum_spacing = spacing
layout.seed = part_seeds[part][source]
domain = _reaching(partitions[part][0][: slot + 1], _SEAM_REACH * spacing)
region, clearance = _around(cell, domain, placed, spacing)
# Turbines placed around it can leave a cell no room at all.
farm = (
None if region.is_empty else _filled(layout, _GeometryBoundary(region))
)
farms.append(farm)
clearances.append(float(clearance))
if farm is not None:
placed = np.concatenate([placed, farm.positions])
filled = _inside_by_cell(
boundary,
[np.empty((0, 2)) if farm is None else farm.positions for farm in farms],
)
kept = np.split(
_kept_apart_mask(np.concatenate(filled), spacing),
np.cumsum([len(values) for values in filled])[:-1],
)
if sum(int(mask.sum()) for mask in kept) < 2:
raise _no_fit_error(spacing, float(boundary.area), self._layout_name)
mixtures = []
for part, (seed, (_, record)) in enumerate(zip(seeds, partitions)):
cell_slice = slice(part * len(active), (part + 1) * len(active))
cells_record = {
"sources": active,
**record,
"clearances": clearances[cell_slice],
# A fill lays every cell out around the turbines before it.
"laid_out_whole": [False] * len(active),
}
part_farms = {
source: farm
for source, farm in zip(active, farms[cell_slice])
if farm is not None
}
positions = {
source: values[mask]
for source, values, mask in zip(
active, filled[cell_slice], kept[cell_slice]
)
}
mixtures.append(
self._mixture(
seed, part_seeds[part], None, part_farms, positions, cells_record
)
)
if len(pieces) == 1:
positions, metadata, guide = mixtures[0]
farm = Farm(positions, boundary, metadata=metadata, guide=guide)
else:
farm = self._filled_parts(boundary, pieces, seeds, root_seed, mixtures)
return self._modified(farm, root_seed, spacing)
def _filled_parts(self, boundary, pieces, seeds, root_seed, mixtures) -> Farm:
"""The multipart farm of each part's filled mixture, one component each."""
components = []
start = 0
for index, (piece, seed, (positions, metadata, _)) in enumerate(
zip(pieces, seeds, mixtures)
):
components.append(
{
"component_index": index,
"area": float(piece.area),
"n_turbines": int(len(positions)),
"start": start,
"stop": start + len(positions),
"seed": int(seed),
"layout": metadata,
}
)
start += len(positions)
source_counts = np.sum(
[metadata["counts"] for _, metadata, _ in mixtures], axis=0
)
return Farm(
np.concatenate([positions for positions, _, _ in mixtures]),
boundary,
metadata={
"kind": "multipart",
"algorithm": "Mixture",
"seed": int(root_seed),
"requested_n": None,
"achieved_n": start,
"source_counts": source_counts.astype(int).tolist(),
"components": components,
"modifiers": [],
},
guide=_combine_guides([guide for _, _, guide in mixtures]),
)
def _allocate_parts(self, areas):
counts = _allocate_counts(self._required_n, areas)
matrix = _component_source_counts(counts, self._source_counts())
return counts, [{"source_counts": row.astype(int).tolist()} for row in matrix]
def _part_layout(self, count, record):
# Each part draws its share of every source from the global apportionment.
child = super()._part_layout(count, record)
child._fixed_counts = tuple(record["source_counts"])
return child
def _multipart_metadata(self):
return {"source_counts": [int(value) for value in self._source_counts()]}
