improve: optionally disregard malformed data

Author: FelixAbrahamsson

kmm/header/header.py

@@ -13,13 +13,15 @@ class Header(kmm.FunctionalBase):
 
     @staticmethod
     @validate_arguments
-    def from_path(path: Path):
+    def from_path(path: Path, raise_on_malformed_data: bool = True):
         """
         Loads header data from .hdr file.
         """
         try:
             tree = ElementTree.parse(path)
         except ElementTree.ParseError:
+            if raise_on_malformed_data:
+                raise ValueError("Unable to parse header file, invalid XML.")
             tree = attempt_to_patch_xml(path)
 
         position, sync = kmm.header.position_sync(tree)

kmm/positions/positions.py

@@ -1,13 +1,13 @@
-from kmm.header.header import Header
 from pathlib import Path
+
 import pandas as pd
 from pydantic import validate_arguments
 
 import kmm
+from kmm.header.header import Header
 
 
 class Positions(kmm.FunctionalBase):
-
     dataframe: pd.DataFrame
 
     class Config:
@@ -34,21 +34,29 @@ def read_sync_adjust(
         kmm_path: Path,
         header_path: Path,
         adjustment: kmm.PositionAdjustment = kmm.PositionAdjustment.WIRE_CAMERA,
+        raise_on_malformed_data: bool = True,
     ):
         """
         Loads positions from .kmm or .kmm2 file + .hdr file, then performs
         frame index sync, position adjustment and geodetic coordinate transformation.
         """
-        header = kmm.Header.from_path(header_path)
+        header = kmm.Header.from_path(header_path, raise_on_malformed_data)
         return (
             Positions.from_path(kmm_path)
-            .sync_frame_index(header, adjustment)
+            .sync_frame_index(header, adjustment, raise_on_malformed_data)
             .geodetic()
         )
 
     @validate_arguments
-    def sync_frame_index(self, header: Header, adjustment: kmm.PositionAdjustment):
-        return kmm.positions.sync_frame_index(self, header, adjustment)
+    def sync_frame_index(
+        self,
+        header: Header,
+        adjustment: kmm.PositionAdjustment,
+        raise_on_malformed_data: bool = True,
+    ):
+        return kmm.positions.sync_frame_index(
+            self, header, adjustment, raise_on_malformed_data
+        )
 
     def geodetic(self):
         return kmm.positions.geodetic(self)

kmm/positions/sync_frame_index.py

@@ -2,33 +2,30 @@
 from pydantic import validate_arguments
 
 from kmm import CarDirection, PositionAdjustment
-from kmm.positions.positions import Positions
 from kmm.header.header import Header
+from kmm.positions.positions import Positions
 
 
 @validate_arguments(config=dict(arbitrary_types_allowed=True))
-def sync_frame_index(positions: Positions, header: Header, adjustment: PositionAdjustment):
-
-    validate_meter_increments(positions)
+def sync_frame_index(
+    positions: Positions,
+    header: Header,
+    adjustment: PositionAdjustment,
+    raise_on_malformed_data: bool = True,
+):
+    if raise_on_malformed_data:
+        validate_meter_increments(positions)
     frame_index = (
-        (
-            positions.dataframe["centimeter"].values
-            + header.position
-            - header.sync
-        ) / 10
+        (positions.dataframe["centimeter"].values + header.position - header.sync) / 10
     ).astype(int)
 
     if header.car_direction == CarDirection.A:
-        dataframe = (
-            positions.dataframe
-            .iloc[:-adjustment]
-            .assign(frame_index=frame_index[adjustment:])
+        dataframe = positions.dataframe.iloc[:-adjustment].assign(
+            frame_index=frame_index[adjustment:]
         )
     elif header.car_direction == CarDirection.B:
-        dataframe = (
-            positions.dataframe
-            .iloc[adjustment:]
-            .assign(frame_index=frame_index[:-adjustment])
+        dataframe = positions.dataframe.iloc[adjustment:].assign(
+            frame_index=frame_index[:-adjustment]
         )
     else:
         raise ValueError(f"Unsupported car direction {header.car_direction}")
@@ -39,9 +36,8 @@ def sync_frame_index(positions: Positions, header: Header, adjustment: PositionA
 
 
 def validate_meter_increments(positions):
-    for (track_section, kilometer), group in (
-        positions.dataframe
-        .groupby(["track_section", "kilometer"])
+    for (track_section, kilometer), group in positions.dataframe.groupby(
+        ["track_section", "kilometer"]
     ):
         diffs = np.sign(group["meter"].values[1:] - group["meter"].values[:-1])
         if len(diffs) >= 10 and (diffs > 0).mean() < 0.9 and (diffs < 0).mean() < 0.9: