Merge pull request #3 from grisxa/feat/brevet-top-strava-cleanup

Use distance functions from the brevet-top-numpy-utils

brevet_top_strava/pyproject.toml

@@ -30,6 +30,7 @@ dependencies = [
   "numpy>=1.21.5",
   "numpy-hirschberg==0.1.1",
   "brevet-top-plot-a-route",
+  "brevet-top-numpy-utils==0.1.1",
 ]
 
 [project.urls]

brevet_top_strava/src/brevet_top_strava/__about__.py

@@ -1,4 +1,4 @@
-__version__ = '0.1.3'
+__version__ = '0.1.4'
 __author__ = 'Grigorii Batalov'
 __license__ = 'MIT'
 __description__ = 'Utils for Strava service in the brevet.top'

brevet_top_strava/src/brevet_top_strava/__init__.py

@@ -4,13 +4,13 @@
 
 import numpy as np
 
-from brevet_top_numpy_utils import FloatArray
+from brevet_top_numpy_utils import FloatArray, np_geo_distance_track
 
 from .api import (auth_token, get_activities, get_activity, get_track_points,  # noqa: F401
                   refresh_tokens, TimeWindow, time_window, tokens_expired)  # noqa: F401
 from .build import build_checkpoint_list  # noqa: F401
 from .exceptions import ActivityError, ActivityNotFound, AthleteNotFound  # noqa: F401
-from .math import np_align_track_to_route, np_geo_distance_track
+from .math import np_align_track_to_route
 from .simplify import (clear_stops, cut_off_epilog, cut_off_prolog,
                        down_sample_mask)
 
@@ -39,7 +39,6 @@ def track_alignment(
     draft: FloatArray,
     checkpoints: FloatArray,
 ) -> FloatArray:
-
     logging.info(f"Full track length {len(draft)}")
 
     start = timer()

brevet_top_strava/src/brevet_top_strava/math.py

@@ -1,66 +1,13 @@
-import sys
 from typing import Tuple
 
-import numpy as np
-from numpy import radians, arccos, sin, cos, sum as np_sum, isnan
-
-from brevet_top_numpy_utils import FloatArray
 from numpy_hirschberg import align
 
-DISTANCE_FACTOR = np.float64(0.001)
-EARTH_RADIUS = np.float64(6371e3)
-MAX_POINT_DISTANCE = 3000
-
-
-def np_geo_distance(
-    point: FloatArray,
-    track: FloatArray,
-    factor: np.float64 = DISTANCE_FACTOR,
-) -> FloatArray:
-    point_latitude, point_longitude = radians(point[0:2]).astype(np.float64)
-    latitude_radians, longitude_radians = radians(track.T[0:2]).astype(np.float64)
-
-    distance_shift = abs(track.T[3] - point[3]) * factor
-    return distance_shift + EARTH_RADIUS * arccos(
-        sin(point_latitude) * sin(latitude_radians)
-        + cos(point_latitude)
-        * cos(latitude_radians)
-        * cos(longitude_radians - point_longitude)
-    )
-
+from brevet_top_numpy_utils import FloatArray, np_geo_distance
 
-def np_geo_distance_track(
-    source: FloatArray,
-    target: FloatArray,
-    factor: np.float64 = DISTANCE_FACTOR,
-) -> np.float64:
-    """
-    Compares two tracks point by point and returns a total difference between them.
-    The distance from start is also considered but may be tuned with factor parameter
-    (say, set to 0).
-
-    :param source: the source track or route
-    :param target: the target track
-    :param factor: distance from the start multiplier
-    :return:
-    """
-    source_lat_radians, source_lon_radians = radians(source.T[0:2].astype(float))
-    target_lat_radians, target_lon_radians = radians(target.T[0:2].astype(float))
-
-    distance_shift: FloatArray = abs(target.T[3] - source.T[3]) * factor
-    difference: FloatArray = distance_shift + EARTH_RADIUS * arccos(
-        sin(source_lat_radians) * sin(target_lat_radians)
-        + cos(source_lat_radians)
-        * cos(target_lat_radians)
-        * cos(target_lon_radians - source_lon_radians)
-    )
-    np.nan_to_num(difference, copy=False, nan=MAX_POINT_DISTANCE)
-    return np_sum(difference[~isnan(difference)])
+MAX_POINT_DISTANCE = 3000
 
 
-def np_align_track_to_route(
-    route: FloatArray, track: FloatArray
-) -> Tuple[float, FloatArray]:
+def np_align_track_to_route(route: FloatArray, track: FloatArray) -> Tuple[float, FloatArray]:
     """
     Compare route and track sequences.
 

brevet_top_strava/src/brevet_top_strava/simplify.py

@@ -1,9 +1,6 @@
-from typing import Tuple
-
 import numpy as np
 
-from brevet_top_numpy_utils import FloatArray
-from .math import np_geo_distance
+from brevet_top_numpy_utils import FloatArray, np_geo_distance
 
 # from plot_a_route import geo_distance
 
@@ -22,13 +19,11 @@ def clear_stops(track: FloatArray, checkpoints: FloatArray) -> FloatArray:
     """
     mask = [True] * track.shape[0]
     for cp in checkpoints:
