chore:revert gpx generate code

run_page/keep_sync.py

@@ -8,14 +8,14 @@
 from datetime import datetime, timedelta
 
 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
+import xml.etree.ElementTree as ET
 
 # need to test
 LOGIN_API = "https://api.gotokeep.com/v1.1/users/login"
@@ -123,9 +123,7 @@ def parse_raw_data_to_nametuple(
                 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
-                )
+                gpx_data = parse_points_to_gpx(run_points_data_gpx, start_time)
                 download_keep_gpx(gpx_data, str(keep_id))
     else:
         print(f"ID {keep_id} no gps data")
@@ -187,7 +185,7 @@ def get_all_keep_tracks(email, password, old_tracks_ids, with_download_gpx=False
     return tracks
 
 
-def parse_points_to_gpx(run_id, run_points_data, start_time):
+def parse_points_to_gpx(run_points_data, start_time):
     """
     Convert run points data to GPX format.
 
@@ -217,18 +215,34 @@ def parse_points_to_gpx(run_id, run_points_data, start_time):
             "hr": point.get("hr"),
         }
         points_dict_list.append(points_dict)
-
-    meta = Metadata(
-        name="Run from Keep",
-        type="Run",
-        desc=f"Run from Keep",
-        link=RUN_LOG_API.format(run_id=run_id),
-    )
-    gpx_data = parse_df_points_to_gpx(meta, DataFrame(points_dict_list))
-    return gpx_data
+    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(
+            latitude=p["latitude"],
+            longitude=p["longitude"],
+            time=p["time"],
+            elevation=p.get("elevation"),
+        )
+        if p.get("hr") is not None:
+            gpx_extension_hr = ET.fromstring(
+                f"""<gpxtpx:TrackPointExtension xmlns:gpxtpx="http://www.garmin.com/xmlschemas/TrackPointExtension/v1">
+                    <gpxtpx:hr>{p["hr"]}</gpxtpx:hr>
+                    </gpxtpx:TrackPointExtension>
+                    """
+            )
+            point.extensions.append(gpx_extension_hr)
+        gpx_segment.points.append(point)
+    return gpx.to_xml()
 
 
-#
 def find_nearest_hr(
     hr_data_list, target_time, start_time, threshold=HR_FRAME_THRESHOLD_IN_DECISECOND
 ):

run_page/utils.py

@@ -4,21 +4,13 @@
 
 import pytz
 
-import gpxpy
-import pandas as pd
-import xml.etree.ElementTree as ET
-import numpy as np
-
 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):
@@ -123,54 +115,3 @@ 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 df_points.loc[idx, col_hr] and not np.isnan(df_points.loc[idx, col_hr]):
-            hr = int(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>{hr}</gpxtpx:hr>
-                </gpxtpx:TrackPointExtension>
-                """
-            )
-            track_point.extensions.append(gpx_extension_hr)
-        gpx_segment.points.append(track_point)
-    return gpx.to_xml()