diff --git a/garminbackup.py b/garminbackup.py index 99b7c01..4f049ef 100755 --- a/garminbackup.py +++ b/garminbackup.py @@ -26,4 +26,4 @@ if __name__ == "__main__": max_retries=args.max_retries) except Exception as e: - log.error(u"failed with exception: %s", str(e)) + log.error("failed with exception: {}".format(e)) diff --git a/garminexport/backup.py b/garminexport/backup.py index 943df5e..709cb28 100644 --- a/garminexport/backup.py +++ b/garminexport/backup.py @@ -8,7 +8,7 @@ from datetime import datetime 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.""" format_suffix = { @@ -20,7 +20,6 @@ format_suffix = { } """A table that maps export formats to their file format extensions.""" - not_found_file = ".not_found" """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. @@ -48,7 +47,7 @@ def export_filename(activity, export_format): id=activity[0], time=activity[1].isoformat(), 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): @@ -64,6 +63,9 @@ def need_backup(activities, backup_dir, export_formats=None): :type activities: list of tuples of `(int, datetime)` :param backup_dir: Destination directory for exported activities. :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. :rtype: set of tuples of `(int, datetime)` """ @@ -86,12 +88,10 @@ def _not_found_activities(backup_dir): if os.path.isfile(_not_found): with open(_not_found, mode="r") as f: failed_activities = [line.strip() for line in f.readlines()] - log.debug("%d tried but failed activities in %s", - len(failed_activities), _not_found) + log.debug("{} tried but failed activities in {}".format(len(failed_activities), _not_found)) return failed_activities - def download(client, activity, retryer, backup_dir, export_formats=None): """Exports a Garmin Connect activity to a given set of formats 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] 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) dest = os.path.join( backup_dir, export_filename(activity, 'json_summary')) with codecs.open(dest, encoding="utf-8", mode="w") as f: - f.write(json.dumps( - activity_summary, ensure_ascii=False, indent=4)) + f.write(json.dumps(activity_summary, ensure_ascii=False, indent=4)) 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) - dest = os.path.join( - backup_dir, export_filename(activity, 'json_details')) + dest = os.path.join(backup_dir, export_filename(activity, 'json_details')) with codecs.open(dest, encoding="utf-8", mode="w") as f: - f.write(json.dumps( - activity_details, ensure_ascii=False, indent=4)) + f.write(json.dumps(activity_details, ensure_ascii=False, indent=4)) not_found_path = os.path.join(backup_dir, not_found_file) with open(not_found_path, mode="a") as not_found: 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) - dest = os.path.join( - backup_dir, export_filename(activity, 'gpx')) + dest = os.path.join(backup_dir, export_filename(activity, 'gpx')) if activity_gpx is None: not_found.write(os.path.basename(dest) + "\n") else: @@ -149,10 +145,9 @@ def download(client, activity, retryer, backup_dir, export_formats=None): f.write(activity_gpx) 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) - dest = os.path.join( - backup_dir, export_filename(activity, 'tcx')) + dest = os.path.join(backup_dir, export_filename(activity, 'tcx')) if activity_tcx is None: not_found.write(os.path.basename(dest) + "\n") else: @@ -160,7 +155,7 @@ def download(client, activity, retryer, backup_dir, export_formats=None): f.write(activity_tcx) 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) dest = os.path.join( backup_dir, export_filename(activity, 'fit')) diff --git a/garminexport/cli.py b/garminexport/cli.py index a8cbd70..6478922 100644 --- a/garminexport/cli.py +++ b/garminexport/cli.py @@ -36,14 +36,15 @@ def parse_args() -> argparse.Namespace: parser.add_argument( "-f", "--format", choices=export_formats, 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( "-E", "--ignore-errors", action='store_true', help="Ignore errors and keep going. Default: FALSE") parser.add_argument( "--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. " - "Exponential backoff will be used, meaning that the delay between successive attempts " - "will double with every retry, starting at one second. DEFAULT: %d") % DEFAULT_MAX_RETRIES) + type=int, + help=("The maximum number of retries to make on failed attempts to fetch an activity. " + "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() diff --git a/garminexport/garminclient.py b/garminexport/garminclient.py index 8fcad4b..6309841 100755 --- a/garminexport/garminclient.py +++ b/garminexport/garminclient.py @@ -6,16 +6,17 @@ parts of the Garmin Connect REST API. import json import logging import os +import os.path import re -import requests -from io import BytesIO import sys import zipfile +from builtins import range +from functools import wraps +from io import BytesIO + import dateutil import dateutil.parser -import os.path -from functools import wraps -from builtins import range +import requests # # 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` methods that need an authenticated session before being called. """ + @wraps(client_function) def check_session(*args, **kwargs): client_object = args[0] if not client_object.session: raise Exception("Attempt to use GarminClient without being connected. Call connect() before first use.'") return client_function(*args, **kwargs) + return check_session @@ -110,31 +113,28 @@ class GarminClient(object): request_params = { "service": "https://connect.garmin.com/modern" } - headers={'origin': 'https://sso.garmin.com'} + headers = {'origin': 'https://sso.garmin.com'} auth_response = self.session.post( 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: - raise ValueError( - "authentication failure: did you enter valid credentials?") - auth_ticket_url = self._extract_auth_ticket_url( - auth_response.text) - log.debug("auth ticket url: '%s'", auth_ticket_url) + raise ValueError("authentication failure: did you enter valid credentials?") + auth_ticket_url = self._extract_auth_ticket_url(auth_response.text) + log.debug("auth ticket url: '{}'".format(auth_ticket_url)) log.info("claiming auth ticket ...") response = self.session.get(auth_ticket_url) if response.status_code != 200: raise RuntimeError( - "auth failure: failed to claim auth ticket: %s: %d\n%s" % - (auth_ticket_url, response.status_code, response.text)) + "auth failure: failed to claim auth ticket: {}: {}\n{}".format( + auth_ticket_url, response.status_code, response.text)) # appears like we need to touch base with the old API to initiate # some form of legacy session. otherwise certain downloads will fail. self.session.get('https://connect.garmin.com/legacy/session') - - - def _extract_auth_ticket_url(self, auth_response): + @staticmethod + def _extract_auth_ticket_url(auth_response): """Extracts an authentication ticket URL from the response of an authentication form submission. The auth ticket URL is typically of form: @@ -143,22 +143,19 @@ class GarminClient(object): :param auth_response: HTML response from an auth form submission. """ - match = re.search( - r'response_url\s*=\s*"(https:[^"]+)"', auth_response) + match = re.search(r'response_url\s*=\s*"(https:[^"]+)"', auth_response) if not match: raise RuntimeError( "auth failure: unable to extract auth ticket URL. did you provide a correct username/password?") auth_ticket_url = match.group(1).replace("\\", "") return auth_ticket_url - @require_session def list_activities(self): """Return all activity ids stored by the logged in user, along with their starting timestamps. - :returns: The full list of activity identifiers (along with their - starting timestamps). + :returns: The full list of activity identifiers (along with their starting timestamps). :rtype: tuples of (int, datetime) """ ids = [] @@ -178,28 +175,24 @@ class GarminClient(object): timestamps) starting at a given index, with index 0 being the user's most recently registered activity. - Should the index be out of bounds or the account empty, an empty - list is returned. + Should the index be out of bounds or the account empty, an empty list is returned. :param start_index: The index of the first activity to retrieve. :type start_index: int :param max_limit: The (maximum) number of activities to retrieve. :type max_limit: int - :returns: A list of activity identifiers (along with their - starting timestamps). + :returns: A list of activity identifiers (along with their starting timestamps). :rtype: tuples of (int, datetime) """ - log.debug("fetching activities {} through {} ...".format( - start_index, start_index+max_limit-1)) + log.debug("fetching activities {} through {} ...".format(start_index, start_index + max_limit - 1)) response = self.session.get( "https://connect.garmin.com/modern/proxy/activitylist-service/activities/search/activities", params={"start": start_index, "limit": max_limit}) if response.status_code != 200: raise Exception( u"failed to fetch activities {} to {} types: {}\n{}".format( - start_index, (start_index+max_limit-1), - response.status_code, response.text)) + start_index, (start_index + max_limit - 1), response.status_code, response.text)) activities = json.loads(response.text) if not activities: # index out of bounds or empty account @@ -211,23 +204,23 @@ class GarminClient(object): timestamp_utc = dateutil.parser.parse(activity["startTimeGMT"]) # make sure UTC timezone gets set 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))) return entries @require_session def get_activity_summary(self, activity_id): - """Return a summary about a given activity. The - summary contains several statistics, such as duration, GPS starting - point, GPS end point, elevation gain, max heart rate, max pace, max - speed, etc). + """Return a summary about a given activity. + The summary contains several statistics, such as duration, GPS starting + point, GPS end point, elevation gain, max heart rate, max pace, max speed, etc). :param activity_id: Activity identifier. :type activity_id: int :returns: The activity summary as a JSON 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: log.error(u"failed to fetch json summary for activity {}: {}\n{}".format( activity_id, response.status_code, response.text)) @@ -247,7 +240,8 @@ class GarminClient(object): :rtype: dict """ # 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: raise Exception(u"failed to fetch json activityDetails for {}: {}\n{}".format( 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. :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 # and is the one used when exporting through the Garmin # 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 # 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)) return response.text - @require_session def get_activity_tcx(self, activity_id): """Return a TCX (Training Center XML) representation of a @@ -298,7 +292,8 @@ class GarminClient(object): :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: return None if response.status_code != 200: @@ -306,7 +301,6 @@ class GarminClient(object): activity_id, response.status_code, response.text)) return response.text - def get_original_activity(self, activity_id): """Return the original file that was uploaded for an activity. 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. :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 # file. As of lately, the endpoint appears to have started to # respond with 500 "NullPointerException" on attempts to download a # .fit file for an activity without one. if response.status_code in [404, 500]: # Manually entered activity, no file source available - return (None,None) + return None, None if response.status_code != 200: raise Exception( 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 # activity_id (should be the only entry!) - zip = zipfile.ZipFile(BytesIO(response.content), mode="r") - for path in zip.namelist(): + zip_file = zipfile.ZipFile(BytesIO(response.content), mode="r") + for path in zip_file.namelist(): fn, ext = os.path.splitext(path) - if fn==str(activity_id): - return ext[1:], zip.open(path).read() - return (None,None) - + if fn == str(activity_id): + return ext[1:], zip_file.open(path).read() + return None, None def get_activity_fit(self, activity_id): """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', # this activity was uploaded in a different format (e.g. gpx/tcx) # and cannot be exported to fit - return orig_file if fmt=='fit' else None + return orig_file if fmt == 'fit' else None @require_session 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 """ - if isinstance(file, basestring): + if isinstance(file, str): file = open(file, "rb") # guess file type if unspecified fn = os.path.basename(file.name) _, ext = os.path.splitext(fn) if format is None: - if ext.lower() in ('.gpx','.tcx','.fit'): + if ext.lower() in ('.gpx', '.tcx', '.fit'): format = ext.lower()[1:] else: raise Exception(u"could not guess file type for {}".format(fn)) @@ -394,15 +388,15 @@ class GarminClient(object): # check response and get activity ID try: j = response.json()["detailedImportResult"] - except (json.JSONDecodeException, KeyError): + except (json.JSONDecodeError, KeyError): raise Exception(u"failed to upload {} for activity: {}\n{}".format( 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( 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( format, len(j["successes"]))) @@ -410,14 +404,20 @@ class GarminClient(object): # add optional fields data = {} - if name is not None: data['activityName'] = name - if description is not None: data['description'] = name - if activity_type is not None: data['activityTypeDTO'] = {"typeKey": activity_type} - if private: data['privacy'] = {"typeKey": "private"} + if name is not None: + data['activityName'] = name + if description is not None: + data['description'] = name + if activity_type is not None: + data['activityTypeDTO'] = {"typeKey": activity_type} + if private: + data['privacy'] = {"typeKey": "private"} if data: data['activityId'] = activity_id - 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) + 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) if response.status_code != 204: raise Exception(u"failed to set metadata for activity {}: {}\n{}".format( activity_id, response.status_code, response.text)) diff --git a/garminexport/incremental_backup.py b/garminexport/incremental_backup.py index 25552fe..181cf8b 100644 --- a/garminexport/incremental_backup.py +++ b/garminexport/incremental_backup.py @@ -35,7 +35,7 @@ def incremental_backup(username: str, """ # if no --format was specified, all formats are to be backed up 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): os.makedirs(backup_dir) @@ -50,23 +50,23 @@ def incremental_backup(username: str, with GarminClient(username, password) as client: # 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)) - 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) 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): 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))) try: garminexport.backup.download(client, activity, retryer, backup_dir, format) except Exception as e: - log.error(u"failed with exception: %s", e) + log.error("failed with exception: {}".format(e)) if not ignore_errors: raise diff --git a/garminexport/retryer.py b/garminexport/retryer.py index df98218..b589c81 100644 --- a/garminexport/retryer.py +++ b/garminexport/retryer.py @@ -1,15 +1,14 @@ import abc - -from datetime import datetime -from datetime import timedelta import logging import time +from datetime import datetime +from datetime import timedelta log = logging.getLogger(__name__) + class GaveUpError(Exception): - """Raised by a :class:`Retryer` that has exceeded its maximum number - of retries.""" + """Raised by a :class:`Retryer` that has exceeded its maximum number of retries.""" pass @@ -22,8 +21,7 @@ class DelayStrategy(object): def next_delay(self, attempts): """Returns the time to wait before the next attempt. - :param attempts: The total number of (failed) attempts performed thus - far. + :param attempts: The total number of (failed) attempts performed thus far. :type attempts: int :return: The delay before the next attempt. @@ -33,8 +31,8 @@ class DelayStrategy(object): class FixedDelayStrategy(DelayStrategy): - """A retry :class:`DelayStrategy` that produces a fixed delay between - attempts.""" + """A retry :class:`DelayStrategy` that produces a fixed delay between attempts.""" + def __init__(self, delay): """ :param delay: Attempt delay. @@ -56,7 +54,7 @@ class ExponentialBackoffDelayStrategy(DelayStrategy): def __init__(self, initial_delay): """ :param initial_delay: Initial delay. - :type delay: `timedelta` + :type initial_delay: `timedelta` """ self.initial_delay = initial_delay @@ -68,25 +66,21 @@ class ExponentialBackoffDelayStrategy(DelayStrategy): class NoDelayStrategy(FixedDelayStrategy): - """A retry :class:`DelayStrategy` that doesn't introduce any delay between - attempts.""" + """A retry :class:`DelayStrategy` that doesn't introduce any delay between attempts.""" + def __init__(self): super(NoDelayStrategy, self).__init__(timedelta(seconds=0)) - - class ErrorStrategy(object): """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 - (re)trying.""" + suppressed and which errors are to be re-raised and thereby end the (re)trying.""" __metaclass__ = abc.ABCMeta @abc.abstractmethod def should_suppress(self, error): """Called after an attempt that raised an exception to determine if - that error should be suppressed (continue retrying) or be re-raised - (and end the retrying). + that error should be suppressed (continue retrying) or be re-raised (and end the retrying). :param error: Error that was raised from an attempt. """ @@ -122,13 +116,14 @@ class StopStrategy(object): class NeverStopStrategy(StopStrategy): """A :class:`StopStrategy` that never gives up.""" + def should_continue(self, attempts, elapsed_time): return True class MaxRetriesStopStrategy(StopStrategy): - """A :class:`StopStrategy` that gives up after a certain number of - retries.""" + """A :class:`StopStrategy` that gives up after a certain number of retries.""" + def __init__(self, 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 the retrying ends with that error). """ + def __init__( self, returnval_predicate=lambda returnval: True, @@ -180,7 +176,6 @@ class Retryer(object): self.stop_strategy = stop_strategy self.error_strategy = error_strategy - def call(self, function, *args, **kw): """Calls the given `function`, with the given arguments, repeatedly until either (1) a satisfactory result is obtained (as indicated by @@ -200,24 +195,21 @@ class Retryer(object): while True: try: attempts += 1 - log.info('{%s}: attempt %d ...', name, attempts) + log.info('{{}}: attempt {} ...'.format(name, attempts)) returnval = function(*args, **kw) if self.returnval_predicate(returnval): # return value satisfies predicate, we're done! - log.debug('{%s}: success: "%s"', name, returnval) + log.debug('{{}}: success: "{}"'.format(name, returnval)) return returnval - log.debug('{%s}: failed: return value: %s', name, returnval) + log.debug('{{}}: failed: return value: {}'.format(name, returnval)) except Exception as e: if not self.error_strategy.should_suppress(e): raise e - log.debug('{%s}: failed: error: %s', name, str(e)) + log.debug('{{}}: failed: error: {}'.format(name, e)) elapsed_time = datetime.now() - start # should we make another attempt? if not self.stop_strategy.should_continue(attempts, elapsed_time): - raise GaveUpError( - '{%s}: gave up after %d failed attempt(s)' % - (name, attempts)) + raise GaveUpError('{{}}: gave up after {} failed attempt(s)'.format(name, attempts)) delay = self.delay_strategy.next_delay(attempts) - log.info('{%s}: waiting %d seconds for next attempt' % - (name, delay.total_seconds())) + log.info('{{}}: waiting {} seconds for next attempt'.format(name, delay.total_seconds())) time.sleep(delay.total_seconds()) diff --git a/get_activity.py b/get_activity.py index 48f20da..44b5348 100755 --- a/get_activity.py +++ b/get_activity.py @@ -6,7 +6,6 @@ import argparse import getpass import logging import os -import sys from datetime import timedelta import dateutil.parser @@ -22,8 +21,8 @@ log = logging.getLogger(__name__) if __name__ == "__main__": parser = argparse.ArgumentParser( - description=("Downloads one particular activity for a given " - "Garmin Connect account.")) + description="Downloads one particular activity for a given Garmin Connect account.") + # positional args parser.add_argument( "username", metavar="", type=str, help="Account user name.") @@ -31,29 +30,30 @@ if __name__ == "__main__": "activity", metavar="", type=int, help="Activity ID.") parser.add_argument( "format", metavar="", type=str, - help="Export format (one of: {}).".format( - garminexport.backup.export_formats)) + help="Export format (one of: {}).".format(garminexport.backup.export_formats)) # optional args parser.add_argument( "--password", type=str, help="Account password.") parser.add_argument( "--destination", metavar="DIR", type=str, - help=("Destination directory for downloaded activity. Default: " - "./activities/"), default=os.path.join(".", "activities")) + help="Destination directory for downloaded activity. Default: ./activities/", + default=os.path.join(".", "activities")) parser.add_argument( "--log-level", metavar="LEVEL", type=str, - help=("Desired log output level (DEBUG, INFO, WARNING, ERROR). " - "Default: INFO."), default="INFO") + help="Desired log output level (DEBUG, INFO, WARNING, ERROR). Default: INFO.", + default="INFO") args = parser.parse_args() - if not args.log_level in LOG_LEVELS: - raise ValueError("Illegal log-level argument: {}".format( - args.log_level)) - if not args.format in garminexport.backup.export_formats: + + if args.log_level not in LOG_LEVELS: + raise ValueError("Illegal log-level argument: {}".format(args.log_level)) + + if args.format not in garminexport.backup.export_formats: raise ValueError( - "Uncrecognized export format: '{}'. Must be one of {}".format( + "Unrecognized export format: '{}'. Must be one of {}".format( args.format, garminexport.backup.export_formats)) + logging.root.setLevel(LOG_LEVELS[args.log_level]) try: @@ -62,20 +62,18 @@ if __name__ == "__main__": if not args.password: args.password = getpass.getpass("Enter password: ") + with GarminClient(args.username, args.password) as client: log.info("fetching activity {} ...".format(args.activity)) summary = client.get_activity_summary(args.activity) - # set up a retryer that will handle retries of failed activity - # downloads + # set up a retryer that will handle retries of failed activity downloads retryer = Retryer( - delay_strategy=ExponentialBackoffDelayStrategy( - initial_delay=timedelta(seconds=1)), + delay_strategy=ExponentialBackoffDelayStrategy(initial_delay=timedelta(seconds=1)), stop_strategy=MaxRetriesStopStrategy(5)) - starttime = dateutil.parser.parse(summary["summaryDTO"]["startTimeGMT"]) + start_time = dateutil.parser.parse(summary["summaryDTO"]["startTimeGMT"]) 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: - exc_type, exc_value, exc_traceback = sys.exc_info() - log.error(u"failed with exception: %s", e) + log.error("failed with exception: {}".format(e)) raise diff --git a/samples/lab.py b/samples/lab.py index 76fd088..425fe4c 100644 --- a/samples/lab.py +++ b/samples/lab.py @@ -10,10 +10,9 @@ Garmin Connect. import argparse import getpass -from garminexport.garminclient import GarminClient -import json import logging -import sys + +from garminexport.garminclient import GarminClient logging.basicConfig(level=logging.INFO, format="%(asctime)-15s [%(levelname)s] %(message)s") log = logging.getLogger(__name__) diff --git a/samples/sample.py b/samples/sample.py index 9438802..7df4afc 100755 --- a/samples/sample.py +++ b/samples/sample.py @@ -2,10 +2,10 @@ import argparse import getpass -from garminexport.garminclient import GarminClient import json import logging -import sys + +from garminexport.garminclient import GarminClient logging.basicConfig(level=logging.INFO, format="%(asctime)-15s [%(levelname)s] %(message)s") log = logging.getLogger(__name__) @@ -36,12 +36,12 @@ if __name__ == "__main__": latest_activity, latest_activity_start = activity_ids[0] activity = client.get_activity_summary(latest_activity) - log.info(u"activity id: %s", activity["activity"]["activityId"]) - log.info(u"activity name: '%s'", activity["activity"]["activityName"]) - log.info(u"activity description: '%s'", activity["activity"]["activityDescription"]) + log.info("activity id: {}".format(activity["activity"]["activityId"])) + log.info("activity name: '{}'".format(activity["activity"]["activityName"])) + log.info("activity description: '{}'".format(activity["activity"]["activityDescription"])) log.info(json.dumps(client.get_activity_details(latest_activity), indent=4)) log.info(client.get_activity_gpx(latest_activity)) except Exception as e: - log.error(u"failed with exception: %s", e) + log.error("failed with exception: {}".format(e)) finally: log.info("done") diff --git a/setup.py b/setup.py index f19737a..c98ac8d 100644 --- a/setup.py +++ b/setup.py @@ -7,7 +7,8 @@ from distutils.core import setup setup(name="Garmin Connect activity exporter", 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(), author="Peter GardfjÀll", author_email="peter.gardfjall.work@gmail.com", diff --git a/upload_activity.py b/upload_activity.py index 462b9ff..10ca513 100755 --- a/upload_activity.py +++ b/upload_activity.py @@ -5,7 +5,6 @@ Connect account. import argparse import getpass import logging -import sys from garminexport.garminclient import GarminClient 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") log = logging.getLogger(__name__) - if __name__ == "__main__": 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 parser.add_argument( "username", metavar="", 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.)") parser.add_argument( "--log-level", metavar="LEVEL", type=str, - help=("Desired log output level (DEBUG, INFO, WARNING, ERROR). " - "Default: INFO."), default="INFO") + help="Desired log output level (DEBUG, INFO, WARNING, ERROR). Default: INFO.", + default="INFO") 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.") - if not args.log_level in LOG_LEVELS: - raise ValueError("Illegal log-level argument: {}".format( - args.log_level)) + + if args.log_level not in LOG_LEVELS: + raise ValueError("Illegal log-level argument: {}".format(args.log_level)) + logging.root.setLevel(LOG_LEVELS[args.log_level]) try: if not args.password: args.password = getpass.getpass("Enter password: ") + with GarminClient(args.username, args.password) as client: for activity in args.activity: log.info("uploading activity file {} ...".format(activity.name)) 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: log.error("upload failed: {!r}".format(e)) else: 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