"""Geometry modifiers shared by every string guide."""
from __future__ import annotations
from abc import ABC, abstractmethod
import numpy as np
import shapely
from shapely.geometry import LineString
from shapely.ops import substring
from .._random import as_rng, resolve_seed
from .._repr import _ArgumentRepr
from .._validation import _validate_integer, _validate_nonnegative, _validate_positive
from .base import _LogicalString, _segment_geometry
__all__ = ["StringModifier", "Breaks", "Meander"]
#: Meander draws retried before giving up on a self-intersecting guide.
_MAX_MEANDER_ATTEMPTS = 32
[docs]
class StringModifier(_ArgumentRepr, ABC):
"""Transform ordered logical strings without mutating the source guide."""
#: Whether the caller gave every random choice of the modifier a seed.
_pinned_seed = True
[docs]
@abstractmethod
def apply(self, strings: tuple[_LogicalString, ...]) -> tuple[_LogicalString, ...]:
"""Return new logical strings, leaving ``strings`` untouched."""
def _reseeded(self, seed: int) -> "StringModifier":
"""Return the modifier redrawn from ``seed`` unless its caller seeded it."""
if self._pinned_seed:
return self
return type(self)(**{**self._arguments(), "seed": seed})
[docs]
class Breaks(StringModifier):
"""Remove random internal sections from logical strings.
Adds ``n_breaks`` gaps, each ``size`` times the original arc length of its
string, with ``size`` in ``(0, 1)``. Without ``seed``, a
:class:`~oowl.layouts.String` layout draws new breaks for every farm.
"""
def __init__(
self,
n_breaks: int = 1,
*,
size: float = 0.05,
seed: int | np.random.Generator | None = None,
) -> None:
self.n_breaks = _validate_integer("n_breaks", n_breaks, minimum=0)
size = _validate_positive("size", size)
if size >= 1:
raise ValueError("size must be between 0 and 1")
self.size = size
self._pinned_seed = seed is not None
self.seed = resolve_seed(seed)
[docs]
def apply(self, strings: tuple[_LogicalString, ...]) -> tuple[_LogicalString, ...]:
"""Return new logical strings with ``n_breaks`` internal gaps removed.
Break locations are weighted by available length. Raises ``ValueError``
when no segment can hold another break.
"""
if self.n_breaks == 0:
return tuple(strings)
rng = as_rng(self.seed)
original_lengths = [logical.length for logical in strings]
segments = [list(logical.segments) for logical in strings]
for _ in range(self.n_breaks):
eligible = []
weights = []
for string_index, string_segments in enumerate(segments):
break_length = self.size * original_lengths[string_index]
available = sum(
segment.length
for segment in string_segments
if segment.length > break_length
)
if available > 0:
eligible.append(string_index)
weights.append(available)
if not eligible:
raise ValueError("breaks do not fit within the remaining guide")
weights = np.asarray(weights, dtype=float)
string_index = int(rng.choice(eligible, p=weights / weights.sum()))
break_length = self.size * original_lengths[string_index]
candidates = [
index
for index, segment in enumerate(segments[string_index])
if segment.length > break_length
]
lengths = np.asarray(
[segments[string_index][index].length for index in candidates]
)
segment_index = int(rng.choice(candidates, p=lengths / lengths.sum()))
segment = segments[string_index][segment_index]
start = float(rng.uniform(0, segment.length - break_length))
stop = start + break_length
replacements = []
if start > 0:
replacements.append(substring(segment, 0, start))
if stop < segment.length:
replacements.append(substring(segment, stop, segment.length))
segments[string_index][segment_index : segment_index + 1] = replacements
return tuple(
_LogicalString(original.guide_type, tuple(updated))
for original, updated in zip(strings, segments)
)
[docs]
class Meander(StringModifier):
"""Add smooth correlated lateral variation while fixing endpoints.
``roughness`` is the peak lateral offset as a fraction of segment arc
length, and ``0`` leaves the guide unchanged. Without ``seed``, a
:class:`~oowl.layouts.String` layout draws a new meander for every farm.
"""
def __init__(
self,
roughness: float = 0.03,
*,
seed: int | np.random.Generator | None = None,
) -> None:
self.roughness = _validate_nonnegative("roughness", roughness)
self._pinned_seed = seed is not None
self.seed = resolve_seed(seed)
[docs]
def apply(self, strings: tuple[_LogicalString, ...]) -> tuple[_LogicalString, ...]:
"""Return new logical strings displaced laterally, endpoints unchanged.
Each segment is offset along its normal by a smoothed random walk.
Candidates are retried a bounded number of times until the guide is
simple, else ``ValueError`` is raised.
"""
if self.roughness == 0:
if not _strings_are_simple(strings):
raise ValueError("Meander cannot produce a simple guide at roughness=0")
return tuple(strings)
rng = as_rng(self.seed)
for _ in range(_MAX_MEANDER_ATTEMPTS):
candidate = tuple(
_LogicalString(
logical.guide_type,
tuple(
_meander_segment(segment, self.roughness, rng)
for segment in logical.segments
),
)
for logical in strings
)
if _strings_are_simple(candidate):
return candidate
raise ValueError(
"Meander could not produce a simple guide "
f"after {_MAX_MEANDER_ATTEMPTS} attempts"
)
def _strings_are_simple(strings: tuple[_LogicalString, ...]) -> bool:
return bool(_segment_geometry(strings).is_simple)
def _meander_segment(segment: LineString, roughness: float, rng) -> LineString:
"""``segment`` offset along its normals by a smoothed random walk."""
n_points = max(257, len(segment.coords))
distances = np.linspace(0.0, segment.length, n_points)
coordinates = shapely.get_coordinates(
shapely.line_interpolate_point(segment, distances)
)
progress = np.linspace(0.0, 1.0, n_points)
# A random walk with its linear trend removed is zero at both ends. An
# 11-point moving average smooths it, and the detrend is repeated because
# smoothing moves the ends.
displacement = np.cumsum(rng.normal(size=n_points))
displacement -= displacement[0] + progress * (displacement[-1] - displacement[0])
window = 11
padded = np.pad(displacement, window // 2, mode="reflect")
displacement = np.convolve(padded, np.ones(window) / window, mode="valid")
displacement -= displacement[0] + progress * (displacement[-1] - displacement[0])
peak = float(np.max(np.abs(displacement)))
if peak > 0:
displacement *= roughness * segment.length / peak
tangents = np.gradient(coordinates, axis=0)
tangent_lengths = np.linalg.norm(tangents, axis=1)
if np.any(tangent_lengths == 0):
raise ValueError("string guide contains a zero-length section")
normals = np.column_stack([-tangents[:, 1], tangents[:, 0]])
normals /= tangent_lengths[:, None]
meandered = coordinates + displacement[:, None] * normals
meandered[[0, -1]] = coordinates[[0, -1]]
return LineString(meandered)
