fitdump

#!/usr/bin/env python
from __future__ import print_function

import argparse
import codecs
import datetime
import itertools
import json
import os.path
import sys
import types

# Python 2 compat
try:
    BrokenPipeError
except NameError:
    import socket
    BrokenPipeError = socket.error

import fitparse


def format_message(num, message, options):
    s = ["{}. {}".format(num, message.name)]
    if options.with_defs:
        s.append(' [{}]'.format(message.type))
    s.append('\n')

    if message.type == 'data':
        for field_data in message:
            s.append(' * {}: {}'.format(field_data.name, field_data.value))
            if field_data.units:
                s.append(' [{}]'.format(field_data.units))
            s.append('\n')

    s.append('\n')
    return "".join(s)


def sort_record_types(record_type: str = '') -> int:
    # Header order
    record_order = {'file_id': 0,
                    'sport': 1,
                    'workout': 2,
                    'activity': 3,
                    'session': 4,
                    'lap': 5,
                    'device_info': 10,
                    'hr_zone': 11,
                    'power_zone': 12,
                    'record': 20,
                    'event': 30,
                    'field_description': 40,
                    'developer_data_id': 100}
    DEFAULT_ORDER = max(record_order.values()) + 1
    try:
        return record_order[record_type]
    except KeyError:
        # Not found, last
        return DEFAULT_ORDER


def field_header_format(field_data) -> str:
    s = field_data.name
    if field_data.units:
        s += ' [{}]'.format(field_data.units)
    return s


def field_data_format(messages, message_header):
    for message in messages:
        # Format this line
        s = [''] * len(message_header)
        for field_data in message:
            s[message_header.index(field_data.name)] = str(field_data.value)
        yield ','.join(s) + '\n'


def write_csv(options, records):
    records = list(records)
    # Collect all the message types in the file
    record_types = set([message.name for message in records])
    # Two-stage sort, header info, then alphabetic
    record_types = list(sorted(sorted(record_types), key=sort_record_types))

    # Cache for all lines
    all_lines = list()

    for write_type in record_types:
        # Collect all messages of this type
        write_messages = [message for message in records if message.name == write_type]
        # Collect the data field information in alphabetic order
        field_names_header = sorted(set([field_header_format(field_data) for message in write_messages for field_data in message]))
        field_names = sorted(set(field_data.name for message in write_messages for field_data in message))
        # Write the overall header line
        all_lines.append(write_type + '\n')
        # Write the field names header line
        all_lines.append(','.join(field_names_header) + '\n')
        # Write the data
        all_lines.extend(field_data_format(write_messages, field_names))
        # Write a blank line
        all_lines.append('\n\n')

    # Write the actual file
    options.output.writelines(all_lines)
    pass


def parse_args(args=None):
    parser = argparse.ArgumentParser(
        description='Dump .FIT files to various formats',
        epilog='python-fitparse version %s' % fitparse.__version__,
    )
    parser.add_argument('-v', '--verbose', action='count', default=0)
    parser.add_argument(
        '-o', '--output', type=argparse.FileType(mode='w'), default="-",
        help='File to output data into (defaults to stdout)',
    )
    parser.add_argument(
        '-t', '--type', choices=('readable', 'json', 'csv'), default='readable',
        '-t', '--type', choices=('readable', 'json', 'gpx', 'csv'), default='readable',
        help='File type to output. (DEFAULT: %(default)s)',
    )
    parser.add_argument(
        '-n', '--name', action='append', help='Message name (or number) to filter',
    )
    parser.add_argument(
        'infile', metavar='FITFILE', type=argparse.FileType(mode='rb'),
        help='Input .FIT file (Use - for stdin)',
    )
    parser.add_argument(
        '--ignore-crc', action='store_const', const=True, help='Some devices seem to write invalid crc\'s, ignore these.'
    )

    options = parser.parse_args(args)

    # Work around argparse.FileType not accepting an `encoding` kwarg in
    # Python < 3.4 by closing and reopening the file (unless it's stdout)
    if options.output is not sys.stdout:
        options.output.close()
        options.output = codecs.open(options.output.name, 'w', encoding='UTF-8')

    options.verbose = options.verbose >= 1
    options.with_defs = (options.type == "readable" and options.verbose)
    options.as_dict = (options.type != "readable" and options.verbose)

    return options


class RecordJSONEncoder(json.JSONEncoder):
    def default(self, obj):
        if isinstance(obj, types.GeneratorType):
            return list(obj)
        if isinstance(obj, (datetime.datetime, datetime.time)):
            return obj.isoformat()
        if isinstance(obj, fitparse.DataMessage):
            return {
                "type": obj.name,
                "data": {
                    data.name: data.value for data in obj
                }
            }
        # Fall back to original to raise a TypeError
        return super(RecordJSONEncoder, self).default(obj)


def generate_gpx(records, filename=None):
    # TODO: Use xml.etree.ElementTree ?

    GPX_TIME_FMT = "%Y-%m-%dT%H:%M:%SZ"  # ISO 8601 format

    records = iter(records)

    # header + open tags
    yield '<?xml version="1.0"?>\n'
    yield '<gpx xmlns="http://www.topografix.com/GPX/1/1" version="1.1" creator="python-fitparse (fitdump)" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://www.topografix.com/GPX/1/1 http://www.topografix.com/GPX/1/1/gpx.xsd">\n'
    yield ' <metadata>\n'

    # file creation time (if a file_id record exists)
    first_record = []
    for message in records:
        if message.name == "file_id":
            for field_data in message:
                if field_data.name == "time_created" and type(field_data.value) == datetime.datetime:
                    yield '  <time>{}</time>\n'.format(field_data.value.strftime(GPX_TIME_FMT))
                    break
            else:
                # No time found in the fields, check next record
                continue
            break
        elif message.name == "record":
            first_record.append(message)
            break

    if filename:
        yield '  <src>{}</src>\n'.format(filename)

    yield ' </metadata>\n'
    yield ' <trk>\n'

    if filename:
        yield '  <name>{}</name>\n'.format(filename)

    yield '  <trkseg>\n'

    # track points
    for message in itertools.chain(first_record, records):
        if message.name != "record":
            continue

        trkpt = {}

        # TODO: support more data types (heart rate, cadence, etc)
        for field_data in message:
            if field_data.name == "position_lat":
                # Units are decimal degrees
                trkpt["lat"] = field_data.value
            elif field_data.name == "position_long":
                # Units are decimal degrees
                trkpt["lon"] = field_data.value
            elif field_data.name == "enhanced_altitude":
                # Units are m
                trkpt["ele"] = field_data.value
            elif field_data.name == "timestamp" and type(field_data.value) == datetime.datetime:
                trkpt["time"] = field_data.value.strftime(GPX_TIME_FMT)
            elif field_data.name == "enhanced_speed" and type(field_data.value) == float:
                # convert from km/h to m/s
                trkpt["speed"] = field_data.value / 3.6

        # Add trackpoint
        if "lat" in trkpt and "lon" in trkpt:
            yield '   <trkpt lat="{lat}" lon="{lon}">\n'.format(**trkpt)
            if "ele" in trkpt:
                yield '    <ele>{ele}</ele>\n'.format(**trkpt)
            if "time" in trkpt:
                yield '    <time>{time}</time>\n'.format(**trkpt)
            if "speed" in trkpt:
                yield '    <speed>{speed}</speed>\n'.format(**trkpt)
            yield '   </trkpt>\n'

    # close tags
    yield '  </trkseg>\n'
    yield ' </trk>\n'
    yield '</gpx>\n'


def main(args=None):
    options = parse_args(args)

    fitfile = fitparse.UncachedFitFile(
        options.infile,
        data_processor=fitparse.StandardUnitsDataProcessor(),
        check_crc=not(options.ignore_crc),
    )
    records = fitfile.get_messages(
        name=options.name,
        with_definitions=options.with_defs,
        as_dict=options.as_dict
    )

    try:
        if options.type == "json":
            json.dump(records, fp=options.output, cls=RecordJSONEncoder)
        elif options.type == "readable":
            options.output.writelines(format_message(n, record, options)
                                      for n, record in enumerate(records, 1))
        elif options.type == 'csv':
            write_csv(options, records)
        elif options.type == "gpx":
            filename = getattr(options.infile, "name")
            if filename:
                filename = os.path.basename(filename)
            options.output.writelines(generate_gpx(records, filename))
    finally:
        try:
            options.output.close()
        except IOError:
            pass


if __name__ == '__main__':
    try:
        main()
    except BrokenPipeError:
        pass