"""Generate interpolated "interlines" between pairs of grouped lines.
This is a port of ``optimal-interlines``' ``InterlinesGenerator``. It reuses the
clustering engine from :mod:`shapelib.tools.lines.grouping`, then for each group
pairs neighboring lines, interpolates an intermediate line scored by its
deviation from a reference distance, and classifies the interlines into
alternating classes. Generation and optimal selection are separate steps:
:func:`generate_interlines` returns *every* interline, and
:func:`choose_optimal_interlines` reduces each group to its better-scoring
class.
"""
from collections import deque
from typing import Dict, List, Sequence
import geopandas as gpd
from geopandas.tools import sjoin
from shapelib.tools.lines.grouping.core import _LineGrouper
from shapelib.tools.lines.interlines.models import _Interline, _Pair
# Output columns that the tool always produces; input attributes colliding with
# these are dropped so the parallel-list build stays consistent.
_RESERVED_COLUMNS = {
"geometry",
"group_id",
"class_id",
"score",
"abs_score",
"scores_per_point",
}
def _optimal_class(
class_ids: Sequence[int], abs_scores: Sequence[float]
) -> int:
"""Return the class id (``1`` or ``-1``) to keep for one group.
The winning class is the one with the lower mean absolute score. This is the
single source of truth for the optimal selection, used by
:func:`choose_optimal_interlines`. An empty (or all-zero-score) class loses to
a non-empty other class, mirroring the original implementation's tie handling.
"""
first = [s for c, s in zip(class_ids, abs_scores) if c == 1]
second = [s for c, s in zip(class_ids, abs_scores) if c != 1]
first_score = sum(first) / max(len(first), 1)
second_score = sum(second) / max(len(second), 1)
if first_score == 0:
return -1
if second_score == 0:
return 1
return 1 if first_score < second_score else -1
class _InterlineEngine:
"""Drives interline generation on top of a :class:`_LineGrouper`."""
def __init__(
self,
lines: gpd.GeoDataFrame,
distance_threshold: float,
reference_distance: float,
) -> None:
self.grouper = _LineGrouper(lines, distance_threshold)
self.distance_threshold = distance_threshold
self.reference_distance = reference_distance
self.crs = lines.crs
def run(self) -> List[_Interline]:
"""Return every generated (classified) interline across every group."""
total: List[_Interline] = []
grouped_lines = self.grouper.classify_lines_by_groups()
groups: Dict[int, list] = {}
for grouped_line in grouped_lines:
groups.setdefault(grouped_line.group, []).append(grouped_line)
for group_lines in groups.values():
interlines = self._generate_for_group(group_lines)
if not interlines:
continue
interlines = self._remove_outliers(interlines, group_lines)
if not interlines:
continue
total.extend(self._classify(interlines))
return total
def _generate_for_group(self, group_lines: list) -> List[_Interline]:
seen_pairs: set = set()
pairs = []
for line in group_lines:
closest = self.grouper.find_line_neighbors(
line, group_lines, max_distance=self.distance_threshold
)
for neighbor in closest:
if neighbor.id == line.id:
continue
key = (min(line.id, neighbor.id), max(line.id, neighbor.id))
if key in seen_pairs:
continue
seen_pairs.add(key)
pairs.append(
_Pair(
line,
neighbor,
self.reference_distance,
self.grouper.next_id(),
)
)
return [p.parallel_line for p in pairs if p.parallel_line is not None]
def _remove_outliers(
self, interlines: List[_Interline], group_lines: list
) -> List[_Interline]:
"""Drop interlines that touch a (slightly buffered) source line."""
buffered = [
line.geom.buffer(self.distance_threshold * 0.1) for line in group_lines
]
lines_gdf = gpd.GeoDataFrame({"geometry": buffered}, crs=self.crs)
interlines_gdf = gpd.GeoDataFrame(
{"geometry": [il.geom for il in interlines]}, crs=self.crs
)
joined = sjoin(interlines_gdf, lines_gdf, how="inner", predicate="intersects")
intersection_indexes = set(joined.index)
return [
interlines[i]
for i in range(len(interlines))
if i not in intersection_indexes
]
def _classify(self, interlines: List[_Interline]) -> List[_Interline]:
"""Assign alternating ``classe`` (±1) across the interline neighbor graph."""
interlines = list(interlines)
biggest = max(interlines, key=lambda line: len(line.geom.coords))
interlines.remove(biggest)
interlines.insert(0, biggest)
class_id = 1
visiteds: set = set()
lines_dict = {line.id: line for line in interlines}
for line in interlines:
if line.id in visiteds:
continue
visiteds.add(line.id)
to_classify = deque([[line, class_id]])
lines_dict[line.id].classe = class_id
while to_classify:
current_line, class_id = to_classify.popleft()
class_id *= -1
neighbors = self.grouper.find_line_neighbors(
current_line, interlines, visiteds, 10
)
for neighbor in neighbors:
if self.grouper.ensure_neighbors(
current_line, neighbor, threshold=0.3
):
visiteds.add(neighbor.id)
to_classify.append([neighbor, class_id])
lines_dict[neighbor.id].classe = class_id
return list(lines_dict.values())
[docs]
def generate_interlines(
lines: gpd.GeoDataFrame,
distance_threshold: float = 1.5,
reference_distance: float = 1.5,
include_point_scores: bool = False,
) -> gpd.GeoDataFrame:
"""
Generate every interpolated interline between pairs of neighboring lines.
Lines are grouped (see :func:`shapelib.tools.lines.grouping.group_lines`),
then within each group neighboring lines are paired and an intermediate line
is interpolated midway between them. Each interline is scored by how far its
sampled points deviate from ``reference_distance`` and classified into
alternating classes (``1`` / ``-1``). *All* interlines are returned; to keep
only the better-scoring class per group, pass the result through
:func:`choose_optimal_interlines`. Filtering lines by talhão is the caller's
responsibility — pass only the lines belonging to a single talhão to process
them in isolation.
Parameters
----------
lines : geopandas.GeoDataFrame
Line geometries. Must be in a projected CRS (metres), since the distance
parameters are in CRS units.
distance_threshold : float, default 1.5
Maximum spacing for two lines to be paired/grouped (CRS units).
reference_distance : float, default 1.5
Target spacing used to score each interline; deviations from it raise the
(absolute) score.
include_point_scores : bool, default False
If ``True``, attach a ``scores_per_point`` object column holding the
per-vertex deviation list for each interline. This column cannot be
written to most vector formats and should be dropped before export.
Returns
-------
geopandas.GeoDataFrame
One row per interline, carrying the source line's original attributes
plus:
- ``group_id`` : integer id of the cluster the interline belongs to
(unique within the call).
- ``class_id`` : alternating class (``1`` / ``-1``).
- ``score`` : mean signed deviation from ``reference_distance``.
- ``abs_score`` : mean absolute deviation from ``reference_distance``.
- ``scores_per_point`` : per-vertex deviations (only if
``include_point_scores=True``).
Examples
--------
>>> import shapelib
>>> lines = shapelib.read_file("lines.shp")
>>> lines.to_utm()
>>> interlines = generate_interlines(lines, distance_threshold=3.5,
... reference_distance=1.5)
>>> sorted(interlines["class_id"].unique())
[-1, 1]
>>> optimal = choose_optimal_interlines(interlines)
>>> sorted(optimal["class_id"].unique())
[1]
"""
base_columns = ["geometry", "group_id", "class_id", "score", "abs_score"]
if include_point_scores:
base_columns.append("scores_per_point")
if lines is None or len(lines) < 2:
return gpd.GeoDataFrame(
{c: [] for c in base_columns},
geometry="geometry",
crs=getattr(lines, "crs", None),
)
engine = _InterlineEngine(lines, distance_threshold, reference_distance)
selected = engine.run()
lines_reset = engine.grouper.lines
geom_name = lines_reset.geometry.name
attr_cols = [
c
for c in lines_reset.columns
if c != geom_name and c not in _RESERVED_COLUMNS
]
records: Dict[str, list] = {c: [] for c in base_columns}
for col in attr_cols:
records[col] = []
for interline in selected:
records["geometry"].append(interline.geom)
records["group_id"].append(interline.group)
records["class_id"].append(interline.classe)
records["score"].append(interline.non_absolute_score)
records["abs_score"].append(interline.absolute_score)
if include_point_scores:
records["scores_per_point"].append(interline.scores_per_point)
source = (
lines_reset.iloc[interline.original_index]
if interline.original_index is not None
else None
)
for col in attr_cols:
records[col].append(source[col] if source is not None else None)
return gpd.GeoDataFrame(records, geometry="geometry", crs=lines.crs)
[docs]
def choose_optimal_interlines(interlines: gpd.GeoDataFrame) -> gpd.GeoDataFrame:
"""
Keep only the better-scoring interline class within each group.
The selection stage of the interlines pipeline: :func:`generate_interlines`
produces *every* interline; this function reduces each group to its optimal
class. Keeping the two stages separate lets callers inspect/adjust the full
set before selecting.
For each ``group_id`` the class (``1`` / ``-1``) with the lower mean
``abs_score`` is kept and the other is dropped (see :func:`_optimal_class`).
Parameters
----------
interlines : geopandas.GeoDataFrame
Interlines as returned by :func:`generate_interlines`. Must carry the
``group_id``, ``class_id`` and ``abs_score`` columns.
Returns
-------
geopandas.GeoDataFrame
The subset holding, per group, only the optimal class. All columns and the
CRS are preserved; the index is reset.
Raises
------
ValueError
If any of the required columns is missing.
Examples
--------
>>> interlines = generate_interlines(lines)
>>> optimal = choose_optimal_interlines(interlines)
"""
required = {"group_id", "class_id", "abs_score"}
missing = required - set(interlines.columns)
if missing:
raise ValueError(
"choose_optimal_interlines needs the columns "
f"{sorted(required)} (produced by generate_interlines); "
f"missing: {sorted(missing)}"
)
if len(interlines) == 0:
return interlines
keep_index: list = []
for _, group in interlines.groupby("group_id", sort=False):
winner = _optimal_class(
group["class_id"].tolist(), group["abs_score"].tolist()
)
selected = group[group["class_id"] == 1] if winner == 1 else group[group["class_id"] != 1]
keep_index.extend(selected.index.tolist())
return interlines.loc[keep_index].reset_index(drop=True)