Fix coordinate parsing for events datafix: harden coordinate parsing and add tests

flight_safety/utils.py

@@ -2,42 +2,92 @@
 Util functions
 """
 
+from __future__ import annotations
+
+import logging
+from typing import Optional
+
 import numpy as np
 
+LOGGER = logging.getLogger(__name__)
 
-def convert_lat(string):
-    try:
-        degs = float(string[0:2])
-        mins = float(string[2:4])
-        secs = float(string[4:6])
-        last = string[6].lower()
-        if last == 's':
-            factor = -1.0
-        elif last == 'n':
-            factor = 1.0
-        else:
-            raise ValueError("invalid hemisphere")
-        return factor * (degs + mins / 60 + secs / 3600)
-    except ValueError:
+
+def _parse_coordinate(
+    value: Optional[str],
+    deg_len: int,
+    hemisphere_positive: str,
+    hemisphere_negative: str,
+) -> float:
+    if value is None:
+        LOGGER.debug("Coordinate is None")
         return np.nan
 
+    if not isinstance(value, str):
+        try:
+            value = str(value)
+        except (TypeError, ValueError):
+            LOGGER.debug("Coordinate has non-string, non-coercible type: %r", value)
+            return np.nan
+
+    value = value.strip()
+    expected_len = deg_len + 2 + 2 + 1
+    if len(value) < expected_len:
+        LOGGER.debug("Coordinate has invalid length: %r", value)
+        return np.nan
 
-def convert_lon(string):
     try:
-        degs = float(string[0:3])
-        mins = float(string[3:5])
-        secs = float(string[5:7])
-        last = string[7].lower()
-        if last == 'w':
-            factor = -1.0
-        elif last == 'e':
-            factor = 1.0
-        else:
-            raise ValueError("invalid direction")
-        return factor * (degs + mins / 60 + secs / 3600)
-    except ValueError:
+        degs = float(value[0:deg_len])
+        mins = float(value[deg_len:deg_len + 2])
+        secs = float(value[deg_len + 2:deg_len + 4])
+        hemisphere = value[deg_len + 4].lower()
+    except (TypeError, ValueError, IndexError):
+        LOGGER.debug("Coordinate has invalid numeric format: %r", value)
+        return np.nan
+
+    if hemisphere == hemisphere_negative:
+        factor = -1.0
+    elif hemisphere == hemisphere_positive:
+        factor = 1.0
+    else:
+        LOGGER.debug("Coordinate has invalid hemisphere: %r", value)
         return np.nan
 
+    return factor * (degs + mins / 60 + secs / 3600)
+
+
+def convert_lat(value: Optional[str]) -> float:
+    """
+    Convert a latitude string in DMS format to decimal degrees.
+
+    Parameters
+    ----------
+    value : Optional[str]
+        Latitude string encoded as DDMMSSN or DDMMSSS.
+
+    Returns
+    -------
+    float
+        Decimal degrees; returns numpy.nan for invalid inputs.
+    """
+    return _parse_coordinate(value, deg_len=2, hemisphere_positive="n", hemisphere_negative="s")
+
+
+def convert_lon(value: Optional[str]) -> float:
+    """
+    Convert a longitude string in DMS format to decimal degrees.
+
+    Parameters
+    ----------
+    value : Optional[str]
+        Longitude string encoded as DDDMMSSE or DDDMMSSW.
+
+    Returns
+    -------
+    float
+        Decimal degrees; returns numpy.nan for invalid inputs.
+    """
+    return _parse_coordinate(value, deg_len=3, hemisphere_positive="e", hemisphere_negative="w")
+
 
 def rename_categories(old_categories, codes_meaning):
     new_categories = []

tests/test_utils.py

@@ -0,0 +1,31 @@
+import math
+
+import numpy as np
+
+from flight_safety.utils import convert_lat, convert_lon
+
+
+def test_convert_lat_valid() -> None:
+    assert math.isclose(convert_lat("372530N"), 37 + 25 / 60 + 30 / 3600)
+    assert math.isclose(convert_lat("372530S"), -(37 + 25 / 60 + 30 / 3600))
+
+
+def test_convert_lon_valid() -> None:
+    assert math.isclose(convert_lon("1223045W"), -(122 + 30 / 60 + 45 / 3600))
+    assert math.isclose(convert_lon("1223045E"), 122 + 30 / 60 + 45 / 3600)
+
+
+def test_convert_lat_invalid() -> None:
+    assert math.isnan(convert_lat(None))
+    assert math.isnan(convert_lat(""))
+    assert math.isnan(convert_lat("123"))
+    assert math.isnan(convert_lat("372530X"))
+    assert math.isnan(convert_lat(123))  # type: ignore[arg-type]
+
+
+def test_convert_lon_invalid() -> None:
+    assert math.isnan(convert_lon(None))
+    assert math.isnan(convert_lon(""))
+    assert math.isnan(convert_lon("123"))
+    assert math.isnan(convert_lon("1223045X"))
+    assert math.isnan(convert_lon(np.nan))  # type: ignore[arg-type]