Source code for shapelib.tools.lines.dedup.core

"""Drop near-coincident (stacked) lines, keeping the longest of each group.

Drone-mapped rows sometimes carry a near-duplicate: two input lines trace the
same physical row a few millimetres apart. That twin poisons downstream tools —
in :mod:`shapelib.tools.lines.merge` it becomes the "rail" of its own sibling and
the merge follows a phantom row. This tool collapses each cluster of stacked
lines into a single representative (the longest one).

A line is dropped when it lies entirely inside the tolerance tube of a longer,
kept line — tested exactly with a buffered ``covers`` predicate evaluated inside
the spatial index, so no point sampling is involved.

The public entry point is :func:`deduplicate_lines`.
"""

import geopandas as gpd
import numpy as np
import shapely

from shapelib.utils.utils import multiline_to_single_line


def _keep_mask(geoms: np.ndarray, tolerance: float) -> np.ndarray:
    """Boolean mask of the lines to keep after collapsing stacked duplicates.

    Lines are visited longest-first; each kept line drops every not-yet-dropped
    line of equal-or-shorter length that its tolerance buffer fully covers. The
    ``covers`` predicate is applied by the STRtree in a single GEOS call per kept
    line, so the covered-by test is exact rather than sampled.
    """
    lengths = shapely.length(geoms)
    buffers = shapely.buffer(geoms, tolerance)
    tree = shapely.STRtree(geoms)

    dropped = np.zeros(len(geoms), dtype=bool)
    for i in np.argsort(lengths)[::-1]:  # longest first
        if dropped[i]:
            continue
        for j in tree.query(buffers[i], predicate="covers"):
            if j != i and not dropped[j] and lengths[j] <= lengths[i]:
                dropped[j] = True
    return ~dropped




def deduplicate_lines(
    lines: gpd.GeoDataFrame,
    tolerance: float = 0.05,
) -> gpd.GeoDataFrame:
    """
    Drop near-coincident (stacked) lines, keeping the longest of each cluster.

    Two input lines that trace the same physical row a few millimetres apart are
    "stacked" duplicates: the shorter one lies entirely within the ``tolerance``
    tube of the longer. This tool collapses every such cluster to its longest
    line and passes all other lines through unchanged, so the output is the same
    line set with the phantom twins removed. Filtering by talhão is the caller's
    responsibility — pass only the lines of a single talhão.

    Parameters
    ----------
    lines : geopandas.GeoDataFrame
        Line geometries. Must be in a projected CRS (metres), since ``tolerance``
        is expressed in CRS units. MultiLineStrings are flattened to single
        LineStrings before the coverage test.
    tolerance : float, default 0.05
        Maximum perpendicular distance (CRS units) between two lines for one to
        count as a stacked duplicate of the other. A line is dropped only when it
        lies entirely within this distance of a longer, kept line.

    Returns
    -------
    geopandas.GeoDataFrame
        A subset of ``lines`` (original attribute columns and CRS preserved) with
        the stacked duplicates removed. When two covering lines have exactly equal
        length, the first in the length ordering is kept.

    Examples
    --------
    >>> import shapelib
    >>> lines = shapelib.read_file("lines.shp")
    >>> lines.to_utm()
    >>> cleaned = deduplicate_lines(lines, tolerance=0.05)
    >>> len(cleaned) <= len(lines)
    True
    """
    if lines is None or len(lines) == 0:
        return lines

    geoms = np.array(
        [multiline_to_single_line(g) for g in lines.geometry.values], dtype=object
    )
    keep = _keep_mask(geoms, tolerance)
    return lines.iloc[keep].copy()