diff --git a/geostructures/structures.py b/geostructures/structures.py index 5c2d678..152aa91 100644 --- a/geostructures/structures.py +++ b/geostructures/structures.py @@ -10,10 +10,12 @@ from abc import ABC import copy +from datetime import timedelta from functools import cached_property import math import statistics from typing import cast, Any, Dict, List, Optional, Tuple, Sequence, TYPE_CHECKING +import warnings import numpy as np @@ -23,7 +25,7 @@ _RE_LINESTRING_WKT, LineLikeMixin, PointLikeMixin, PolygonLikeMixin, SingleShapeBase, SimpleShapeMixin ) -from geostructures.time import GEOTIME_TYPE +from geostructures.time import GEOTIME_TYPE, TimeInterval from geostructures.coordinates import Coordinate from geostructures.calc import ( inverse_haversine_radians, @@ -1479,6 +1481,127 @@ def intersects_shape(self, shape: 'GeoShape', **kwargs) -> bool: # because the centroid may fall in a hole return o_edges[0][0][0] in self or s_edges[0][0][0] in shape + def split(self, distance_meters: float) -> List['GeoLineString']: + """ + Splits a GeoLineString into smaller segments based on a specified distance. + + Args: + distance_meters (float): The maximum distance for each segment in meters. + + Returns: + List[GeoLineString]: A list of GeoLineString objects, each of which is no longer + than the specified distance. If the total length of the + line is less than the specified distance, the original line + is returned as a single segment. + + Notes: + - If the GeoLineString is time-bounded (has a datetime interval), the resulting + segments will have proportional datetime intervals based on the segment's length. + - If the specified distance is greater than the total length of the line, + a warning is issued, and the original line is returned. + """ + + out = [] # List to store resulting GeoLineString segments + segments: List[Tuple[Coordinate, Coordinate]] = self.segments.copy() # Copy of all line segments + vertices = [segments[0][0]] # Initialize the first vertex from the starting point of the first segment + remaining_distance_meters = None # Remaining distance from a previous iteration + total_length_meters = sum(haversine_distance_meters(*segment) for segment in segments) # Total line length + + # Handle case where the total line length is less than the specified distance + if total_length_meters <= distance_meters: + warnings.warn( + f'Total length ({total_length_meters}) is less than the distance ({distance_meters}); returning line.' + ) + return [self] + + properties = self._properties.copy() # Copy of properties for the GeoLineString + cumulative_length = 0 # Tracks cumulative length traversed + dt = None # Placeholder for datetime interval for each segment + + start_time = end_time = total_duration_seconds = None + # If time interval is provided, calculate total duration in seconds + if self.dt and self.dt.start != self.dt.end: + start_time, end_time = self.dt.start, self.dt.end + total_duration_seconds = (end_time - start_time).total_seconds() + + # Iterate through segments and split accordingly + while segments: + remaining_segment_length = haversine_distance_meters(*segments[0]) # Length of the current segment + + if remaining_distance_meters is not None: + # If there's a remaining distance from the previous iteration, process it + cumulative_length += distance_meters - remaining_distance_meters + end_point = inverse_haversine_degrees( + vertices[-1], + bearing_degrees(vertices[-1], segments[0][1]), + remaining_distance_meters + ) + # Calculate the time interval proportionally if applicable + if total_duration_seconds is not None: + segment_start_time = start_time + timedelta( + seconds=cumulative_length / total_length_meters * total_duration_seconds + ) + dt = TimeInterval(dt.end, segment_start_time) + elif self.dt: + dt = self.dt + + vertices.append(end_point) # Add the calculated endpoint to vertices + out.append(GeoLineString(vertices, properties=properties.copy(), dt=dt)) # Store the segment + vertices = [end_point] # Reset vertices for the next segment + remaining_segment_length = haversine_distance_meters(vertices[-1], segments[0][1]) + remaining_distance_meters = None + + while distance_meters < remaining_segment_length: + # Handle cases where the current segment is longer than the specified distance + cumulative_length += distance_meters + end_point = inverse_haversine_degrees( + vertices[-1], + bearing_degrees(vertices[-1], segments[0][1]), + distance_meters + ) + # Calculate the time interval proportionally if applicable + if total_duration_seconds is not None: + segment_start_time = start_time + timedelta( + seconds=(cumulative_length - distance_meters) + / total_length_meters * total_duration_seconds + ) + segment_end_time = start_time + timedelta( + seconds=cumulative_length / total_length_meters * total_duration_seconds + ) + dt = TimeInterval(segment_start_time, segment_end_time) + elif self.dt: + dt = self.dt + + vertices.append(end_point) # Add the calculated endpoint to vertices + out.append(GeoLineString(vertices, properties=properties.copy(), dt=dt)) # Store the segment + vertices = [end_point] # Reset vertices for the next segment + remaining_segment_length = haversine_distance_meters(vertices[-1], segments[0][1]) + + # Calculate the remaining distance after processing the current segment + remaining_distance_meters = distance_meters - remaining_segment_length + cumulative_length += remaining_distance_meters + vertices.append(segments[0][1]) # Add the endpoint of the current segment + + if len(segments) == 1: + break + + segments.pop(0) # Remove the processed segment + + # Handle the final segment if there are remaining distances + if remaining_distance_meters: + last_segment = out.pop() if out else None + if last_segment: + vertices = last_segment.vertices[:-1] + [vertices[-1]] + # Assign the time interval for the final segment if applicable + if total_duration_seconds is not None: + dt = TimeInterval(last_segment.dt.start, end_time) + elif self.dt: + dt = self.dt + + out.append(GeoLineString(vertices, properties=properties.copy(), dt=dt)) + + return out + def to_geo_interface(self, **kwargs): return { **self.__geo_interface__, diff --git a/tests/test_structures.py b/tests/test_structures.py index 1983e08..137edfa 100644 --- a/tests/test_structures.py +++ b/tests/test_structures.py @@ -1593,3 +1593,83 @@ def test_geopoint_from_wkt(): def test_geopoint_to_wkt(geopoint): assert geopoint.to_wkt() == 'POINT(0.0 0.0)' + +@pytest.fixture +def basic_line(): + # A basic GeoLineString without time + return GeoLineString([ + Coordinate(0, 0), + Coordinate(1, 1), + Coordinate(2, 2) + ]) + + +@pytest.fixture +def timed_line(): + # A GeoLineString with a time interval + return GeoLineString( + [ + Coordinate(0, 0), + Coordinate(1, 1), + Coordinate(2, 2) + ], + dt=TimeInterval( + start=datetime(2025, 1, 1, 0, 0, 0), + end=datetime(2025, 1, 1, 2, 0, 0) + ) + ) + + +def test_split_no_split_needed(basic_line): + # Distance is greater than the total length of the line + result = basic_line.split(200000) + assert len(result) == 1 + assert result[0] == basic_line + + +def test_split_even_segments(basic_line): + # Splitting into segments close to half the total length + half_length = 157237.40665500844 # Approximate 1/2 total length of line + result = basic_line.split(half_length) + + # Check if the segments are created + assert len(result) == 2 + + # Validate the first segment + assert result[0].vertices[0] == basic_line.vertices[0] + assert result[0].vertices[-1].longitude != basic_line.vertices[-1].longitude # Shouldn't reach the end + + # Validate the second segment starts where the first ended + assert result[1].vertices[0] == result[0].vertices[-1] + assert result[1].vertices[-1] == basic_line.vertices[-1] + + +def test_split_with_remainder(basic_line): + # Distance doesn't evenly divide the total length + result = basic_line.split(5000) + assert len(result) > 1 + assert all(len(segment.vertices) > 1 for segment in result) + + +def test_split_with_time_intervals(timed_line): + # Ensure time intervals are proportional + result = timed_line.split(157249) + assert len(result) == 2 + assert result[0].dt.start == timed_line.dt.start + assert result[1].dt.end == timed_line.dt.end + assert result[0].dt.end == result[1].dt.start + assert result[0].dt.end < result[1].dt.end + assert result[0].dt.start < result[1].dt.start + + +def test_warning_on_large_distance(basic_line): + # Warning if the split distance exceeds the total length + with pytest.warns(UserWarning): + result = basic_line.split(400000) + assert len(result) == 1 + + +def test_split_exact_division(basic_line): + # Distance perfectly divides the line + result = basic_line.split(157249 * 2) + assert len(result) == 1 \ No newline at end of file