Fix fluctuating timestamps in loaded activity file (#30)

Implement #24

firome/codecs/fit/parse.py

@@ -1,9 +1,13 @@
+from datetime import datetime, timezone
+
 from fitdecode import FitReader, FitDataMessage, FIT_FRAME_DATA
 
-from ..zip import unzip
 from ..errors import UnsupportedFileExtError
-from ...logger import LOGGER
+from ..zip import unzip
 from ...classes.points import DataPoint
+from ...logger import LOGGER
+
+__max_delta_days = 120  # expected activity date range from now
 
 
 def parse_fit(src: str) -> list[DataPoint]:
@@ -24,6 +28,7 @@ def __init__(self, reader: FitReader):
         self._lap = 1  # track increasing lap value
         self._fit = reader
         self._closed = False
+        self._reference_date = None
 
     def process(self):
         if self._closed:
@@ -34,15 +39,38 @@ def process(self):
         for data in self._fit:
             point = self.__frame_to_point(data)
 
-            if point is None:
+            if point is None or point.distance is None:
                 continue
 
+            if point.timestamp.tzinfo is None:
+                point.timestamp = point.timestamp.replace(tzinfo=timezone.utc)
+
             result.append(point)
 
         self._closed = True
 
+        result.sort(key=lambda x: x.timestamp)
+        result.sort(key=lambda x: x.distance)
+
+        # expecting sorted list here
+        self.__fix_timestamps(result)
+
         return result
 
+    @staticmethod
+    def __fix_timestamps(points: list[DataPoint]):
+        err_index = []
+
+        for i in range(1, len(points) - 1):
+            if not _ts_ok(points[i - 1].timestamp, points[i].timestamp, points[i + 1].timestamp):
+                err_index.append(i)  # not mutating slice during iteration
+
+        if size := len(err_index):
+            LOGGER.error("found %d broken timestamps in activity", size)
+
+        for idx in err_index:
+            points[idx].timestamp = _fix_ts(points[idx].timestamp, points[idx - 1].timestamp, points[idx + 1].timestamp)
+
     def __frame_to_point(self, data: FitDataMessage) -> DataPoint | None:
         if data.frame_type != FIT_FRAME_DATA:
             return None
@@ -57,11 +85,36 @@ def __frame_to_point(self, data: FitDataMessage) -> DataPoint | None:
             return None
 
         return DataPoint(
-            timestamp=data.get_value("timestamp"),
+            timestamp=data.get_value("timestamp").replace(tzinfo=timezone.utc),
             distance=data.get_value("distance"),
             speed=data.get_value("speed", fallback=None),
             power=data.get_value("power", fallback=None),
             heart_rate=data.get_value("heart_rate", fallback=None),
             cadence=data.get_value("cadence"),
             lap=self._lap,
         )
+
+
+def _ts_ok(prev, curr, nxt):
+    if prev > curr:
+        return False
+
+    # if next < prev, then next is broken
+    if (curr > nxt) and (nxt >= prev):
+        return False
+
+    if (curr - prev).days > 0:
+        return False
+
+    return True
+
+
+def _fix_ts(current: datetime, normal_previous: datetime, next_ts: datetime) -> datetime:
+    current_fix_date = current.replace(year=normal_previous.year, month=normal_previous.month, day=normal_previous.day)
+
+    if _ts_ok(normal_previous, current_fix_date, next_ts):
+        LOGGER.debug("only date is broken: %s", current.ctime())
+        return current_fix_date
+
+    LOGGER.debug("full replace %s with %s", current.ctime(), normal_previous.ctime())
+    return normal_previous

firome/ui/__main__.py

@@ -1,12 +1,31 @@
+import argparse
+import logging
 import sys
 
 from PySide6.QtWidgets import QApplication
 
+from firome import __version__
+from firome.logger import LOGGER
 from firome.ui.main import MainWindow
 
 app = QApplication(sys.argv)
 
+parser = argparse.ArgumentParser(
+    description="Combines GPX file with training data with GPX file with position data based on the distance"
+)
+parser.add_argument("--debug", action="store_true", help="Enable debug logging")
+parser.add_argument("--version", action="store_true", help="Print app version")
+
 if __name__ == "__main__":
+    args = parser.parse_args()
+
+    if args.version:
+        print(__version__)
+        sys.exit(0)
+
+    if args.debug:
+        LOGGER.setLevel(logging.DEBUG)
+
     mw = MainWindow()
     mw.show()