source: OpenWorkouts-current/ow/fit.py @ 737eb6c

from fitparse import FitFile
import gpxpy
import gpxpy.gpx

try:  # pragma: no cover
    import lxml.etree as mod_etree  # Load LXML or fallback to cET or ET
except ImportError:  # pragma: no cover
    try:
        import xml.etree.cElementTree as mod_etree
    except ImportError:
        import xml.etree.ElementTree as mod_etree

from ow.utilities import semicircles_to_degrees


class Fit(object):
    """
    A fit object, that can be used to load, parse, convert, etc ANT fit
    files
    """
    def __init__(self, path):
        self.path = path
        # FitFile object containing all data after opening the file with
        # python-fitparse
        self.obj = FitFile(path)
        # data is a temporary place to store info we got from the fit file,
        # so we don't have to loop over data in self.obj multiple times
        self.data = {}

    def load(self):
        """
        Load contents from the fit file
        """
        # get some basic info from the fit file
        #
        # IMPORTANT: only single-session fit files are supported right now
        sessions = list(self.obj.get_messages('session'))
        session = sessions[0]
        values = session.get_values()
        self.data['sport'] = values.get('sport', None)

        # in some garmin devices (520), you can have named profiles, which
        # can be used for different bikes or for different types of workouts
        # (training/races/etc)
        self.data['profile'] = values.get('unknown_110', None)

        # naive datetime object
        self.data['start'] = values.get('start_time', None)

        # seconds
        self.data['duration'] = values.get('total_timer_time', None)
        self.data['elapsed'] = values.get('total_elapsed_time', None)

        # meters
        self.data['distance'] = values.get('total_distance', None)

        # meters
        self.data['uphill'] = values.get('total_ascent', None)
        self.data['downhill'] = values.get('total_descent', None)

        self.data['calories'] = values.get('total_calories', None)

        # store here a list with all hr values
        self.data['hr'] = []
        self.data['min_hr'] = None
        self.data['max_hr'] = values.get('max_heart_rate', None)
        self.data['avg_hr'] = values.get('avg_heart_rate', None)

        # store here a list with all cad values
        self.data['cad'] = []
        self.data['min_cad'] = None
        self.data['max_cad'] = values.get('max_cadence', None)
        self.data['avg_cad'] = values.get('avg_cadence', None)

        self.data['max_speed'] = values.get('enhanced_max_speed', None)
        if self.data['max_speed'] is None:
            self.data['max_speed'] = values.get('max_speed', None)
            if self.data['max_speed'] is not None:
                self.data['max_speed'] = self.data['max_speed'] / 1000

        self.data['avg_speed'] = values.get('enhanced_avg_speed', None)
        if self.data['avg_speed'] is None:
            self.data['avg_speed'] = values.get('avg_speed', None)
            if self.data['avg_speed'] is not None:
                self.data['avg_speed'] = self.data['avg_speed'] / 1000

        # no temp values yet
        # store here a list with all temp values
        self.data['atemp'] = []
        self.data['min_atemp'] = None
        self.data['max_atemp'] = 0
        self.data['avg_atemp'] = 0

    @property
    def name(self):
        """
        Build a name based on some info from the fit file
        """
        if self.data['profile']:
            return self.data['profile'] + ' ' + self.data['sport']
        return self.data['sport']

    def _calculate_avg_atemp(self):
        temps = [t[0] for t in self.data['atemp']]
        if temps:
            self.data['avg_atemp'] = int(round(sum(temps)/len(temps)))
        return self.data['avg_atemp']

    @property
    def gpx(self):
        """
        Return valid xml-formatted gpx contents from the loaded fit file
        """
        # Create a gpx object, adding some basic metadata to it.
        #
        # The schema namespaces info is very important, in order to support
        # all the data we will extract from the fit file.

        gpx = gpxpy.gpx.GPX()
        gpx.creator = 'OpenWorkouts'
        gpx.schema_locations = [
            'http://www.topografix.com/GPX/1/1',
            'http://www.topografix.com/GPX/1/1/gpx.xsd',
            'http://www.garmin.com/xmlschemas/GpxExtensions/v3',
            'http://www.garmin.com/xmlschemas/GpxExtensionsv3.xsd',
            'http://www.garmin.com/xmlschemas/TrackPointExtension/v1',
            'http://www.garmin.com/xmlschemas/TrackPointExtensionv1.xsd'
        ]
        garmin_schemas = 'http://www.garmin.com/xmlschemas'
        gpx.nsmap['gpxtpx'] = garmin_schemas + '/TrackPointExtension/v1'
        gpx.nsmap['gpxx'] = garmin_schemas + '/GpxExtensions/v3'

        # Create first track in our GPX:
        gpx_track = gpxpy.gpx.GPXTrack()
        gpx_track.name = self.name
        gpx.tracks.append(gpx_track)

        # Create first segment in our GPX track:
        gpx_segment = gpxpy.gpx.GPXTrackSegment()
        gpx_track.segments.append(gpx_segment)

        # namespace for the additional gpx extensions
        # (temperature, cadence, heart rate)
        namespace = '{gpxtpx}'
        # nsmap = {'ext': namespace[1:-1]}

        # loop over the 'record' messages in the fit file, which are the actual
        # data "recorded" by the device
        for message in self.obj.get_messages('record'):
            values = message.get_values()

            # Create an entry where we add the "gpx extensions" data, that will
            # be added to the track point later.
            extensions_root = mod_etree.Element(
                namespace + 'TrackPointExtension')

            atemp = mod_etree.Element(namespace + 'atemp')
            atemp_value = values.get('temperature', 0)
            atemp.text = str(atemp_value)
            atemp.tail = ''
            extensions_root.append(atemp)
            self.data['atemp'].append((atemp_value, values['timestamp']))
            if self.data['min_atemp'] is None:
                self.data['min_atemp'] = atemp_value
            elif atemp_value < self.data['min_atemp']:
                self.data['min_atemp'] = atemp_value
            if atemp_value > self.data['max_atemp']:
                self.data['max_atemp'] = atemp_value

            hr = mod_etree.Element(namespace + 'hr')
            hr_value = values.get('heart_rate', 0)
            if hr_value == 0:
                # don't allow hr values of 0, something may have gone wrong
                # with the heart rate monitor, so we use the previous value,
                # if any
                if self.data['hr']:
                    hr_value = self.data['hr'][-1][0]
            hr.text = str(hr_value)
            hr.tail = ''
            extensions_root.append(hr)
            self.data['hr'].append((hr_value, values['timestamp']))
            # min_hr None means we are on the first value, set it as min, we
            # don't want 0 values here. If it is not None, check if the current
            # value is lower than the current min, update accordingly
            if self.data['min_hr'] is None or hr_value < self.data['min_hr']:
                self.data['min_hr'] = hr_value

            cad = mod_etree.Element(namespace + 'cad')
            cad_value = values.get('cadence', 0)
            cad.text = str(cad_value)
            cad.tail = ''
            extensions_root.append(cad)
            self.data['cad'].append((cad_value, values['timestamp']))
            if self.data['min_cad'] is None:
                self.data['min_cad'] = cad_value
            elif cad_value < self.data['min_cad']:
                self.data['min_cad'] = cad_value

            # Create a gpx track point, that holds the "recorded" geo and speed
            # data, as well as the "gpx extensions" object
            if values.get('position_lat') and values.get('position_long'):
                track_point = gpxpy.gpx.GPXTrackPoint(
                    latitude=semicircles_to_degrees(values['position_lat']),
                    longitude=semicircles_to_degrees(values['position_long']),
                    time=values['timestamp']
                )

                if 'enhanced_altitude' in values.keys():
                    track_point.elevation = values['enhanced_altitude']

                if 'enhanced_speed' in values.keys():
                    track_point.speed = values['enhanced_speed']

                track_point.extensions.append(extensions_root)

                gpx_segment.points.append(track_point)

        # if the fit file has temperature data, calculate the avg
        if self.data['atemp']:
            self._calculate_avg_atemp()

        # return a string containing the xml representation of the gpx object
        return gpx.to_xml()