Source code for oowl.boundaries.composites

"""Compositions of logical farm boundaries and modifier chains."""

from __future__ import annotations

from collections.abc import Iterable
from copy import copy
from functools import cached_property

import numpy as np
from shapely import affinity
from shapely.ops import unary_union

from .._random import spawn_seeds
from .._repr import _describe
from .base import (
    Boundary,
    _centroid,
    _GeometryBoundary,
    _validate_target_area,
)
from .modifiers import BoundaryModifier, ExclusionZones, Setback
from .partition import _CUT_GRIDS, _PartitionModifier, _scaled_details

__all__ = ["Composite", "Union"]

#: Relative error allowed between the area a resized composite reaches and
#: the one requested. A partition cuts its gaps by snap rounding on a grid of
#: up to ``_CUT_GRIDS[-1]`` of its span. As the base grows, a vertex jumps by
#: a grid step and the area by that step times the edges at the vertex, a
#: relative jump a few times the relative grid. No base side reaches a target
#: inside such a jump, so the nearest side is accepted within ten coarsest
#: grid steps, while the far larger jump of a cut that moves elsewhere still
#: fails.
_AREA_TOLERANCE = 10.0 * _CUT_GRIDS[-1]

#: Relative area error at which a resize search stops. Far below the
#: tolerance, so the result moves smoothly with the requested area and a
#: search nested around it, such as one for a spacing, sees no steps.
_AREA_SETTLED = 1e-12

#: Steps allowed to bracket the base side that reaches a target area.
_MAX_BRACKET_STEPS = 16

#: Share of a buffer's distance within which GEOS simplifies the buffered
#: input, its default ``simplifyFactor``. A vertex that drops in or out as a
#: site grows by micrometers moved a 194 m setback's edge by 1.7 m, and the
#: area of a 40 km2 site jumped 4000 m2.
_BUFFER_SIMPLIFICATION = 0.01


def _buffered_distance(boundary) -> float:
    """Total distance in meters the modifiers of ``boundary`` buffer by.

    A setback buffers by its distance and a partition by half its gap, the
    largest member's sum for a union.
    """
    if isinstance(boundary, Union):
        return max(
            (_buffered_distance(member) for member in boundary.boundaries), default=0.0
        )
    if not isinstance(boundary, Composite):
        return 0.0
    total = _buffered_distance(boundary.base)
    for modifier in boundary.modifiers:
        if isinstance(modifier, Setback):
            total += modifier.distance
        elif isinstance(modifier, _PartitionModifier):
            total += 0.5 * modifier.gap
    return total


class _Reached(Exception):
    """Ends Brent's method at the first base side that settles the area."""


