diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..a85ebeb --- /dev/null +++ b/.gitignore @@ -0,0 +1,2 @@ +*__pycache__/* +test/__pycache__/ \ No newline at end of file diff --git a/geostructures/collections.py b/geostructures/collections.py index f3a4f99..0280800 100644 --- a/geostructures/collections.py +++ b/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()) diff --git a/tests/test_collections.py b/tests/test_collections.py index e6e7b00..b952d1e 100644 --- a/tests/test_collections.py +++ b/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))