text: #! /usr/bin/env python
  """A module for authenticating against and communicating with selected
  parts of the Garmin Connect REST API.
  """
  
  import json
  import logging
  import os
  import os.path
  import re
  import sys
  import zipfile
  from datetime import timedelta, datetime
  from builtins import range
  from functools import wraps
  from io import BytesIO
  
  import dateutil
  import dateutil.parser
  import requests
  import cloudscraper
  
  from garminexport.retryer import Retryer, ExponentialBackoffDelayStrategy, MaxRetriesStopStrategy
  
  #
  # Note: For more detailed information about the API services
  # used by this module, log in to your Garmin Connect account
  # through the web browser and visit the API documentation page
  # for the REST service of interest. For example:
  #   https://connect.garmin.com/proxy/activity-service-1.3/index.html
  #   https://connect.garmin.com/proxy/activity-search-service-1.2/index.html
  #
  
  #
  # Other useful references:
  #   https://github.com/cpfair/tapiriik/blob/master/tapiriik/services/GarminConnect/garminconnect.py
  #   https://forums.garmin.com/showthread.php?72150-connect-garmin-com-signin-question/page2
  #
  
  log = logging.getLogger(__name__)
  
  # reduce logging noise from requests library
  logging.getLogger("requests").setLevel(logging.ERROR)
  
  SSO_LOGIN_URL = "https://sso.garmin.com/sso/login"
  """Garmin Connect's Single-Sign On login URL."""
  SSO_SIGNIN_URL = "https://sso.garmin.com/sso/signin"
  """The Garmin Connect Single-Sign On sign-in URL. This is where the login form
  gets POSTed."""
  
  
  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
  
  
  class GarminClient(object):
      """A client class used to authenticate with Garmin Connect and
      extract data from the user account.
  
      Since this class implements the context manager protocol, this object
      can preferably be used together with the with-statement. This will
      automatically take care of logging in to Garmin Connect before any
      further interactions and logging out after the block completes or
      a failure occurs.
  
      Example of use: ::
        with GarminClient("my.sample@sample.com", "secretpassword") as client:
            ids = client.list_activity_ids()
            for activity_id in ids:
                 gpx = client.get_activity_gpx(activity_id)
  
      """
  
      def __init__(self, username, password, user_agent_fn=None):
          """Initialize a :class:`GarminClient` instance.
  
          :param username: Garmin Connect user name or email address.
          :type username: str
          :param password: Garmin Connect account password.
          :type password: str
          :keyword user_agent_fn: A function that, when called, produces a
          `User-Agent` string to be used as `User-Agent` for the remainder of the
          session. If set to None, the default user agent of the http request
          library is used.
          :type user_agent_fn: Callable[[], str]
  
          """
          self.username = username
          self.password = password
          self._user_agent_fn = user_agent_fn
  
          self.session = None
  
  
      def __enter__(self):
          self.connect()
          return self
  
      def __exit__(self, exc_type, exc_value, traceback):
          self.disconnect()
  
      def connect(self):
          self.session = cloudscraper.create_scraper()
          self._authenticate()
  
      def disconnect(self):
          if self.session:
              self.session.close()
              self.session = None
  
      def _authenticate(self):
          log.info("authenticating user ...")
  
          form_data = {
              "username": self.username,
              "password": self.password,
              "embed": "false",
              "_csrf": self._get_csrf_token(),
          }
          headers = {
              'origin': 'https://sso.garmin.com',
          }
          if self._user_agent_fn:
              user_agent = self._user_agent_fn()
              if not user_agent:
                  raise ValueError("user_agent_fn didn't produce a value")
              headers['User-Agent'] = user_agent
  
          auth_response = self.session.post(
              SSO_SIGNIN_URL, headers=headers, params=self._auth_params(), data=form_data)
          log.debug("got auth response: %s", 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)
  
          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: {}: {}\n{}".format(
                      auth_ticket_url, response.status_code, response.text))
  
          # appears like we need to touch base with the main page to complete the
          # login ceremony.
          self.session.get('https://connect.garmin.com/modern')
  
  
      def _get_csrf_token(self):
          """Retrieves a Cross-Site Request Forgery (CSRF) token from Garmin's login
          page. The token is passed along in the login form for increased
          security."""
          log.info("fetching CSRF token ...")
          resp = self.session.get(SSO_LOGIN_URL, params=self._auth_params())
          if resp.status_code != 200:
              raise ValueError("auth failure: could not load {}".format(SSO_LOGIN_URL))
          # extract CSRF token
          csrf_token = re.search(r'<input type="hidden" name="_csrf" value="(\w+)"',
                                 resp.content.decode('utf-8'))
          if not csrf_token:
              raise ValueError("auth failure: no CSRF token in {}".format(SSO_LOGIN_URL))
          return csrf_token.group(1)
  
      def _auth_params(self):
          """A set of request query parameters that need to be present for Garmin to
          accept our login attempt.
          """
          return {
              "service": "https://connect.garmin.com/modern/",
              "gauthHost": "https://sso.garmin.com/sso",
          }
  
  
      @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:
  
            https://connect.garmin.com/modern?ticket=ST-0123456-aBCDefgh1iJkLmN5opQ9R-cas
  
          :param auth_response: HTML response from an auth form submission.
          """
          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).
          :rtype: tuples of (int, datetime)
          """
          ids = []
          batch_size = 100
          # fetch in batches since the API doesn't allow more than a certain
          # number of activities to be retrieved on every invocation
          for start_index in range(0, sys.maxsize, batch_size):
              next_batch = self._fetch_activity_ids_and_ts(start_index, batch_size)
              if not next_batch:
                  break
              ids.extend(next_batch)
          return ids
  
      @require_session
      def _fetch_activity_ids_and_ts(self, start_index, max_limit=100):
          """Return a sequence of activity ids (along with their starting
          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.
  
          :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).
          :rtype: tuples of (int, datetime)
          """
          log.debug("fetching activities %d through %d ...", start_index, start_index + max_limit - 1)
          response = self.session.get(
              "https://connect.garmin.com/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))
          activities = json.loads(response.text)
          if not activities:
              # index out of bounds or empty account
              return []
  
          entries = []
          for activity in activities:
              id = int(activity["activityId"])
              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))
          log.debug("got %d activities.", 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).
  
          :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/proxy/activity-service/activity/{}".format(activity_id))
          if response.status_code != 200:
              log.error(u"failed to fetch json summary for activity %s: %d\n%s",
                        activity_id, response.status_code, response.text)
              raise Exception(u"failed to fetch json summary for activity {}: {}\n{}".format(
                  activity_id, response.status_code, response.text))
          return json.loads(response.text)
  
      @require_session
      def get_activity_details(self, activity_id):
          """Return a JSON representation of a given activity including
          available measurements such as location (longitude, latitude),
          heart rate, distance, pace, speed, elevation.
  
          :param activity_id: Activity identifier.
          :type activity_id: int
          :returns: The activity details as a JSON dict.
          :rtype: dict
          """
          # mounted at xml or json depending on result encoding
          response = self.session.get(
              "https://connect.garmin.com/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))
          return json.loads(response.text)
  
      @require_session
      def get_activity_gpx(self, activity_id):
          """Return a GPX (GPS Exchange Format) representation of a
          given activity. If the activity cannot be exported to GPX
          (not yet observed in practice, but that doesn't exclude the
          possibility), a :obj:`None` value is returned.
  
          :param activity_id: Activity identifier.
          :type activity_id: int
          :returns: The GPX representation of the activity as an XML string
            or ``None`` if the activity couldn't be exported to GPX.
          :rtype: str
          """
          response = self.session.get(
              "https://connect.garmin.com/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))
  
          # 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
          # example be a manually entered activity without any device data.
          if response.status_code in (404, 204):
              return None
          if response.status_code != 200:
              raise Exception(u"failed to fetch GPX for activity {}: {}\n{}".format(
                  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
          given activity. If the activity doesn't have a TCX source (for
          example, if it was originally uploaded in GPX format, Garmin
          won't try to synthesize a TCX file) a :obj:`None` value is
          returned.
  
          :param activity_id: Activity identifier.
          :type activity_id: int
          :returns: The TCX representation of the activity as an XML string
            or ``None`` if the activity cannot be exported to TCX.
          :rtype: str
          """
  
          response = self.session.get(
              "https://connect.garmin.com/proxy/download-service/export/tcx/activity/{}".format(activity_id))
          if response.status_code == 404:
              return None
          if response.status_code != 200:
              raise Exception(u"failed to fetch TCX for activity {}: {}\n{}".format(
                  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,
          if it was entered manually rather than imported from a Garmin
          device) then :obj:`(None,None)` is returned.
  
          :param activity_id: Activity identifier.
          :type activity_id: int
          :returns: A tuple of the file type (e.g. 'fit', 'tcx', 'gpx') and
            its contents, or :obj:`(None,None)` if no file is found.
          :rtype: (str, str)
          """
          response = self.session.get(
              "https://connect.garmin.com/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
          if response.status_code != 200:
              raise Exception(
                  u"failed to get original activity file for {}: {}\n{}".format(
                      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_file = zipfile.ZipFile(BytesIO(response.content), mode="r")
          for path in zip_file.namelist():
              fn, ext = os.path.splitext(path)
              if fn.startswith(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
          doesn't have a FIT source (for example, if it was entered manually
          rather than imported from a Garmin device) a :obj:`None` value is
          returned.
  
          :param activity_id: Activity identifier.
          :type activity_id: int
          :returns: A string with a FIT file for the activity or :obj:`None`
            if no FIT source exists for this activity (e.g., entered manually).
          :rtype: str
          """
          fmt, orig_file = self.get_original_activity(activity_id)
          # 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
  
      @require_session
      def _poll_upload_completion(self, uuid, creation_date):
          """Poll for completion of an upload. If Garmin connect returns
          HTTP status 202 ("Accepted") after initial upload, then we must poll
          until the upload has either succeeded or failed. Raises an
          :class:`Exception` if the upload has failed.
  
          :param uuid: uploadUuid returned on initial upload.
          :type uuid: str
          :param creation_date: creationDate returned from initial upload (e.g.
            "2020-01-01 12:34:56.789 GMT")
          :type creation_date: str
          :returns: Garmin's internalId for the newly-created activity, or
            :obj:`None` if upload is still processing.
          :rtype: int
          """
          response = self.session.get("https://connect.garmin.com/proxy/activity-service/activity/status/{}/{}?_={}".format(
              creation_date[:10], uuid.replace("-",""), int(datetime.now().timestamp()*1000)), headers={"nk": "NT"})
          if response.status_code == 201 and response.headers["location"]:
              # location should be https://connectapi.garmin.com/activity-service/activity/ACTIVITY_ID
              return int(response.headers["location"].split("/")[-1])
          elif response.status_code == 202:
              return None # still processing
          else:
              response.raise_for_status()
  
      @require_session
      def upload_activity(self, file, format=None, name=None, description=None, activity_type=None, private=None):
          """Upload a GPX, TCX, or FIT file for an activity.
  
          :param file: Path or open file
          :param format: File format (gpx, tcx, or fit); guessed from filename if :obj:`None`
          :type format: str
          :param name: Optional name for the activity on Garmin Connect
          :type name: str
          :param description: Optional description for the activity on Garmin Connect
          :type description: str
          :param activity_type: Optional activityType key (lowercase: e.g. running, cycling)
          :type activityType: str
          :param private: If true, then activity will be set as private.
          :type private: bool
          :returns: ID of the newly-uploaded activity
          :rtype: int
          """
  
          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'):
                  format = ext.lower()[1:]
              else:
                  raise Exception(u"could not guess file type for {}".format(fn))
  
          # upload it
          files = dict(data=(fn, file))
          response = self.session.post("https://connect.garmin.com/proxy/upload-service/upload/.{}".format(format),
                                       files=files, headers={"nk": "NT"})
  
          # check response and get activity ID
          try:
              j = response.json()["detailedImportResult"]
          except (json.JSONDecodeError, KeyError):
              raise Exception(u"failed to upload {} for activity: {}\n{}".format(
                  format, response.status_code, response.text))
  
          # single activity, immediate success
          if len(j["successes"]) == 1 and len(j["failures"]) == 0:
              activity_id = j["successes"][0]["internalId"]
  
          # duplicate of existing activity
          elif len(j["failures"]) == 1 and len(j["successes"]) == 0 and response.status_code == 409:
              log.info(u"duplicate activity uploaded, continuing")
              activity_id = j["failures"][0]["internalId"]
  
          # need to poll until success/failure
          elif len(j["failures"]) == 0 and len(j["successes"]) == 0 and response.status_code == 202:
              retryer = Retryer(
                  returnval_predicate=bool,
                  delay_strategy=ExponentialBackoffDelayStrategy(initial_delay=timedelta(seconds=1)),
                  stop_strategy=MaxRetriesStopStrategy(6), # wait for up to 64 seconds (2**6)
                  error_strategy=None
              )
              activity_id = retryer.call(self._poll_upload_completion, j["uploadUuid"]["uuid"], j["creationDate"])
  
          # don't know how to handle multiple activities
          elif len(j["successes"]) > 1:
              raise Exception(u"uploading {} resulted in multiple activities ({})".format(
                  format, len(j["successes"])))
  
          # all other errors
          else:
              raise Exception(u"failed to upload {} for activity: {}\n{}".format(
                  format, response.status_code, j["failures"]))
  
          # add optional fields
          data = {}
          if name is not None:
              data['activityName'] = name
          if description is not None:
              data['description'] = description
          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)
              if response.status_code != 204:
                  raise Exception(u"failed to set metadata for activity {}: {}\n{}".format(
                      activity_id, response.status_code, response.text))
  
          return activity_id
  