-        mask = mask & (np_geo_distance(cp, track, factor=0) > CHECKPOINT_RADIUS)
+        mask = mask & (np_geo_distance(cp, track, factor=np.float64(0.0)) > CHECKPOINT_RADIUS)
     return track[mask]
 
 
-def cut_off_epilog(
-    track: FloatArray, end: Tuple[float, float, float, float]
-) -> FloatArray:
+def cut_off_epilog(track: FloatArray, end: FloatArray) -> FloatArray:
     """
     Remove points after the last route point.
 
@@ -40,15 +35,11 @@ def cut_off_epilog(
         return track
     backwards: FloatArray = np.flipud(track)
     # index of the nearest point to the finish
-    offset: int = np.argmax(  # type: ignore[assignment]
-        np_geo_distance(end, backwards) < CHECKPOINT_RADIUS
-    )
+    offset: int = np.argmax(np_geo_distance(end, backwards) < CHECKPOINT_RADIUS)  # type: ignore[assignment]
     return np.flipud(backwards[offset:])
 
 
-def cut_off_prolog(
-    track: FloatArray, start: Tuple[float, float, float, float]
-) -> FloatArray:
+def cut_off_prolog(track: FloatArray, start: FloatArray) -> FloatArray:
     """
     Remove points before the start route point.
 
@@ -59,32 +50,14 @@ def cut_off_prolog(
     if len(track) < 2:
         return track
     # index of the nearest point to the start
-    offset: int = np.argmax(  # type: ignore[assignment]
-        np_geo_distance(start, track) < CHECKPOINT_RADIUS
-    )
+    offset: int = np.argmax(np_geo_distance(start, track) < CHECKPOINT_RADIUS)  # type: ignore[assignment]
     # how many meters before the start
     prolog: np.float64 = track[offset][3]
     # decrease the distance column
     track[offset:, 3] = track[offset:, 3] - prolog
     return track[offset:]
 
 
-"""
-def down_sample_mask_pure(track: FloatArray) -> List[bool]:
-    mask = [True] + [False] * (track.shape[0] - 1)
-    i = 0
-    while i < track.shape[0] - 1:
-        j = i
-        while j < track.shape[0] - 1:
-            j += 1
-            if geo_distance(*track[i][0:2], *track[j][0:2]) > DOWN_SAMPLE_INTERVAL:
-                mask[j] = True
-                break
-        i = j
-    return mask
-"""
-
-
 def down_sample_mask(
     track: FloatArray,
     ahead: int = LOOKUP_AHEAD_POINTS,
@@ -106,13 +79,9 @@ def down_sample_mask(
         # accept the current point
         mask[i] = True
         # build a distance vector ahead : [ i+1, ... i+ahead ]
-        distance: FloatArray = np_geo_distance(
-            track[i], track[i + 1 : i + ahead + 1]  # noqa: E203
-        )
+        distance: FloatArray = np_geo_distance(track[i], track[i + 1 : i + ahead + 1])  # noqa: E203
         # find the first point far enough
-        offset: int = np.argmax(  # type: ignore[assignment]
-            ~np.isnan(distance) & (distance > interval)
-        )
+        offset: int = np.argmax(~np.isnan(distance) & (distance > interval))  # type: ignore[assignment]
         if offset == 0:
             # the first point is good
             if not np.isnan(distance[0]) and distance[0] > interval:

brevet_top_strava/tests/test_track_n_route.py

@@ -6,41 +6,24 @@
 import pytest
 from matplotlib import pyplot as plt
 
-from brevet_top_numpy_utils import FloatArray
-from brevet_top_strava import (
-    cut_off_prolog,
-    cut_off_epilog,
-    np_align_track_to_route,
-    clear_stops,
-)
-from brevet_top_strava.math import (
-    np_geo_distance_track,
-    DISTANCE_FACTOR,
-)
+from brevet_top_numpy_utils import (DISTANCE_FACTOR, FloatArray,
+                                    np_geo_distance_track)
+from brevet_top_strava import (clear_stops, cut_off_epilog, cut_off_prolog,
+                               np_align_track_to_route)
 from brevet_top_strava.simplify import down_sample_mask
 
 
 @pytest.fixture
 def route() -> FloatArray:
-    file_path = (
-        pathlib.Path(__file__).parent.absolute()
-        / "files"
-        / "track_n_route"
-        / "route.csv"
-    )
+    file_path = pathlib.Path(__file__).parent.absolute() / "files" / "track_n_route" / "route.csv"
     with open(file_path, newline="", encoding="utf-8") as csv_file:
         table = csv.reader(csv_file, delimiter=",", quotechar='"')
         return np.array([row for row in table], dtype=np.float64)
 
 
 @pytest.fixture
 def track() -> FloatArray:
-    file_path = (
-        pathlib.Path(__file__).parent.absolute()
-        / "files"
-        / "track_n_route"
-        / "track.csv"
-    )
+    file_path = pathlib.Path(__file__).parent.absolute() / "files" / "track_n_route" / "track.csv"
     with open(file_path, newline="", encoding="utf-8") as csv_file:
         table = csv.reader(csv_file, delimiter=",", quotechar='"')
         return np.array([row for row in table], dtype=np.float64)
@@ -73,22 +56,12 @@ def checkpoints() -> FloatArray:
 
 @pytest.fixture
 def csv_file():
-    file_path = (
-        pathlib.Path(__file__).parent.absolute()
-        / "files"
-        / "track_n_route"
-        / "reduced.csv"
-    )
+    file_path = pathlib.Path(__file__).parent.absolute() / "files" / "track_n_route" / "reduced.csv"
     with open(file_path, "a", newline="", encoding="utf-8") as csv_file_handler:
-        yield csv.writer(
-            csv_file_handler, delimiter=",", quotechar='"', quoting=csv.QUOTE_MINIMAL
-        )
-
+        yield csv.writer(csv_file_handler, delimiter=",", quotechar='"', quoting=csv.QUOTE_MINIMAL)
 
-def test_track_n_route(
-    route: FloatArray, track: FloatArray, checkpoints: FloatArray  # , csv_file
-):
 
+def test_track_n_route(route: FloatArray, track: FloatArray, checkpoints: FloatArray):  # , csv_file
     assert len(route) == 92
     assert len(track) == 13176
     assert len(checkpoints) == 16

strava_watcher/main.py

@@ -13,12 +13,10 @@
 from pytz import utc
 from requests import HTTPError
 
-from brevet_top_gcp_utils import (create_document, firestore_to_track_point,
-                                  get_checkpoints)
-from brevet_top_strava import (ActivityError, ActivityNotFound,
-                               AthleteNotFound, auth_token,
-                               build_checkpoint_list, get_activity,
-                               refresh_tokens, tokens_expired, track_alignment)
+from brevet_top_gcp_utils import create_document, firestore_to_track_point, get_checkpoints
+from brevet_top_numpy_utils import FloatArray
+from brevet_top_strava import (ActivityError, ActivityNotFound, AthleteNotFound, auth_token, build_checkpoint_list,
+                               get_activity, get_track_points, refresh_tokens, tokens_expired, track_alignment)
 
 log_client = google.cloud.logging.Client()
 log_client.get_default_handler()
@@ -143,8 +141,10 @@ def strava_compare(athlete_id: int, activity_id: int, secret: dict):
 
         # start searching
         try:
+            # retrieve activities and transform to a track
+            track: FloatArray = get_track_points(sorted([activity], key=lambda a: a["start_date"]), auth_token(riders[0]["strava"]))
             points = track_alignment(
-                brevet_dict, riders[0]["strava"], [activity], checkpoints
+                brevet_dict, track, checkpoints
             )
         except (ActivityNotFound, ActivityError) as error:
             logging.error(f"Activity error {error}")

strava_watcher/requirements.txt

@@ -9,4 +9,4 @@ pytz==2021.3
 rdp==0.8
 numpy>=1.21.5
 brevet-top-gcp-utils==0.1.4
-brevet-top-strava==0.1.1
+brevet-top-strava==0.1.4