implement climb detection

1 files changed, 101 insertions, 0 deletions

diff --git a/fietsboek/geo.py b/fietsboek/geo.py
index e6abb71..e7a58a9 100644
--- a/fietsboek/geo.py
+++ b/fietsboek/geo.py
@@ -5,6 +5,7 @@ import io
 from dataclasses import dataclass
 from itertools import islice
 from math import cos, radians, sin, sqrt
+from typing import Iterator
 
 from . import util
 
@@ -17,6 +18,15 @@ EARTH_RADIUS = 6378137.0
 MOVING_THRESHOLD = 1.1
 """Speed which is considered to be the moving threshold, in m/s."""
 
+CLIMB_MIN_LENGTH = 500
+"""Minimum length (in m) for an ascent to be considered a climb."""
+
+CLIMB_MIN_ASCENSION = 0.03
+"""Minimum relative ascension for a segment to be considered a climb."""
+
+CLIMB_MERGE_THRESHOLD = 5
+"""Window for when climbs are merged (in m)."""
+
 
 @dataclass
 class Waypoint:
@@ -120,6 +130,10 @@ class Point:
             return 0.0
         return sqrt(radicand)
 
+    def latlon(self) -> tuple[float, float]:
+        """Returns the latitude and longitude of this point as a tuple."""
+        return (self.latitude, self.longitude)
+
 
 class Path:
     """A GPS path, that is a series of GPS points."""
@@ -162,6 +176,93 @@ class Path:
             movement_data.average_speed = 0.0
         return movement_data
 
+    def climbs(self) -> Iterator[tuple[int, int]]:
+        """Returns an iterator over all climbs of this path.
+
+        :return: An iterator over the climbs, represented by the starting index
+                 and ending index of points included in the climb.
+        """
+        current_climb = None
+        for climb in self._climbs():
+            if current_climb is None:
+                current_climb = climb
+            else:
+                path_between = Path([self.points[i] for i in range(current_climb[1], climb[0])])
+                if path_between.movement_data().length > CLIMB_MERGE_THRESHOLD:
+                    yield current_climb
+                    current_climb = climb
+                else:
+                    current_climb = (current_climb[0], climb[1])
+        if current_climb is not None:
+            yield current_climb
+
+    def _climbs(self):
+        # Detection of climbs:
+        # We use a single pass over the track with two pointers, the "leader"
+        # and the "chaser". The chaser always points behind the leader. In a
+        # loop, one of each is always advanced:
+        #
+        # If the distance between leader and chaser is less than the minimum
+        # climb length, we advance the leader to enlarge the window. This
+        # does not constitute a climb.
+        #
+        # If the current ascension is below the ascension threshold, we advance
+        # the chaser. This moves the window to advance the search. If the
+        # ascension was above the threshold before the last advancement, we emit
+        # the climb (which is now over).
+        #
+        # If the current ascension is above the threshold, we advance the
+        # leader. This enlarges the window while constituting a climb.
+        try:
+            front = enumerate(self.points)
+            li, lead = next(front)
+            back = enumerate(self.points)
+            ci, chaser = next(back)
+        except StopIteration:
+            # Happens if there are no points
+            return
+
+        length = 0.0
+        elevation_diff = 0.0
+        climb_start = None
+
+        try:
+            while True:
+                if length < CLIMB_MIN_LENGTH:
+                    old_point = lead
+                    li, lead = next(front)
+                    length += old_point.distance(lead)
+                    elevation_diff += lead.elevation - old_point.elevation
+                    continue
+
+                ascension = elevation_diff / length
+                if ascension < CLIMB_MIN_ASCENSION:
+                    if climb_start is not None:
+                        yield (climb_start, li)
+                        climb_start = None
+                        # Skip to the end of the just emitted climb
+                        while ci != li:
+                            old_point = chaser
+                            ci, chaser = next(back)
+                            length -= old_point.distance(chaser)
+                            elevation_diff -= chaser.elevation - old_point.elevation
+                    else:
+                        old_point = chaser
+                        ci, chaser = next(back)
+                        length -= old_point.distance(chaser)
+                        elevation_diff -= chaser.elevation - old_point.elevation
+                else:
+                    if climb_start is None:
+                        climb_start = ci
+                    # See how much farther we can extend this climb
+                    old_point = lead
+                    li, lead = next(front)
+                    length += old_point.distance(lead)
+                    elevation_diff += lead.elevation - old_point.elevation
+        except StopIteration:
+            if climb_start is not None:
+                yield (climb_start, li)
+
 
 def gpx_xml(
     title: str | None,