diff --git a/.github/workflows/gh-pages.yml b/.github/workflows/gh-pages.yml
index c2ae7ed79ca..1c1db5eada6 100644
--- a/.github/workflows/gh-pages.yml
+++ b/.github/workflows/gh-pages.yml
@@ -75,7 +75,7 @@ jobs:
with:
node-version: '20'
- - uses: pnpm/action-setup@v2
+ - uses: pnpm/action-setup@v3
name: Install pnpm
with:
version: 8
@@ -100,10 +100,10 @@ jobs:
run: PATH_PREFIX=/${{ github.event.repository.name }} pnpm build
- name: Upload artifact
- uses: actions/upload-pages-artifact@v2
+ uses: actions/upload-pages-artifact@v3
with:
# Upload dist repository
path: './dist'
- name: Deploy to GitHub Pages
id: deployment
- uses: actions/deploy-pages@v3
+ uses: actions/deploy-pages@v4
diff --git a/.github/workflows/run_data_sync.yml b/.github/workflows/run_data_sync.yml
index 7d595359573..52919bcb610 100644
--- a/.github/workflows/run_data_sync.yml
+++ b/.github/workflows/run_data_sync.yml
@@ -17,13 +17,15 @@ on:
- run_page/keep_sync.py
- run_page/gpx_sync.py
- run_page/tcx_sync.py
+ - run_page/tcx_to_garmin_sync.py
- run_page/garmin_to_strava_sync.py
- run_page/keep_to_strava_sync.py
+ - run_page/oppo_sync.py
- requirements.txt
env:
# please change to your own config.
- RUN_TYPE: pass # support strava/nike/garmin/coros/garmin_cn/garmin_sync_cn_global/keep/only_gpx/only_fit/nike_to_strava/strava_to_garmin/strava_to_garmin_cn/garmin_to_strava/garmin_to_strava_cn/codoon, Please change the 'pass' it to your own
+ RUN_TYPE: pass # support strava/nike/garmin/coros/garmin_cn/garmin_sync_cn_global/keep/only_gpx/only_fit/nike_to_strava/strava_to_garmin/tcx_to_garmin/strava_to_garmin_cn/garmin_to_strava/garmin_to_strava_cn/codoon/oppo, Please change the 'pass' it to your own
ATHLETE: ben_29
TITLE: Workouts
MIN_GRID_DISTANCE: 10 # change min distance here
@@ -121,6 +123,12 @@ jobs:
run: |
python run_page/codoon_sync.py ${{ secrets.CODOON_MOBILE }} ${{ secrets.CODOON_PASSWORD }}
+ - name: Run sync tcx to Garmin script
+ if: env.RUN_TYPE == 'tcx_to_garmin'
+ run: |
+ # python run_page/tcx_to_garmin_sync.py ${{ secrets.GARMIN_SECRET_STRING }}
+ python run_page/tcx_to_garmin_sync.py ${{ secrets.GARMIN_SECRET_STRING_CN }} --is-cn
+
# for garmin if you want generate `tcx` you can add --tcx command in the args.
- name: Run sync Garmin script
if: env.RUN_TYPE == 'garmin'
@@ -187,6 +195,13 @@ jobs:
run: |
python run_page/tulipsport_sync.py ${{ secrets.TULIPSPORT_TOKEN }} --with-gpx
+ - name: Run sync Oppo heytap script, note currently this script is not worked
+ if: env.RUN_TYPE == 'oppo'
+ run: |
+ python run_page/oppo_sync.py ${{ secrets.OPPO_ID }} ${{ secrets.OPPO_CLIENT_SECRET }} ${{ secrets.OPPO_CLIENT_REFRESH_TOKEN }} --with-tcx
+ # If you want to sync fit activity in gpx format, please consider the following script:
+ # python run_page/oppo_sync.py ${{ secrets.OPPO_ID }} ${{ secrets.OPPO_CLIENT_SECRET }} ${{ secrets.OPPO_CLIENT_REFRESH_TOKEN }} --with-gpx
+
- name: Make svg GitHub profile
if: env.RUN_TYPE != 'pass'
run: |
diff --git a/assets/github_2024.svg b/assets/github_2024.svg
index 441af32ea60..25dc1d0e57d 100644
--- a/assets/github_2024.svg
+++ b/assets/github_2024.svg
@@ -1,2 +1,2 @@
-
\ No newline at end of file
+
\ No newline at end of file
diff --git a/assets/year_2024.svg b/assets/year_2024.svg
index 18ca7218da7..1d476eff514 100644
--- a/assets/year_2024.svg
+++ b/assets/year_2024.svg
@@ -1,2 +1,2 @@
-
\ No newline at end of file
+
\ No newline at end of file
diff --git a/run_page/codoon_sync.py b/run_page/codoon_sync.py
index 80a3a33b162..e0b1020c2ed 100755
--- a/run_page/codoon_sync.py
+++ b/run_page/codoon_sync.py
@@ -9,6 +9,7 @@
import xml.etree.ElementTree as ET
from collections import namedtuple
from datetime import datetime, timedelta
+from xml.dom import minidom
import eviltransform
import gpxpy
@@ -45,6 +46,8 @@
# device info
user_agent = "CodoonSport(8.9.0 1170;Android 7;Sony XZ1)"
did = "24-00000000-03e1-7dd7-0033-c5870033c588"
+# May be Forerunner 945?
+CONNECT_API_PART_NUMBER = "006-D2449-00"
# fixed params
base_url = "https://api.codoon.com"
@@ -61,9 +64,9 @@
# for tcx type
TCX_TYPE_DICT = {
- 0: "Hike",
+ 0: "Hiking",
1: "Running",
- 2: "Ride",
+ 2: "Biking",
}
# only for running sports, if you want others, please change the True to False
@@ -127,6 +130,9 @@ def formated_input(
def tcx_output(fit_array, run_data):
+ """
+ If you want to make a more detailed tcx file, please refer to oppo_sync.py
+ """
# route ID
fit_id = str(run_data["id"])
# local time
@@ -149,7 +155,7 @@ def tcx_output(fit_array, run_data):
},
)
# xml tree
- tree = ET.ElementTree(training_center_database)
+ ET.ElementTree(training_center_database)
# Activities
activities = ET.Element("Activities")
training_center_database.append(activities)
@@ -163,12 +169,15 @@ def tcx_output(fit_array, run_data):
activity_id.text = fit_start_time # Codoon use start_time as ID
activity.append(activity_id)
# Creator
- activity_creator = ET.Element("Creator")
+ activity_creator = ET.Element("Creator", {"xsi:type": "Device_t"})
activity.append(activity_creator)
# Name
activity_creator_name = ET.Element("Name")
- activity_creator_name.text = "咕咚"
+ activity_creator_name.text = "Codoon"
activity_creator.append(activity_creator_name)
+ activity_creator_product = ET.Element("ProductID")
+ activity_creator_product.text = "3441"
+ activity_creator.append(activity_creator_product)
# Lap
activity_lap = ET.Element("Lap", {"StartTime": fit_start_time})
activity.append(activity_lap)
@@ -215,11 +224,22 @@ def tcx_output(fit_array, run_data):
altitude_meters = ET.Element("AltitudeMeters")
altitude_meters.text = bytes.decode(i["elevation"])
tp.append(altitude_meters)
-
+ # Author
+ author = ET.Element("Author", {"xsi:type": "Application_t"})
+ training_center_database.append(author)
+ author_name = ET.Element("Name")
+ author_name.text = "Connect Api"
+ author.append(author_name)
+ author_lang = ET.Element("LangID")
+ author_lang.text = "en"
+ author.append(author_lang)
+ author_part = ET.Element("PartNumber")
+ author_part.text = CONNECT_API_PART_NUMBER
+ author.append(author_part)
# write to TCX file
- tree.write(
- TCX_FOLDER + "/" + fit_id + ".tcx", encoding="utf-8", xml_declaration=True
- )
+ xml_str = minidom.parseString(ET.tostring(training_center_database)).toprettyxml()
+ with open(TCX_FOLDER + "/" + fit_id + ".tcx", "w") as f:
+ f.write(str(xml_str))
# TODO time complexity is too heigh, need to be reduced
diff --git a/run_page/config.py b/run_page/config.py
index daac91c752b..e5befaa2f42 100644
--- a/run_page/config.py
+++ b/run_page/config.py
@@ -26,7 +26,7 @@
BASE_TIMEZONE = "Asia/Shanghai"
-
+UTC_TIMEZONE = "UTC"
start_point = namedtuple("start_point", "lat lon")
run_map = namedtuple("polyline", "summary_polyline")
diff --git a/run_page/generator/__init__.py b/run_page/generator/__init__.py
index 381e82edc57..de66d2e70fd 100644
--- a/run_page/generator/__init__.py
+++ b/run_page/generator/__init__.py
@@ -84,13 +84,8 @@ def sync_from_data_dir(self, data_dir, file_suffix="gpx"):
return
synced_files = []
- if file_suffix == "fit":
- name_mapping = load_fit_name_mapping()
for t in tracks:
- activity_id = t.file_names[0].split(".")[0]
- if file_suffix == "fit" and activity_id in name_mapping:
- t.name = name_mapping[activity_id]
created = update_or_create_activity(self.session, t.to_namedtuple())
if created:
sys.stdout.write("+")
@@ -204,3 +199,16 @@ def get_old_tracks_ids(self):
# pass the error
print(f"something wrong with {str(e)}")
return []
+
+ def get_old_tracks_dates(self):
+ try:
+ activities = (
+ self.session.query(Activity)
+ .order_by(Activity.start_date_local.desc())
+ .all()
+ )
+ return [str(a.start_date_local) for a in activities]
+ except Exception as e:
+ # pass the error
+ print(f"something wrong with {str(e)}")
+ return []
diff --git a/run_page/gpxtrackposter/track.py b/run_page/gpxtrackposter/track.py
index a79a1358635..3df75e100e2 100644
--- a/run_page/gpxtrackposter/track.py
+++ b/run_page/gpxtrackposter/track.py
@@ -64,7 +64,7 @@ def load_gpx(self, file_name):
# (for example, treadmill runs pulled via garmin-connect-export)
if os.path.getsize(file_name) == 0:
raise TrackLoadError("Empty GPX file")
- with open(file_name, "rb") as file:
+ with open(file_name, "r", encoding="utf-8", errors="ignore") as file:
self._load_gpx_data(mod_gpxpy.parse(file))
except Exception as e:
print(
@@ -291,10 +291,6 @@ def _load_fit_data(self, fit: dict):
lng = record["position_long"] / SEMICIRCLE
_polylines.append(s2.LatLng.from_degrees(lat, lng))
self.polyline_container.append([lat, lng])
- for record in fit["device_info_mesgs"]:
- if "device_index" in record and record["device_index"] == "creator":
- self.source = f'{record["manufacturer"]} {record["garmin_product"]} fit'
- break
if self.polyline_container:
self.start_time_local, self.end_time_local = parse_datetime_to_local(
self.start_time, self.end_time, self.polyline_container[0]
diff --git a/run_page/joyrun_sync.py b/run_page/joyrun_sync.py
index 44abe025581..6b8093bab13 100644
--- a/run_page/joyrun_sync.py
+++ b/run_page/joyrun_sync.py
@@ -2,6 +2,8 @@
import argparse
import json
import os
+import subprocess
+import sys
import time
from collections import namedtuple
from datetime import datetime, timedelta
@@ -337,12 +339,95 @@ def get_all_joyrun_tracks(self, old_tracks_ids, with_gpx=False):
return tracks
+def _generate_svg_profile(athlete, min_grid_distance):
+ # To generate svg for 'Total' in the left-up map
+ if not athlete:
+ # Skip to avoid override
+ print("Skipping gen_svg. Fill your name with --athlete if you don't want skip")
+ return
+ print(
+ f"Running scripts for [Make svg GitHub profile] with athlete={athlete} min_grid_distance={min_grid_distance}"
+ )
+ cmd_args_list = [
+ [
+ sys.executable,
+ "run_page/gen_svg.py",
+ "--from-db",
+ "--title",
+ f"{athlete} Running",
+ "--type",
+ "github",
+ "--athlete",
+ athlete,
+ "--special-distance",
+ "10",
+ "--special-distance2",
+ "20",
+ "--special-color",
+ "yellow",
+ "--special-color2",
+ "red",
+ "--output",
+ "assets/github.svg",
+ "--use-localtime",
+ "--min-distance",
+ "0.5",
+ ],
+ [
+ sys.executable,
+ "run_page/gen_svg.py",
+ "--from-db",
+ "--title",
+ f"Over {min_grid_distance} Running",
+ "--type",
+ "grid",
+ "--athlete",
+ athlete,
+ "--special-distance",
+ "20",
+ "--special-distance2",
+ "40",
+ "--special-color",
+ "yellow",
+ "--special-color2",
+ "red",
+ "--output",
+ "assets/grid.svg",
+ "--use-localtime",
+ "--min-distance",
+ str(min_grid_distance),
+ ],
+ [
+ sys.executable,
+ "run_page/gen_svg.py",
+ "--from-db",
+ "--type",
+ "circular",
+ "--use-localtime",
+ ],
+ ]
+ for cmd_args in cmd_args_list:
+ subprocess.run(cmd_args, check=True)
+
+
if __name__ == "__main__":
parser = argparse.ArgumentParser()
parser.add_argument("phone_number_or_uid", help="joyrun phone number or uid")
parser.add_argument(
"identifying_code_or_sid", help="joyrun identifying_code from sms or sid"
)
+ parser.add_argument(
+ "--athlete",
+ dest="athlete",
+ help="athlete, keep same with {env.ATHLETE}",
+ )
+ parser.add_argument(
+ "--min_grid_distance",
+ dest="min_grid_distance",
+ help="min_grid_distance, keep same with {env.MIN_GRID_DISTANCE}",
+ type=int,
+ default=10,
+ )
parser.add_argument(
"--with-gpx",
dest="with_gpx",
@@ -375,3 +460,6 @@ def get_all_joyrun_tracks(self, old_tracks_ids, with_gpx=False):
activities_list = generator.load()
with open(JSON_FILE, "w") as f:
json.dump(activities_list, f, indent=0)
+
+ print("Data export to DB done")
+ _generate_svg_profile(options.athlete, options.min_grid_distance)
diff --git a/run_page/keep_to_strava_sync.py b/run_page/keep_to_strava_sync.py
index 38647b1871b..ecbd29a31a6 100644
--- a/run_page/keep_to_strava_sync.py
+++ b/run_page/keep_to_strava_sync.py
@@ -1,16 +1,13 @@
import argparse
import json
import os
-from sre_constants import SUCCESS
import time
from collections import namedtuple
-import requests
from config import GPX_FOLDER
-from Crypto.Cipher import AES
from config import OUTPUT_DIR
from stravalib.exc import ActivityUploadFailed, RateLimitTimeout
from utils import make_strava_client, upload_file_to_strava
-from keep_sync import KEEP_DATA_TYPE_API, get_all_keep_tracks
+from keep_sync import KEEP_SPORT_TYPES, get_all_keep_tracks
from strava_sync import run_strava_sync
"""
@@ -72,10 +69,10 @@ def run_keep_sync(email, password, keep_sports_data_api, with_download_gpx=False
)
options = parser.parse_args()
- for api in options.sync_types:
+ for _tpye in options.sync_types:
assert (
- api in KEEP_DATA_TYPE_API
- ), f"{api} are not supported type, please make sure that the type entered in the {KEEP_DATA_TYPE_API}"
+ _tpye in KEEP_SPORT_TYPES
+ ), f"{_tpye} are not supported type, please make sure that the type entered in the {KEEP_SPORT_TYPES}"
new_tracks = run_keep_sync(
options.phone_number, options.password, options.sync_types, True
)
diff --git a/run_page/nike_sync.py b/run_page/nike_sync.py
index d7599e0fd63..0f6242cbf81 100644
--- a/run_page/nike_sync.py
+++ b/run_page/nike_sync.py
@@ -40,20 +40,21 @@ class Nike:
def __init__(self, refresh_token):
self.client = httpx.Client()
- response = self.client.post(
- TOKEN_REFRESH_URL,
- headers=NIKE_HEADERS,
- json={
- "refresh_token": refresh_token,
- "client_id": b64decode(NIKE_CLIENT_ID).decode(),
- "grant_type": "refresh_token",
- "ux_id": b64decode(NIKE_UX_ID).decode(),
- },
- timeout=60,
- )
- response.raise_for_status()
-
- access_token = response.json()["access_token"]
+ # response = self.client.post(
+ # TOKEN_REFRESH_URL,
+ # headers=NIKE_HEADERS,
+ # json={
+ # "refresh_token": refresh_token,
+ # "client_id": b64decode(NIKE_CLIENT_ID).decode(),
+ # "grant_type": "refresh_token",
+ # "ux_id": b64decode(NIKE_UX_ID).decode(),
+ # },
+ # timeout=60,
+ # )
+ # response.raise_for_status()
+ #
+ # access_token = response.json()["access_token"]
+ access_token = "The content of 'access_token' that you just copied."
self.client.headers.update({"Authorization": f"Bearer {access_token}"})
def get_activities_since_timestamp(self, timestamp):
diff --git a/run_page/oppo_sync.py b/run_page/oppo_sync.py
new file mode 100644
index 00000000000..cf5bf803f6b
--- /dev/null
+++ b/run_page/oppo_sync.py
@@ -0,0 +1,727 @@
+import argparse
+import hashlib
+import json
+import os
+import time
+import xml.etree.ElementTree as ET
+from collections import namedtuple
+from datetime import datetime, timedelta
+from xml.dom import minidom
+
+import gpxpy
+import polyline
+import requests
+from tzlocal import get_localzone
+
+from config import (
+ GPX_FOLDER,
+ JSON_FILE,
+ SQL_FILE,
+ run_map,
+ start_point,
+ TCX_FOLDER,
+ UTC_TIMEZONE,
+)
+from generator import Generator
+from utils import adjust_time
+
+TOKEN_REFRESH_URL = "https://sport.health.heytapmobi.com/open/v1/oauth/token"
+OPPO_HEADERS = {
+ "User-Agent": "Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:78.0) Gecko/20100101 Firefox/78.0",
+ "Content-Type": "application/json",
+ "Accept": "application/json",
+}
+
+# Query brief version of sports records
+# The query range cannot exceed one month!
+"""
+Return value is like:
+[
+ {
+ "dataType": 2,//运动数类型 1=健身类 2=其他运动类
+ "startTime": 1630323565000, //开始时间 单位毫秒
+ "endTime": 1630337130000,//结束时间 单位毫秒
+ "sportMode": 10,//运动模式 室内跑 详情见文档附录
+ "otherSportData": {
+ "avgHeartRate": 153,//平均心率 单位：count/min
+ "avgPace": 585,//平均配速 单位s/km
+ "avgStepRate": 115,//平均步频 单位step/min
+ "bestStepRate": 135,//最佳步频 单位step/min
+ "bestPace": 572,//最佳配速 单位s/km
+ "totalCalories": 2176000,//总消耗 单位卡
+ "totalDistance": 23175,//总距离 单位米
+ "totalSteps": 26062,//总步数
+ "totalTime": 13562000,//总时长，单位:毫秒
+ "totalClimb": 100//累计爬升高度，单位:米
+ },
+ },
+ {
+ "dataType": 1,//运动数类型 1=健身类 2=其他运动类
+ "startTime": 1630293981497 //开始时间 单位毫秒
+ "endTime": 1630294218127,//结束时间 单位毫秒
+ "sportMode": 9,//运动模式 健身 详情见文档附录
+ "fitnessData": {
+ "avgHeartRate": 90,//平均心率 单位：count/min
+ "courseName": "零基础减脂碎片练习",//课程名称
+ "finishNumber": 1,//课程完成次数
+ "trainedCalorie": 13554,//训练消耗的卡路里，单位:卡
+ "trainedDuration": 176000//实际训练时间，单位:ms
+ },
+ }
+]
+"""
+BRIEF_SPORT_DATA_API = "https://sport.health.heytapmobi.com/open/v1/data/sport/record?startTimeMillis={start_time}&endTimeMillis={end_time}"
+
+# Query detailed sports records
+# The query range cannot exceed one day!
+DETAILED_SPORT_DATA_API = "https://sport.health.heytapmobi.com/open/v2/data/sport/record?startTimeMillis={start_time}&endTimeMillis={end_time}"
+
+TIMESTAMP_THRESHOLD_IN_MILLISECOND = 5000
+
+# If your points need trans from gcj02 to wgs84 coordinate which use by Mapbox
+TRANS_GCJ02_TO_WGS84 = True
+
+# May be Forerunner 945?
+CONNECT_API_PART_NUMBER = "006-D2449-00"
+
+AVAILABLE_OUTDOOR_SPORT_MODE = [
+ 1, # WALK
+ 2, # RUN
+ 3, # RIDE
+ 13, # OUTDOOR_PHYSICAL_RUN
+ 15, # OUTDOOR_5KM_RELAX_RUN
+ 17, # OUTDOOR_FAT_REDUCE_RUN
+ 22, # MARATHON
+ 36, # MOUNTAIN_CLIMBING
+ 37, # CROSS_COUNTRY
+]
+
+AVAILABLE_INDOOR_SPORT_MODE = [
+ 10, # INDOOR_RUN
+ 14, # INDOOR_PHYSICAL_RUN
+ 16, # INDOOR_5KM_RELAX_RUN
+ 18, # INDOOR_FAT_REDUCE_RUN
+ 19, # INDOOR_FITNESS_WALK
+ 21, # TREADMILL_RUN
+]
+
+
+def get_access_token(session, client_id, client_secret, refresh_token):
+ headers = {
+ "User-Agent": "Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:78.0) Gecko/20100101 Firefox/78.0",
+ "Content-Type": "application/json",
+ "Accept": "application/json",
+ }
+ data = {
+ "clientId": client_id,
+ "clientSecret": client_secret,
+ "refreshToken": refresh_token,
+ "grantType": "refreshToken",
+ }
+ r = session.post(TOKEN_REFRESH_URL, headers=headers, json=data)
+ if r.ok:
+ token = r.json()["body"]["accessToken"]
+ headers["access-token"] = token
+ return session, headers
+
+
+def get_to_download_runs_ranges(session, sync_months, headers, start_timestamp):
+ result = []
+ current_time = datetime.now()
+ start_datatime = datetime.fromtimestamp(start_timestamp / 1000)
+
+ if start_datatime < current_time + timedelta(days=-30 * sync_months):
+ """retrieve the data of last 6 months."""
+ while sync_months >= 0:
+ temp_end = int(current_time.timestamp() * 1000)
+ current_time = current_time + timedelta(days=-30)
+ temp_start = int(current_time.timestamp() * 1000)
+ sync_months = sync_months - 1
+ result.extend(
+ parse_brief_sport_data(session, headers, temp_start, temp_end)
+ )
+ else:
+ while start_datatime < current_time:
+ temp_start = int(start_datatime.timestamp() * 1000)
+ start_datatime = start_datatime + timedelta(days=30)
+ temp_end = int(start_datatime.timestamp() * 1000)
+ result.extend(
+ parse_brief_sport_data(session, headers, temp_start, temp_end)
+ )
+ return result
+
+
+def parse_brief_sport_data(session, headers, temp_start, temp_end):
+ result = []
+ r = session.get(
+ BRIEF_SPORT_DATA_API.format(end_time=temp_end, start_time=temp_start),
+ headers=headers,
+ )
+ if r.ok:
+ sport_logs = r.json()["body"]
+ for i in sport_logs:
+ if (
+ i["sportMode"] in AVAILABLE_INDOOR_SPORT_MODE
+ or i["sportMode"] in AVAILABLE_OUTDOOR_SPORT_MODE
+ ):
+ result.append((i["startTime"], i["endTime"]))
+ print(f"sync record: start_time: " + str(i["startTime"]))
+ time.sleep(1) # spider rule
+ return result
+
+
+def get_single_run_data(session, headers, start, end):
+ r = session.get(
+ DETAILED_SPORT_DATA_API.format(end_time=end, start_time=start), headers=headers
+ )
+ if r.ok:
+ return r.json()
+
+
+def parse_raw_data_to_name_tuple(sport_data, with_gpx, with_tcx):
+ sport_data = sport_data["body"][0]
+ m = hashlib.md5()
+ m.update(str.encode(str(sport_data)))
+ oppo_id_str = str(int(m.hexdigest(), 16))[0:16]
+ oppo_id = int(oppo_id_str)
+
+ sport_data["id"] = oppo_id
+ start_time = sport_data["startTime"]
+ other_data = sport_data["otherSportData"]
+ avg_heart_rate = None
+ if other_data:
+ avg_heart_rate = other_data.get("avgHeartRate", None)
+ # fix #66
+ if avg_heart_rate and avg_heart_rate < 0:
+ avg_heart_rate = None
+
+ # if TRANS_GCJ02_TO_WGS84:
+ # run_points_data = [
+ # list(eviltransform.gcj2wgs(p["latitude"], p["longitude"]))
+ # for p in run_points_data
+ # ]
+ # for i, p in enumerate(run_points_data_gpx):
+ # p["latitude"] = run_points_data[i][0]
+ # p["longitude"] = run_points_data[i][1]
+
+ point_dict = prepare_track_points(sport_data, with_gpx)
+
+ if with_gpx is True:
+ gpx_data = parse_points_to_gpx(sport_data, point_dict)
+ download_keep_gpx(gpx_data, str(oppo_id))
+ if with_tcx is True:
+ parse_points_to_tcx(sport_data, point_dict)
+
+ else:
+ print(f"ID {oppo_id} no gps data")
+
+ gps_data = [
+ (item["latitude"], item["longitude"]) for item in other_data["gpsPoint"]
+ ]
+ polyline_str = polyline.encode(gps_data) if gps_data else ""
+ start_latlng = start_point(*gps_data[0]) if gps_data else None
+ start_date = datetime.utcfromtimestamp(start_time / 1000)
+ start_date_local = adjust_time(start_date, str(get_localzone()))
+ end = datetime.utcfromtimestamp(sport_data["endTime"] / 1000)
+ end_local = adjust_time(end, str(get_localzone()))
+ location_country = None
+ if not other_data["totalTime"]:
+ print(f"ID {oppo_id} has no total time just ignore please check")
+ return
+ d = {
+ "id": int(oppo_id),
+ "name": "activity from oppo",
+ # future to support others workout now only for run
+ "type": map_oppo_fit_type_to_strava_activity_type(sport_data["sportMode"]),
+ "start_date": datetime.strftime(start_date, "%Y-%m-%d %H:%M:%S"),
+ "end": datetime.strftime(end, "%Y-%m-%d %H:%M:%S"),
+ "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": other_data["totalDistance"],
+ "average_heartrate": int(avg_heart_rate) if avg_heart_rate else None,
+ "map": run_map(polyline_str),
+ "start_latlng": start_latlng,
+ "distance": other_data["totalDistance"],
+ "moving_time": timedelta(seconds=other_data["totalTime"]),
+ "elapsed_time": timedelta(
+ seconds=int((sport_data["endTime"] - sport_data["startTime"]) / 1000)
+ ),
+ "average_speed": other_data["totalDistance"] / other_data["totalTime"] * 1000,
+ "location_country": location_country,
+ "source": sport_data["deviceName"],
+ }
+ return namedtuple("x", d.keys())(*d.values())
+
+
+def get_all_oppo_tracks(
+ client_id,
+ client_secret,
+ refresh_token,
+ sync_months,
+ last_track_date,
+ with_download_gpx,
+ with_download_tcx,
+):
+ if with_download_gpx and not os.path.exists(GPX_FOLDER):
+ os.mkdir(GPX_FOLDER)
+ s = requests.Session()
+ s, headers = get_access_token(s, client_id, client_secret, refresh_token)
+
+ last_timestamp = (
+ 0
+ if (last_track_date == 0)
+ else int(
+ datetime.timestamp(datetime.strptime(last_track_date, "%Y-%m-%d %H:%M:%S"))
+ * 1000
+ )
+ )
+
+ runs = get_to_download_runs_ranges(s, sync_months, headers, last_timestamp + 1000)
+ print(f"{len(runs)} new oppo runs to generate")
+ tracks = []
+ for start, end in runs:
+ print(f"parsing oppo id {str(start)}-{str(end)}")
+ try:
+ run_data = get_single_run_data(s, headers, start, end)
+ track = parse_raw_data_to_name_tuple(
+ run_data, with_download_gpx, with_download_tcx
+ )
+ tracks.append(track)
+ except Exception as e:
+ print(f"Something wrong paring keep id {str(start)}-{str(end)}" + str(e))
+ return tracks
+
+
+def switch(v):
+ yield lambda *c: v in c
+
+
+def map_oppo_fit_type_to_gpx_type(oppo_type):
+ for case in switch(oppo_type):
+ if case(1): # WALK
+ return "Walking"
+ if case(2, 13, 15, 17, 22, 10, 14, 16, 18, 21, 37):
+ # RUN |
+ # OUTDOOR_PHYSICAL_RUN |
+ # OUTDOOR_5KM_RELAX_RUN |
+ # OUTDOOR_FAT_REDUCE_RUN |
+ # MARATHON
+ # INDOOR_RUN, etc.
+ # CROSS_COUNTRY
+ return "Running"
+ if case(19): # MOUNTAIN_CLIMBING
+ return "Hiking"
+ if case(3): # Ride
+ return "Biking"
+
+
+def map_oppo_fit_type_to_strava_activity_type(oppo_type):
+ """
+ Note: should consider the supported strava activity type:
+ Link: https://developers.strava.com/docs/reference/#api-models-ActivityType
+ """
+ for case in switch(oppo_type):
+ if case(1): # WALK
+ return "Walk"
+ if case(2, 13, 15, 17, 22, 10, 14, 16, 18, 21, 37):
+ # RUN |
+ # OUTDOOR_PHYSICAL_RUN |
+ # OUTDOOR_5KM_RELAX_RUN |
+ # OUTDOOR_FAT_REDUCE_RUN |
+ # MARATHON
+ # INDOOR_RUN, etc.
+ # CROSS_COUNTRY
+ return "Run"
+ if case(19): # MOUNTAIN_CLIMBING
+ return "Hike"
+ if case(3): # Ride
+ return "Ride"
+
+
+def parse_points_to_gpx(sport_data, points_dict_list):
+ gpx = gpxpy.gpx.GPX()
+ gpx.nsmap["gpxtpx"] = "http://www.garmin.com/xmlschemas/TrackPointExtension/v1"
+ gpx_track = gpxpy.gpx.GPXTrack()
+ gpx_track.name = f"""gpx from {sport_data["deviceName"]}"""
+ gpx_track.type = map_oppo_fit_type_to_gpx_type(sport_data["sportMode"])
+ 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"),
+ )
+ hr = p.get("hr")
+ cad = p.get("cad")
+ if hr is not None or cad is not None:
+ hr_str = f"""{hr}""" if hr is not None else ""
+ cad_str = (
+ f"""{p["cad"]}""" if cad is not None else ""
+ )
+ gpx_extension = ET.fromstring(
+ f"""
+ {hr_str}
+ {cad_str}
+
+ """
+ )
+ point.extensions.append(gpx_extension)
+ gpx_segment.points.append(point)
+ return gpx.to_xml()
+
+
+def download_keep_gpx(gpx_data, keep_id):
+ try:
+ print(f"downloading keep_id {str(keep_id)} gpx")
+ file_path = os.path.join(GPX_FOLDER, str(keep_id) + ".gpx")
+ with open(file_path, "w") as fb:
+ fb.write(gpx_data)
+ except:
+ print(f"wrong id {keep_id}")
+ pass
+
+
+def prepare_track_points(sport_data, with_gpx):
+ """
+ Convert run points data to GPX format.
+
+ Args:
+ sport_data (map of dict): A map of run data points.
+ with_gpx (boolean): export to gpx file or not.
+
+ Returns:
+ points_dict_list (list): data with need to parse.
+ """
+ other_data = sport_data["otherSportData"]
+ decoded_hr_data = other_data.get("heartRate", None)
+ points_dict_list = []
+
+ if other_data.get("gpsPoint"):
+ timestamp_list = [item["timestamp"] for item in decoded_hr_data]
+ other_data = sport_data["otherSportData"]
+ value_size = len(other_data.get("gpsPoint", None))
+
+ for i in range(value_size):
+ temp_timestamp = other_data.get("gpsPoint")[i]["timestamp"]
+ j = timestamp_list.index(temp_timestamp)
+
+ points_dict = {
+ "latitude": other_data.get("gpsPoint")[i]["latitude"],
+ "longitude": other_data.get("gpsPoint")[i]["longitude"],
+ "time": datetime.utcfromtimestamp(temp_timestamp / 1000),
+ "hr": other_data.get("heartRate")[j]["value"],
+ }
+ points_dict_list.append(get_value(j, points_dict, other_data))
+ elif with_gpx is False:
+ value_size = len(other_data.get("heartRate", None))
+
+ for i in range(value_size):
+ temp_timestamp = other_data.get("heartRate")[i]["timestamp"]
+ temp_date = datetime.utcfromtimestamp(temp_timestamp / 1000)
+ points_dict = {
+ "time": temp_date,
+ "hr": other_data.get("heartRate")[i]["value"],
+ }
+ points_dict_list.append(get_value(i, points_dict, other_data))
+
+ return points_dict_list
+
+
+def get_value(index, points_dict, other_data):
+ if other_data.get("pace"):
+ pace = other_data.get("pace")[index]["value"]
+ points_dict["speed"] = 0 if pace == 0 else 1000 / pace
+ if other_data.get("frequency"):
+ points_dict["cad"] = other_data.get("frequency")[index]["value"]
+ if other_data.get("distance"):
+ points_dict["distance"] = other_data.get("distance")[index]["value"]
+ if other_data.get("elevation"):
+ points_dict["elevation"] = other_data.get("elevation")[index]["value"]
+ return points_dict
+
+
+def parse_points_to_tcx(sport_data, points_dict_list):
+ # route ID
+ fit_id = str(sport_data["id"])
+ # local time
+ start_time = sport_data["startTime"]
+ start_date = datetime.utcfromtimestamp(start_time / 1000)
+ fit_start_time = datetime.strftime(
+ adjust_time(start_date, UTC_TIMEZONE), "%Y-%m-%dT%H:%M:%SZ"
+ )
+
+ # Root node
+ training_center_database = ET.Element(
+ "TrainingCenterDatabase",
+ {
+ "xmlns": "http://www.garmin.com/xmlschemas/TrainingCenterDatabase/v2",
+ "xmlns:ns5": "http://www.garmin.com/xmlschemas/ActivityGoals/v1",
+ "xmlns:ns3": "http://www.garmin.com/xmlschemas/ActivityExtension/v2",
+ "xmlns:ns2": "http://www.garmin.com/xmlschemas/UserProfile/v2",
+ "xmlns:xsi": "http://www.w3.org/2001/XMLSchema-instance",
+ "xmlns:ns4": "http://www.garmin.com/xmlschemas/ProfileExtension/v1",
+ "xsi:schemaLocation": "http://www.garmin.com/xmlschemas/TrainingCenterDatabase/v2 http://www.garmin.com/xmlschemas/TrainingCenterDatabasev2.xsd",
+ },
+ )
+ # xml tree
+ ET.ElementTree(training_center_database)
+ # Activities
+ activities = ET.Element("Activities")
+ training_center_database.append(activities)
+ # sport type
+ sports_type = map_oppo_fit_type_to_gpx_type(sport_data["sportMode"])
+ # activity
+ activity = ET.Element("Activity", {"Sport": sports_type})
+ activities.append(activity)
+ # Id
+ activity_id = ET.Element("Id")
+ activity_id.text = fit_start_time # Codoon use start_time as ID
+ activity.append(activity_id)
+ # Creator
+ activity_creator = ET.Element("Creator", {"xsi:type": "Device_t"})
+ activity.append(activity_creator)
+ # Name
+ activity_creator_name = ET.Element("Name")
+ activity_creator_name.text = sport_data["deviceName"]
+ activity_creator.append(activity_creator_name)
+ activity_creator_product = ET.Element("ProductID")
+ activity_creator_product.text = "3441"
+ activity_creator.append(activity_creator_product)
+
+ """
+ first, find distance split index
+ """
+ lap_split_indexes = [0]
+ points_dict_list_chunks = []
+
+ for idx, item in enumerate(points_dict_list):
+ size = len(lap_split_indexes)
+ if sports_type == "Running":
+ target_distance = 1000 * size
+ elif sports_type == "Biking":
+ target_distance = 5000 * size
+ else:
+ break
+
+ if idx + 1 != len(points_dict_list):
+ if (
+ item["distance"]
+ < target_distance
+ <= points_dict_list[idx + 1]["distance"]
+ ):
+ lap_split_indexes.append(idx)
+
+ if len(lap_split_indexes) == 1:
+ points_dict_list_chunks = [points_dict_list]
+ else:
+ for idx, item in enumerate(lap_split_indexes):
+ if idx + 1 == len(lap_split_indexes):
+ points_dict_list_chunks.append(
+ points_dict_list[item : len(points_dict_list) - 1]
+ )
+ else:
+ points_dict_list_chunks.append(
+ points_dict_list[item : lap_split_indexes[idx + 1]]
+ )
+
+ current_distance = 0
+ current_time = start_date
+
+ for item in points_dict_list_chunks:
+ # Lap
+ lap_start_time = datetime.strftime(
+ adjust_time(item[0]["time"], UTC_TIMEZONE), "%Y-%m-%dT%H:%M:%SZ"
+ )
+ activity_lap = ET.Element("Lap", {"StartTime": lap_start_time})
+ activity.append(activity_lap)
+
+ # DistanceMeters
+ total_distance_node = ET.Element("DistanceMeters")
+ total_distance_node.text = str(item[-1]["distance"] - current_distance)
+ current_distance = item[-1]["distance"]
+ activity_lap.append(total_distance_node)
+ # TotalTimeSeconds
+ chile_node = ET.Element("TotalTimeSeconds")
+ chile_node.text = str((item[-1]["time"] - current_time).total_seconds())
+ current_time = item[-1]["time"]
+ activity_lap.append(chile_node)
+ # MaximumSpeed
+ chile_node = ET.Element("MaximumSpeed")
+ chile_node.text = str(max(node["speed"] for node in item))
+ activity_lap.append(chile_node)
+ # # Calories
+ # chile_node = ET.Element("Calories")
+ # chile_node.text = str(int(other_data["totalCalories"] / 1000))
+ # activity_lap.append(chile_node)
+ # AverageHeartRateBpm
+ # bpm = ET.Element("AverageHeartRateBpm")
+ # bpm_value = ET.Element("Value")
+ # bpm.append(bpm_value)
+ # bpm_value.text = str(other_data["avgHeartRate"])
+ # heartrate_list = [item["value"] for item in other_data["heartRate"]]
+ # bpm_value.text = str(round(statistics.mean(heartrate_list)))
+ # activity_lap.append(bpm)
+ # # MaximumHeartRateBpm
+ # bpm = ET.Element("MaximumHeartRateBpm")
+ # bpm_value = ET.Element("Value")
+ # bpm.append(bpm_value)
+ # bpm_value.text = str(max(node["hr"] for node in item))
+ # activity_lap.append(bpm)
+
+ # Track
+ track = ET.Element("Track")
+ activity_lap.append(track)
+
+ for p in item:
+ tp = ET.Element("Trackpoint")
+ track.append(tp)
+ # Time
+ time_stamp = datetime.strftime(
+ adjust_time(p["time"], UTC_TIMEZONE), "%Y-%m-%dT%H:%M:%SZ"
+ )
+ time_label = ET.Element("Time")
+ time_label.text = time_stamp
+
+ tp.append(time_label)
+ if sports_type == "Biking" and p.get("cad"):
+ cadence_label = ET.Element("Cadence")
+ cadence_label.text = str(p["cad"])
+ tp.append(cadence_label)
+ if p.get("distance"):
+ distance_label = ET.Element("DistanceMeters")
+ distance_label.text = str(p["distance"])
+ tp.append(distance_label)
+ # HeartRateBpm
+ # None was converted to bytes by np.dtype, becoming a string "None" after decode...-_-
+ # as well as LatitudeDegrees and LongitudeDegrees below
+ if p.get("hr"):
+ bpm = ET.Element("HeartRateBpm")
+ bpm_value = ET.Element("Value")
+ bpm.append(bpm_value)
+ bpm_value.text = str(p["hr"])
+ tp.append(bpm)
+ # AltitudeMeters
+ if p.get("elevation"):
+ altitude_meters = ET.Element("AltitudeMeters")
+ altitude_meters.text = str(p["elevation"] / 10)
+ tp.append(altitude_meters)
+ if p.get("latitude"):
+ position = ET.Element("Position")
+ tp.append(position)
+ # LatitudeDegrees
+ lati = ET.Element("LatitudeDegrees")
+ lati.text = str(p["latitude"])
+ position.append(lati)
+ # LongitudeDegrees
+ longi = ET.Element("LongitudeDegrees")
+ longi.text = str(p["longitude"])
+ position.append(longi)
+ # Extensions
+ if p.get("speed") is not None or (
+ p.get("cad") is not None and sports_type == "Running"
+ ):
+ extensions = ET.Element("Extensions")
+ tp.append(extensions)
+ tpx = ET.Element("ns3:TPX")
+ extensions.append(tpx)
+ # LatitudeDegrees
+ # LatitudeDegrees
+ if p.get("speed") is not None:
+ speed = ET.Element("ns3:Speed")
+ speed.text = str(p["speed"])
+ tpx.append(speed)
+ if p.get("cad") is not None and sports_type == "Running":
+ cad = ET.Element("ns3:RunCadence")
+ cad.text = str(round(p["cad"] / 2))
+ tpx.append(cad)
+ # Author
+ author = ET.Element("Author", {"xsi:type": "Application_t"})
+ training_center_database.append(author)
+ author_name = ET.Element("Name")
+ author_name.text = "Connect Api"
+ author.append(author_name)
+ author_lang = ET.Element("LangID")
+ author_lang.text = "en"
+ author.append(author_lang)
+ author_part = ET.Element("PartNumber")
+ author_part.text = CONNECT_API_PART_NUMBER
+ author.append(author_part)
+ # write to TCX file
+ xml_str = minidom.parseString(ET.tostring(training_center_database)).toprettyxml()
+ with open(TCX_FOLDER + "/" + fit_id + ".tcx", "w") as f:
+ f.write(str(xml_str))
+
+
+def formated_input(
+ run_data, run_data_label, tcx_label
+): # load run_data from run_data_label, parse to tcx_label, return xml node
+ fit_data = str(run_data[run_data_label])
+ chile_node = ET.Element(tcx_label)
+ chile_node.text = fit_data
+ return chile_node
+
+
+def run_oppo_sync(
+ client_id,
+ client_secret,
+ refresh_token,
+ sync_months=6,
+ with_download_gpx=False,
+ with_download_tcx=True,
+):
+ generator = Generator(SQL_FILE)
+ old_tracks_dates = generator.get_old_tracks_dates()
+ new_tracks = get_all_oppo_tracks(
+ client_id,
+ client_secret,
+ refresh_token,
+ sync_months,
+ old_tracks_dates[0] if old_tracks_dates else 0,
+ with_download_gpx,
+ with_download_tcx,
+ )
+ generator.sync_from_app(new_tracks)
+
+ activities_list = generator.load()
+ with open(JSON_FILE, "w") as f:
+ json.dump(activities_list, f, indent=0)
+
+
+if __name__ == "__main__":
+ parser = argparse.ArgumentParser()
+ parser.add_argument("client_id", help="oppo heytap fit client id")
+ parser.add_argument("client_secret", help="oppo heytap fit client secret")
+ parser.add_argument("refresh_token", help="oppo heytap fit refresh token")
+ parser.add_argument(
+ "--with-gpx",
+ dest="with_gpx",
+ action="store_true",
+ help="get all oppo fit data to gpx and download",
+ )
+ parser.add_argument(
+ "--with-tcx",
+ dest="with_tcx",
+ action="store_true",
+ help="get all oppo fit data to tcx and download",
+ )
+ parser.add_argument(
+ "-m" "--months",
+ type=int,
+ default=6,
+ dest="sync_months",
+ help="oppo has limited the data retrieve, so the default months we can sync is 6.",
+ )
+ options = parser.parse_args()
+ run_oppo_sync(
+ options.client_id,
+ options.client_secret,
+ options.refresh_token,
+ options.sync_months,
+ options.with_gpx,
+ options.with_tcx,
+ )
diff --git a/run_page/tcx_to_garmin_sync.py b/run_page/tcx_to_garmin_sync.py
new file mode 100644
index 00000000000..6a3a42ba64a
--- /dev/null
+++ b/run_page/tcx_to_garmin_sync.py
@@ -0,0 +1,81 @@
+import argparse
+import asyncio
+import os
+from datetime import datetime
+
+from tcxreader.tcxreader import TCXReader
+
+from config import TCX_FOLDER
+from garmin_sync import Garmin
+
+
+def get_to_generate_files(last_time):
+ """
+ return to one sorted list for next time upload
+ """
+ file_names = os.listdir(TCX_FOLDER)
+ tcx = TCXReader()
+ tcx_files = [
+ (
+ tcx.read(os.path.join(TCX_FOLDER, i), only_gps=False),
+ os.path.join(TCX_FOLDER, i),
+ )
+ for i in file_names
+ if i.endswith(".tcx")
+ ]
+ tcx_files_dict = {
+ int(i[0].trackpoints[0].time.timestamp()): i[1]
+ for i in tcx_files
+ if len(i[0].trackpoints) > 0
+ and int(i[0].trackpoints[0].time.timestamp()) > last_time
+ }
+
+ dict(sorted(tcx_files_dict.items()))
+
+ return tcx_files_dict.values()
+
+
+async def upload_tcx_files_to_garmin(options):
+ print("Need to load all tcx files maybe take some time")
+ garmin_auth_domain = "CN" if options.is_cn else ""
+ garmin_client = Garmin(options.secret_string, garmin_auth_domain)
+
+ last_time = 0
+ if not options.all:
+ print("upload new tcx to Garmin")
+ last_activity = await garmin_client.get_activities(0, 1)
+ if not last_activity:
+ print("no garmin activity")
+ else:
+ after_datetime_str = last_activity[0]["startTimeGMT"]
+ after_datetime = datetime.strptime(after_datetime_str, "%Y-%m-%d %H:%M:%S")
+ last_time = datetime.timestamp(after_datetime)
+ else:
+ print("Need to load all tcx files maybe take some time")
+ to_upload_dict = get_to_generate_files(last_time)
+
+ await garmin_client.upload_activities_files(to_upload_dict)
+
+
+if __name__ == "__main__":
+ if not os.path.exists(TCX_FOLDER):
+ os.mkdir(TCX_FOLDER)
+ parser = argparse.ArgumentParser()
+ parser.add_argument(
+ "secret_string", nargs="?", help="secret_string fro get_garmin_secret.py"
+ )
+ parser.add_argument(
+ "--all",
+ dest="all",
+ action="store_true",
+ help="if upload to strava all without check last time",
+ )
+ parser.add_argument(
+ "--is-cn",
+ dest="is_cn",
+ action="store_true",
+ help="if garmin account is cn",
+ )
+ loop = asyncio.get_event_loop()
+ future = asyncio.ensure_future(upload_tcx_files_to_garmin(parser.parse_args()))
+ loop.run_until_complete(future)