[docs] class Composite(Boundary): """A base boundary with an ordered chain of boundary modifiers. Modifiers apply left to right. A chain holds at most one partition, and modifiers after it apply to each partition cell separately. ``area`` sets the area of the finished boundary in square meters, ``None`` to leave it as the chain makes it. Composing a composite flattens the chains and keeps its area unless ``area`` is given. Raises ``TypeError`` for a non-boundary base or a non-modifier item. Resizing a composite whose modifiers are given in meters, such as a setback distance, a partition gap or exclusion zones, resizes the base and re-applies the chain to it, so those stay in meters while the result has the requested area to rounding. Where GEOS's buffer simplification makes the area jump, the nearest size is kept, within a hundredth of the buffered distance times the edge length. A chain without one scales as a whole with its cells. """ # The base carries its own size, see ``_sized``. _unit_shape = False def __init__( self, base: Boundary, modifiers: Iterable[BoundaryModifier], *, area: float | None = None, ): if not isinstance(base, Boundary): raise TypeError("base must be a Boundary") modifiers = tuple(modifiers) for modifier in modifiers: if not isinstance(modifier, BoundaryModifier): raise TypeError("boundary modifiers must be BoundaryModifier objects") if isinstance(base, Composite): self.base = base.base self.modifiers = base.modifiers + modifiers if area is None: area = base.target_area else: self.base = base self.modifiers = modifiers self.target_area = None if area is None else _validate_target_area(area) self._positioned = getattr(self.base, "_positioned", True) # Reproducible only when all random components are seeded. A base that # draws nothing has no seed to give, so it leaves the chain pinned. self._pinned_seed = ( self._is_random and (self.base._pinned_seed or not self.base._is_random) and all( getattr(item, "_pinned_seed", False) for item in self.modifiers if hasattr(item, "seed") or getattr(item, "_is_random", False) ) ) if sum(isinstance(item, _PartitionModifier) for item in self.modifiers) > 1: raise ValueError("a boundary may contain only one partition modifier") # The base the chain was last built on, resized when a target area in # meters made the build search for its size. self._sized_base = self.base # The base area a search for a spacing set, see ``_at_base_area``. self._base_area: float | None = None self._unscaled_parts = None self._unscaled_details = None self._partition_parts = None self._partition_details = None @property def _is_random(self) -> bool: return self.base._is_random or any( hasattr(item, "seed") or getattr(item, "_is_random", False) for item in self.modifiers ) @property def _in_meters(self) -> bool: """Whether a modifier depends on the meter scale of its input.""" return any(modifier._in_meters for modifier in self.modifiers) @property def _sized(self) -> bool: """Whether the chain was given an area, or its base was.""" return ( self.target_area is not None or self._base_area is not None or self.base._sized ) @property def _gapless_cells(self) -> bool: """Whether the chain ends in cells cut with no gap, which tile the shape.""" last = self.modifiers[-1] if self.modifiers else None return isinstance(last, _PartitionModifier) and last.gap == 0 def _modified(self, geometry, modifiers=None): """Return ``geometry`` through the chain with its cells and partition record. ``modifiers`` stands in for the chain, as when partitions carry sites. """ parts = None details = None for modifier in self.modifiers if modifiers is None else modifiers: if isinstance(modifier, _PartitionModifier): parts = list(modifier.partition(geometry)) details = dict(modifier.details) # Cells cut with no gap tile the shape, and their union would # only add slivers where shared edges are noded differently. if modifier.gap > 0: geometry = unary_union(parts) elif parts is None: geometry = modifier.apply(geometry) else: parts = [modifier.apply(part) for part in parts] if any(part.is_empty or part.area <= 0 for part in parts): raise ValueError("a boundary modifier removes a partition cell") geometry = unary_union(parts) return geometry, None if parts is None else tuple(parts), details @cached_property def _natural(self): """The chain applied to the base as given, or the ``ValueError`` it raises. Resized copies keep it, so a search for each new area starts from it. The sites each partition samples are kept in ``_natural_sites``. """ try: result = self._modified(self.base.geometry) except ValueError as exc: self._natural_sites = None return exc self._natural_sites = [ np.asarray(modifier.details["sites"], dtype=float) if isinstance(modifier, _PartitionModifier) and modifier._samples_sites else None for modifier in self.modifiers ] return result def _chain_on(self, base_geometry): """The chain applied to ``base_geometry``, the base at another size. A partition that sampled its sites on the base as given cuts its cells around those sites moved with the base, centroid to centroid, so the cells grow with it rather than being drawn again. Where the carried sites leave a cell to a sliver the gap removes, or a weight fit fails, the sites are drawn at this size, as at the size given. Carried alone, the search could not shrink a farm past the size where a cell first vanished. """ _ = self._natural carried = self._natural_sites if carried is None or all(sites is None for sites in carried): return self._modified(base_geometry) natural = self.base.geometry factor = float(np.sqrt(base_geometry.area / natural.area)) start = np.asarray(_centroid(natural)) end = np.asarray(_centroid(base_geometry)) modifiers = [ modifier._carrying(end + factor * (sites - start)) if isinstance(modifier, _PartitionModifier) and sites is not None else modifier for modifier, sites in zip(self.modifiers, carried) ] try: return self._modified(base_geometry, modifiers) except ValueError: return self._modified(base_geometry) def _modified_at_area(self, target_area: float): """Return the chain applied to the base resized to reach ``target_area``. Returns the base the chain was applied to and its result. A base the chain leaves nothing of, or refuses, counts as a result of zero area, so the search grows it. Raises the chain's own ``ValueError`` when no base side leaves a site, and ``ValueError`` when none reaches the target. """ base = self.base natural = float(np.sqrt(base.geometry.area)) side, result = _searched_side( lambda side: self._chain_on(base._resized(side**2).geometry), natural, self._natural, target_area, lambda results, last: self._unreachable_area(target_area, results, last), buffered=_buffered_distance(self), ) return base if side == natural else base._resized(side**2), result def _unreachable_area(self, target_area: float, results: dict, last: float): """The error for a search that ends at base side ``last`` short of the area. A chain that left a site at no side tried raises its own error. """ natural = self._natural if isinstance(natural, ValueError) and all( isinstance(result, ValueError) or result[0].area <= 0.0 for result in results.values() ): return natural.with_traceback(None) return _area_not_reached(self, "its modifiers", target_area, results[last]) def _build_geometry(self): target_area = getattr(self, "target_area", None) base_area = self._base_area if base_area is not None: self._sized_base = self.base._resized(base_area) geometry, parts, details = self._chain_on(self._sized_base.geometry) # The area reached stands as the target, so the boundary prints # as the call that rebuilds it. self.target_area = float(geometry.area) elif target_area is None or not self._in_meters: natural = self._natural if isinstance(natural, ValueError): raise natural.with_traceback(None) geometry, parts, details = natural self._sized_base = self.base else: self._sized_base, (geometry, parts, details) = self._modified_at_area( target_area ) # Cells and their record are kept unscaled beside the shape, so a # resized copy reusing the shape scales them afresh. self._unscaled_parts = parts self._unscaled_details = details return geometry def _scale_to_target_area(self, geometry): target_area = getattr(self, "target_area", None) parts = self._unscaled_parts details = self._unscaled_details if target_area is None or self._in_meters: # Not resized, or already built at the target area. self._partition_parts = None if parts is None else list(parts) self._partition_details = details return geometry if parts is None: self._partition_parts = None self._partition_details = details return super()._scale_to_target_area(geometry) # The cells take the similarity transform the union would, so the # parts a farm is built from keep the requested area. Cells cut with # no gap tile the unscaled shape, which scales as it is. The union of # cells with gaps is taken afresh. factor = float(np.sqrt(target_area / geometry.area)) origin = _centroid(geometry) self._partition_parts = [ affinity.scale(part, xfact=factor, yfact=factor, origin=origin) for part in parts ] self._partition_details = ( None if details is None else _scaled_details(details, factor, origin) ) if self._gapless_cells: return affinity.scale(geometry, xfact=factor, yfact=factor, origin=origin) return unary_union(self._partition_parts) def _resized(self, target_area: float) -> Boundary: if not self._in_meters: return super()._resized(target_area) # The chain is re-applied to a resized base, whose shape the base # itself builds once and shares. Each search starts from the chain on # the base as given, which the copies share too. target_area = _validate_target_area(target_area) _ = self._natural boundary = self._uncached_copy() setattr(boundary, "target_area", target_area) setattr(boundary, "_base_area", None) return boundary def _at_base_area(self, base_area: float) -> Boundary: """This boundary with its base at ``base_area`` and the chain applied once. A search for a spacing needs some size of the site, not a given area, so it sizes the base directly rather than nesting a search for an area in each of its draws, which cost one farm on seven channels 58 chain builds for 9 draws. Partition sites carry over from the base as given, as in a resize. The copy records the area it reaches as its own. """ if not self._in_meters: return super()._at_base_area(base_area) _ = self._natural boundary = self._uncached_copy() setattr(boundary, "target_area", None) setattr(boundary, "_base_area", _validate_target_area(base_area)) return boundary @property def parts(self) -> tuple[Boundary, ...]: """Partition cells as boundaries, or this composite alone without one.""" _ = self.geometry if self._partition_parts is None: return (self,) return tuple(_GeometryBoundary(part) for part in self._partition_parts) @property def partition_metadata(self) -> dict | None: """JSON-compatible record of the partition, or ``None`` without one.""" _ = self.geometry return ( None if self._partition_details is None else dict(self._partition_details) )
[docs] def modify(self, *modifiers: BoundaryModifier) -> "Composite": """Return a new composite with ``modifiers`` appended to this chain.""" return Composite(self, modifiers)
def __repr__(self) -> str: if self.target_area is None: return f"{self.base!r}.modify({', '.join(map(repr, self.modifiers))})" modifiers = ", ".join(map(repr, self.modifiers)) if len(self.modifiers) == 1: modifiers += "," return ( f"Composite(base={self.base!r}, modifiers=({modifiers}), " f"area={_describe(self.target_area)})" ) def _reseeded(self, seed: int): """Return a copy re-deriving only its unpinned parts. A part the caller seeded keeps that seed, so a derived seed never overrides an explicit one. Positioning is left to ``_resample``, as for a plain boundary. """ seeds = spawn_seeds(seed, 1 + len(self.modifiers)) modifiers = [] for modifier, child_seed in zip(self.modifiers, seeds[1:]): if isinstance(modifier, ExclusionZones): item = modifier._reseeded(child_seed) else: item = copy(modifier) if hasattr(item, "seed") and not getattr(item, "_pinned_seed", False): setattr(item, "seed", int(child_seed)) if hasattr(item, "details"): setattr(item, "details", {}) modifiers.append(item) return _with_target_area( self, Composite(_reseeded_child(self.base, seeds[0]), modifiers) )
def _searched_side( evaluate, natural_side: float, natural, target_area: float, unreachable, buffered: float = 0.0, ): """Return the base side whose result reaches ``target_area`` and that result. ``evaluate`` builds the result for a base side, a tuple led by its geometry, and ``natural`` is the result, or the ``ValueError``, at ``natural_side``. The side of the result, the square root of its area, grows about one for one with the side of the base, as a setback or a gap removes a strip of fixed width. Steps of slope one, then steps fitting the area as the base area less a line in the side, bracket the root quickly, and Brent's method finishes where they stall. A base side whose result raises ``ValueError`` counts as a result of zero area, so the search grows it. ``buffered`` is the distance in meters the chain buffers by, whose simplification makes the area jump, so where no side settles the nearest within that jump is kept. Raises ``unreachable(results, last)`` when no side reaches the target. """ target_side = float(np.sqrt(target_area)) results: dict[float, tuple | ValueError] = {} def error(result) -> float: if isinstance(result, ValueError): return float("inf") return abs(result[0].area / target_area - 1.0) def settled(result) -> bool: return error(result) <= _AREA_SETTLED def closest() -> float | None: """The side whose result is nearest the target, if within the tolerance.""" best = min(results, key=lambda key: error(results[key])) result = results[best] if isinstance(result, ValueError): return None # GEOS simplifies a buffer's input within a share of its distance, so # the area can jump by about that times the edge length. simplified = _BUFFER_SIMPLIFICATION * buffered * result[0].length tolerance = _AREA_TOLERANCE + simplified / target_area return best if error(result) <= tolerance else None def side(base_side: float) -> float: if base_side not in results: try: results[base_side] = evaluate(base_side) except ValueError as exc: results[base_side] = exc result = results[base_side] if isinstance(result, ValueError): return 0.0 return float(np.sqrt(result[0].area)) def step(x1: float, f1: float, x0: float, f0: float) -> float: """The next base side to try after ``x1``, which missed by ``f1``.""" if f1 <= -target_side: # Nothing left, so grow to at least the target, as the chain # only removes area. return target_side if x1 < target_side else 2.0 * x1 if x0 == x1 or f0 <= -target_side: # One result alone, which a strip of fixed width leaves about # one for one behind its base. return max(x1 - f1, 0.5 * x1) # A strip or a zone of fixed size takes an area linear in the base # side, so the area is the base area ``x**2`` less a line through # both results. a0, a1 = (f0 + target_side) ** 2, (f1 + target_side) ** 2 slope = ((a1 - x1**2) - (a0 - x0**2)) / (x1 - x0) constant = a1 - x1**2 - slope * x1 - target_area x2 = _larger_root(slope, constant) if x2 is None: secant = (f1 - f0) / (x1 - x0) x2 = x1 - f1 / secant if secant > 0.0 else x1 - f1 return max(x2, 0.5 * x1) results[natural_side] = natural x1 = natural_side f1 = side(x1) - target_side x0, f0 = x1, f1 # The latest sides left short of the target and beyond it. below: float | None = None above: float | None = None for attempt in range(_MAX_BRACKET_STEPS + 1): if settled(results[x1]): return x1, results[x1] if f1 < 0.0: below = x1 else: above = x1 bracket = None if below is None or above is None else (below, above) if attempt == _MAX_BRACKET_STEPS: if bracket is None: if (best := closest()) is not None: return best, results[best] raise unreachable(results, x1) break x2 = step(x1, f1, x0, f0) # Model steps go on inside a bracket while they at least halve the # miss, and Brent's method takes over otherwise. if bracket is not None and ( not min(bracket) < x2 < max(bracket) or abs(f1) > 0.5 * abs(f0) ): break x0, f0 = x1, f1 x1 = x2 f1 = side(x1) - target_side assert below is not None and above is not None # SciPy loads only when a resize needs it, keeping ``import oowl`` light. from scipy.optimize import brentq def residual(x: float) -> float: value = side(x) - target_side if settled(results[x]): raise _Reached(x) return value low, high = min(below, above), max(below, above) try: root = brentq(residual, low, high, xtol=1e-11 * low, rtol=1e-11) except _Reached as reached: return reached.args[0], results[reached.args[0]] side(root) if (best := closest()) is None: raise unreachable(results, root) return best, results[best] def _area_not_reached( boundary: Boundary, removers: str, target_area: float, last ) -> ValueError: """The error for a resize whose last result ``last`` fell short of the area.""" error = ValueError( f"{boundary!r} cannot be resized to an area of {target_area:.6g} m2, " f"as the area left by {removers} does not reach it" ) if isinstance(last, ValueError): error.__cause__ = last return error def _grown(boundary: Boundary, scale: float, offset: tuple[float, float]): """Return the shape of ``boundary`` mapped by ``x -> scale * x + offset``. A composite in meters maps its base and re-applies its chain, and a union in meters maps its members so, as resizing either does. Distances given in meters keep their size and exclusion zones stay fixed in the plane. Raises the ``ValueError`` of a chain that leaves nothing of the mapped base. """ if isinstance(boundary, Composite) and boundary._in_meters: _ = boundary.geometry geometry, _, _ = boundary._chain_on(_grown(boundary._sized_base, scale, offset)) return geometry if isinstance(boundary, Union) and boundary._in_meters: _ = boundary.geometry # The members of the union as built are mapped by its own factor about # its origin first. factor = boundary._factor x, y = boundary._origin return boundary._members_grown( scale * factor, ( scale * (1.0 - factor) * x + offset[0], scale * (1.0 - factor) * y + offset[1], ), ) return affinity.affine_transform( boundary.geometry, (scale, 0.0, 0.0, scale, offset[0], offset[1]) ) def _bases(boundary: Boundary): """Return the shape of ``boundary`` as built, without its chains in meters. Its area is what the bases of a resized union cover, as the base area of a resized composite is. """ if isinstance(boundary, Composite) and boundary._in_meters: _ = boundary.geometry return _bases(boundary._sized_base) if isinstance(boundary, Union) and boundary._in_meters: _ = boundary.geometry factor = boundary._factor x, y = boundary._origin return affinity.affine_transform( boundary._natural_bases, (factor, 0.0, 0.0, factor, (1.0 - factor) * x, (1.0 - factor) * y), ) return boundary.geometry def _larger_root(slope: float, constant: float) -> float | None: """The larger root of ``x**2 + slope * x + constant``, ``None`` if not real.""" discriminant = slope**2 - 4.0 * constant if discriminant < 0.0: return None root = float(np.sqrt(discriminant)) if slope <= 0.0: return 0.5 * (root - slope) # The same root, without cancelling two close terms. return -2.0 * constant / (slope + root) def _reseeded_child(boundary: Boundary, seed: int) -> Boundary: """Re-derive a child without overriding randomness its caller pinned.""" configured_seed = getattr(boundary, "seed", None) if getattr(boundary, "_pinned_seed", False) and configured_seed is not None: seed = configured_seed return boundary._reseeded(seed) def _with_target_area(source: Boundary, boundary: Boundary) -> Boundary: """Return ``boundary`` scaled to the area ``source`` was resized to, if any.""" target_area = getattr(source, "target_area", None) if target_area is not None: setattr(boundary, "target_area", target_area) return boundary
[docs] class Union(Boundary): """One logical boundary formed by the geometric union of other boundaries. Overlapping members merge into a single polygon. Disconnected members remain components of one ``MultiPolygon`` farm. Each member is scaled to its own ``area`` before the union. ``area`` then scales the union to that many square meters, ``None`` to keep the members' sizes. Resizing a union with a member whose modifiers are given in meters, such as a setback distance, a partition gap or exclusion zones, scales the bases of all members by one factor about the centroid of the union and re-applies their chains, so those stay in meters while the result has the requested area, as for a :class:`Composite`. A union without one scales as a whole. """ # The members carry their own sizes, see ``_sized``. _unit_shape = False def __init__(self, boundaries: Iterable[Boundary], *, area: float | None = None): try: members = tuple(boundaries) except TypeError as exc: raise TypeError( "boundaries must be an iterable of Boundary objects" ) from exc if not members: raise ValueError("Union requires at least one boundary") for index, boundary in enumerate(members): if not isinstance(boundary, Boundary): raise TypeError( f"union member {index} must be a Boundary, " f"not {type(boundary).__name__}" ) self.boundaries = members self.target_area = None if area is None else _validate_target_area(area) self._positioned = all( getattr(boundary, "_positioned", True) for boundary in members ) # Reproducible only when all random members are seeded. A member that # draws nothing has no seed to give, so it leaves the union pinned. self._pinned_seed = any(boundary._is_random for boundary in members) and all( boundary._pinned_seed or not boundary._is_random for boundary in members ) # The factor the members were scaled by about the origin of the # union, other than one when a target area in meters set it. self._factor = 1.0 @property def _is_random(self) -> bool: return any(boundary._is_random for boundary in self.boundaries) @property def _in_meters(self) -> bool: """Whether a member keeps something in meters when resized.""" return any(boundary._in_meters for boundary in self.boundaries) @property def _sized(self) -> bool: """Whether the union was given an area or all its members were.""" return self.target_area is not None or all( boundary._sized for boundary in self.boundaries ) @cached_property def _natural(self): """The union of the members as given. Resized copies keep it, so a search for each new area starts from it. """ return unary_union([boundary.geometry for boundary in self.boundaries]) @property def _origin(self) -> tuple[float, float]: """The point the members are scaled about, the centroid of their union.""" return _centroid(self._natural) @cached_property def _natural_bases(self): """The union of the members as given, without their chains in meters.""" return unary_union([_bases(boundary) for boundary in self.boundaries]) def _members_grown(self, scale: float, offset: tuple[float, float]): """Return the union of the members mapped by ``x -> scale * x + offset``.""" return unary_union( [_grown(boundary, scale, offset) for boundary in self.boundaries] ) def _build_geometry(self): target_area = getattr(self, "target_area", None) if target_area is None or not self._in_meters: self._factor = 1.0 return self._natural natural = self._natural x, y = self._origin # The side of the bases as given, which grows with the factor, so a # strip of fixed width leaves the side of the union about one for one # behind it, as it does for a composite. natural_side = float(np.sqrt(self._natural_bases.area)) def factor(side: float) -> float: return side / natural_side def evaluate(side: float) -> tuple: scale = factor(side) return (self._members_grown(scale, ((1.0 - scale) * x, (1.0 - scale) * y)),) side, (geometry,) = _searched_side( evaluate, natural_side, (natural,), target_area, lambda results, last: _area_not_reached( self, "the modifiers of its members", target_area, results[last] ), buffered=_buffered_distance(self), ) self._factor = factor(side) return geometry def _scale_to_target_area(self, geometry): if self._in_meters: # Not resized, or already built at the target area. return geometry return super()._scale_to_target_area(geometry) def _resized(self, target_area: float) -> Boundary: if not self._in_meters: return super()._resized(target_area) # The chains of the members are re-applied to their scaled bases. Each # search starts from the members as given, which the copies share. target_area = _validate_target_area(target_area) _ = self._natural, self._natural_bases boundary = self._uncached_copy() setattr(boundary, "target_area", target_area) return boundary def _reseeded(self, seed: int): """Return a union with only its unpinned members re-derived.""" seeds = spawn_seeds(seed, len(self.boundaries)) return _with_target_area( self, Union( _reseeded_child(boundary, child_seed) for boundary, child_seed in zip(self.boundaries, seeds) ), )