import warnings
from pathlib import Path
from typing import Any
import geopandas as gpd
import numpy as np
import shapely
from rasterio.io import DatasetReader
from rasterio.windows import Window
from shapely.geometry.linestring import LineString
from shapelib.core.geometries.base import BaseShapes
from shapelib.core.geometries.polygons import Polygons
from shapelib.tools.converters import lines_to_mask, mask_to_lines
from shapelib.tools.lines.distance import estimate_line_spacing
from shapelib.tools.lines.grouping import group_lines
[docs]
class Lines(BaseShapes):
"""
A class for managing LineString geometries in a GeoDataFrame.
This class provides specific functionality for handling LineString geometries,
including calculating lengths and managing geometry-specific metadata.
Parameters
----------
data : gpd.GeoDataFrame or pd.DataFrame, optional
The input data for the GeoDataFrame, can be a GeoDataFrame or a DataFrame.
geometry : shapely.geometry.LineString, optional
The geometry type to use (LineString in this case). Defaults to None.
crs : str, int, or pyproj.CRS, optional
Coordinate reference system (CRS) to use for the GeoDataFrame.
path : str or pathlib.Path, optional
The file path for the data (if reading from a file).
driver : str, optional
The file driver to use when reading/writing data.
Attributes
----------
has_spacement_info : bool
Indicates whether spacement information is available.
length_column_name : str
The name of the column storing the calculated length values.
geometry_type : type
The expected geometry type for this class (LineString).
length_precision : int
The precision for rounding length values when calculating the total length.
Examples
--------
Create a Lines object:
>>> lines = Lines(data=gdf)
>>> total_length = lines.get_total_length() # Get total length of all lines
>>> print(f"Total: {total_length} km")
"""
_metadata = BaseShapes._metadata + [
"has_spacement_info",
"length_column_name",
"geometry_type",
"length_precision",
]
def __init__(
self,
data: Any = None,
*args: Any,
geometry: Any = None,
crs: Any = None,
path: str | Path | None = None,
driver: str | None = None,
**kwargs: Any,
) -> None:
super().__init__(data, *args, geometry=geometry, crs=crs, path=path, driver=driver, **kwargs)
self.has_spacement_info: bool = False
self.length_column_name: str = "length"
self.geometry_type = LineString
self.length_precision: int = 6
@property
def length(self) -> gpd.GeoSeries:
"""
Return the length of each LineString geometry (GeoPandas native behavior).
.. deprecated:: 0.1.0
Accessing `.length` now returns a GeoSeries with individual line lengths,
following GeoPandas' native behavior. Use `.get_total_length()` to obtain
the sum of all line lengths.
This property returns a GeoSeries containing the length of each individual
LineString in the GeoDataFrame, computed using the current CRS. This is the
standard GeoPandas behavior.
Returns
-------
gpd.GeoSeries
A GeoSeries with the length of each geometry in the current CRS units.
See Also
--------
get_total_length : Get the sum of all line lengths with geodetic accuracy.
calculate_length : Calculate geodetic length for each line and store in a column.
Examples
--------
>>> # Get individual lengths (GeoSeries)
>>> lengths = lines.length
>>> print(lengths)
0 150.5
1 230.8
dtype: float64
>>> # Get total length of all lines
>>> total = lines.get_total_length()
>>> print(f"Total: {total} km")
Notes
-----
This property emits a DeprecationWarning to alert users of the behavior change.
In previous versions, `.length` returned the total sum of all line lengths.
"""
warnings.warn(
"The `.length` property now returns a GeoSeries with individual line "
"lengths (GeoPandas native behavior). To get the total sum of all line "
"lengths, use `.get_total_length()` instead. This warning will be removed "
"in a future version.",
DeprecationWarning,
stacklevel=2,
)
return super().length
def _geom_length(self, geom: LineString) -> float:
"""
Calculate the length of a single LineString geometry.
This method calculates the length of the given LineString geometry using geodetic
methods for accuracy. It returns the length in kilometers.
Parameters
----------
geom : shapely.geometry.LineString
The LineString geometry to calculate the length for.
Returns
-------
float or np.nan
The length of the LineString in kilometers. Returns np.nan if the geometry is
invalid.
Examples
--------
>>> length = lines._geom_length(geom)
>>> print(f"Line Length: {length} km")
"""
return self.geod.geometry_length(geometry=geom)
[docs]
def calculate_length(self, column_name: str = "length") -> None:
"""
Calculate the length of each line in the GeoDataFrame and store the result in a
new column.
This method transforms the GeoDataFrame to WGS84 to ensure accurate length
calculations, computes the length of each line, and stores the result in a column
with the specified name.
Parameters
----------
column_name : str, optional
The name of the column to store the calculated lengths. Defaults to "length".
Returns
-------
None
The method modifies the GeoDataFrame in place by adding a new column with the
length values.
Examples
--------
>>> lines.calculate_length()
>>> print(lines["length"]) # See the calculated length for each line.
"""
with self.temp_crs(self.wgs84_crs):
self[column_name] = self.geometry.apply(self._geom_length)
self.length_column_name = column_name
[docs]
def get_total_length(self) -> float:
"""
Return the total length of all lines in the GeoDataFrame.
This method recalculates the length for each line, sums the values, and rounds
to the precision specified by `length_precision`.
Returns
-------
float
The total length of all lines, rounded to the specified precision.
Returns 0.0 if the GeoDataFrame is empty.
Examples
--------
>>> total_length = lines.get_total_length()
>>> print(f"Total Length: {total_length} km")
"""
if len(self) == 0:
return 0.0
self.calculate_length()
return round(float(self[self.length_column_name].sum()), self.length_precision)
[docs]
def check_spacement_info(self) -> str | None:
"""
Check if the GeoDataFrame has a column related to spacement information.
This method searches for common spacement-related column names
(e.g., "spacing", "espaçamento") in the GeoDataFrame and sets the
`has_spacement_info` flag accordingly. If a matching column is found, its name
is returned.
Returns
-------
str or None
The name of the spacement-related column if found, otherwise None.
Examples
--------
>>> column = lines.check_spacement_info()
>>> if column:
... print(f"Spacement info available in column: {column}")
... else:
... print("No spacement info found.")
"""
options: tuple[str, ...] = (
"espac",
"spac",
"espacamento",
"espaçamento",
"space",
"spacing",
)
for option in options:
if self.has_column(option):
self.has_spacement_info = True
return option
self.has_spacement_info = False
return None
def _reduce_geom_length(self, geom: LineString, length: float) -> LineString:
"""
Reduce the length of a LineString by trimming it from both ends.
This method shortens the length of a LineString by trimming it from both ends
using the specified length.
Parameters
----------
geom : shapely.geometry.LineString
The LineString geometry to reduce.
length : float
The length by which to reduce the LineString.
Returns
-------
shapely.geometry.LineString
The shortened LineString.
Examples
--------
>>> reduced_geom = lines._reduce_geom_length(geom, length=100)
>>> print(f"Reduced Line Length: {reduced_geom.length} km")
"""
new_point_end = geom.interpolate(geom.length - length)
new_point_end = new_point_end.buffer(0.0001)
new_point_start = geom.interpolate(length)
new_point_start = new_point_start.buffer(0.0001)
geom = shapely.ops.split(geom, new_point_end).geoms[0]
geom = shapely.ops.split(geom, new_point_start).geoms[-1]
return geom
[docs]
def reduce_length(self, length: float) -> None:
"""
Reduce the length of all LineString geometries in the GeoDataFrame by the
specified length.
This method applies `_reduce_geom_length` to each geometry, reducing its length
by the specified value.
Parameters
----------
length : float
The length by which to reduce each LineString geometry. If the value is less
than or equal to zero, no reduction is performed.
Returns
-------
None
The method modifies the geometries in place.
Examples
--------
>>> lines.reduce_length(100)
>>> print("Length reduced by 100 meters for all geometries.")
"""
if length <= 0 or len(self) == 0:
return
# Ensure we have a valid UTM CRS to perform metric operations
if self.utm_crs is None:
self.estimate_utm_crs()
if self.utm_crs is None:
raise ValueError("UTM CRS could not be determined.")
with self.temp_crs(self.utm_crs):
self.geometry = self.geometry.apply(self._reduce_geom_length, args=(length,))
[docs]
def filter_by_length(self, length: float) -> None:
"""
Filter the GeoDataFrame to retain only LineString geometries that are longer than
the specified length.
This method removes any geometries whose length is less than the specified value.
Parameters
----------
length : float
The minimum length required for a geometry to be retained. If the
value is less than or equal to zero, no filtering is performed.
Returns
-------
None
The method modifies the GeoDataFrame in place.
Examples
--------
>>> lines.filter_by_length(500)
>>> print(f"Retained lines longer than 500 meters.")
"""
if length <= 0 or len(self) == 0:
return
self.calculate_length()
self.drop(self[self[self.length_column_name] < length].index, inplace=True)
[docs]
def buffer_m(self, length_m: float, cap_style: int = 2) -> Polygons:
"""
Create a buffer around each LineString geometry in meters and return the result
as a Polygons object.
This method creates a buffer of the specified length (in meters) around each
LineString geometry and returns a new `Polygons` object with the buffered
geometries.
Parameters
----------
length_m : float
The buffer distance in meters.
cap_style : int, optional
The style of the buffer's cap (e.g., round or square). Defaults to 2 (round).
Returns
-------
Polygons
A new `Polygons` object containing the buffered geometries.
Examples
--------
>>> buffered_polys = lines.buffer_m(100)
>>> buffered_polys.plot()
"""
original_crs = self.crs
# Ensure a valid UTM CRS to do metric buffering
if self.utm_crs is None:
self.estimate_utm_crs()
if self.utm_crs is None:
raise ValueError("UTM CRS could not be determined.")
with self.temp_crs(self.utm_crs):
buffered_series = self.buffer(length_m, cap_style=cap_style)
buffered_gdf = gpd.GeoDataFrame(geometry=buffered_series, crs=self.crs)
buffered_polys = Polygons(buffered_gdf)
buffered_polys.to_crs(original_crs, inplace=True)
return buffered_polys
[docs]
def simplify_lines(self, tolerance: float) -> None:
"""
Simplify the geometries in the GeoDataFrame by reducing the number of vertices.
This method simplifies the geometries based on the specified tolerance, reducing
their complexity and improving performance for rendering or spatial operations.
Parameters
----------
tolerance : float
The tolerance for simplifying the geometries. Higher values result in greater
simplification.
Returns
-------
None
The method modifies the geometries in place.
Examples
--------
>>> lines.simplify_lines(0.01)
>>> lines.plot()
"""
# Delegate to BaseShapes.simplify, which uses temp_crs(self.utm_crs)
# and restores the original CRS automatically.
self.simplify(tolerance)
[docs]
def split_by_touching_polygons(self, polygons: Polygons) -> tuple["Lines", "Lines"]:
"""
Separate LineStrings into two groups: those that touch the polygon boundaries and
those that do not.
Parameters
----------
polygons : Polygons
A Polygons object containing the polygon geometries.
Returns
-------
tuple[Lines, Lines]
A tuple where the first element is a Lines object containing LineStrings
touching the polygon boundaries, and the second element is a Lines object
containing LineStrings not touching the polygon boundaries.
Examples
--------
>>> touching_lines, non_touching_lines = \
lines.split_by_touching_polygons(polygons)
>>> print(f"Lines touching: {len(touching_lines)}, Lines not touching:
{len(non_touching_lines)}")
"""
if polygons.empty or self.empty:
return self.iloc[[]], self.iloc[[]]
# Ensure both objects use the same CRS
if polygons.crs != self.crs:
polygons = polygons.to_crs(self.crs)
# Compute polygon boundaries efficiently
polygon_boundaries = polygons.boundary.unary_union
# Identify lines that touch the polygon boundaries
touching_mask = self.geometry.touches(polygon_boundaries)
return self.loc[touching_mask], self.loc[~touching_mask]
[docs]
@classmethod
def from_mask(
cls,
mask: np.ndarray[Any, Any],
dataset_reader: DatasetReader,
window: Window | None = None,
dilation_kernel_size: int = 5,
min_distance_points: int = 10,
look_ahead_distance: int = 8,
min_line_size: int = 10,
min_object_size: int = 0,
) -> "Lines":
"""
Create a Lines object from a labeled or binary mask and a rasterio dataset reader.
This method extracts line geometries from a 2D NumPy array (mask), where nonzero
values represent features of interest (e.g., 1 for foreground, 0 for background,
or any labeled regions).
Parameters
----------
mask : np.ndarray
A 2D NumPy array where 0 represents background and any nonzero (or labeled)
value represents features to be converted into lines.
dataset_reader : rasterio.io.DatasetReader
A rasterio dataset that the mask is derived from, used for transform and CRS.
window : rasterio.windows.Window, optional
If provided, the transform will be adjusted for the window region.
The mask should correspond to the windowed region.
dilation_kernel_size : int, optional
Size of the morphological dilation kernel. Larger values connect nearby
features but may merge distinct lines. Set to 0 to skip dilation. Defaults
to 5.
min_distance_points : int, optional
Minimum distance between consecutive points in the line tracing algorithm.
Higher values produce smoother lines with fewer vertices. Defaults to 10.
look_ahead_distance : int, optional
Look-ahead distance for the line tracing algorithm.
Higher values produce smoother but potentially less accurate lines. Defaults
to 8.
min_line_size : int, optional
Minimum number of pixels required for a line to be included in the output.
Defaults to 10.
min_object_size : int, optional
Minimum size (in pixels) of connected components to retain after removing
small objects. Components smaller than this will be removed. Defaults to 0
(no removal).
Returns
-------
Lines
A Lines object containing the extracted lines as georeferenced geometries.
Example
-------
>>> with rasterio.open('image.tif') as src:
... mask = model.predict(src.read(1)) # Get a binary or labeled mask
... lines = Lines.from_mask(mask, src)
... print(lines)
... # Extract lines from a specific window
... window = rasterio.windows.Window(0, 0, 100, 100)
... mask_window = mask[:100, :100]
... lines_window = Lines.from_mask(mask_window, src, window=window)
... print(lines_window)
"""
gdf = mask_to_lines(
mask,
dataset_reader,
window=window,
dilation_kernel_size=dilation_kernel_size,
min_distance_points=min_distance_points,
look_ahead_distance=look_ahead_distance,
min_line_size=min_line_size,
min_object_size=min_object_size,
)
return cls(gdf)
[docs]
def to_mask(
self,
dataset_reader: DatasetReader,
fill_value: int = 1,
background_value: int = 0,
all_touched: bool = True,
window: Window = None,
min_size: tuple[int, int] | None = None,
) -> np.ndarray:
"""
Convert the Lines object to a binary mask using a rasterio dataset reader.
This method rasterizes the line geometries in this GeoDataFrame into a binary
mask with the same dimensions and coordinate reference system as the provided
rasterio dataset reader. If a window is provided, only the region inside the
window is rasterized.
Parameters
----------
dataset_reader : rasterio.io.DatasetReader
Rasterio dataset reader that defines the target raster properties
(transform, CRS, height, width) for the output mask.
fill_value : int, optional
Value to assign to pixels that intersect with line geometries.
Defaults to 1.
background_value : int, optional
Value to assign to pixels that do not intersect with any line.
Defaults to 0.
all_touched : bool, optional
If True, all pixels touched by geometries will be burned in.
If False, only pixels whose center is within the line or that
are selected by Bresenham's line algorithm will be burned in.
Defaults to False for more precise boundary adherence.
window : rasterio.windows.Window, optional
If provided, only rasterize the region inside the window. The output mask
will have shape (window.height, window.width).
min_size : tuple[int, int], optional
Minimum size (height, width) for the output mask. If the generated mask is
smaller than this size, it will be padded with background_value to match
the exact minimum size. Defaults to None.
Returns
-------
np.ndarray
A 2D numpy array (mask) with the same dimensions as the input raster or
window, where pixels intersecting with lines have the fill_value and
background pixels have the background_value.
Examples
--------
>>> import rasterio
>>> with rasterio.open('image.tif') as src:
... mask = lines.to_mask(src)
... print(mask.shape) # Should match src.shape
... print(mask.dtype) # uint8
...
... window = rasterio.windows.Window(0, 0, 50, 50)
... mask_window = lines.to_mask(src, window=window)
... print(mask_window.shape) # Should be (50, 50)
...
... # With minimum size padding
... mask_padded = lines.to_mask(src, window=window, min_size=(256, 256))
... print(mask_padded.shape) # Should be (256, 256)
"""
return lines_to_mask(
self,
dataset_reader,
fill_value=fill_value,
background_value=background_value,
all_touched=all_touched,
window=window,
min_size=min_size,
)
[docs]
def estimate_line_spacing(
self,
sample: float = 0.1,
min_samples: int = 30,
seed: int | None = None,
) -> float:
"""
Estimate the spacing between parallel planting lines (e.g. rows every
1.5 m or 2.4 m).
Samples a few lines, measures each one's center-to-nearest-line
distance, and returns the median. See
:func:`shapelib.tools.lines.distance.estimate_line_spacing`.
Parameters
----------
sample : float, default 0.1
Fraction of lines to sample if ``<= 1``, else an absolute count.
min_samples : int, default 30
Minimum lines to sample, clamped to the total.
seed : int, optional
Seed for reproducible sampling.
Returns
-------
float
Estimated spacing in meters.
"""
return estimate_line_spacing(self, sample, min_samples, seed)
[docs]
def group_lines(
self,
distance_threshold: float = 2.5,
) -> "Lines":
"""
Cluster these lines into proximity groups.
Thin wrapper around :func:`shapelib.tools.lines.grouping.group_lines`.
Lines must be in a projected CRS (metres); call ``to_utm()`` first.
Filtering by talhão is the caller's responsibility — call this on the
lines of a single talhão to group them in isolation.
Parameters
----------
distance_threshold : float, default 2.5
Maximum spacing (in CRS units) for two lines to be neighbors.
Returns
-------
Lines
A Lines object with the original attributes plus ``group_id`` and
``line_id`` columns.
Examples
--------
>>> lines.to_utm()
>>> grouped = lines.group_lines(distance_threshold=3.5)
>>> grouped["group_id"].nunique()
2
"""
gdf = group_lines(self, distance_threshold=distance_threshold)
return Lines(gdf, crs=self.crs)