improve: better errors when crashing

Author: FelixAbrahamsson

kmm/header/car_direction.py

@@ -20,7 +20,11 @@ def car_direction(tree: ElementTree):
     car_direction = re.search(
         r"CarDirection = \"(.*)\"",
         start_tag,
-    ).group(1)
+    )
+    if car_direction is None:
+        raise ValueError("""Did not find a "CarDirection" field under the Start tag.""")
+    else:
+        car_direction = car_direction.group(1)
 
     if not any(car_direction == item.value for item in kmm.CarDirection):
         raise ValueError(f"Unknown car direction {car_direction}")

kmm/header/header.py

@@ -16,7 +16,11 @@ def from_path(path: Path):
         """
         Loads header data from .hdr file.
         """
-        tree = ElementTree.parse(path)
+        try:
+            tree = ElementTree.parse(path)
+        except ElementTree.ParseError as e:
+            raise ValueError("Unable to parse header file, invalid XML.") from e
+
         position, sync = kmm.header.position_sync(tree)
         return Header(
             position=position,

kmm/header/position_sync.py

@@ -12,18 +12,27 @@ def position_sync(tree: ElementTree):
     ]
 
     if len(sync_tags) == 0:
-        raise ValueError("Did not find a sync tag in header.")
+        raise ValueError("Did not find a Sync tag in header.")
 
     sync_tag = sync_tags[0]
 
-    position = int(re.search(
+    position = re.search(
         r"Position = \"(\d*)\"",
         sync_tag,
-    ).group(1))
+    )
+    if position is not None:
+        position = int(position.group(1))
+    else:
+        raise ValueError("""Did not find a "Position" field under the Sync tag.""")
 
-    sync = int(re.search(
+    sync = re.search(
         r"Sync = \"(\d*)\"",
         sync_tag,
-    ).group(1))
+    )
+
+    if sync is not None:
+        sync = int(sync.group(1))
+    else:
+        raise ValueError("""Did not find a "Sync" field under the Sync tag.""")
 
     return position, sync

kmm/positions/read_kmm.py

@@ -6,32 +6,35 @@
 
 @validate_arguments
 def read_kmm(path: Path):
-    return pd.read_csv(
-        path,
-        sep="\t",
-        encoding="latin1",
-        names=[
-            "centimeter",
-            "track_section",
-            "kilometer",
-            "meter",
-            "track_lane",
-            "1?",
-            "2?",
-            "3?",
-            "4?",
-            "5?",
-            "sweref99_tm_x",
-            "sweref99_tm_y",
-            "8?",
-            "9?",
-        ],
-        dtype=dict(
-            track_section=str,
-            kilometer=np.int32,
-            meter=np.int32,
-            track_lane=str,
-            sweref99_tm_x=np.float32,
-            sweref99_tm_y=np.float32,
-        ),
-    )
+    try:
+        return pd.read_csv(
+            path,
+            sep="\t",
+            encoding="latin1",
+            names=[
+                "centimeter",
+                "track_section",
+                "kilometer",
+                "meter",
+                "track_lane",
+                "1?",
+                "2?",
+                "3?",
+                "4?",
+                "5?",
+                "sweref99_tm_x",
+                "sweref99_tm_y",
+                "8?",
+                "9?",
+            ],
+            dtype=dict(
+                track_section=str,
+                kilometer=np.int32,
+                meter=np.int32,
+                track_lane=str,
+                sweref99_tm_x=np.float32,
+                sweref99_tm_y=np.float32,
+            ),
+        )
+    except Exception as e:
+        raise ValueError("Unable to parse kmm2 file, invalid csv.") from e

kmm/positions/read_kmm2.py

@@ -18,45 +18,48 @@ def read_kmm2(path: Path):
         if pattern.match(line) or pattern2.match(line)
     ]
 
-    return pd.read_csv(
-        path,
-        skiprows=[0] + skiprows,
-        delimiter="\t",
-        encoding="latin1",
-        names=[
-            "code",
-            "centimeter",
-            "track_section",
-            "kilometer",
-            "meter",
-            "track_lane",
-            "1?",
-            "2?",
-            "3?",
-            "4?",
-            "sweref99_tm_x",
-            "sweref99_tm_y",
-            "contact_wire_material",
-            "rail_model",
-            "sliper_model",
-            "between_stations",
-            "5?",
-            "6?",
-            "7?",
-            "8?",
-            "max_speed",
-        ],
-        dtype=dict(
-            centimeter=np.int64,
-            track_section=str,
-            kilometer=np.int32,
-            meter=np.int32,
-            track_lane=str,
-            sweref99_tm_x=np.float32,
-            sweref99_tm_y=np.float32,
-        ),
-        low_memory=False,
-    )
+    try:
+        return pd.read_csv(
+            path,
+            skiprows=[0] + skiprows,
+            delimiter="\t",
+            encoding="latin1",
+            names=[
+                "code",
+                "centimeter",
+                "track_section",
+                "kilometer",
+                "meter",
+                "track_lane",
+                "1?",
+                "2?",
+                "3?",
+                "4?",
+                "sweref99_tm_x",
+                "sweref99_tm_y",
+                "contact_wire_material",
+                "rail_model",
+                "sliper_model",
+                "between_stations",
+                "5?",
+                "6?",
+                "7?",
+                "8?",
+                "max_speed",
+            ],
+            dtype=dict(
+                centimeter=np.int64,
+                track_section=str,
+                kilometer=np.int32,
+                meter=np.int32,
+                track_lane=str,
+                sweref99_tm_x=np.float32,
+                sweref99_tm_y=np.float32,
+            ),
+            low_memory=False,
+        )
+    except Exception as e:
+        raise ValueError("Unable to parse kmm2 file, invalid csv.") from e
 
 
 def test_patterns():

kmm/positions/sync_frame_index.py

@@ -31,7 +31,7 @@ def sync_frame_index(positions: Positions, header: Header, adjustment: PositionA
             .assign(frame_index=frame_index[:-adjustment])
         )
     else:
-        raise ValueError(header.car_direction)
+        raise ValueError(f"Unsupported car direction {header.car_direction}")
 
     return positions.replace(
         dataframe=dataframe.assign(car_direction=header.car_direction)