andre/scripts
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
333a200bf771ddef1507ae7b9b13bf0b9cdb120e
scripts/garmin-connect-export/gcexport.py
Andre Geißler 4eaceb8b4d init V
2024-10-14 00:11:55 +02:00

1346 lines
66 KiB
Python
Raw Blame History

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
File: gcexport.py
Original author: Kyle Krafka (https://github.com/kjkjava/)
Date: April 28, 2015
Fork author: Michael P (https://github.com/moderation/)
Date: February 21, 2016
Fork author: Peter Steiner (https://github.com/pe-st/)
Date: June 2017
Date: March 2020 - Python3 support by Thomas Th. (https://github.com/telemaxx/)
Description: Use this script to export your fitness data from Garmin Connect.
See README.md for more information, CHANGELOG.md for a history of the changes
Activity & event types:
https://connect.garmin.com/modern/main/js/properties/event_types/event_types.properties
https://connect.garmin.com/modern/main/js/properties/activity_types/activity_types.properties
"""
# Standard library imports
import argparse
import csv
import http.cookiejar
import io
import json
import logging
import os
import os.path
import re
import string
import sys
import unicodedata
import urllib.request
import zipfile
from datetime import datetime, timedelta, tzinfo
from getpass import getpass
from math import floor
from platform import python_version
from subprocess import call
from timeit import default_timer as timer
from urllib.error import HTTPError, URLError
from urllib.parse import urlencode
from urllib.request import Request
# Local application/library specific imports
from filtering import read_exclude, update_download_stats
COOKIE_JAR = http.cookiejar.CookieJar()
OPENER = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(COOKIE_JAR), urllib.request.HTTPSHandler(debuglevel=0))
SCRIPT_VERSION = '4.0.0'
# This version here should correspond to what is written in CONTRIBUTING.md#python-3x-versions
MINIMUM_PYTHON_VERSION = (3, 7)
# this is almost the datetime format Garmin used in the activity-search-service
# JSON 'display' fields (Garmin didn't zero-pad the date and the hour, but %d and %H do)
ALMOST_RFC_1123 = "%a, %d %b %Y %H:%M"
# used by sanitize_filename()
VALID_FILENAME_CHARS = f'-_.() {string.ascii_letters}{string.digits}'
# map the numeric parentTypeId to its name for the CSV output
PARENT_TYPE_ID = {
1: 'running',
2: 'cycling',
3: 'hiking',
4: 'other',
9: 'walking',
17: 'any',
26: 'swimming',
29: 'fitness_equipment',
71: 'motorcycling',
83: 'transition',
144: 'diving',
149: 'yoga',
165: 'winter_sports',
}
# typeId values using pace instead of speed
USES_PACE = {1, 3, 9} # running, hiking, walking
HR_ZONES_EMPTY = [None, None, None, None, None]
# Maximum number of activities you can request at once.
# Used to be 100 and enforced by Garmin for older endpoints; for the current endpoint 'URL_GC_LIST'
# the limit is not known (I have less than 1000 activities and could get them all in one go)
LIMIT_MAXIMUM = 1000
MAX_TRIES = 3
CSV_TEMPLATE = os.path.join(os.path.dirname(os.path.realpath(__file__)), "csv_header_default.properties")
WEBHOST = "https://connect.garmin.com"
REDIRECT = "https://connect.garmin.com/modern/"
BASE_URL = "https://connect.garmin.com/en-US/signin"
SSO = "https://sso.garmin.com/sso"
CSS = "https://static.garmincdn.com/com.garmin.connect/ui/css/gauth-custom-v1.2-min.css"
DATA = {
'service': REDIRECT,
'webhost': WEBHOST,
'source': BASE_URL,
'redirectAfterAccountLoginUrl': REDIRECT,
'redirectAfterAccountCreationUrl': REDIRECT,
'gauthHost': SSO,
'locale': 'en_US',
'id': 'gauth-widget',
'cssUrl': CSS,
'clientId': 'GarminConnect',
'rememberMeShown': 'true',
'rememberMeChecked': 'false',
'createAccountShown': 'true',
'openCreateAccount': 'false',
'displayNameShown': 'false',
'consumeServiceTicket': 'false',
'initialFocus': 'true',
'embedWidget': 'false',
'generateExtraServiceTicket': 'true',
'generateTwoExtraServiceTickets': 'false',
'generateNoServiceTicket': 'false',
'globalOptInShown': 'true',
'globalOptInChecked': 'false',
'mobile': 'false',
'connectLegalTerms': 'true',
'locationPromptShown': 'true',
'showPassword': 'true',
}
# URLs for various services.
URL_GC_LOGIN = 'https://sso.garmin.com/sso/signin?' + urlencode(DATA)
URL_GC_POST_AUTH = 'https://connect.garmin.com/modern/activities?'
URL_GC_PROFILE = 'https://connect.garmin.com/modern/profile'
URL_GC_USERSTATS = 'https://connect.garmin.com/modern/proxy/userstats-service/statistics/'
URL_GC_LIST = 'https://connect.garmin.com/modern/proxy/activitylist-service/activities/search/activities?'
URL_GC_ACTIVITY = 'https://connect.garmin.com/modern/proxy/activity-service/activity/'
URL_GC_DEVICE = 'https://connect.garmin.com/modern/proxy/device-service/deviceservice/app-info/'
URL_GC_GEAR = 'https://connect.garmin.com/modern/proxy/gear-service/gear/filterGear?activityId='
URL_GC_ACT_PROPS = 'https://connect.garmin.com/modern/main/js/properties/activity_types/activity_types.properties'
URL_GC_EVT_PROPS = 'https://connect.garmin.com/modern/main/js/properties/event_types/event_types.properties'
URL_GC_GPX_ACTIVITY = 'https://connect.garmin.com/modern/proxy/download-service/export/gpx/activity/'
URL_GC_TCX_ACTIVITY = 'https://connect.garmin.com/modern/proxy/download-service/export/tcx/activity/'
URL_GC_ORIGINAL_ACTIVITY = 'http://connect.garmin.com/proxy/download-service/files/activity/'
def resolve_path(directory, subdir, time):
"""
Replace time variables and returns changed path. Supported place holders are {YYYY} and {MM}
:param directory: export root directory
:param subdir: subdirectory, can have place holders.
:param time: date-time-string
:return: Updated dictionary string
"""
ret = os.path.join(directory, subdir)
if re.compile(".*{YYYY}.*").match(ret):
ret = ret.replace("{YYYY}", time[0:4])
if re.compile(".*{MM}.*").match(ret):
ret = ret.replace("{MM}", time[5:7])
return ret
def hhmmss_from_seconds(sec):
"""Helper function that converts seconds to HH:MM:SS time format."""
if isinstance(sec, (float, int)):
formatted_time = str(timedelta(seconds=int(sec))).zfill(8)
else:
formatted_time = "0.000"
return formatted_time
def kmh_from_mps(mps):
"""Helper function that converts meters per second (mps) to km/h."""
return str(mps * 3.6)
def sanitize_filename(name, max_length=0):
"""
Remove or replace characters that are unsafe for filename
"""
# inspired by https://stackoverflow.com/a/698714/3686
cleaned_filename = unicodedata.normalize('NFKD', name) if name else ''
stripped_filename = ''.join(c for c in cleaned_filename if c in VALID_FILENAME_CHARS).replace(' ', '_')
return stripped_filename[:max_length] if max_length > 0 else stripped_filename
def write_to_file(filename, content, mode='w', file_time=None):
"""
Helper function that persists content to a file.
:param filename: name of the file to write
:param content: content to write; can be 'bytes' or 'str'.
If it's 'bytes' and the mode 'w', it will be converted/decoded
:param mode: 'w' or 'wb'
:param file_time: if given use as timestamp for the file written (in seconds since 1970-01-01)
"""
if mode == 'w':
if isinstance(content, bytes):
content = content.decode('utf-8')
with io.open(filename, mode, encoding='utf-8') as text_file:
text_file.write(content)
elif mode == 'wb':
with io.open(filename, 'wb') as binary_file:
binary_file.write(content)
else:
raise Exception('Unsupported file mode: ', mode)
if file_time:
os.utime(filename, (file_time, file_time))
def http_req(url, post=None, headers=None):
"""
Helper function that makes the HTTP requests.
:param url: URL for the request
:param post: dictionary of POST parameters
:param headers: dictionary of headers
:return: response body (type 'bytes')
"""
request = Request(url)
# Tell Garmin we're some supported browser.
request.add_header(
'User-Agent',
'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/54.0.2816.0 Safari/537.36',
)
request.add_header('nk', 'NT') # necessary since 2021-02-23 to avoid http error code 402
if headers:
for header_key, header_value in headers.items():
request.add_header(header_key, header_value)
if post:
post = urlencode(post) # Convert dictionary to POST parameter string.
post = post.encode("utf-8")
start_time = timer()
try:
response = OPENER.open(request, data=post)
except HTTPError as ex:
if hasattr(ex, 'code'):
logging.error('Server couldn\'t fulfill the request, url %s, code %s, error: %s', url, ex.code, ex)
logging.info('Headers returned:\n%s', ex.info())
raise
except URLError as ex:
if hasattr(ex, 'reason'):
logging.error('Failed to reach url %s, error: %s', url, ex)
raise
logging.debug('Got %s in %s s from %s', response.getcode(), timer() - start_time, url)
logging.debug('Headers returned:\n%s', response.info())
# N.B. urllib2 will follow any 302 redirects.
# print(response.getcode())
if response.getcode() == 204:
# 204 = no content, e.g. for activities without GPS coordinates there is no GPX download.
# Write an empty file to prevent redownloading it.
logging.info('Got 204 for %s, returning empty response', url)
return b''
if response.getcode() != 200:
raise Exception(f'Bad return code ({response.getcode()}) for: {url}')
return response.read()
def http_req_as_string(url, post=None, headers=None):
"""Helper function that makes the HTTP requests, returning a string instead of bytes."""
return http_req(url, post, headers).decode()
# idea stolen from https://stackoverflow.com/a/31852401/3686
def load_properties(multiline, separator='=', comment_char='#', keys=None):
"""
Read a multiline string of properties (key/value pair separated by *separator*) into a dict
:param multiline: input string of properties
:param separator: separator between key and value
:param comment_char: lines starting with this char are considered comments, not key/value pairs
:param keys: list to append the keys to
:return:
"""
props = {}
for line in multiline.splitlines():
stripped_line = line.strip()
if stripped_line and not stripped_line.startswith(comment_char):
key_value = stripped_line.split(separator)
key = key_value[0].strip()
value = separator.join(key_value[1:]).strip().strip('"')
props[key] = value
if keys is not None:
keys.append(key)
return props
def value_if_found_else_key(some_dict, key):
"""Lookup a value in some_dict and use the key itself as fallback"""
return some_dict.get(key, key)
def present(element, act):
"""Return True if act[element] is valid and not None"""
if not act:
return False
if element not in act:
return False
return act[element]
def absent_or_null(element, act):
"""Return False only if act[element] is valid and not None"""
if not act:
return True
if element not in act:
return True
if act[element]:
return False
return True
def from_activities_or_detail(element, act, detail, detail_container):
"""Return detail[detail_container][element] if valid and act[element] (or None) otherwise"""
if absent_or_null(detail_container, detail) or absent_or_null(element, detail[detail_container]):
return None if absent_or_null(element, act) else act[element]
return detail[detail_container][element]
def trunc6(some_float):
"""Return the given float as string formatted with six digit precision"""
return f'{floor(some_float * 1000000) / 1000000:12.6f}'.lstrip()
# A class building tzinfo objects for fixed-offset time zones.
# (copied from https://docs.python.org/2/library/datetime.html)
class FixedOffset(tzinfo):
"""Fixed offset in minutes east from UTC."""
def __init__(self, offset, name):
super().__init__()
self.__offset = timedelta(minutes=offset)
self.__name = name
def utcoffset(self, dt):
return self.__offset
def tzname(self, dt):
return self.__name
def dst(self, dt):
return timedelta(0)
def offset_date_time(time_local, time_gmt):
"""
Build an 'aware' datetime from two 'naive' datetime objects (that is timestamps
as present in the activitylist-service.json), using the time difference as offset.
"""
local_dt = datetime_from_iso(time_local)
gmt_dt = datetime_from_iso(time_gmt)
offset = local_dt - gmt_dt
offset_tz = FixedOffset(offset.seconds // 60, "LCL")
return local_dt.replace(tzinfo=offset_tz)
def datetime_from_iso(iso_date_time):
"""
Call 'datetime.strptime' supporting different ISO time formats
(with or without 'T' between date and time, with or without microseconds,
but without offset)
:param iso_date_time: timestamp string in ISO format
:return: a 'naive` datetime
"""
pattern = re.compile(r"(\d{4}-\d{2}-\d{2})[T ](\d{2}:\d{2}:\d{2})(\.\d+)?")
match = pattern.match(iso_date_time)
if not match:
raise Exception(f'Invalid ISO timestamp {iso_date_time}.')
micros = match.group(3) if match.group(3) else ".0"
iso_with_micros = f'{match.group(1)} {match.group(2)}{micros}'
return datetime.strptime(iso_with_micros, "%Y-%m-%d %H:%M:%S.%f")
def epoch_seconds_from_summary(summary):
"""
Determine the start time in epoch seconds (seconds since 1970-01-01)
:param summary: summary dict
:return: epoch seconds as integer
"""
if present('beginTimestamp', summary):
return summary['beginTimestamp'] // 1000
if present('startTimeLocal', summary) and present('startTimeGMT', summary):
date_time = offset_date_time(summary['startTimeLocal'], summary['startTimeGMT'])
return int(date_time.timestamp())
logging.info('No timestamp found in activity %s', summary['activityId'])
return None
def pace_or_speed_raw(type_id, parent_type_id, mps):
"""Convert speed (m/s) to speed (km/h) or pace (min/km) depending on type and parent type"""
kmh = 3.6 * mps
if (type_id in USES_PACE) or (parent_type_id in USES_PACE):
return 60 / kmh
return kmh
def pace_or_speed_formatted(type_id, parent_type_id, mps):
"""
Convert speed (m/s) to string: speed (km/h as x.x) or
pace (min/km as MM:SS), depending on type and parent type
"""
kmh = 3.6 * mps
if (type_id in USES_PACE) or (parent_type_id in USES_PACE):
# format seconds per kilometer as MM:SS, see https://stackoverflow.com/a/27751293
div_mod = divmod(int(round(3600 / kmh)), 60)
return f'{div_mod[0]:02d}:{div_mod[1]:02d}'
return f'{round(kmh, 1):.1f}'
class CsvFilter:
"""Collects, filters and writes CSV."""
def __init__(self, csv_file, csv_header_properties):
self.__csv_file = csv_file
with open(csv_header_properties, 'r', encoding='utf-8') as prop:
csv_header_props = prop.read()
self.__csv_columns = []
self.__csv_headers = load_properties(csv_header_props, keys=self.__csv_columns)
self.__csv_field_names = []
for column in self.__csv_columns:
self.__csv_field_names.append(self.__csv_headers[column])
self.__writer = csv.DictWriter(self.__csv_file, fieldnames=self.__csv_field_names, quoting=csv.QUOTE_ALL)
self.__current_row = {}
def write_header(self):
"""Write the active column names as CSV header"""
self.__writer.writeheader()
def write_row(self):
"""Write the prepared CSV record"""
self.__writer.writerow(self.__current_row)
self.__current_row = {}
def set_column(self, name, value):
"""
Store a column value (if the column is active) into
the record prepared for the next write_row call
"""
if value and name in self.__csv_columns:
self.__current_row[self.__csv_headers[name]] = value
def is_column_active(self, name):
"""Return True if the column is present in the header template"""
return name in self.__csv_columns
def parse_arguments(argv):
"""
Setup the argument parser and parse the command line arguments.
"""
current_date = datetime.now().strftime('%Y-%m-%d')
activities_directory = f'./{current_date}_garmin_connect_export'
parser = argparse.ArgumentParser(description='Garmin Connect Exporter')
# fmt: off
parser.add_argument('--version', action='version', version='%(prog)s ' + SCRIPT_VERSION,
help='print version and exit')
parser.add_argument('-v', '--verbosity', action='count', default=0,
help='increase output and log verbosity, save more intermediate files')
parser.add_argument('--username',
help='your Garmin Connect username or email address (otherwise, you will be prompted)')
parser.add_argument('--password',
help='your Garmin Connect password (otherwise, you will be prompted)')
parser.add_argument('-c', '--count', default='1',
help='number of recent activities to download, or \'all\' (default: 1)')
parser.add_argument('-e', '--external',
help='path to external program to pass CSV file too')
parser.add_argument('-a', '--args',
help='additional arguments to pass to external program')
parser.add_argument('-f', '--format', choices=['gpx', 'tcx', 'original', 'json'], default='gpx',
help="export format; can be 'gpx', 'tcx', 'original' or 'json' (default: 'gpx')")
parser.add_argument('-d', '--directory', default=activities_directory,
help='the directory to export to (default: \'./YYYY-MM-DD_garmin_connect_export\')')
parser.add_argument('-s', '--subdir',
help='the subdirectory for activity files (tcx, gpx etc.), supported placeholders are {YYYY} and {MM} (default: export directory)')
parser.add_argument('-lp', '--logpath',
help='the directory to store logfiles (default: same as for --directory)')
parser.add_argument('-u', '--unzip', action='store_true',
help='if downloading ZIP files (format: \'original\'), unzip the file and remove the ZIP file')
parser.add_argument('-ot', '--originaltime', action='store_true',
help='will set downloaded (and possibly unzipped) file time to the activity start time')
parser.add_argument('--desc', type=int, nargs='?', const=0, default=None,
help='append the activity\'s description to the file name of the download; limit size if number is given')
parser.add_argument('-t', '--template', default=CSV_TEMPLATE,
help='template file with desired columns for CSV output')
parser.add_argument('-fp', '--fileprefix', action='count', default=0,
help='set the local time as activity file name prefix')
parser.add_argument('-sa', '--start_activity_no', type=int, default=1,
help='give index for first activity to import, i.e. skipping the newest activities')
parser.add_argument('-ex', '--exclude', metavar='FILE',
help='JSON file with Array of activity IDs to exclude from download. Format example: {"ids": ["6176888711"]}')
# fmt: on
return parser.parse_args(argv[1:])
def login_to_garmin_connect(args):
"""
Perform all HTTP requests to login to Garmin Connect.
"""
username = args.username if args.username else input('Username: ')
password = args.password if args.password else getpass()
logging.debug("Login params: %s", urlencode(DATA))
# Initially, we need to get a valid session cookie, so we pull the login page.
print('Connecting to Garmin Connect...', end='')
logging.info('Connecting to %s', URL_GC_LOGIN)
connect_response = http_req_as_string(URL_GC_LOGIN)
if args.verbosity > 0:
write_to_file(os.path.join(args.directory, 'connect_response.html'), connect_response, 'w')
for cookie in COOKIE_JAR:
logging.debug("Cookie %s : %s", cookie.name, cookie.value)
print(' Done.')
# Now we'll actually login.
# Fields that are passed in a typical Garmin login.
post_data = {
'username': username,
'password': password,
'embed': 'false',
'rememberme': 'on',
}
headers = {'referer': URL_GC_LOGIN}
print('Requesting Login ticket...', end='')
logging.info('Requesting Login ticket')
login_response = http_req_as_string(f'{URL_GC_LOGIN}#', post_data, headers)
for cookie in COOKIE_JAR:
logging.debug("Cookie %s : %s", cookie.name, cookie.value)
if args.verbosity > 0:
write_to_file(os.path.join(args.directory, 'login_response.html'), login_response, 'w')
# extract the ticket from the login response
pattern = re.compile(r".*\?ticket=([-\w]+)\";.*", re.MULTILINE | re.DOTALL)
match = pattern.match(login_response)
if not match:
raise Exception(
'Couldn\'t find ticket in the login response. Cannot log in. Did you enter the correct username and password?'
)
login_ticket = match.group(1)
print(' Done. Ticket=', login_ticket, sep='')
print("Authenticating...", end='')
logging.info('Authentication URL %s', f'{URL_GC_POST_AUTH}ticket={login_ticket}')
http_req(f'{URL_GC_POST_AUTH}ticket={login_ticket}')
print(' Done.')
def csv_write_record(csv_filter, extract, actvty, details, activity_type_name, event_type_name):
"""
Write out the given data as a CSV record
"""
type_id = 4 if absent_or_null('activityType', actvty) else actvty['activityType']['typeId']
parent_type_id = 4 if absent_or_null('activityType', actvty) else actvty['activityType']['parentTypeId']
if present(parent_type_id, PARENT_TYPE_ID):
parent_type_key = PARENT_TYPE_ID[parent_type_id]
else:
parent_type_key = None
logging.warning("Unknown parentType %s, please tell script author", str(parent_type_id))
# get some values from detail if present, from a otherwise
start_latitude = from_activities_or_detail('startLatitude', actvty, details, 'summaryDTO')
start_longitude = from_activities_or_detail('startLongitude', actvty, details, 'summaryDTO')
end_latitude = from_activities_or_detail('endLatitude', actvty, details, 'summaryDTO')
end_longitude = from_activities_or_detail('endLongitude', actvty, details, 'summaryDTO')
# fmt: off
csv_filter.set_column('id', str(actvty['activityId']))
csv_filter.set_column('url', 'https://connect.garmin.com/modern/activity/' + str(actvty['activityId']))
csv_filter.set_column('activityName', actvty['activityName'] if present('activityName', actvty) else None)
csv_filter.set_column('description', actvty['description'] if present('description', actvty) else None)
csv_filter.set_column('startTimeIso', extract['start_time_with_offset'].isoformat())
csv_filter.set_column('startTime1123', extract['start_time_with_offset'].strftime(ALMOST_RFC_1123))
csv_filter.set_column('startTimeMillis', str(actvty['beginTimestamp']) if present('beginTimestamp', actvty) else None)
csv_filter.set_column('startTimeRaw', details['summaryDTO']['startTimeLocal'] if present('startTimeLocal', details['summaryDTO']) else None)
csv_filter.set_column('endTimeIso', extract['end_time_with_offset'].isoformat() if extract['end_time_with_offset'] else None)
csv_filter.set_column('endTime1123', extract['end_time_with_offset'].strftime(ALMOST_RFC_1123) if extract['end_time_with_offset'] else None)
csv_filter.set_column('endTimeMillis', str(actvty['beginTimestamp'] + extract['elapsed_seconds'] * 1000) if present('beginTimestamp', actvty) else None)
csv_filter.set_column('durationRaw', str(round(actvty['duration'], 3)) if present('duration', actvty) else None)
csv_filter.set_column('duration', hhmmss_from_seconds(round(actvty['duration'])) if present('duration', actvty) else None)
csv_filter.set_column('elapsedDurationRaw', str(round(extract['elapsed_duration'], 3)) if extract['elapsed_duration'] else None)
csv_filter.set_column('elapsedDuration', hhmmss_from_seconds(round(extract['elapsed_duration'])) if extract['elapsed_duration'] else None)
csv_filter.set_column('movingDurationRaw', str(round(details['summaryDTO']['movingDuration'], 3)) if present('movingDuration', details['summaryDTO']) else None)
csv_filter.set_column('movingDuration', hhmmss_from_seconds(round(details['summaryDTO']['movingDuration'])) if present('movingDuration', details['summaryDTO']) else None)
csv_filter.set_column('distanceRaw', f"{actvty['distance'] / 1000:.5f}" if present('distance', actvty) else None)
csv_filter.set_column('averageSpeedRaw', kmh_from_mps(details['summaryDTO']['averageSpeed']) if present('averageSpeed', details['summaryDTO']) else None)
csv_filter.set_column('averageSpeedPaceRaw', trunc6(pace_or_speed_raw(type_id, parent_type_id, actvty['averageSpeed'])) if present('averageSpeed', actvty) else None)
csv_filter.set_column('averageSpeedPace', pace_or_speed_formatted(type_id, parent_type_id, actvty['averageSpeed']) if present('averageSpeed', actvty) else None)
csv_filter.set_column('averageMovingSpeedRaw', kmh_from_mps(details['summaryDTO']['averageMovingSpeed']) if present('averageMovingSpeed', details['summaryDTO']) else None)
csv_filter.set_column('averageMovingSpeedPaceRaw', trunc6(pace_or_speed_raw(type_id, parent_type_id, details['summaryDTO']['averageMovingSpeed'])) if present('averageMovingSpeed', details['summaryDTO']) else None)
csv_filter.set_column('averageMovingSpeedPace', pace_or_speed_formatted(type_id, parent_type_id, details['summaryDTO']['averageMovingSpeed']) if present('averageMovingSpeed', details['summaryDTO']) else None)
csv_filter.set_column('maxSpeedRaw', kmh_from_mps(details['summaryDTO']['maxSpeed']) if present('maxSpeed', details['summaryDTO']) else None)
csv_filter.set_column('maxSpeedPaceRaw', trunc6(pace_or_speed_raw(type_id, parent_type_id, details['summaryDTO']['maxSpeed'])) if present('maxSpeed', details['summaryDTO']) else None)
csv_filter.set_column('maxSpeedPace', pace_or_speed_formatted(type_id, parent_type_id, details['summaryDTO']['maxSpeed']) if present('maxSpeed', details['summaryDTO']) else None)
csv_filter.set_column('elevationLoss', str(round(details['summaryDTO']['elevationLoss'], 2)) if present('elevationLoss', details['summaryDTO']) else None)
csv_filter.set_column('elevationLossUncorr', str(round(details['summaryDTO']['elevationLoss'], 2)) if not actvty['elevationCorrected'] and present('elevationLoss', details['summaryDTO']) else None)
csv_filter.set_column('elevationLossCorr', str(round(details['summaryDTO']['elevationLoss'], 2)) if actvty['elevationCorrected'] and present('elevationLoss', details['summaryDTO']) else None)
csv_filter.set_column('elevationGain', str(round(details['summaryDTO']['elevationGain'], 2)) if present('elevationGain', details['summaryDTO']) else None)
csv_filter.set_column('elevationGainUncorr', str(round(details['summaryDTO']['elevationGain'], 2)) if not actvty['elevationCorrected'] and present('elevationGain', details['summaryDTO']) else None)
csv_filter.set_column('elevationGainCorr', str(round(details['summaryDTO']['elevationGain'], 2)) if actvty['elevationCorrected'] and present('elevationGain', details['summaryDTO']) else None)
csv_filter.set_column('minElevation', str(round(details['summaryDTO']['minElevation'], 2)) if present('minElevation', details['summaryDTO']) else None)
csv_filter.set_column('minElevationUncorr', str(round(details['summaryDTO']['minElevation'], 2)) if not actvty['elevationCorrected'] and present('minElevation', details['summaryDTO']) else None)
csv_filter.set_column('minElevationCorr', str(round(details['summaryDTO']['minElevation'], 2)) if actvty['elevationCorrected'] and present('minElevation', details['summaryDTO']) else None)
csv_filter.set_column('maxElevation', str(round(details['summaryDTO']['maxElevation'], 2)) if present('maxElevation', details['summaryDTO']) else None)
csv_filter.set_column('maxElevationUncorr', str(round(details['summaryDTO']['maxElevation'], 2)) if not actvty['elevationCorrected'] and present('maxElevation', details['summaryDTO']) else None)
csv_filter.set_column('maxElevationCorr', str(round(details['summaryDTO']['maxElevation'], 2)) if actvty['elevationCorrected'] and present('maxElevation', details['summaryDTO']) else None)
csv_filter.set_column('elevationCorrected', 'true' if actvty['elevationCorrected'] else 'false')
# csv_record += empty_record # no minimum heart rate in JSON
csv_filter.set_column('maxHRRaw', str(details['summaryDTO']['maxHR']) if present('maxHR', details['summaryDTO']) else None)
csv_filter.set_column('maxHR', f"{actvty['maxHR']:.0f}" if present('maxHR', actvty) else None)
csv_filter.set_column('averageHRRaw', str(details['summaryDTO']['averageHR']) if present('averageHR', details['summaryDTO']) else None)
csv_filter.set_column('averageHR', f"{actvty['averageHR']:.0f}" if present('averageHR', actvty) else None)
csv_filter.set_column('caloriesRaw', str(details['summaryDTO']['calories']) if present('calories', details['summaryDTO']) else None)
csv_filter.set_column('calories', f"{details['summaryDTO']['calories']:.0f}" if present('calories', details['summaryDTO']) else None)
csv_filter.set_column('vo2max', str(actvty['vO2MaxValue']) if present('vO2MaxValue', actvty) else None)
csv_filter.set_column('aerobicEffect', str(round(details['summaryDTO']['trainingEffect'], 2)) if present('trainingEffect', details['summaryDTO']) else None)
csv_filter.set_column('anaerobicEffect', str(round(details['summaryDTO']['anaerobicTrainingEffect'], 2)) if present('anaerobicTrainingEffect', details['summaryDTO']) else None)
csv_filter.set_column('hrZone1Low', str(extract['hrZones'][0]['zoneLowBoundary']) if present('zoneLowBoundary', extract['hrZones'][0]) else None)
csv_filter.set_column('hrZone1Seconds', f"{extract['hrZones'][0]['secsInZone']:.0f}" if present('secsInZone', extract['hrZones'][0]) else None)
csv_filter.set_column('hrZone2Low', str(extract['hrZones'][1]['zoneLowBoundary']) if present('zoneLowBoundary', extract['hrZones'][1]) else None)
csv_filter.set_column('hrZone2Seconds', f"{extract['hrZones'][1]['secsInZone']:.0f}" if present('secsInZone', extract['hrZones'][1]) else None)
csv_filter.set_column('hrZone3Low', str(extract['hrZones'][2]['zoneLowBoundary']) if present('zoneLowBoundary', extract['hrZones'][2]) else None)
csv_filter.set_column('hrZone3Seconds', f"{extract['hrZones'][2]['secsInZone']:.0f}" if present('secsInZone', extract['hrZones'][2]) else None)
csv_filter.set_column('hrZone4Low', str(extract['hrZones'][3]['zoneLowBoundary']) if present('zoneLowBoundary', extract['hrZones'][3]) else None)
csv_filter.set_column('hrZone4Seconds', f"{extract['hrZones'][3]['secsInZone']:.0f}" if present('secsInZone', extract['hrZones'][3]) else None)
csv_filter.set_column('hrZone5Low', str(extract['hrZones'][4]['zoneLowBoundary']) if present('zoneLowBoundary', extract['hrZones'][4]) else None)
csv_filter.set_column('hrZone5Seconds', f"{extract['hrZones'][4]['secsInZone']:.0f}" if present('secsInZone', extract['hrZones'][4]) else None)
csv_filter.set_column('averageRunCadence', str(round(details['summaryDTO']['averageRunCadence'], 2)) if present('averageRunCadence', details['summaryDTO']) else None)
csv_filter.set_column('maxRunCadence', str(details['summaryDTO']['maxRunCadence']) if present('maxRunCadence', details['summaryDTO']) else None)
csv_filter.set_column('strideLength', str(round(details['summaryDTO']['strideLength'], 2)) if present('strideLength', details['summaryDTO']) else None)
csv_filter.set_column('steps', str(actvty['steps']) if present('steps', actvty) else None)
csv_filter.set_column('averageCadence', str(actvty['averageBikingCadenceInRevPerMinute']) if present('averageBikingCadenceInRevPerMinute', actvty) else None)
csv_filter.set_column('maxCadence', str(actvty['maxBikingCadenceInRevPerMinute']) if present('maxBikingCadenceInRevPerMinute', actvty) else None)
csv_filter.set_column('strokes', str(actvty['strokes']) if present('strokes', actvty) else None)
csv_filter.set_column('averageTemperature', str(details['summaryDTO']['averageTemperature']) if present('averageTemperature', details['summaryDTO']) else None)
csv_filter.set_column('minTemperature', str(details['summaryDTO']['minTemperature']) if present('minTemperature', details['summaryDTO']) else None)
csv_filter.set_column('maxTemperature', str(details['summaryDTO']['maxTemperature']) if present('maxTemperature', details['summaryDTO']) else None)
csv_filter.set_column('device', extract['device'] if extract['device'] else None)
csv_filter.set_column('gear', extract['gear'] if extract['gear'] else None)
csv_filter.set_column('activityTypeKey', actvty['activityType']['typeKey'].title() if present('typeKey', actvty['activityType']) else None)
csv_filter.set_column('activityType', value_if_found_else_key(activity_type_name, 'activity_type_' + actvty['activityType']['typeKey']) if present('activityType', actvty) else None)
csv_filter.set_column('activityParent', value_if_found_else_key(activity_type_name, 'activity_type_' + parent_type_key) if parent_type_key else None)
csv_filter.set_column('eventTypeKey', actvty['eventType']['typeKey'].title() if present('typeKey', actvty['eventType']) else None)
csv_filter.set_column('eventType', value_if_found_else_key(event_type_name, actvty['eventType']['typeKey']) if present('eventType', actvty) else None)
csv_filter.set_column('privacy', details['accessControlRuleDTO']['typeKey'] if present('typeKey', details['accessControlRuleDTO']) else None)
csv_filter.set_column('fileFormat', details['metadataDTO']['fileFormat']['formatKey'] if present('fileFormat', details['metadataDTO']) and present('formatKey', details['metadataDTO']['fileFormat']) else None)
csv_filter.set_column('tz', details['timeZoneUnitDTO']['timeZone'] if present('timeZone', details['timeZoneUnitDTO']) else None)
csv_filter.set_column('tzOffset', extract['start_time_with_offset'].isoformat()[-6:])
csv_filter.set_column('locationName', details['locationName'] if present('locationName', details) else None)
csv_filter.set_column('startLatitudeRaw', str(start_latitude) if start_latitude else None)
csv_filter.set_column('startLatitude', trunc6(start_latitude) if start_latitude else None)
csv_filter.set_column('startLongitudeRaw', str(start_longitude) if start_longitude else None)
csv_filter.set_column('startLongitude', trunc6(start_longitude) if start_longitude else None)
csv_filter.set_column('endLatitudeRaw', str(end_latitude) if end_latitude else None)
csv_filter.set_column('endLatitude', trunc6(end_latitude) if end_latitude else None)
csv_filter.set_column('endLongitudeRaw', str(end_longitude) if end_longitude else None)
csv_filter.set_column('endLongitude', trunc6(end_longitude) if end_longitude else None)
csv_filter.set_column('sampleCount', str(extract['samples']['metricsCount']) if present('metricsCount', extract['samples']) else None)
# fmt: on
csv_filter.write_row()
def extract_device(device_dict, details, start_time_seconds, args, http_caller, file_writer):
"""
Try to get the device details (and cache them, as they're used for multiple activities)
:param device_dict: cache (dict) of already known devices
:param details: dict with the details of an activity, should contain a device ID
:param start_time_seconds: if given use as timestamp for the file written (in seconds since 1970-01-01)
:param args: command-line arguments (for the file_writer callback)
:param http_caller: callback to perform the HTTP call for downloading the device details
:param file_writer: callback that saves the device details in a file
:return: string with the device name
"""
if not present('metadataDTO', details):
logging.warning("no metadataDTO")
return None
metadata = details['metadataDTO']
device_app_inst_id = (
metadata['deviceApplicationInstallationId'] if present('deviceApplicationInstallationId', metadata) else None
)
if device_app_inst_id:
if device_app_inst_id not in device_dict:
# observed from my stock of activities:
# details['metadataDTO']['deviceMetaDataDTO']['deviceId'] == null -> device unknown
# details['metadataDTO']['deviceMetaDataDTO']['deviceId'] == '0' -> device unknown
# details['metadataDTO']['deviceMetaDataDTO']['deviceId'] == 'someid' -> device known
device_dict[device_app_inst_id] = None
device_meta = metadata['deviceMetaDataDTO'] if present('deviceMetaDataDTO', metadata) else {}
device_id = device_meta['deviceId'] if present('deviceId', device_meta) else None
if 'deviceId' not in device_meta or device_id and device_id != '0':
device_json = http_caller(URL_GC_DEVICE + str(device_app_inst_id))
file_writer(os.path.join(args.directory, f'device_{device_app_inst_id}.json'), device_json, 'w', start_time_seconds)
if not device_json:
logging.warning("Device Details %s are empty", device_app_inst_id)
device_dict[device_app_inst_id] = "device-id:" + str(device_app_inst_id)
else:
device_details = json.loads(device_json)
if present('productDisplayName', device_details):
device_dict[device_app_inst_id] = (
device_details['productDisplayName'] + ' ' + device_details['versionString']
)
else:
logging.warning("Device details %s incomplete", device_app_inst_id)
return device_dict[device_app_inst_id]
return None
def load_zones(activity_id, start_time_seconds, args, http_caller, file_writer):
"""
Try to get the heart rate zones
:param activity_id: ID of the activity (as string)
:param start_time_seconds: if given use as timestamp for the file written (in seconds since 1970-01-01)
:param args: command-line arguments (for the file_writer callback)
:param http_caller: callback to perform the HTTP call for downloading the device details
:param file_writer: callback that saves the device details in a file
:return: array with the heart rate zones
"""
zones = HR_ZONES_EMPTY
zones_json = http_caller(f'{URL_GC_ACTIVITY}{activity_id}/hrTimeInZones')
file_writer(os.path.join(args.directory, f'activity_{activity_id}_zones.json'), zones_json, 'w', start_time_seconds)
zones_raw = json.loads(zones_json)
if not zones_raw:
logging.warning("HR Zones %s are empty", activity_id)
else:
for raw_zone in zones_raw:
if present('zoneNumber', raw_zone):
index = raw_zone['zoneNumber'] - 1
zones[index] = {}
zones[index]['secsInZone'] = raw_zone['secsInZone']
zones[index]['zoneLowBoundary'] = raw_zone['zoneLowBoundary']
return zones
def load_gear(activity_id, args):
"""Retrieve the gear/equipment for an activity"""
try:
gear_json = http_req_as_string(URL_GC_GEAR + activity_id)
gear = json.loads(gear_json)
if gear:
if args.verbosity > 0:
write_to_file(os.path.join(args.directory, f'activity_{activity_id}-gear.json'), gear_json, 'w')
gear_display_name = gear[0]['displayName'] if present('displayName', gear[0]) else None
gear_model = gear[0]['customMakeModel'] if present('customMakeModel', gear[0]) else None
logging.debug("Gear for %s = %s/%s", activity_id, gear_display_name, gear_model)
return gear_display_name if gear_display_name else gear_model
return None
except HTTPError as ex:
logging.info("Unable to get gear for %d, error: %s", activity_id, ex)
# logging.exception(ex)
return None
def export_data_file(activity_id, activity_details, args, file_time, append_desc, date_time):
"""
Write the data of the activity to a file, depending on the chosen data format
The default filename is 'activity_' + activity_id, but this can be modified
by the '--fileprefix' option and the 'append_desc' parameter; the directory
to write the file into can be modified by the '--subdir' option.
:param activity_id: ID of the activity (as string)
:param activity_details: details of the activity (for format 'json')
:param args: command-line arguments
:param file_time: if given the desired time stamp for the activity file (in seconds since 1970-01-01)
:param append_desc: suffix to the default filename
:param date_time: datetime in ISO format used for '--fileprefix' and '--subdir' options
:return: True if the file was written, False if the file existed already
"""
# Time dependent subdirectory for activity files, e.g. '{YYYY}'
if args.subdir is not None:
directory = resolve_path(args.directory, args.subdir, date_time)
# export activities to root directory
else:
directory = args.directory
if not os.path.isdir(directory):
os.makedirs(directory)
# timestamp as prefix for filename
if args.fileprefix > 0:
prefix = f'{date_time.replace("-", "").replace(":", "").replace(" ", "-")}-'
else:
prefix = ""
original_basename = None
if args.format == 'gpx':
data_filename = os.path.join(directory, f'{prefix}activity_{activity_id}{append_desc}.gpx')
download_url = f'{URL_GC_GPX_ACTIVITY}{activity_id}?full=true'
file_mode = 'w'
elif args.format == 'tcx':
data_filename = os.path.join(directory, f'{prefix}activity_{activity_id}{append_desc}.tcx')
download_url = f'{URL_GC_TCX_ACTIVITY}{activity_id}?full=true'
file_mode = 'w'
elif args.format == 'original':
data_filename = os.path.join(directory, f'{prefix}activity_{activity_id}{append_desc}.zip')
# not all 'original' files are in FIT format, some are GPX or TCX...
original_basename = os.path.join(directory, f'{prefix}activity_{activity_id}{append_desc}')
download_url = URL_GC_ORIGINAL_ACTIVITY + activity_id
file_mode = 'wb'
elif args.format == 'json':
data_filename = os.path.join(directory, f'{prefix}activity_{activity_id}{append_desc}_summary.json')
file_mode = 'w'
else:
raise Exception('Unrecognized format.')
if os.path.isfile(data_filename):
logging.debug('Data file for %s already exists', activity_id)
print('\tData file already exists; skipping...')
# Inform the main program that the file already exists
return False
# Regardless of unzip setting, don't redownload if the ZIP or FIT/GPX/TCX original file exists.
if args.format == 'original' and (
os.path.isfile(original_basename + '.fit')
or os.path.isfile(original_basename + '.gpx')
or os.path.isfile(original_basename + '.tcx')
):
logging.debug('Original data file for %s already exists', activity_id)
print('\tOriginal data file already exists; skipping...')
# Inform the main program that the file already exists
return False
if args.format != 'json':
# Download the data file from Garmin Connect. If the download fails (e.g., due to timeout),
# this script will die, but nothing will have been written to disk about this activity, so
# just running it again should pick up where it left off.
try:
data = http_req(download_url)
except HTTPError as ex:
# Handle expected (though unfortunate) error codes; die on unexpected ones.
if ex.code == 500 and args.format == 'tcx':
# Garmin will give an internal server error (HTTP 500) when downloading TCX files
# if the original was a manual GPX upload. Writing an empty file prevents this file
# from being redownloaded, similar to the way GPX files are saved even when there
# are no tracks. One could be generated here, but that's a bit much. Use the GPX
# format if you want actual data in every file, as I believe Garmin provides a GPX
# file for every activity.
logging.info('Writing empty file since Garmin did not generate a TCX file for this activity...')
data = ''
elif ex.code == 404 and args.format == 'original':
# For manual activities (i.e., entered in online without a file upload), there is
# no original file. # Write an empty file to prevent redownloading it.
logging.info('Writing empty file since there was no original activity data...')
data = ''
else:
logging.info('Got %s for %s', ex.code, download_url)
raise Exception(f'Failed. Got an HTTP error {ex.code} for {download_url}') from ex
else:
data = activity_details
# Persist file
write_to_file(data_filename, data, file_mode, file_time)
# Success: Add activity ID to downloaded_ids.json
update_download_stats(activity_id, args.directory)
if args.format == 'original':
# Even manual upload of a GPX file is zipped, but we'll validate the extension.
if args.unzip and data_filename[-3:].lower() == 'zip':
logging.debug('Unzipping and removing original file, size is %s', os.stat(data_filename).st_size)
if os.stat(data_filename).st_size > 0:
with open(data_filename, 'rb') as zip_file, zipfile.ZipFile(zip_file) as zip_obj:
for name in zip_obj.namelist():
unzipped_name = zip_obj.extract(name, directory)
# prepend 'activity_' and append the description to the base name
name_base, name_ext = os.path.splitext(name)
# sometimes in 2020 Garmin added '_ACTIVITY' to the name in the ZIP. Remove it...
# note that 'new_name' should match 'original_basename' elsewhere in this script to
# avoid downloading the same files again
name_base = name_base.replace('_ACTIVITY', '')
new_name = os.path.join(directory, f'{prefix}activity_{name_base}{append_desc}{name_ext}')
logging.debug('renaming %s to %s', unzipped_name, new_name)
os.rename(unzipped_name, new_name)
if file_time:
os.utime(new_name, (file_time, file_time))
else:
print('\tSkipping 0Kb zip file.')
os.remove(data_filename)
# Inform the main program that the file is new
return True
def setup_logging(args):
"""Setup logging"""
logpath = args.logpath if args.logpath else args.directory
if not os.path.isdir(logpath):
os.makedirs(logpath)
logging.basicConfig(
filename=os.path.join(logpath, 'gcexport.log'), level=logging.DEBUG, format='%(asctime)s [%(levelname)-7.7s] %(message)s'
)
# set up logging to console
console = logging.StreamHandler()
console.setLevel(logging.WARN)
formatter = logging.Formatter('[%(levelname)s] %(message)s')
console.setFormatter(formatter)
logging.getLogger('').addHandler(console)
def logging_verbosity(verbosity):
"""Adapt logging verbosity, separately for logfile and console output"""
logger = logging.getLogger()
for handler in logger.handlers:
if isinstance(handler, logging.FileHandler):
# this is the logfile handler
level = logging.DEBUG if verbosity > 0 else logging.INFO
handler.setLevel(level)
logging.info('New logfile level: %s', logging.getLevelName(level))
elif isinstance(handler, logging.StreamHandler):
# this is the console handler
level = logging.DEBUG if verbosity > 1 else (logging.INFO if verbosity > 0 else logging.WARN)
handler.setLevel(level)
logging.debug('New console log level: %s', logging.getLevelName(level))
def fetch_userstats(args):
"""
Http request for getting user statistic like total number of activities. The json will be saved as file
'userstats.json'
:param args: command-line arguments (for args.directory etc)
:return: json with user statistics
"""
print('Getting display name...', end='')
logging.info('Profile page %s', URL_GC_PROFILE)
profile_page = http_req_as_string(URL_GC_PROFILE)
if args.verbosity > 0:
write_to_file(os.path.join(args.directory, 'profile.html'), profile_page, 'w')
display_name = extract_display_name(profile_page)
print(' Done. displayName=', display_name, sep='')
print('Fetching user stats...', end='')
logging.info('Userstats page %s', URL_GC_USERSTATS + display_name)
result = http_req_as_string(URL_GC_USERSTATS + display_name)
print(' Done.')
# Persist JSON
write_to_file(os.path.join(args.directory, 'userstats.json'), result, 'w')
return json.loads(result)
def extract_display_name(profile_page):
"""
Extract the display name from the profile page HTML document
:param profile_page: HTML document
:return: the display name
"""
# the display name should be in the HTML document as
# "displayName":"John.Doe"
pattern = re.compile(r".*\"displayName\":\"(.+?)\".*", re.MULTILINE | re.DOTALL)
match = pattern.match(profile_page)
if not match:
raise Exception('Did not find the display name in the profile page.')
display_name = match.group(1)
return display_name
def fetch_activity_list(args, total_to_download):
"""
Fetch the first 'total_to_download' activity summaries; as a side effect save them in json format.
:param args: command-line arguments (for args.directory etc)
:param total_to_download: number of activities to download
:return: List of activity summaries
"""
# This while loop will download data from the server in multiple chunks, if necessary.
activities = []
total_downloaded = 0
while total_downloaded < total_to_download:
# Maximum chunk size 'LIMIT_MAXIMUM' ... 400 return status if over maximum. So download
# maximum or whatever remains if less than maximum.
# As of 2018-03-06 I get return status 500 if over maximum
if total_to_download - total_downloaded > LIMIT_MAXIMUM:
num_to_download = LIMIT_MAXIMUM
else:
num_to_download = total_to_download - total_downloaded
chunk = fetch_activity_chunk(args, num_to_download, total_downloaded)
activities.extend(chunk)
total_downloaded += num_to_download
# it seems that parent multisport activities are not counted in userstats
if len(activities) != total_to_download:
logging.info('Expected %s activities, got %s.', total_to_download, len(activities))
return activities
def annotate_activity_list(activities, start, exclude_list):
"""
Creates an action list with a tuple per activity summary
The tuple per activity contains three values:
- index: the index of the activity summary in the activities argument
(the first gets index 0, the second index 1 etc)
- activity the activity summary from the activites argument
- action the action to take for this activity (d=download, s=skip, e=exclude)
:param activities: List of activity summaries
:param start: One-based index of the first non-skipped activity
(i.e. with 1 no activity gets skipped, with 2 the first activity gets skipped etc)
:param exclude_list: List of activity ids that have to be skipped explicitly
:return: List of action tuples
"""
action_list = []
for index, activity in enumerate(activities):
if index < (start - 1):
action = 's'
elif str(activity['activityId']) in exclude_list:
action = 'e'
else:
action = 'd'
action_list.append(dict(index=index, action=action, activity=activity))
return action_list
def fetch_activity_chunk(args, num_to_download, total_downloaded):
"""
Fetch a chunk of activity summaries; as a side effect save them in json format.
:param args: command-line arguments (for args.directory etc)
:param num_to_download: number of summaries to download in this chunk
:param total_downloaded: number of already downloaded summaries in previous chunks
:return: List of activity summaries
"""
search_params = {'start': total_downloaded, 'limit': num_to_download}
# Query Garmin Connect
print('Querying list of activities ', total_downloaded + 1, '..', total_downloaded + num_to_download, '...', sep='', end='')
logging.info('Activity list URL %s', URL_GC_LIST + urlencode(search_params))
result = http_req_as_string(URL_GC_LIST + urlencode(search_params))
print(' Done.')
# Persist JSON activities list
current_index = total_downloaded + 1
activities_list_filename = f'activities-{current_index}-{total_downloaded+num_to_download}.json'
write_to_file(os.path.join(args.directory, activities_list_filename), result, 'w')
activity_summaries = json.loads(result)
fetch_multisports(activity_summaries, http_req_as_string, args)
return activity_summaries
def fetch_multisports(activity_summaries, http_caller, args):
"""
Search 'activity_summaries' for multisport activities and then
fetch the information for the activity parts (child activities)
and insert them into the 'activity_summaries' just after the multisport
activity
:param activity_summaries: list of activity summaries, will be modified in-place
:param http_caller: callback to perform the HTTP call for downloading the activity details
:param args: command-line arguments (for args.directory etc)
"""
for idx, child_summary in enumerate(activity_summaries):
type_key = None if absent_or_null('activityType', child_summary) else child_summary['activityType']['typeKey']
if type_key == 'multi_sport':
_, details = fetch_details(child_summary['activityId'], http_caller)
child_ids = (
details['metadataDTO']['childIds'] if 'metadataDTO' in details and 'childIds' in details['metadataDTO'] else None
)
# insert the children in reversed order always at the same index to get
# the correct order in activity_summaries
for child_id in reversed(child_ids):
child_string, child_details = fetch_details(child_id, http_caller)
if args.verbosity > 0:
write_to_file(os.path.join(args.directory, f'child_{child_id}.json'), child_string, 'w')
child_summary = {}
copy_details_to_summary(child_summary, child_details)
activity_summaries.insert(idx + 1, child_summary)
def fetch_details(activity_id, http_caller):
"""
Try to get the activity details for an activity
:param activity_id: id of the activity to fetch
:param http_caller: callback to perform the HTTP call for downloading the activity details
:return details_as_string, details_as_json_dict:
"""
activity_details = None
details = None
tries = MAX_TRIES
while tries > 0:
activity_details = http_caller(f'{URL_GC_ACTIVITY}{activity_id}')
details = json.loads(activity_details)
# I observed a failure to get a complete JSON detail in about 5-10 calls out of 1000
# retrying then statistically gets a better JSON ;-)
if details['summaryDTO']:
tries = 0
else:
logging.info("Retrying activity details download %s", URL_GC_ACTIVITY + str(activity_id))
tries -= 1
if tries == 0:
raise Exception(f'Didn\'t get "summaryDTO" after {MAX_TRIES} tries for {activity_id}')
return activity_details, details
def copy_details_to_summary(summary, details):
"""
Add some activity properties from the 'details' dict to the 'summary' dict
The choice of which properties are copied is determined by the properties
used by the 'csv_write_record' method.
This particularly useful for childs of multisport activities, as I don't
know how to get these activity summaries otherwise
:param summary: summary dict, will be modified in-place
:param details: details dict
"""
# fmt: off
summary['activityId'] = details['activityId']
summary['activityName'] = details['activityName']
summary['description'] = details['description'] if present('description', details) else None
summary['activityType'] = {}
summary['activityType']['typeId'] = details['activityTypeDTO']['typeId'] if 'activityTypeDTO' in details and present('typeId', details['activityTypeDTO']) else None
summary['activityType']['typeKey'] = details['activityTypeDTO']['typeKey'] if 'activityTypeDTO' in details and present('typeKey', details['activityTypeDTO']) else None
summary['activityType']['parentTypeId'] = details['activityTypeDTO']['parentTypeId'] if 'activityTypeDTO' in details and present('parentTypeId', details['activityTypeDTO']) else None
summary['eventType'] = {}
summary['eventType']['typeKey'] = details['eventType']['typeKey'] if 'eventType' in details and present('typeKey', details['eventType']) else None
summary['startTimeLocal'] = details['summaryDTO']['startTimeLocal'] if 'summaryDTO' in details and 'startTimeLocal' in details['summaryDTO'] else None
summary['startTimeGMT'] = details['summaryDTO']['startTimeGMT'] if 'summaryDTO' in details and 'startTimeGMT' in details['summaryDTO'] else None
summary['duration'] = details['summaryDTO']['duration'] if 'summaryDTO' in details and 'duration' in details['summaryDTO'] else None
summary['distance'] = details['summaryDTO']['distance'] if 'summaryDTO' in details and 'distance' in details['summaryDTO'] else None
summary['averageSpeed'] = details['summaryDTO']['averageSpeed'] if 'summaryDTO' in details and 'averageSpeed' in details['summaryDTO'] else None
summary['maxHR'] = details['summaryDTO']['maxHR'] if 'summaryDTO' in details and 'maxHR' in details['summaryDTO'] else None
summary['averageHR'] = details['summaryDTO']['averageHR'] if 'summaryDTO' in details and 'averageHR' in details['summaryDTO'] else None
summary['elevationCorrected'] = details['metadataDTO']['elevationCorrected'] if 'metadataDTO' in details and 'elevationCorrected' in details['metadataDTO'] else None
# fmt: on
def process_activity_item(item, number_of_items, device_dict, activity_type_name, event_type_name, csv_filter, args):
"""
Process one activity item: download the data, parse it and write a line to the CSV file
:param item: activity item tuple, see `annotate_activity_list()`
:param number_of_items: total number of items (for progress output)
:param device_dict: cache (dict) of already known devices
:param activity_type_name: lookup table for activity type descriptions
:param event_type_name: lookup table for event type descriptions
:param csv_filter: object encapsulating CSV file access
:param args: command-line arguments
"""
current_index = item['index'] + 1
actvty = item['activity']
action = item['action']
# Action: skipping
if action == 's':
# Display which entry we're skipping.
print('Skipping : Garmin Connect activity ', end='')
print(f"({current_index}/{number_of_items}) [{actvty['activityId']}]")
return
# Action: excluding
if action == 'e':
# Display which entry we're skipping.
print('Excluding : Garmin Connect activity ', end='')
print(f"({current_index}/{number_of_items}) [{actvty['activityId']}]")
return
# Action: download
# Display which entry we're working on.
print('Downloading: Garmin Connect activity ', end='')
print(f"({current_index}/{number_of_items}) [{actvty['activityId']}] {actvty['activityName']}")
# Retrieve also the detail data from the activity (the one displayed on
# the https://connect.garmin.com/modern/activity/xxx page), because some
# data are missing from 'actvty' (or are even different, e.g. for my activities
# 86497297 or 86516281)
activity_details, details = fetch_details(actvty['activityId'], http_req_as_string)
extract = {}
extract['start_time_with_offset'] = offset_date_time(actvty['startTimeLocal'], actvty['startTimeGMT'])
if 'summaryDTO' in details and 'elapsedDuration' in details['summaryDTO']:
elapsed_duration = details['summaryDTO']['elapsedDuration']
else:
elapsed_duration = None
extract['elapsed_duration'] = elapsed_duration if elapsed_duration else actvty['duration']
extract['elapsed_seconds'] = int(round(extract['elapsed_duration']))
extract['end_time_with_offset'] = extract['start_time_with_offset'] + timedelta(seconds=extract['elapsed_seconds'])
print('\t', extract['start_time_with_offset'].isoformat(), ', ', sep='', end='')
print(hhmmss_from_seconds(extract['elapsed_seconds']), ', ', sep='', end='')
if 'distance' in actvty and isinstance(actvty['distance'], float):
print(f"{actvty['distance'] / 1000:.3f} km")
else:
print('0.000 km')
if args.desc is not None:
append_desc = '_' + sanitize_filename(actvty['activityName'], args.desc)
else:
append_desc = ''
if args.originaltime:
start_time_seconds = epoch_seconds_from_summary(actvty)
else:
start_time_seconds = None
extract['device'] = extract_device(device_dict, details, start_time_seconds, args, http_req_as_string, write_to_file)
# try to get the JSON with all the samples (not all activities have it...),
# but only if it's really needed for the CSV output
extract['samples'] = None
if csv_filter.is_column_active('sampleCount'):
try:
# TODO implement retries here, I have observed temporary failures
activity_measurements = http_req_as_string(f"{URL_GC_ACTIVITY}{actvty['activityId']}/details")
write_to_file(
os.path.join(args.directory, f"activity_{actvty['activityId']}_samples.json"),
activity_measurements,
'w',
start_time_seconds,
)
samples = json.loads(activity_measurements)
extract['samples'] = samples
except HTTPError as ex:
logging.info("Unable to get samples for %d", actvty['activityId'])
logging.exception(ex)
extract['gear'] = None
if csv_filter.is_column_active('gear'):
extract['gear'] = load_gear(str(actvty['activityId']), args)
extract['hrZones'] = HR_ZONES_EMPTY
if csv_filter.is_column_active('hrZone1Low') or csv_filter.is_column_active('hrZone1Seconds'):
extract['hrZones'] = load_zones(str(actvty['activityId']), start_time_seconds, args, http_req_as_string, write_to_file)
# Save the file and inform if it already existed. If the file already existed, do not apped the record to the csv
if export_data_file(
str(actvty['activityId']), activity_details, args, start_time_seconds, append_desc, actvty['startTimeLocal']
):
# Write stats to CSV.
csv_write_record(csv_filter, extract, actvty, details, activity_type_name, event_type_name)
def main(argv):
"""
Main entry point for gcexport.py
"""
args = parse_arguments(argv)
setup_logging(args)
logging.info("Starting %s version %s, using Python version %s", argv[0], SCRIPT_VERSION, python_version())
logging_verbosity(args.verbosity)
print('Welcome to Garmin Connect Exporter!')
if sys.version_info < MINIMUM_PYTHON_VERSION:
logging.warning(
"Python version %s is older than %s.%s.x, results might be unexpected",
python_version(),
MINIMUM_PYTHON_VERSION[0],
MINIMUM_PYTHON_VERSION[1],
)
# Get filter list with IDs to exclude
if args.exclude is not None:
exclude_list = read_exclude(args.exclude)
if exclude_list is None:
sys.exit(1)
else:
exclude_list = []
# Create directory for data files.
if os.path.isdir(args.directory):
logging.warning(
'Output directory %s already exists. Will skip already-downloaded files and append to the CSV file.', args.directory
)
else:
os.mkdir(args.directory)
login_to_garmin_connect(args)
# Query the userstats (activities totals on the profile page). Needed for
# filtering and for downloading 'all' to know how many activities are available
userstats_json = fetch_userstats(args)
if args.count == 'all':
total_to_download = int(userstats_json['userMetrics'][0]['totalActivities'])
else:
total_to_download = int(args.count)
device_dict = {}
# load some dictionaries with lookup data from REST services
activity_type_props = http_req_as_string(URL_GC_ACT_PROPS)
if args.verbosity > 0:
write_to_file(os.path.join(args.directory, 'activity_types.properties'), activity_type_props, 'w')
activity_type_name = load_properties(activity_type_props)
event_type_props = http_req_as_string(URL_GC_EVT_PROPS)
if args.verbosity > 0:
write_to_file(os.path.join(args.directory, 'event_types.properties'), activity_type_props, 'w')
event_type_name = load_properties(event_type_props)
activities = fetch_activity_list(args, total_to_download)
action_list = annotate_activity_list(activities, args.start_activity_no, exclude_list)
csv_filename = os.path.join(args.directory, 'activities.csv')
csv_existed = os.path.isfile(csv_filename)
with open(csv_filename, mode='a', encoding='utf-8') as csv_file:
csv_filter = CsvFilter(csv_file, args.template)
# Write header to CSV file
if not csv_existed:
csv_filter.write_header()
# Process each activity.
for item in action_list:
process_activity_item(item, len(action_list), device_dict, activity_type_name, event_type_name, csv_filter, args)
if args.external:
print('Open CSV output.')
print(csv_filename)
call([args.external, "--" + args.args, csv_filename])
print('Done!')
if __name__ == "__main__":
try:
main(sys.argv)
except KeyboardInterrupt:
print('Interrupted')
sys.exit(0)
Reference in New Issue View Git Blame Copy Permalink