improve: remove adjust, shift rows to adjust frame index

Author: FelixAbrahamsson

README.rst

@@ -19,7 +19,6 @@ Usage
     positions = (
         kmm.Positions.from_path(kmm_path)
         .sync_frame_index(header)
-        .adjust(kmm.PositionAdjustment.WIRE_CAMERA, header)
         .geodetic()
     )
 

kmm/__init__.py

@@ -1,6 +1,5 @@
 from kmm.functional_base import FunctionalBase
 from kmm.car_direction import CarDirection
-from kmm.position_adjustment import PositionAdjustment
 from kmm import positions
 from kmm import header
 from kmm.positions.positions import Positions

kmm/position_adjustment.py

@@ -1,5 +1,5 @@
 from kmm.positions.read_kmm import read_kmm
 from kmm.positions.read_kmm2 import read_kmm2
 from kmm.positions.geodetic import geodetic
-from kmm.positions.wire_camera_positions import wire_camera_positions
+from kmm.positions.sync_frame_index import sync_frame_index
 from kmm.positions.positions import Positions

kmm/positions/__init__.py

@@ -32,7 +32,6 @@ def from_path(path: Path):
     def read_sync_adjust(
         kmm_path: Path,
         header_path: Path,
-        adjustment: kmm.PositionAdjustment = kmm.PositionAdjustment.WIRE_CAMERA,
     ):
         """
         Loads positions from .kmm or .kmm2 file + .hdr file, then performs
@@ -42,28 +41,11 @@ def read_sync_adjust(
         return (
             Positions.from_path(kmm_path)
             .sync_frame_index(header)
-            .adjust(adjustment, header)
             .geodetic()
         )
 
     def sync_frame_index(self, header):
-        return self.replace(
-            dataframe=self.dataframe.assign(
-                frame_index=(
-                    (
-                        self.dataframe["centimeter"]
-                        + header.position
-                        - header.sync
-                    ) / 10
-                ).astype(int)
-            )
-        )
-
-    def adjust(self, adjustment: kmm.PositionAdjustment, header):
-        if adjustment == kmm.PositionAdjustment.WIRE_CAMERA:
-            return kmm.positions.wire_camera_positions(self, header.car_direction)
-        else:
-            raise ValueError(f"Unknown adjustment option {adjustment}")
+        return kmm.positions.sync_frame_index(self, header)
 
     def geodetic(self):
         return kmm.positions.geodetic(self)

kmm/positions/positions.py

@@ -0,0 +1,53 @@
+import numpy as np
+from pydantic import validate_arguments
+
+from kmm import CarDirection
+from kmm.positions.positions import Positions
+from kmm.header.header import Header
+
+
+@validate_arguments(config=dict(arbitrary_types_allowed=True))
+def sync_frame_index(positions: Positions, header: Header):
+
+    frame_index = (
+        (
+            positions.dataframe["centimeter"].values
+            + header.position
+            - header.sync
+        ) / 10
+    ).astype(int)
+    pad = np.zeros(8)
+    pad.fill(np.nan)
+
+    if header.car_direction == CarDirection.A:
+        frame_index = np.concatenate([pad, frame_index[8:]])
+    elif header.car_direction == CarDirection.B:
+        frame_index = np.concatenate([frame_index[:-8], pad])
+    else:
+        raise ValueError(header.car_direction)
+
+    return positions.replace(
+        dataframe=positions.dataframe.assign(frame_index=frame_index)
+    )
+
+
+def test_sync_frame_index_kmm():
+    from kmm import Header
+
+    positions = Positions.from_path("tests/ascending_B.kmm")
+    header = Header.from_path("tests/ascending_B.hdr")
+    assert (
+        sync_frame_index(positions, header)
+        .dataframe["frame_index"].iloc[0] == 659
+    )
+
+
+def test_sync_frame_index_kmm2():
+    from kmm import Header
+
+    positions = Positions.from_path("tests/ascending_B.kmm2")
+    header = Header.from_path("tests/ascending_B.hdr")
+    assert (
+        sync_frame_index(positions, header)
+        .dataframe["frame_index"].iloc[0] == -808
+    )