Created
March 1, 2020 21:09
-
-
Save anatolebeuzon/2e16aee606a16481047dd2b42d95436d to your computer and use it in GitHub Desktop.
Batch upload your Garmin .fit files to trainingpeaks.com using this Python script. I had 5000 of them, and their customer support suggested I drag and drop each of them individually. So I did, kind of :-)
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import logging | |
import os | |
from selenium import webdriver | |
from selenium.webdriver.firefox.options import Options | |
from selenium.webdriver.support import expected_conditions | |
from selenium.webdriver.support.ui import WebDriverWait | |
from selenium.webdriver.common.by import By | |
from selenium.common.exceptions import NoSuchElementException, TimeoutException | |
from joblib import Parallel, delayed | |
PARALLELISM = 8 | |
TIMEOUT_SEC = 10 | |
HEADLESS = True | |
URL = "https://app.trainingpeaks.com/#calendar" | |
FIT_FILE_DIR = os.environ["FIT_FILE_DIR"] # Path to the directory containing your .fit files | |
APP_USERNAME = os.environ["APP_USERNAME"] # Your trainingpeaks.com username | |
APP_PASSWORD = os.environ["APP_PASSWORD"] # Your trainingpeaks.com password | |
logging.basicConfig() | |
logger = logging.getLogger() | |
logger.setLevel(logging.DEBUG) # not effective when using joblib.Parallel? | |
def get_browser(headless): | |
options = Options() | |
options.headless = headless | |
return webdriver.Firefox(options=options) | |
def login(browser): | |
browser.get(URL) | |
form = browser.find_element_by_xpath("/html/body/main/div[2]/div/div[2]/form") | |
form.find_element_by_name("Username").send_keys(APP_USERNAME) | |
form.find_element_by_name("Password").send_keys(APP_PASSWORD) | |
form.find_element_by_id("btnSubmit").click() | |
def open_upload_popup(browser): | |
upload_button_xpath = "/html/body/div[1]/div/div/div[2]/div/div/div[1]/div/div/div/div[3]/div/label" | |
WebDriverWait(browser, TIMEOUT_SEC).until( | |
expected_conditions.presence_of_element_located((By.XPATH, upload_button_xpath)) | |
) | |
browser.find_element_by_xpath(upload_button_xpath).click() | |
def upload_file(browser, filepath): | |
input_file_xpath = "/html/body/div[6]/section/div/input" | |
browser.find_element_by_xpath(input_file_xpath).send_keys(filepath) | |
def remove(filepath): | |
os.remove(filepath) | |
logger.info(f"File {filepath} removed") | |
def upload_files(files): | |
browser = get_browser(HEADLESS) | |
try: | |
login(browser) | |
for filepath in files: | |
if not filepath.endswith(".fit"): | |
logger.warning(f"Skipped file {filepath}") | |
continue | |
open_upload_popup(browser) | |
upload_file(browser, filepath) | |
# Here, one of two things can happen: | |
# A) the upload fails, and the browser shows an error popup with a "userConfirm" button | |
# B) the upload succeeds and the browser shows a popup div with id "workOutQuickView" | |
# | |
# Apparently WebDriverWait's API cannot wait for A OR B. | |
# So we wait for A, and if we reach the timeout, then we try B. | |
# This is *extremely* sub-optimal. | |
try: | |
WebDriverWait(browser, TIMEOUT_SEC).until( | |
expected_conditions.presence_of_element_located((By.ID, "userConfirm")) | |
) | |
# if no exception is thrown, an error popup is on the page | |
error_reason_xpath = "/html/body/div[8]/div/div[2]/p" | |
error_reason = browser.find_element_by_xpath(error_reason_xpath).text | |
logger.warning(f"File {filepath}: error: {error_reason}") | |
if error_reason in ["No workouts found in this file", "File has already been uploaded"]: | |
remove(filepath) | |
# For some reason the error popup can't be closed easily. | |
# (when targetting <button id="userConfirm">, getting error: "not clickable because another element obscures it") | |
# So we just reload the page instead. This is sub-optimal. | |
browser.refresh() | |
except TimeoutException: | |
try: | |
browser.find_element_by_id("workOutQuickView") | |
# if no exception is thrown, upload succeeded | |
logger.warning(f"File {filepath}: upload succeeded") | |
browser.find_element_by_id("close").click() | |
remove(filepath) | |
except NoSuchElementException: | |
logger.error(f"File {filepath}: unknown failure") | |
browser.refresh() | |
finally: | |
browser.close() | |
# Get .fit files and launch n parallel instances of Firefox | |
# to upload those to app.trainingpeaks.com. | |
def dispatch(): | |
# Get all Garmin .fit files | |
# Some might not contain any workout. | |
# Empirically, the heavier the file, the more likely the file contains a workout. | |
# So we sort them by size (heaviest first) | |
all_files = map(lambda file: os.path.join(FIT_FILE_DIR, os.fsdecode(file)), os.listdir(os.fsencode(FIT_FILE_DIR))) | |
sorted_files = list(reversed(sorted(all_files, key=os.path.getsize))) | |
# Create PARALLELISM groups of files to be uploaded concurrently | |
# We don't want one worker uploading all the heaviest ones while the others are slacking off. | |
# So we distribute progressively over all the file groups | |
file_batches = [[] for i in range(PARALLELISM)] | |
for file_index in range(len(sorted_files)): | |
file_batches[file_index % PARALLELISM].append(sorted_files[file_index]) | |
Parallel(n_jobs=PARALLELISM)(delayed(upload_files)(file_batch) for file_batch in file_batches) | |
return "Done!" | |
if __name__ == "__main__": | |
print(dispatch()) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment