piezza/garminexport
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

cosmetic code cleanup

Browse Source 
Co-authored-by: Stanislav Khrapov <stanislav.khrapov@dbschenker.com>
This commit is contained in:
Stanislav Khrapov 2020-03-08 20:05:56 +01:00 committed by GitHub
parent 8cd27fcb19
commit 4ee0cda314
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
11 changed files with 160 additions and 172 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
garminbackup.py
View File

@ -26,4 +26,4 @@ if __name__ == "__main__":
max_retries=args.max_retries) max_retries=args.max_retries)
except Exception as e: except Exception as e:
log.error(u"failed with exception: %s", str(e)) log.error("failed with exception: {}".format(e))

37
garminexport/backup.py
View File

@ -8,7 +8,7 @@ from datetime import datetime
log = logging.getLogger(__name__) log = logging.getLogger(__name__)
export_formats=["json_summary", "json_details", "gpx", "tcx", "fit"] export_formats = ["json_summary", "json_details", "gpx", "tcx", "fit"]
"""The range of supported export formats for activities.""" """The range of supported export formats for activities."""
format_suffix = { format_suffix = {
@ -20,7 +20,6 @@ format_suffix = {
} }
"""A table that maps export formats to their file format extensions.""" """A table that maps export formats to their file format extensions."""
not_found_file = ".not_found" not_found_file = ".not_found"
"""A file that lists all tried but failed export attempts. The lines in """A file that lists all tried but failed export attempts. The lines in
the file are the would-have-been file names, had the exports been successful. the file are the would-have-been file names, had the exports been successful.
@ -48,7 +47,7 @@ def export_filename(activity, export_format):
id=activity[0], id=activity[0],
time=activity[1].isoformat(), time=activity[1].isoformat(),
suffix=format_suffix[export_format]) suffix=format_suffix[export_format])
return fn.replace(':','_') if os.name=='nt' else fn return fn.replace(':', '_') if os.name == 'nt' else fn
def need_backup(activities, backup_dir, export_formats=None): def need_backup(activities, backup_dir, export_formats=None):
@ -64,6 +63,9 @@ def need_backup(activities, backup_dir, export_formats=None):
:type activities: list of tuples of `(int, datetime)` :type activities: list of tuples of `(int, datetime)`
:param backup_dir: Destination directory for exported activities. :param backup_dir: Destination directory for exported activities.
:type backup_dir: str :type backup_dir: str
:keyword export_formats: Which format(s) to export to. Could be any
of: 'json_summary', 'json_details', 'gpx', 'tcx', 'fit'.
:type export_formats: list of str
:return: All activities that need to be backed up. :return: All activities that need to be backed up.
:rtype: set of tuples of `(int, datetime)` :rtype: set of tuples of `(int, datetime)`
""" """
@ -86,12 +88,10 @@ def _not_found_activities(backup_dir):
if os.path.isfile(_not_found): if os.path.isfile(_not_found):
with open(_not_found, mode="r") as f: with open(_not_found, mode="r") as f:
failed_activities = [line.strip() for line in f.readlines()] failed_activities = [line.strip() for line in f.readlines()]
log.debug("%d tried but failed activities in %s", log.debug("{} tried but failed activities in {}".format(len(failed_activities), _not_found))
len(failed_activities), _not_found)
return failed_activities return failed_activities
def download(client, activity, retryer, backup_dir, export_formats=None): def download(client, activity, retryer, backup_dir, export_formats=None):
"""Exports a Garmin Connect activity to a given set of formats """Exports a Garmin Connect activity to a given set of formats
and saves the resulting file(s) to a given backup directory. and saves the resulting file(s) to a given backup directory.
@ -117,31 +117,27 @@ def download(client, activity, retryer, backup_dir, export_formats=None):
id = activity[0] id = activity[0]
if 'json_summary' in export_formats: if 'json_summary' in export_formats:
log.debug("getting json summary for %s", id) log.debug("getting json summary for {}".format(id))
activity_summary = retryer.call(client.get_activity_summary, id) activity_summary = retryer.call(client.get_activity_summary, id)
dest = os.path.join( dest = os.path.join(
backup_dir, export_filename(activity, 'json_summary')) backup_dir, export_filename(activity, 'json_summary'))
with codecs.open(dest, encoding="utf-8", mode="w") as f: with codecs.open(dest, encoding="utf-8", mode="w") as f:
f.write(json.dumps( f.write(json.dumps(activity_summary, ensure_ascii=False, indent=4))
activity_summary, ensure_ascii=False, indent=4))
if 'json_details' in export_formats: if 'json_details' in export_formats:
log.debug("getting json details for %s", id) log.debug("getting json details for {}".format(id))
activity_details = retryer.call(client.get_activity_details, id) activity_details = retryer.call(client.get_activity_details, id)
dest = os.path.join( dest = os.path.join(backup_dir, export_filename(activity, 'json_details'))
backup_dir, export_filename(activity, 'json_details'))
with codecs.open(dest, encoding="utf-8", mode="w") as f: with codecs.open(dest, encoding="utf-8", mode="w") as f:
f.write(json.dumps( f.write(json.dumps(activity_details, ensure_ascii=False, indent=4))
activity_details, ensure_ascii=False, indent=4))
not_found_path = os.path.join(backup_dir, not_found_file) not_found_path = os.path.join(backup_dir, not_found_file)
with open(not_found_path, mode="a") as not_found: with open(not_found_path, mode="a") as not_found:
if 'gpx' in export_formats: if 'gpx' in export_formats:
log.debug("getting gpx for %s", id) log.debug("getting gpx for {}".format(id))
activity_gpx = retryer.call(client.get_activity_gpx, id) activity_gpx = retryer.call(client.get_activity_gpx, id)
dest = os.path.join( dest = os.path.join(backup_dir, export_filename(activity, 'gpx'))
backup_dir, export_filename(activity, 'gpx'))
if activity_gpx is None: if activity_gpx is None:
not_found.write(os.path.basename(dest) + "\n") not_found.write(os.path.basename(dest) + "\n")
else: else:
@ -149,10 +145,9 @@ def download(client, activity, retryer, backup_dir, export_formats=None):
f.write(activity_gpx) f.write(activity_gpx)
if 'tcx' in export_formats: if 'tcx' in export_formats:
log.debug("getting tcx for %s", id) log.debug("getting tcx for {}".format(id))
activity_tcx = retryer.call(client.get_activity_tcx, id) activity_tcx = retryer.call(client.get_activity_tcx, id)
dest = os.path.join( dest = os.path.join(backup_dir, export_filename(activity, 'tcx'))
backup_dir, export_filename(activity, 'tcx'))
if activity_tcx is None: if activity_tcx is None:
not_found.write(os.path.basename(dest) + "\n") not_found.write(os.path.basename(dest) + "\n")
else: else:
@ -160,7 +155,7 @@ def download(client, activity, retryer, backup_dir, export_formats=None):
f.write(activity_tcx) f.write(activity_tcx)
if 'fit' in export_formats: if 'fit' in export_formats:
log.debug("getting fit for %s", id) log.debug("getting fit for {}".format(id))
activity_fit = retryer.call(client.get_activity_fit, id) activity_fit = retryer.call(client.get_activity_fit, id)
dest = os.path.join( dest = os.path.join(
backup_dir, export_filename(activity, 'fit')) backup_dir, export_filename(activity, 'fit'))

9
garminexport/cli.py
View File

@ -36,14 +36,15 @@ def parse_args() -> argparse.Namespace:
parser.add_argument( parser.add_argument(
"-f", "--format", choices=export_formats, "-f", "--format", choices=export_formats,
default=None, action='append', default=None, action='append',
help="Desired output formats (" + ', '.join(export_formats) + "). Default: ALL.") help="Desired output formats ({}). Default: ALL.".format(', '.join(export_formats)))
parser.add_argument( parser.add_argument(
"-E", "--ignore-errors", action='store_true', "-E", "--ignore-errors", action='store_true',
help="Ignore errors and keep going. Default: FALSE") help="Ignore errors and keep going. Default: FALSE")
parser.add_argument( parser.add_argument(
"--max-retries", metavar="NUM", default=DEFAULT_MAX_RETRIES, "--max-retries", metavar="NUM", default=DEFAULT_MAX_RETRIES,
type=int, help=("The maximum number of retries to make on failed attempts to fetch an activity. " type=int,
"Exponential backoff will be used, meaning that the delay between successive attempts " help=("The maximum number of retries to make on failed attempts to fetch an activity. "
"will double with every retry, starting at one second. DEFAULT: %d") % DEFAULT_MAX_RETRIES) "Exponential backoff will be used, meaning that the delay between successive attempts "
"will double with every retry, starting at one second. DEFAULT: {}").format(DEFAULT_MAX_RETRIES))
return parser.parse_args() return parser.parse_args()

126
garminexport/garminclient.py
View File

@ -6,16 +6,17 @@ parts of the Garmin Connect REST API.
import json import json
import logging import logging
import os import os
import os.path
import re import re
import requests
from io import BytesIO
import sys import sys
import zipfile import zipfile
from builtins import range
from functools import wraps
from io import BytesIO
import dateutil import dateutil
import dateutil.parser import dateutil.parser
import os.path import requests
from functools import wraps
from builtins import range
# #
# Note: For more detailed information about the API services # Note: For more detailed information about the API services
@ -45,12 +46,14 @@ def require_session(client_function):
"""Decorator that is used to annotate :class:`GarminClient` """Decorator that is used to annotate :class:`GarminClient`
methods that need an authenticated session before being called. methods that need an authenticated session before being called.
""" """
@wraps(client_function) @wraps(client_function)
def check_session(*args, **kwargs): def check_session(*args, **kwargs):
client_object = args[0] client_object = args[0]
if not client_object.session: if not client_object.session:
raise Exception("Attempt to use GarminClient without being connected. Call connect() before first use.'") raise Exception("Attempt to use GarminClient without being connected. Call connect() before first use.'")
return client_function(*args, **kwargs) return client_function(*args, **kwargs)
return check_session return check_session
@ -110,31 +113,28 @@ class GarminClient(object):
request_params = { request_params = {
"service": "https://connect.garmin.com/modern" "service": "https://connect.garmin.com/modern"
} }
headers={'origin': 'https://sso.garmin.com'} headers = {'origin': 'https://sso.garmin.com'}
auth_response = self.session.post( auth_response = self.session.post(
SSO_LOGIN_URL, headers=headers, params=request_params, data=form_data) SSO_LOGIN_URL, headers=headers, params=request_params, data=form_data)
log.debug("got auth response: %s", auth_response.text) log.debug("got auth response: {}".format(auth_response.text))
if auth_response.status_code != 200: if auth_response.status_code != 200:
raise ValueError( raise ValueError("authentication failure: did you enter valid credentials?")
"authentication failure: did you enter valid credentials?") auth_ticket_url = self._extract_auth_ticket_url(auth_response.text)
auth_ticket_url = self._extract_auth_ticket_url( log.debug("auth ticket url: '{}'".format(auth_ticket_url))
auth_response.text)
log.debug("auth ticket url: '%s'", auth_ticket_url)
log.info("claiming auth ticket ...") log.info("claiming auth ticket ...")
response = self.session.get(auth_ticket_url) response = self.session.get(auth_ticket_url)
if response.status_code != 200: if response.status_code != 200:
raise RuntimeError( raise RuntimeError(
"auth failure: failed to claim auth ticket: %s: %d\n%s" % "auth failure: failed to claim auth ticket: {}: {}\n{}".format(
(auth_ticket_url, response.status_code, response.text)) auth_ticket_url, response.status_code, response.text))
# appears like we need to touch base with the old API to initiate # appears like we need to touch base with the old API to initiate
# some form of legacy session. otherwise certain downloads will fail. # some form of legacy session. otherwise certain downloads will fail.
self.session.get('https://connect.garmin.com/legacy/session') self.session.get('https://connect.garmin.com/legacy/session')
@staticmethod
def _extract_auth_ticket_url(auth_response):
def _extract_auth_ticket_url(self, auth_response):
"""Extracts an authentication ticket URL from the response of an """Extracts an authentication ticket URL from the response of an
authentication form submission. The auth ticket URL is typically authentication form submission. The auth ticket URL is typically
of form: of form:
@ -143,22 +143,19 @@ class GarminClient(object):
:param auth_response: HTML response from an auth form submission. :param auth_response: HTML response from an auth form submission.
""" """
match = re.search( match = re.search(r'response_url\s*=\s*"(https:[^"]+)"', auth_response)
r'response_url\s*=\s*"(https:[^"]+)"', auth_response)
if not match: if not match:
raise RuntimeError( raise RuntimeError(
"auth failure: unable to extract auth ticket URL. did you provide a correct username/password?") "auth failure: unable to extract auth ticket URL. did you provide a correct username/password?")
auth_ticket_url = match.group(1).replace("\\", "") auth_ticket_url = match.group(1).replace("\\", "")
return auth_ticket_url return auth_ticket_url
@require_session @require_session
def list_activities(self): def list_activities(self):
"""Return all activity ids stored by the logged in user, along """Return all activity ids stored by the logged in user, along
with their starting timestamps. with their starting timestamps.
:returns: The full list of activity identifiers (along with their :returns: The full list of activity identifiers (along with their starting timestamps).
starting timestamps).
:rtype: tuples of (int, datetime) :rtype: tuples of (int, datetime)
""" """
ids = [] ids = []
@ -178,28 +175,24 @@ class GarminClient(object):
timestamps) starting at a given index, with index 0 being the user's timestamps) starting at a given index, with index 0 being the user's
most recently registered activity. most recently registered activity.
Should the index be out of bounds or the account empty, an empty Should the index be out of bounds or the account empty, an empty list is returned.
list is returned.
:param start_index: The index of the first activity to retrieve. :param start_index: The index of the first activity to retrieve.
:type start_index: int :type start_index: int
:param max_limit: The (maximum) number of activities to retrieve. :param max_limit: The (maximum) number of activities to retrieve.
:type max_limit: int :type max_limit: int
:returns: A list of activity identifiers (along with their :returns: A list of activity identifiers (along with their starting timestamps).
starting timestamps).
:rtype: tuples of (int, datetime) :rtype: tuples of (int, datetime)
""" """
log.debug("fetching activities {} through {} ...".format( log.debug("fetching activities {} through {} ...".format(start_index, start_index + max_limit - 1))
start_index, start_index+max_limit-1))
response = self.session.get( response = self.session.get(
"https://connect.garmin.com/modern/proxy/activitylist-service/activities/search/activities", "https://connect.garmin.com/modern/proxy/activitylist-service/activities/search/activities",
params={"start": start_index, "limit": max_limit}) params={"start": start_index, "limit": max_limit})
if response.status_code != 200: if response.status_code != 200:
raise Exception( raise Exception(
u"failed to fetch activities {} to {} types: {}\n{}".format( u"failed to fetch activities {} to {} types: {}\n{}".format(
start_index, (start_index+max_limit-1), start_index, (start_index + max_limit - 1), response.status_code, response.text))
response.status_code, response.text))
activities = json.loads(response.text) activities = json.loads(response.text)
if not activities: if not activities:
# index out of bounds or empty account # index out of bounds or empty account
@ -211,23 +204,23 @@ class GarminClient(object):
timestamp_utc = dateutil.parser.parse(activity["startTimeGMT"]) timestamp_utc = dateutil.parser.parse(activity["startTimeGMT"])
# make sure UTC timezone gets set # make sure UTC timezone gets set
timestamp_utc = timestamp_utc.replace(tzinfo=dateutil.tz.tzutc()) timestamp_utc = timestamp_utc.replace(tzinfo=dateutil.tz.tzutc())
entries.append( (id, timestamp_utc) ) entries.append((id, timestamp_utc))
log.debug("got {} activities.".format(len(entries))) log.debug("got {} activities.".format(len(entries)))
return entries return entries
@require_session @require_session
def get_activity_summary(self, activity_id): def get_activity_summary(self, activity_id):
"""Return a summary about a given activity. The """Return a summary about a given activity.
summary contains several statistics, such as duration, GPS starting The summary contains several statistics, such as duration, GPS starting
point, GPS end point, elevation gain, max heart rate, max pace, max point, GPS end point, elevation gain, max heart rate, max pace, max speed, etc).
speed, etc).
:param activity_id: Activity identifier. :param activity_id: Activity identifier.
:type activity_id: int :type activity_id: int
:returns: The activity summary as a JSON dict. :returns: The activity summary as a JSON dict.
:rtype: dict :rtype: dict
""" """
response = self.session.get("https://connect.garmin.com/modern/proxy/activity-service/activity/{}".format(activity_id)) response = self.session.get(
"https://connect.garmin.com/modern/proxy/activity-service/activity/{}".format(activity_id))
if response.status_code != 200: if response.status_code != 200:
log.error(u"failed to fetch json summary for activity {}: {}\n{}".format( log.error(u"failed to fetch json summary for activity {}: {}\n{}".format(
activity_id, response.status_code, response.text)) activity_id, response.status_code, response.text))
@ -247,7 +240,8 @@ class GarminClient(object):
:rtype: dict :rtype: dict
""" """
# mounted at xml or json depending on result encoding # mounted at xml or json depending on result encoding
response = self.session.get("https://connect.garmin.com/modern/proxy/activity-service/activity/{}/details".format(activity_id)) response = self.session.get(
"https://connect.garmin.com/modern/proxy/activity-service/activity/{}/details".format(activity_id))
if response.status_code != 200: if response.status_code != 200:
raise Exception(u"failed to fetch json activityDetails for {}: {}\n{}".format( raise Exception(u"failed to fetch json activityDetails for {}: {}\n{}".format(
activity_id, response.status_code, response.text)) activity_id, response.status_code, response.text))
@ -266,11 +260,12 @@ class GarminClient(object):
or ``None`` if the activity couldn't be exported to GPX. or ``None`` if the activity couldn't be exported to GPX.
:rtype: str :rtype: str
""" """
response = self.session.get("https://connect.garmin.com/modern/proxy/download-service/export/gpx/activity/{}".format(activity_id)) response = self.session.get(
"https://connect.garmin.com/modern/proxy/download-service/export/gpx/activity/{}".format(activity_id))
# An alternate URL that seems to produce the same results # An alternate URL that seems to produce the same results
# and is the one used when exporting through the Garmin # and is the one used when exporting through the Garmin
# Connect web page. # Connect web page.
#response = self.session.get("https://connect.garmin.com/proxy/activity-service-1.1/gpx/activity/{}?full=true".format(activity_id)) # response = self.session.get("https://connect.garmin.com/proxy/activity-service-1.1/gpx/activity/{}?full=true".format(activity_id))
# A 404 (Not Found) or 204 (No Content) response are both indicators # A 404 (Not Found) or 204 (No Content) response are both indicators
# of a gpx file not being available for the activity. It may, for # of a gpx file not being available for the activity. It may, for
@ -282,7 +277,6 @@ class GarminClient(object):
activity_id, response.status_code, response.text)) activity_id, response.status_code, response.text))
return response.text return response.text
@require_session @require_session
def get_activity_tcx(self, activity_id): def get_activity_tcx(self, activity_id):
"""Return a TCX (Training Center XML) representation of a """Return a TCX (Training Center XML) representation of a
@ -298,7 +292,8 @@ class GarminClient(object):
:rtype: str :rtype: str
""" """
response = self.session.get("https://connect.garmin.com/modern/proxy/download-service/export/tcx/activity/{}".format(activity_id)) response = self.session.get(
"https://connect.garmin.com/modern/proxy/download-service/export/tcx/activity/{}".format(activity_id))
if response.status_code == 404: if response.status_code == 404:
return None return None
if response.status_code != 200: if response.status_code != 200:
@ -306,7 +301,6 @@ class GarminClient(object):
activity_id, response.status_code, response.text)) activity_id, response.status_code, response.text))
return response.text return response.text
def get_original_activity(self, activity_id): def get_original_activity(self, activity_id):
"""Return the original file that was uploaded for an activity. """Return the original file that was uploaded for an activity.
If the activity doesn't have any file source (for example, If the activity doesn't have any file source (for example,
@ -319,28 +313,28 @@ class GarminClient(object):
its contents, or :obj:`(None,None)` if no file is found. its contents, or :obj:`(None,None)` if no file is found.
:rtype: (str, str) :rtype: (str, str)
""" """
response = self.session.get("https://connect.garmin.com/modern/proxy/download-service/files/activity/{}".format(activity_id)) response = self.session.get(
"https://connect.garmin.com/modern/proxy/download-service/files/activity/{}".format(activity_id))
# A 404 (Not Found) response is a clear indicator of a missing .fit # A 404 (Not Found) response is a clear indicator of a missing .fit
# file. As of lately, the endpoint appears to have started to # file. As of lately, the endpoint appears to have started to
# respond with 500 "NullPointerException" on attempts to download a # respond with 500 "NullPointerException" on attempts to download a
# .fit file for an activity without one. # .fit file for an activity without one.
if response.status_code in [404, 500]: if response.status_code in [404, 500]:
# Manually entered activity, no file source available # Manually entered activity, no file source available
return (None,None) return None, None
if response.status_code != 200: if response.status_code != 200:
raise Exception( raise Exception(
u"failed to get original activity file for {}: {}\n{}".format( u"failed to get original activity file for {}: {}\n{}".format(
activity_id, response.status_code, response.text)) activity_id, response.status_code, response.text))
# return the first entry from the zip archive where the filename is # return the first entry from the zip archive where the filename is
# activity_id (should be the only entry!) # activity_id (should be the only entry!)
zip = zipfile.ZipFile(BytesIO(response.content), mode="r") zip_file = zipfile.ZipFile(BytesIO(response.content), mode="r")
for path in zip.namelist(): for path in zip_file.namelist():
fn, ext = os.path.splitext(path) fn, ext = os.path.splitext(path)
if fn==str(activity_id): if fn == str(activity_id):
return ext[1:], zip.open(path).read() return ext[1:], zip_file.open(path).read()
return (None,None) return None, None
def get_activity_fit(self, activity_id): def get_activity_fit(self, activity_id):
"""Return a FIT representation for a given activity. If the activity """Return a FIT representation for a given activity. If the activity
@ -358,7 +352,7 @@ class GarminClient(object):
# if the file extension of the original activity file isn't 'fit', # if the file extension of the original activity file isn't 'fit',
# this activity was uploaded in a different format (e.g. gpx/tcx) # this activity was uploaded in a different format (e.g. gpx/tcx)
# and cannot be exported to fit # and cannot be exported to fit
return orig_file if fmt=='fit' else None return orig_file if fmt == 'fit' else None
@require_session @require_session
def upload_activity(self, file, format=None, name=None, description=None, activity_type=None, private=None): def upload_activity(self, file, format=None, name=None, description=None, activity_type=None, private=None):
@ -374,14 +368,14 @@ class GarminClient(object):
:rtype: int :rtype: int
""" """
if isinstance(file, basestring): if isinstance(file, str):
file = open(file, "rb") file = open(file, "rb")
# guess file type if unspecified # guess file type if unspecified
fn = os.path.basename(file.name) fn = os.path.basename(file.name)
_, ext = os.path.splitext(fn) _, ext = os.path.splitext(fn)
if format is None: if format is None:
if ext.lower() in ('.gpx','.tcx','.fit'): if ext.lower() in ('.gpx', '.tcx', '.fit'):
format = ext.lower()[1:] format = ext.lower()[1:]
else: else:
raise Exception(u"could not guess file type for {}".format(fn)) raise Exception(u"could not guess file type for {}".format(fn))
@ -394,15 +388,15 @@ class GarminClient(object):
# check response and get activity ID # check response and get activity ID
try: try:
j = response.json()["detailedImportResult"] j = response.json()["detailedImportResult"]
except (json.JSONDecodeException, KeyError): except (json.JSONDecodeError, KeyError):
raise Exception(u"failed to upload {} for activity: {}\n{}".format( raise Exception(u"failed to upload {} for activity: {}\n{}".format(
format, response.status_code, response.text)) format, response.status_code, response.text))
if len(j["failures"]) or len(j["successes"])<1: if len(j["failures"]) or len(j["successes"]) < 1:
raise Exception(u"failed to upload {} for activity: {}\n{}".format( raise Exception(u"failed to upload {} for activity: {}\n{}".format(
format, response.status_code, j["failures"])) format, response.status_code, j["failures"]))
if len(j["successes"])>1: if len(j["successes"]) > 1:
raise Exception(u"uploading {} resulted in multiple activities ({})".format( raise Exception(u"uploading {} resulted in multiple activities ({})".format(
format, len(j["successes"]))) format, len(j["successes"])))
@ -410,14 +404,20 @@ class GarminClient(object):
# add optional fields # add optional fields
data = {} data = {}
if name is not None: data['activityName'] = name if name is not None:
if description is not None: data['description'] = name data['activityName'] = name
if activity_type is not None: data['activityTypeDTO'] = {"typeKey": activity_type} if description is not None:
if private: data['privacy'] = {"typeKey": "private"} data['description'] = name
if activity_type is not None:
data['activityTypeDTO'] = {"typeKey": activity_type}
if private:
data['privacy'] = {"typeKey": "private"}
if data: if data:
data['activityId'] = activity_id data['activityId'] = activity_id
encoding_headers = {"Content-Type": "application/json; charset=UTF-8"} # see Tapiriik encoding_headers = {"Content-Type": "application/json; charset=UTF-8"} # see Tapiriik
response = self.session.put("https://connect.garmin.com/proxy/activity-service/activity/{}".format(activity_id), data=json.dumps(data), headers=encoding_headers) response = self.session.put(
"https://connect.garmin.com/proxy/activity-service/activity/{}".format(activity_id),
data=json.dumps(data), headers=encoding_headers)
if response.status_code != 204: if response.status_code != 204:
raise Exception(u"failed to set metadata for activity {}: {}\n{}".format( raise Exception(u"failed to set metadata for activity {}: {}\n{}".format(
activity_id, response.status_code, response.text)) activity_id, response.status_code, response.text))

14
garminexport/incremental_backup.py
View File

@ -35,7 +35,7 @@ def incremental_backup(username: str,
""" """
# if no --format was specified, all formats are to be backed up # if no --format was specified, all formats are to be backed up
format = format if format else export_formats format = format if format else export_formats
log.info("backing up formats: %s", ", ".join(format)) log.info("backing up formats: {}".format(", ".join(format)))
if not os.path.isdir(backup_dir): if not os.path.isdir(backup_dir):
os.makedirs(backup_dir) os.makedirs(backup_dir)
@ -50,23 +50,23 @@ def incremental_backup(username: str,
with GarminClient(username, password) as client: with GarminClient(username, password) as client:
# get all activity ids and timestamps from Garmin account # get all activity ids and timestamps from Garmin account
log.info("scanning activities for %s ...", username) log.info("scanning activities for {} ...".format(username))
activities = set(retryer.call(client.list_activities)) activities = set(retryer.call(client.list_activities))
log.info("account has a total of %d activities", len(activities)) log.info("account has a total of {} activities".format(len(activities)))
missing_activities = garminexport.backup.need_backup(activities, backup_dir, format) missing_activities = garminexport.backup.need_backup(activities, backup_dir, format)
backed_up = activities - missing_activities backed_up = activities - missing_activities
log.info("%s contains %d backed up activities", backup_dir, len(backed_up)) log.info("{} contains {} backed up activities".format(backup_dir, len(backed_up)))
log.info("activities that aren't backed up: %d", len(missing_activities)) log.info("activities that aren't backed up: {}".format(len(missing_activities)))
for index, activity in enumerate(missing_activities): for index, activity in enumerate(missing_activities):
id, start = activity id, start = activity
log.info("backing up activity %d from %s (%d out of %d) ..." % ( log.info("backing up activity {} from {} ({} out of {}) ...".format(
id, start, index + 1, len(missing_activities))) id, start, index + 1, len(missing_activities)))
try: try:
garminexport.backup.download(client, activity, retryer, backup_dir, format) garminexport.backup.download(client, activity, retryer, backup_dir, format)
except Exception as e: except Exception as e:
log.error(u"failed with exception: %s", e) log.error("failed with exception: {}".format(e))
if not ignore_errors: if not ignore_errors:
raise raise

52
garminexport/retryer.py
View File

@ -1,15 +1,14 @@
import abc import abc
from datetime import datetime
from datetime import timedelta
import logging import logging
import time import time
from datetime import datetime
from datetime import timedelta
log = logging.getLogger(__name__) log = logging.getLogger(__name__)
class GaveUpError(Exception): class GaveUpError(Exception):
"""Raised by a :class:`Retryer` that has exceeded its maximum number """Raised by a :class:`Retryer` that has exceeded its maximum number of retries."""
of retries."""
pass pass
@ -22,8 +21,7 @@ class DelayStrategy(object):
def next_delay(self, attempts): def next_delay(self, attempts):
"""Returns the time to wait before the next attempt. """Returns the time to wait before the next attempt.
:param attempts: The total number of (failed) attempts performed thus :param attempts: The total number of (failed) attempts performed thus far.
far.
:type attempts: int :type attempts: int
:return: The delay before the next attempt. :return: The delay before the next attempt.
@ -33,8 +31,8 @@ class DelayStrategy(object):
class FixedDelayStrategy(DelayStrategy): class FixedDelayStrategy(DelayStrategy):
"""A retry :class:`DelayStrategy` that produces a fixed delay between """A retry :class:`DelayStrategy` that produces a fixed delay between attempts."""
attempts."""
def __init__(self, delay): def __init__(self, delay):
""" """
:param delay: Attempt delay. :param delay: Attempt delay.
@ -56,7 +54,7 @@ class ExponentialBackoffDelayStrategy(DelayStrategy):
def __init__(self, initial_delay): def __init__(self, initial_delay):
""" """
:param initial_delay: Initial delay. :param initial_delay: Initial delay.
:type delay: `timedelta` :type initial_delay: `timedelta`
""" """
self.initial_delay = initial_delay self.initial_delay = initial_delay
@ -68,25 +66,21 @@ class ExponentialBackoffDelayStrategy(DelayStrategy):
class NoDelayStrategy(FixedDelayStrategy): class NoDelayStrategy(FixedDelayStrategy):
"""A retry :class:`DelayStrategy` that doesn't introduce any delay between """A retry :class:`DelayStrategy` that doesn't introduce any delay between attempts."""
attempts."""
def __init__(self): def __init__(self):
super(NoDelayStrategy, self).__init__(timedelta(seconds=0)) super(NoDelayStrategy, self).__init__(timedelta(seconds=0))
class ErrorStrategy(object): class ErrorStrategy(object):
"""Used by a :class:`Retryer` to determine which errors are to be """Used by a :class:`Retryer` to determine which errors are to be
suppressed and which errors are to be re-raised and thereby end the suppressed and which errors are to be re-raised and thereby end the (re)trying."""
(re)trying."""
__metaclass__ = abc.ABCMeta __metaclass__ = abc.ABCMeta
@abc.abstractmethod @abc.abstractmethod
def should_suppress(self, error): def should_suppress(self, error):
"""Called after an attempt that raised an exception to determine if """Called after an attempt that raised an exception to determine if
that error should be suppressed (continue retrying) or be re-raised that error should be suppressed (continue retrying) or be re-raised (and end the retrying).
(and end the retrying).
:param error: Error that was raised from an attempt. :param error: Error that was raised from an attempt.
""" """
@ -122,13 +116,14 @@ class StopStrategy(object):
class NeverStopStrategy(StopStrategy): class NeverStopStrategy(StopStrategy):
"""A :class:`StopStrategy` that never gives up.""" """A :class:`StopStrategy` that never gives up."""
def should_continue(self, attempts, elapsed_time): def should_continue(self, attempts, elapsed_time):
return True return True
class MaxRetriesStopStrategy(StopStrategy): class MaxRetriesStopStrategy(StopStrategy):
"""A :class:`StopStrategy` that gives up after a certain number of """A :class:`StopStrategy` that gives up after a certain number of retries."""
retries."""
def __init__(self, max_retries): def __init__(self, max_retries):
self.max_retries = max_retries self.max_retries = max_retries
@ -149,6 +144,7 @@ class Retryer(object):
to decide if the error should be suppressed or re-raised (in which case to decide if the error should be suppressed or re-raised (in which case
the retrying ends with that error). the retrying ends with that error).
""" """
def __init__( def __init__(
self, self,
returnval_predicate=lambda returnval: True, returnval_predicate=lambda returnval: True,
@ -180,7 +176,6 @@ class Retryer(object):
self.stop_strategy = stop_strategy self.stop_strategy = stop_strategy
self.error_strategy = error_strategy self.error_strategy = error_strategy
def call(self, function, *args, **kw): def call(self, function, *args, **kw):
"""Calls the given `function`, with the given arguments, repeatedly """Calls the given `function`, with the given arguments, repeatedly
until either (1) a satisfactory result is obtained (as indicated by until either (1) a satisfactory result is obtained (as indicated by
@ -200,24 +195,21 @@ class Retryer(object):
while True: while True:
try: try:
attempts += 1 attempts += 1
log.info('{%s}: attempt %d ...', name, attempts) log.info('{{}}: attempt {} ...'.format(name, attempts))
returnval = function(*args, **kw) returnval = function(*args, **kw)
if self.returnval_predicate(returnval): if self.returnval_predicate(returnval):
# return value satisfies predicate, we're done! # return value satisfies predicate, we're done!
log.debug('{%s}: success: "%s"', name, returnval) log.debug('{{}}: success: "{}"'.format(name, returnval))
return returnval return returnval
log.debug('{%s}: failed: return value: %s', name, returnval) log.debug('{{}}: failed: return value: {}'.format(name, returnval))
except Exception as e: except Exception as e:
if not self.error_strategy.should_suppress(e): if not self.error_strategy.should_suppress(e):
raise e raise e
log.debug('{%s}: failed: error: %s', name, str(e)) log.debug('{{}}: failed: error: {}'.format(name, e))
elapsed_time = datetime.now() - start elapsed_time = datetime.now() - start
# should we make another attempt? # should we make another attempt?
if not self.stop_strategy.should_continue(attempts, elapsed_time): if not self.stop_strategy.should_continue(attempts, elapsed_time):
raise GaveUpError( raise GaveUpError('{{}}: gave up after {} failed attempt(s)'.format(name, attempts))
'{%s}: gave up after %d failed attempt(s)' %
(name, attempts))
delay = self.delay_strategy.next_delay(attempts) delay = self.delay_strategy.next_delay(attempts)
log.info('{%s}: waiting %d seconds for next attempt' % log.info('{{}}: waiting {} seconds for next attempt'.format(name, delay.total_seconds()))
(name, delay.total_seconds()))
time.sleep(delay.total_seconds()) time.sleep(delay.total_seconds())

42
get_activity.py
View File

@ -6,7 +6,6 @@ import argparse
import getpass import getpass
import logging import logging
import os import os
import sys
from datetime import timedelta from datetime import timedelta
import dateutil.parser import dateutil.parser
@ -22,8 +21,8 @@ log = logging.getLogger(__name__)
if __name__ == "__main__": if __name__ == "__main__":
parser = argparse.ArgumentParser( parser = argparse.ArgumentParser(
description=("Downloads one particular activity for a given " description="Downloads one particular activity for a given Garmin Connect account.")
"Garmin Connect account."))
# positional args # positional args
parser.add_argument( parser.add_argument(
"username", metavar="<username>", type=str, help="Account user name.") "username", metavar="<username>", type=str, help="Account user name.")
@ -31,29 +30,30 @@ if __name__ == "__main__":
"activity", metavar="<activity>", type=int, help="Activity ID.") "activity", metavar="<activity>", type=int, help="Activity ID.")
parser.add_argument( parser.add_argument(
"format", metavar="<format>", type=str, "format", metavar="<format>", type=str,
help="Export format (one of: {}).".format( help="Export format (one of: {}).".format(garminexport.backup.export_formats))
garminexport.backup.export_formats))
# optional args # optional args
parser.add_argument( parser.add_argument(
"--password", type=str, help="Account password.") "--password", type=str, help="Account password.")
parser.add_argument( parser.add_argument(
"--destination", metavar="DIR", type=str, "--destination", metavar="DIR", type=str,
help=("Destination directory for downloaded activity. Default: " help="Destination directory for downloaded activity. Default: ./activities/",
"./activities/"), default=os.path.join(".", "activities")) default=os.path.join(".", "activities"))
parser.add_argument( parser.add_argument(
"--log-level", metavar="LEVEL", type=str, "--log-level", metavar="LEVEL", type=str,
help=("Desired log output level (DEBUG, INFO, WARNING, ERROR). " help="Desired log output level (DEBUG, INFO, WARNING, ERROR). Default: INFO.",
"Default: INFO."), default="INFO") default="INFO")
args = parser.parse_args() args = parser.parse_args()
if not args.log_level in LOG_LEVELS:
raise ValueError("Illegal log-level argument: {}".format( if args.log_level not in LOG_LEVELS:
args.log_level)) raise ValueError("Illegal log-level argument: {}".format(args.log_level))
if not args.format in garminexport.backup.export_formats:
if args.format not in garminexport.backup.export_formats:
raise ValueError( raise ValueError(
"Uncrecognized export format: '{}'. Must be one of {}".format( "Unrecognized export format: '{}'. Must be one of {}".format(
args.format, garminexport.backup.export_formats)) args.format, garminexport.backup.export_formats))
logging.root.setLevel(LOG_LEVELS[args.log_level]) logging.root.setLevel(LOG_LEVELS[args.log_level])
try: try:
@ -62,20 +62,18 @@ if __name__ == "__main__":
if not args.password: if not args.password:
args.password = getpass.getpass("Enter password: ") args.password = getpass.getpass("Enter password: ")
with GarminClient(args.username, args.password) as client: with GarminClient(args.username, args.password) as client:
log.info("fetching activity {} ...".format(args.activity)) log.info("fetching activity {} ...".format(args.activity))
summary = client.get_activity_summary(args.activity) summary = client.get_activity_summary(args.activity)
# set up a retryer that will handle retries of failed activity # set up a retryer that will handle retries of failed activity downloads
# downloads
retryer = Retryer( retryer = Retryer(
delay_strategy=ExponentialBackoffDelayStrategy( delay_strategy=ExponentialBackoffDelayStrategy(initial_delay=timedelta(seconds=1)),
initial_delay=timedelta(seconds=1)),
stop_strategy=MaxRetriesStopStrategy(5)) stop_strategy=MaxRetriesStopStrategy(5))
starttime = dateutil.parser.parse(summary["summaryDTO"]["startTimeGMT"]) start_time = dateutil.parser.parse(summary["summaryDTO"]["startTimeGMT"])
garminexport.backup.download( garminexport.backup.download(
client, (args.activity, starttime), retryer, args.destination, export_formats=[args.format]) client, (args.activity, start_time), retryer, args.destination, export_formats=[args.format])
except Exception as e: except Exception as e:
exc_type, exc_value, exc_traceback = sys.exc_info() log.error("failed with exception: {}".format(e))
log.error(u"failed with exception: %s", e)
raise raise

5
samples/lab.py
View File

@ -10,10 +10,9 @@ Garmin Connect.
import argparse import argparse
import getpass import getpass
from garminexport.garminclient import GarminClient
import json
import logging import logging
import sys
from garminexport.garminclient import GarminClient
logging.basicConfig(level=logging.INFO, format="%(asctime)-15s [%(levelname)s] %(message)s") logging.basicConfig(level=logging.INFO, format="%(asctime)-15s [%(levelname)s] %(message)s")
log = logging.getLogger(__name__) log = logging.getLogger(__name__)

12
samples/sample.py
View File

@ -2,10 +2,10 @@
import argparse import argparse
import getpass import getpass
from garminexport.garminclient import GarminClient
import json import json
import logging import logging
import sys
from garminexport.garminclient import GarminClient
logging.basicConfig(level=logging.INFO, format="%(asctime)-15s [%(levelname)s] %(message)s") logging.basicConfig(level=logging.INFO, format="%(asctime)-15s [%(levelname)s] %(message)s")
log = logging.getLogger(__name__) log = logging.getLogger(__name__)
@ -36,12 +36,12 @@ if __name__ == "__main__":
latest_activity, latest_activity_start = activity_ids[0] latest_activity, latest_activity_start = activity_ids[0]
activity = client.get_activity_summary(latest_activity) activity = client.get_activity_summary(latest_activity)
log.info(u"activity id: %s", activity["activity"]["activityId"]) log.info("activity id: {}".format(activity["activity"]["activityId"]))
log.info(u"activity name: '%s'", activity["activity"]["activityName"]) log.info("activity name: '{}'".format(activity["activity"]["activityName"]))
log.info(u"activity description: '%s'", activity["activity"]["activityDescription"]) log.info("activity description: '{}'".format(activity["activity"]["activityDescription"]))
log.info(json.dumps(client.get_activity_details(latest_activity), indent=4)) log.info(json.dumps(client.get_activity_details(latest_activity), indent=4))
log.info(client.get_activity_gpx(latest_activity)) log.info(client.get_activity_gpx(latest_activity))
except Exception as e: except Exception as e:
log.error(u"failed with exception: %s", e) log.error("failed with exception: {}".format(e))
finally: finally:
log.info("done") log.info("done")

3
setup.py
View File

@ -7,7 +7,8 @@ from distutils.core import setup
setup(name="Garmin Connect activity exporter", setup(name="Garmin Connect activity exporter",
version="1.0.0", version="1.0.0",
description=("A program that downloads all activities for a given Garmin Connect account and stores them locally on the user's computer."), description=("A program that downloads all activities for a given Garmin Connect account "
"and stores them locally on the user's computer."),
long_description=open('README.md').read(), long_description=open('README.md').read(),
author="Peter Gardfjäll", author="Peter Gardfjäll",
author_email="peter.gardfjall.work@gmail.com", author_email="peter.gardfjall.work@gmail.com",

30
upload_activity.py
View File

@ -5,7 +5,6 @@ Connect account.
import argparse import argparse
import getpass import getpass
import logging import logging
import sys
from garminexport.garminclient import GarminClient from garminexport.garminclient import GarminClient
from garminexport.logging_config import LOG_LEVELS from garminexport.logging_config import LOG_LEVELS
@ -13,11 +12,11 @@ from garminexport.logging_config import LOG_LEVELS
logging.basicConfig(level=logging.INFO, format="%(asctime)-15s [%(levelname)s] %(message)s") logging.basicConfig(level=logging.INFO, format="%(asctime)-15s [%(levelname)s] %(message)s")
log = logging.getLogger(__name__) log = logging.getLogger(__name__)
if __name__ == "__main__": if __name__ == "__main__":
parser = argparse.ArgumentParser( parser = argparse.ArgumentParser(
description=("Uploads an activity file to a Garmin Connect account.")) description="Uploads an activity file to a Garmin Connect account.")
# positional args # positional args
parser.add_argument( parser.add_argument(
"username", metavar="<username>", type=str, help="Account user name.") "username", metavar="<username>", type=str, help="Account user name.")
@ -38,31 +37,34 @@ if __name__ == "__main__":
'-T', '--type', help="Override activity type (running, cycling, walking, hiking, strength_training, etc.)") '-T', '--type', help="Override activity type (running, cycling, walking, hiking, strength_training, etc.)")
parser.add_argument( parser.add_argument(
"--log-level", metavar="LEVEL", type=str, "--log-level", metavar="LEVEL", type=str,
help=("Desired log output level (DEBUG, INFO, WARNING, ERROR). " help="Desired log output level (DEBUG, INFO, WARNING, ERROR). Default: INFO.",
"Default: INFO."), default="INFO") default="INFO")
args = parser.parse_args() args = parser.parse_args()
if len(args.activity)>1 and (args.description is not None or args.name is not None):
if len(args.activity) > 1 and (args.description is not None or args.name is not None):
parser.error("When uploading multiple activities, --name or --description cannot be used.") parser.error("When uploading multiple activities, --name or --description cannot be used.")
if not args.log_level in LOG_LEVELS:
raise ValueError("Illegal log-level argument: {}".format( if args.log_level not in LOG_LEVELS:
args.log_level)) raise ValueError("Illegal log-level argument: {}".format(args.log_level))
logging.root.setLevel(LOG_LEVELS[args.log_level]) logging.root.setLevel(LOG_LEVELS[args.log_level])
try: try:
if not args.password: if not args.password:
args.password = getpass.getpass("Enter password: ") args.password = getpass.getpass("Enter password: ")
with GarminClient(args.username, args.password) as client: with GarminClient(args.username, args.password) as client:
for activity in args.activity: for activity in args.activity:
log.info("uploading activity file {} ...".format(activity.name)) log.info("uploading activity file {} ...".format(activity.name))
try: try:
id = client.upload_activity(activity, name=args.name, description=args.description, private=args.private, activity_type=args.type) id = client.upload_activity(activity, name=args.name, description=args.description,
private=args.private, activity_type=args.type)
except Exception as e: except Exception as e:
log.error("upload failed: {!r}".format(e)) log.error("upload failed: {!r}".format(e))
else: else:
log.info("upload successful: https://connect.garmin.com/modern/activity/{}".format(id)) log.info("upload successful: https://connect.garmin.com/modern/activity/{}".format(id))
except Exception as e:
exc_type, exc_value, exc_traceback = sys.exc_info()
log.error(u"failed with exception: %s", e)
raise
except Exception as e:
log.error("failed with exception: {}".format(e))
raise