feat:suport heart_rate in export gpx file

run_page/keep_sync.py

@@ -1,21 +1,22 @@
 import argparse
 import base64
 import json
-import math
 import os
 import time
 import zlib
 from collections import namedtuple
 from datetime import datetime, timedelta
+from Crypto.Cipher import AES
 
 import eviltransform
 import gpxpy
 import polyline
 import requests
 from config import GPX_FOLDER, JSON_FILE, SQL_FILE, run_map, start_point
-from Crypto.Cipher import AES
 from generator import Generator
 from utils import adjust_time
+from utils import parse_df_points_to_gpx, Metadata
+from pandas import DataFrame
 
 # need to test
 LOGIN_API = "https://api.gotokeep.com/v1.1/users/login"
@@ -88,6 +89,17 @@ def parse_raw_data_to_nametuple(
     keep_id = run_data["id"].split("_")[1]
 
     start_time = run_data["startTime"]
+    avg_heart_rate = None
+    decoded_hr_data = []
+    if run_data["heartRate"]:
+        avg_heart_rate = run_data["heartRate"].get("averageHeartRate", None)
+        heart_rate_data = run_data["heartRate"].get("heartRates", None)
+        if heart_rate_data is not None:
+            decoded_hr_data = decode_runmap_data(heart_rate_data)
+        # fix #66
+        if avg_heart_rate and avg_heart_rate < 0:
+            avg_heart_rate = None
+
     if run_data["geoPoints"]:
         run_points_data = decode_runmap_data(run_data["geoPoints"], True)
         run_points_data_gpx = run_points_data
@@ -99,20 +111,22 @@ def parse_raw_data_to_nametuple(
             for i, p in enumerate(run_points_data_gpx):
                 p["latitude"] = run_points_data[i][0]
                 p["longitude"] = run_points_data[i][1]
+                p_hr_data = find_nearest_hr_data(
+                    decoded_hr_data, int(p["timestamp"]), start_time
+                )
+                if p_hr_data is not None:
+                    p["hr"] = p_hr_data["beatsPerMinute"]
         else:
             run_points_data = [[p["latitude"], p["longitude"]] for p in run_points_data]
         if with_download_gpx:
-            if str(keep_id) not in old_gpx_ids:
-                gpx_data = parse_points_to_gpx(run_points_data_gpx, start_time)
+            if (
+                str(keep_id) not in old_gpx_ids
+                and run_data["dataType"] == "outdoorRunning"
+            ):
+                gpx_data = parse_points_to_gpx(run_data["id"], run_points_data_gpx, start_time)
                 download_keep_gpx(gpx_data, str(keep_id))
     else:
         print(f"ID {keep_id} no gps data")
-    heart_rate = None
-    if run_data["heartRate"]:
-        heart_rate = run_data["heartRate"].get("averageHeartRate", None)
-        # fix #66
-        if heart_rate and heart_rate < 0:
-            heart_rate = None
     polyline_str = polyline.encode(run_points_data) if run_points_data else ""
     start_latlng = start_point(*run_points_data[0]) if run_points_data else None
     start_date = datetime.utcfromtimestamp(start_time / 1000)
@@ -133,7 +147,7 @@ def parse_raw_data_to_nametuple(
         "start_date_local": datetime.strftime(start_date_local, "%Y-%m-%d %H:%M:%S"),
         "end_local": datetime.strftime(end_local, "%Y-%m-%d %H:%M:%S"),
         "length": run_data["distance"],
-        "average_heartrate": int(heart_rate) if heart_rate else None,
+        "average_heartrate": int(avg_heart_rate) if avg_heart_rate else None,
         "map": run_map(polyline_str),
         "start_latlng": start_latlng,
         "distance": run_data["distance"],
@@ -143,6 +157,7 @@ def parse_raw_data_to_nametuple(
         ),
         "average_speed": run_data["distance"] / run_data["duration"],
         "location_country": str(run_data.get("region", "")),
+        "source": "Keep",
     }
     return namedtuple("x", d.keys())(*d.values())
 
@@ -171,34 +186,55 @@ def get_all_keep_tracks(email, password, old_tracks_ids, with_download_gpx=False
     return tracks
 
 
-def parse_points_to_gpx(run_points_data, start_time):
-    # future to support heart rate
+def parse_points_to_gpx(run_id, run_points_data, start_time):
     points_dict_list = []
+    # early timestamp fields in keep's data stands for delta time, but in newly data timestamp field stands for exactly time,
+    # so it does'nt need to plus extra start_time
+    # the 3_600_000 stands for 100 hours sports time. 100h = 100 * 60 * 60 * 10
+    if run_points_data[0]["timestamp"] > 3_600_000:
+        start_time = 0
+
     for point in run_points_data:
         points_dict = {
             "latitude": point["latitude"],
             "longitude": point["longitude"],
             "time": datetime.utcfromtimestamp(
                 (point["timestamp"] * 100 + start_time) / 1000
             ),
+            "elevation": point.get("verticalAccuracy"),
+            "hr": point.get("hr"),
         }
-        if "verticalAccuracy" in point:
-            points_dict["elevation"] = point["verticalAccuracy"]
         points_dict_list.append(points_dict)
-    gpx = gpxpy.gpx.GPX()
-    gpx.nsmap["gpxtpx"] = "http://www.garmin.com/xmlschemas/TrackPointExtension/v1"
-    gpx_track = gpxpy.gpx.GPXTrack()
-    gpx_track.name = "gpx from keep"
-    gpx.tracks.append(gpx_track)
-
-    # Create first segment in our GPX track:
-    gpx_segment = gpxpy.gpx.GPXTrackSegment()
-    gpx_track.segments.append(gpx_segment)
-    for p in points_dict_list:
-        point = gpxpy.gpx.GPXTrackPoint(**p)
-        gpx_segment.points.append(point)
-
-    return gpx.to_xml()
+
+    meta = Metadata(
+        name="Run from Keep",
+        type="Run",
+        desc=f"Run from Keep",
+        link=RUN_LOG_API.format(run_id=run_id),
+    )
+    df_points = DataFrame(points_dict_list)
+    gpx_data = parse_df_points_to_gpx(meta, df_points)
+    return gpx_data
+
+
+# if cannot found suitable HR data within the specified time frame (within 10 seconds by default), there will be no hr data return
+def find_nearest_hr_data(hr_data_list, target_timestamp, start_time, threshold=1000):
+    closest_element = None
+    # init difference value
+    min_difference = float("inf")
+    delta_time = target_timestamp
+    if target_timestamp > 3_600_000:
+        delta_time = (target_timestamp * 100 - start_time) / 100
+
+    for item in hr_data_list:
+        timestamp = item["timestamp"]
+        difference = abs(timestamp - delta_time)
+
+        if difference <= threshold and difference < min_difference:
+            closest_element = item
+            min_difference = difference
+
+    return closest_element
 
 
 def download_keep_gpx(gpx_data, keep_id):

run_page/nike_sync.py

@@ -22,6 +22,8 @@
 )
 from generator import Generator
 from utils import adjust_time, make_activities_file
+from pandas import DataFrame
+from utils import parse_df_points_to_gpx, Metadata
 
 # logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger("nike_sync")
@@ -248,24 +250,10 @@ def update_points(points, update_data, update_name):
     if heart_rate_data:
         update_points(points_dict_list, heart_rate_data, "heart_rate")
 
-    for p in points_dict_list:
-        # delete useless attr
-        del p["start_time"]
-        if p.get("heart_rate") is None:
-            point = gpxpy.gpx.GPXTrackPoint(**p)
-        else:
-            heart_rate_num = p.pop("heart_rate")
-            point = gpxpy.gpx.GPXTrackPoint(**p)
-            gpx_extension_hr = ElementTree.fromstring(
-                f"""<gpxtpx:TrackPointExtension xmlns:gpxtpx="http://www.garmin.com/xmlschemas/TrackPointExtension/v1">
-                <gpxtpx:hr>{heart_rate_num}</gpxtpx:hr>
-                </gpxtpx:TrackPointExtension>
-            """
-            )
-            point.extensions.append(gpx_extension_hr)
-        gpx_segment.points.append(point)
-
-    return gpx.to_xml()
+    meta = Metadata(name=title, type="Run", desc="Run from NRC")
+    df_points = DataFrame(points_dict_list)
+    gpx_doc = parse_df_points_to_gpx(meta, df_points, col_hr="heart_rate")
+    return gpx_doc
 
 
 def parse_activity_data(activity):

run_page/utils.py

@@ -4,13 +4,20 @@
 
 import pytz
 
+import gpxpy
+import pandas as pd
+import xml.etree.ElementTree as ET
+
 try:
     from rich import print
 except:
     pass
 from generator import Generator
 from stravalib.client import Client
 from stravalib.exc import RateLimitExceeded
+from collections import namedtuple
+
+Metadata = namedtuple("Metadata", ("name", "type", "desc", "link"))
 
 
 def adjust_time(time, tz_name):
@@ -115,3 +122,53 @@ def upload_file_to_strava(client, file_name, data_type, force_to_run=True):
         print(
             f"Uploading {data_type} file: {file_name} to strava, upload_id: {r.upload_id}."
         )
+
+
+def parse_df_points_to_gpx(
+    metadata,
+    df_points,
+    col_lat="latitude",
+    col_long="longitude",
+    col_time="time",
+    col_ele="elevation",
+    col_hr="hr",
+):
+    """
+    Convert a pandas dataframe to gpx
+    Parameters:
+        metadata(Metadata): metadata of the gpx track
+        df_points (pd.DataFrame): pandas dataframe containing at minimum lat,long,time info of points
+    """
+    gpx = gpxpy.gpx.GPX()
+    gpx.nsmap["gpxtpx"] = "http://www.garmin.com/xmlschemas/TrackPointExtension/v1"
+    gpx_track = gpxpy.gpx.GPXTrack()
+    gpx.tracks.append(gpx_track)
+
+    gpx_segment = gpxpy.gpx.GPXTrackSegment()
+    gpx_track.segments.append(gpx_segment)
+
+    # support multi tracks in the future
+    gpx.tracks[0].name = metadata.name
+    gpx.tracks[0].type = metadata.type
+    gpx.tracks[0].description = metadata.desc
+    gpx.tracks[0].link = metadata.link
+
+    for idx in df_points.index:
+        track_point = gpxpy.gpx.GPXTrackPoint(
+            latitude=df_points.loc[idx, col_lat],
+            longitude=df_points.loc[idx, col_long],
+            time=pd.Timestamp(df_points.loc[idx, col_time]),
+            elevation=df_points.loc[idx, col_ele] if col_ele else None,
+        )
+
+        # gpx extensions
+        if col_hr and df_points.loc[idx, col_hr]:
+            gpx_extension_hr = ET.fromstring(
+                f"""<gpxtpx:TrackPointExtension xmlns:gpxtpx="http://www.garmin.com/xmlschemas/TrackPointExtension/v1">
+                <gpxtpx:hr>{df_points.loc[idx, col_hr]}</gpxtpx:hr>
+                </gpxtpx:TrackPointExtension>
+                """
+            )
+            track_point.extensions.append(gpx_extension_hr)
+        gpx_segment.points.append(track_point)
+    return gpx.to_xml()