Handle zfilled track sections

kmm/functional_base.py

@@ -1,15 +1,13 @@
-from pydantic import BaseModel, Extra
+from pydantic import BaseModel, ConfigDict
 
 
 class FunctionalBase(BaseModel):
-    class Config:
-        allow_mutation = False
-        extra = Extra.forbid
+    model_config = ConfigDict(frozen=True, extra="forbid")
 
     def map(self, fn, *args, **kwargs):
         return fn(self, *args, **kwargs)
 
     def replace(self, **kwargs):
-        new_dict = self.dict()
+        new_dict = self.model_dump()
         new_dict.update(**kwargs)
         return type(self)(**new_dict)

kmm/header/header.py

@@ -1,7 +1,7 @@
 from pathlib import Path
 from xml.etree import ElementTree
 
-from pydantic import validate_arguments
+from pydantic import validate_call
 
 import kmm
 
@@ -12,7 +12,7 @@ class Header(kmm.FunctionalBase):
     sync: int
 
     @staticmethod
-    @validate_arguments
+    @validate_call
     def from_path(path: Path, raise_on_malformed_data: bool = True):
         """
         Loads header data from .hdr file.

kmm/positions/positions.py

@@ -1,7 +1,7 @@
 from pathlib import Path
 
 import pandas as pd
-from pydantic import validate_arguments
+from pydantic import ConfigDict, validate_call
 
 import kmm
 from kmm.header.header import Header
@@ -10,11 +10,10 @@
 class Positions(kmm.FunctionalBase):
     dataframe: pd.DataFrame
 
-    class Config:
-        arbitrary_types_allowed = True
+    model_config = ConfigDict(arbitrary_types_allowed=True)
 
     @staticmethod
-    @validate_arguments
+    @validate_call
     def from_path(
         path: Path,
         raise_on_malformed_data: bool = True,
@@ -37,7 +36,7 @@ def from_path(
         return Positions(dataframe=dataframe)
 
     @staticmethod
-    @validate_arguments
+    @validate_call
     def read_sync_adjust(
         kmm_path: Path,
         header_path: Path,
@@ -60,7 +59,7 @@ def read_sync_adjust(
             .geodetic()
         )
 
-    @validate_arguments
+    @validate_call
     def sync_frame_index(
         self,
         header: Header,

kmm/positions/read_kmm.py

@@ -3,10 +3,10 @@
 
 import numpy as np
 import pandas as pd
-from pydantic import validate_arguments
+from pydantic import validate_call
 
 
-@validate_arguments
+@validate_call
 def read_kmm(path: Path, replace_commas: bool = True):
     try:
         if replace_commas:

kmm/positions/read_kmm2.py

@@ -4,7 +4,7 @@
 
 import numpy as np
 import pandas as pd
-from pydantic import validate_arguments
+from pydantic import validate_call
 
 pattern = re.compile(r".+\[.+\]")
 pattern2 = re.compile(r"CMAST")
@@ -46,7 +46,7 @@
 )
 
 
-@validate_arguments
+@validate_call
 def read_kmm2(
     path: Path, raise_on_malformed_data: bool = True, replace_commas: bool = True
 ):
@@ -72,39 +72,39 @@ def read_kmm2(
             file_obj = path
 
         try:
-            df = pd.read_csv(
-                file_obj,
-                skiprows=skiprows,
-                delimiter="\t",
+            parser_kwargs = dict(
+                sep="\t",
                 encoding="latin1" if not replace_commas else None,
-                low_memory=False,
                 header=None,
+                skiprows=skiprows,
+                low_memory=False,
+            )
+            n_columns = len(pd.read_csv(file_obj, **parser_kwargs).columns)
+
+            # Reset file pointer to beginning for StringIO objects
+            if hasattr(file_obj, "seek"):
+                file_obj.seek(0)
+
+            if n_columns > len(expected_columns):
+                columns = [f"{i+1}?" for i in range(len(expected_columns), n_columns)]
+            elif n_columns < len(expected_columns):
+                columns = expected_columns[:n_columns]
+            else:
+                columns = expected_columns
+
+            return pd.read_csv(
+                file_obj,
+                **parser_kwargs,
+                names=columns,
+                dtype=expected_dtypes,
             )
         except pd.errors.EmptyDataError:
             return pd.DataFrame(columns=expected_columns)
-        else:
-            return with_column_names(df)
     except Exception as e:
-        raise ValueError("Unable to parse kmm2 file, invalid csv.") from e
-
+        import traceback
 
-def with_column_names(df):
-    length_diff = len(df.columns) - len(expected_columns)
-    if length_diff > 0:
-        columns = expected_columns + [f"{i}?" for i in range(8, 8 + length_diff)]
-    elif length_diff < 0:
-        columns = expected_columns[:length_diff]
-    else:
-        columns = expected_columns
-    df.columns = columns
-    df.astype(
-        {
-            column: dtype
-            for column, dtype in expected_dtypes.items()
-            if column in df.columns
-        }
-    )
-    return df
+        traceback.print_exc()
+        raise ValueError("Unable to parse kmm2 file, invalid csv.") from e
 
 
 def test_patterns():

kmm/positions/sync_frame_index.py

@@ -1,12 +1,12 @@
 import numpy as np
-from pydantic import validate_arguments
+from pydantic import validate_call
 
 from kmm import CarDirection, PositionAdjustment
 from kmm.header.header import Header
 from kmm.positions.positions import Positions
 
 
-@validate_arguments(config=dict(arbitrary_types_allowed=True))
+@validate_call(config=dict(arbitrary_types_allowed=True))
 def sync_frame_index(
     positions: Positions,
     header: Header,