bicycle-statistics/gpx_parser/__init__.py

import datetime
import glob
import os
import gpxpy
import gpxpy.gpx
from geopy import distance
from geopy import Point
import pandas as pd

class Segment(object):
    start_time = None
    end_time = None
    distance = 0.0     # [m]

class Track(object):
    start_time = None
    end_time = None
    distance = 0.0     # [m]
    avg_speed = 0.0    # [km/h]
    duration = None

    def __init__(self, raw_track):
        for segment in raw_track.segments:
            seg = Segment()
            for i in range(1, len(segment.points)):
                if self.start_time is None:
                    self.start_time = segment.points[i - 1].time
                if seg.start_time is None:
                    seg.start_time = segment.points[i - 1].time
                seg.end_time = segment.points[i - 1].time
                point1 = Point(str(segment.points[i - 1].latitude) + \
                    ' ' + str(segment.points[i - 1].longitude))
                point2 = Point(str(segment.points[i].latitude) + \
                    ' ' + str(segment.points[i].longitude))
                seg.distance += distance.distance(point1, point2).meters

            try:
                if self.duration is None:
                    self.duration = seg.end_time - seg.start_time
                else:
                    self.duration += seg.end_time - seg.start_time
            except Exception:
                # TODO: Add logging mechanism.
                pass
            self.end_time = seg.end_time
            self.distance += seg.distance
        self.avg_speed = self.distance / self.duration.total_seconds() * 3.6


class Tracks(object):
    __distance = dict()
    __duration = dict()
    __avg_speed = dict()
    __tracks = list()
    __files = list()

    def __init__(self, logger):
        self.logger = logger

    def add(self, filename):
        if filename not in self.__files:
            self.logger.info("Adding file %s.", filename)
            with open(filename, 'r') as f:
                try:
                    self.__files.append(filename)
                    gpx = gpxpy.parse(f)
                    for raw in gpx.tracks:
                        track = Track(raw)
                        self.__tracks.append(track)
                        trk_month = track.start_time.month
                        trk_year = track.start_time.year

                        if trk_year not in self.__distance:
                            self.__distance[trk_year] = {1: 0, 2: 0, 3: 0, 4: 0, 5: 0, 6: 0, 7: 0, 8: 0, 9: 0, 10: 0, 11: 0, 12: 0}
                        self.__distance[trk_year][trk_month] += track.distance / 1000

                        if trk_year not in self.__duration:
                            self.__duration[trk_year] = {1: 0, 2: 0, 3: 0, 4: 0, 5: 0, 6: 0, 7: 0, 8: 0, 9: 0, 10: 0, 11: 0, 12: 0}
                        self.__duration[trk_year][trk_month] += track.duration.total_seconds()

                        if trk_year not in self.__avg_speed:
                            self.__avg_speed[trk_year] = {1: 0, 2: 0, 3: 0, 4: 0, 5: 0, 6: 0, 7: 0, 8: 0, 9: 0, 10: 0, 11: 0, 12: 0}
                        self.__avg_speed[trk_year][trk_month] = self.__distance[trk_year][trk_month] / (self.__duration[trk_year][trk_month] / 3600)
                    self.logger.info("Adding done.")
                except Exception as exception:
                    # TODO: Add logging mechanism.
                    pass

    def years(self):
        ret = None
        try:
            ret = sorted(self.__distance.keys())
        except Exception:
            pass
        return ret

    def distances(self, year):
        ret = 0
        try:
            ret = self.__distance[year].values()
        except Exception:
            pass
        return ret

    def avg_speeds(self, year):
        ret = None
        try:
            ret = self.__avg_speed[year].values()
        except Exception:
            pass
        return ret

    def tracks(self, start_date, end_date):
        tracks = list()
        dates = pd.date_range(start_date.date(), end_date.date())
        for track in self.__tracks:
            if track.start_time.date() in dates.date:
                tracks.append(track)
        return tracks