Rckleisley/extrap3

.gitignore

@@ -0,0 +1,2 @@
+*__pycache__/*
+test/__pycache__/

geostructures/collections.py

@@ -18,7 +18,7 @@
 from geostructures import Coordinate, LOGGER
 from geostructures._base import PolygonLikeMixin, PointLikeMixin, LineLikeMixin, MultiShapeBase, BaseShape
 from geostructures._geometry import convex_hull
-from geostructures.calc import haversine_distance_meters
+from geostructures.calc import bearing_degrees, haversine_distance_meters, inverse_haversine_degrees
 from geostructures.multistructures import MultiGeoLineString, MultiGeoPoint, MultiGeoPolygon
 from geostructures.structures import GeoLineString, GeoPoint, GeoPolygon
 from geostructures.time import TimeInterval
@@ -229,6 +229,9 @@ def _get_dt(rec):
             ):
                 return None
 
+            if isinstance(dt_start, str) and isinstance(dt_start, str):
+                return TimeInterval.from_str(dt_start, dt_end)
+
             if not (dt_start and dt_end) or dt_start == dt_end:
                 return dt_start or dt_end
 
@@ -746,6 +749,63 @@ def time_start_diffs(self):
             for x, y in zip(self.geoshapes, self.geoshapes[1:])
         ])
 
+    def extrapolate(self, time_traveled: 'timedelta') -> 'Track':
+        last_shape = self.geoshapes[-1].copy()
+        if not isinstance(last_shape, (GeoPoint, GeoLineString)):
+            raise TypeError('Can only extrapolate a Points or LineStrings.')
+
+        if isinstance(last_shape, GeoLineString):
+            segments = last_shape.segments
+            start_time = last_shape.dt.start
+            end_time = last_shape.dt.end
+            total_duration_seconds = (end_time - start_time).total_seconds()
+            total_length_meters = sum(haversine_distance_meters(*segment) for segment in segments)
+            speed = total_length_meters / total_duration_seconds
+            bearing = bearing_degrees(*segments[0])
+            extrapolated_point = inverse_haversine_degrees(
+                segments[0][1],
+                bearing,
+                speed * time_traveled.total_seconds()
+            )
+            new_track = Track([
+                GeoLineString(
+                    [last_shape.vertices[-1], extrapolated_point],
+                    dt=TimeInterval(end_time, end_time + time_traveled),
+                    properties=last_shape.properties.copy()
+                )
+            ])
+
+        if isinstance(last_shape, GeoPoint):
+            second_last_shape = self.geoshapes[-2]
+            if last_shape.dt.start == last_shape.dt.end:
+                start_time = second_last_shape.dt.end
+                end_time = last_shape.dt.end
+
+            else:
+                start_time, end_time = last_shape.dt.start, last_shape.dt.end
+
+            total_duration_seconds = (end_time - start_time).total_seconds()
+            total_length_meters = haversine_distance_meters(
+                second_last_shape.coordinate,
+                last_shape.coordinate
+            )
+            speed = total_length_meters / total_duration_seconds
+            bearing = bearing_degrees(second_last_shape.coordinate, last_shape.coordinate)
+            extrapolated_point = inverse_haversine_degrees(
+                last_shape.coordinate,
+                bearing,
+                speed * time_traveled.total_seconds()
+            )
+            new_track = Track([
+                GeoPoint(
+                    extrapolated_point,
+                    dt=TimeInterval(end_time, end_time + time_traveled),
+                    properties=last_shape.properties.copy()
+                )
+            ])
+
+        return self.__add__(new_track)
+
     def copy(self):
         """Returns a shallow copy of self"""
         return Track(self.geoshapes.copy())

tests/test_collections.py

@@ -1,5 +1,5 @@
 
-from datetime import datetime, time, timezone
+from datetime import datetime, time, timezone, timedelta
 import os
 import tempfile
 from zipfile import ZipFile
@@ -1065,3 +1065,93 @@ def test_track_intersection():
 
     gbox = GeoBox(Coordinate(0., 2.), Coordinate(2., 0.), dt=datetime(2020, 1, 1, 3))
     assert len(track1.filter_by_intersection(gbox)) == 1
+
+
+@pytest.fixture
+def basic_track():
+    """Fixture to create a simple track with GeoPoints."""
+    points = [
+        GeoPoint(
+            Coordinate(0, 0),
+            dt=TimeInterval(
+                start=datetime(2025, 1, 1, 0, 0, 0),
+                end=datetime(2025, 1, 1, 0, 30, 0)
+            ),
+            properties={"id": "point_0"}
+        ),
+        GeoPoint(
+            Coordinate(1, 1),
+            dt=TimeInterval(
+                start=datetime(2025, 1, 1, 0, 30, 0),
+                end=datetime(2025, 1, 1, 1, 0, 0)
+            ),
+            properties={"id": "point_1"}
+        )
+    ]
+    return Track(points)
+
+@pytest.fixture
+def line_track():
+    """Fixture to create a track with a GeoLineString."""
+    line = GeoLineString(
+        [
+            Coordinate(0, 0),
+            Coordinate(1, 1)
+        ],
+        dt=TimeInterval(
+            start=datetime(2025, 1, 1, 0, 0, 0),
+            end=datetime(2025, 1, 1, 1, 0, 0)
+        ),
+        properties={"id": "line_0"}
+    )
+    return Track([line])
+
+
+def test_extrapolate_geopoint(basic_track):
+    """Test extrapolation from a track ending in a GeoPoint."""
+    track = basic_track
+    extrapolated_time = timedelta(minutes=30)
+
+    result = track.extrapolate(extrapolated_time)
+
+    assert len(result.geoshapes) == 3
+    assert isinstance(result.geoshapes[-1], GeoPoint)
+
+    last_point = result.geoshapes[-1]
+    assert last_point.dt.start == datetime(2025, 1, 1, 1, 0, 0, tzinfo=timezone.utc)
+    assert last_point.dt.end == datetime(2025, 1, 1, 1, 30, 0, tzinfo=timezone.utc)
+    assert "id" in last_point.properties
+
+
+def test_extrapolate_geoline(line_track):
+    """Test extrapolation from a track ending in a GeoLineString."""
+    track = line_track
+    extrapolated_time = timedelta(minutes=30)
+
+    result = track.extrapolate(extrapolated_time)
+
+    assert len(result.geoshapes) == 2
+    assert isinstance(result.geoshapes[-1], GeoLineString)
+
+    last_line = result.geoshapes[-1]
+    assert last_line.dt.start == datetime(2025, 1, 1, 1, 0, 0, tzinfo=timezone.utc)
+    assert last_line.dt.end == datetime(2025, 1, 1, 1, 30, 0, tzinfo=timezone.utc)
+    assert "id" in last_line.properties
+
+
+def test_extrapolate_invalid_shape():
+    """Test extrapolation fails on invalid shape type."""
+    invalid_track = Track(
+        [GeoPolygon([
+            Coordinate(0, 0),
+            Coordinate(0, 1),
+            Coordinate(1, 1),
+            Coordinate(0, 0)
+        ], dt=TimeInterval(
+            datetime(2025, 1, 1),
+            datetime(2025, 1, 1)
+        ))]
+    )
+
+    with pytest.raises(TypeError, match="Can only extrapolate a Points or LineStrings."):
+        invalid_track.extrapolate(timedelta(minutes=30))