chg: improve memory usage and avoid repetitive work

Author: nicokant

src/gps_logger_parser/accelerometer/__init__.py

@@ -31,6 +31,15 @@ class AcceleratorParser(AccelerometerHarmonizationMixin, Parser):
         AccelerometerHarmonizedColumn.Z: "Z",
     }
 
+    @classmethod
+    def can_parse(cls, parsable):
+        """Check if the file starts with the expected HEAD bytes."""
+        try:
+            with parsable.get_stream(binary=False) as stream:
+                return stream_starts_with(stream, cls.HEAD)
+        except (UnicodeDecodeError, OSError):
+            return False
+
     def __init__(self, parsable: Parsable):
         super().__init__(parsable)
 
@@ -46,7 +55,7 @@ def __init__(self, parsable: Parsable):
 
             stream.seek(0)
 
-            for row in stream.readlines():
+            for row in stream:
                 if [v.strip() for v in row.split(",")] == self.FIELDS:
                     break
                 row_count += 1

src/gps_logger_parser/gps/axytrek.py

@@ -89,8 +89,10 @@ def __init__(self, parsable: Parsable):
         parse_options = pacsv.ParseOptions(
             delimiter=self.SEPARATOR, invalid_row_handler=skip
         )
-        with self.file.get_stream(binary=True) as stream:
-            self.data = pacsv.read_csv(stream, parse_options=parse_options).to_pandas()
+        with self.file.get_stream(binary=True) as binary_stream:
+            self.data = pacsv.read_csv(
+                binary_stream, parse_options=parse_options
+            ).to_pandas()
 
 
 PARSERS = [

src/gps_logger_parser/gps/catlog.py

@@ -3,7 +3,7 @@
 
 import pandas as pd
 
-from ..helpers import stream_chunk_contains, stream_starts_with
+from ..helpers import stream_starts_with
 from ..parser_base import CSVParser, Parsable
 from .columns import GPSHarmonizedColumn
 from .mixin import GPSHarmonizationMixin
@@ -54,6 +54,17 @@ class GPSCatTrackParser(GPSHarmonizationMixin, CSVParser):
         GPSHarmonizedColumn.TRIP_NR: None,
     }
 
+    @classmethod
+    def can_parse(cls, parsable):
+        """Check if the file starts with the expected START_WITH prefix."""
+        if not cls.START_WITH:
+            return True
+        try:
+            with parsable.get_stream(binary=False) as stream:
+                return stream_starts_with(stream, cls.START_WITH)
+        except (UnicodeDecodeError, OSError):
+            return False
+
     def harmonize_data(self, data):
         # Combine Date and Time columns into timestamp
         data["timestamp"] = pd.to_datetime(
@@ -73,13 +84,13 @@ def __init__(self, parsable: Parsable):
                 self._raise_not_supported("Stream must start with Name:CatLog")
 
             if self.DIVIDER:
-                if stream_chunk_contains(stream, 500, self.DIVIDER):
-                    _intro, data = stream.read().split(self.DIVIDER)
-                    content = io.StringIO(data)
-                else:
+                full_content = stream.read()
+                if self.DIVIDER not in full_content:
                     self._raise_not_supported(
                         f"Stream doesn't have the divider {self.DIVIDER}"
                     )
+                _intro, data = full_content.split(self.DIVIDER, 1)
+                content = io.StringIO(data)
             else:
                 content = stream
 
@@ -191,13 +202,13 @@ def __init__(self, parsable: Parsable):
                 self._raise_not_supported("Stream must start with Name:CatLog")
 
             if self.DIVIDER:
-                if stream_chunk_contains(stream, 500, self.DIVIDER):
-                    _intro, data = stream.read().split(self.DIVIDER)
-                    content = io.StringIO(data)
-                else:
+                full_content = stream.read()
+                if self.DIVIDER not in full_content:
                     self._raise_not_supported(
                         f"Stream doesn't have the divider {self.DIVIDER}"
                     )
+                _intro, data = full_content.split(self.DIVIDER, 1)
+                content = io.StringIO(data)
             else:
                 content = stream
 

src/gps_logger_parser/gps/ecotone.py

@@ -1,3 +1,5 @@
+import csv
+
 import geoarrow.pyarrow as ga
 import numpy as np
 import pandas as pd
@@ -49,6 +51,21 @@ def _check_headers(self, header):
                 f"{len(header)} != {len(self.FIELDS)}"
             )
 
+    @classmethod
+    def can_parse(cls, parsable):
+        """Check if the CSV header has the expected number of columns."""
+        try:
+            with parsable.get_stream(binary=False) as stream:
+                if not stream.seekable():
+                    return False
+                reader = csv.reader(
+                    stream, delimiter=cls.SEPARATOR, skipinitialspace=True
+                )
+                header = next(reader)
+                return len(header) == len(cls.FIELDS)
+        except (StopIteration, UnicodeDecodeError):
+            return False
+
     def harmonize_data(self, data):
         # Call parent harmonization — applies MAPPINGS, enforces GPS schema,
         # creates geometry, and drops raw source columns
@@ -81,10 +98,8 @@ def harmonize_data(self, data):
                 data["meters_north"].str[:-1].astype(float).values,
                 crs="EPSG:32633",  # UTM zone 33N
             )
-            # Convert to pandas ArrowExtensionArray
-            result["geometry"] = pd.array(
-                points.to_pylist(), dtype=pd.ArrowDtype(points.type)
-            )
+            # Convert to pandas ArrowExtensionArray directly from Arrow array
+            result["geometry"] = pd.array(points, dtype=pd.ArrowDtype(points.type))
         else:
             # Create empty geometry column if lat/lon don't exist or are all null
             empty_points = ga.make_point(
@@ -93,7 +108,7 @@ def harmonize_data(self, data):
                 crs="EPSG:32633",  # UTM zone 33N
             )
             result["geometry"] = pd.array(
-                empty_points.to_pylist(), dtype=pd.ArrowDtype(empty_points.type)
+                empty_points, dtype=pd.ArrowDtype(empty_points.type)
             )
 
         return result

src/gps_logger_parser/gps/gpx.py

@@ -42,6 +42,17 @@ class GPXParser(GPSHarmonizationMixin, Parser):
         GPSHarmonizedColumn.TRIP_NR: None,
     }
 
+    @classmethod
+    def can_parse(cls, parsable):
+        """Check if the file starts with <?xml within the first 30 bytes."""
+        try:
+            with parsable.get_stream(binary=False) as stream:
+                if not stream.seekable():
+                    return False
+                return stream_chunk_contains(stream, 30, "<?xml")
+        except (UnicodeDecodeError, OSError):
+            return False
+
     def harmonize_data(self, data):
         data["time"] = pd.to_datetime(data["time"], utc=True)
         return super().harmonize_data(data)
@@ -61,6 +72,6 @@ def __init__(self, stream):
             for track in gpx.tracks:
                 for segment in track.segments:
                     for point in segment.points:
-                        points.append(getattr(point, f) for f in self.FIELDS)
+                        points.append(tuple(getattr(point, f) for f in self.FIELDS))
 
             self.data = pd.DataFrame(points, columns=self.FIELDS)

src/gps_logger_parser/gps/jm.py

@@ -68,6 +68,15 @@ class GPS2JMParser7_5(GPSHarmonizationMixin, Parser):
         GPSHarmonizedColumn.TRIP_NR: None,
     }
 
+    @classmethod
+    def can_parse(cls, parsable):
+        """Check if the file contains the 2JmGPS-LOG marker."""
+        try:
+            with parsable.get_stream(binary=False, errors="backslashreplace") as stream:
+                return stream_chunk_contains(stream, 30, "2JmGPS-LOG")
+        except (UnicodeDecodeError, OSError):
+            return False
+
     def harmonize_data(self, data):
         # Call parent harmonization — applies MAPPINGS, enforces GPS schema,
         # creates geometry, and drops raw source columns
@@ -276,8 +285,18 @@ class GPS2JMParser8Alternative(GPSHarmonizationMixin, Parser):
         GPSHarmonizedColumn.TRIP_NR: None,
     }
 
-    def harmonize_data(self, data):
+    @classmethod
+    def can_parse(cls, parsable):
+        """Check if the file contains the GPS DATA marker."""
+        try:
+            with parsable.get_stream(binary=False, errors="backslashreplace") as stream:
+                return stream_chunk_contains(
+                    stream, 50, "************* GPS DATA *************"
+                )
+        except (UnicodeDecodeError, OSError):
+            return False
 
+    def harmonize_data(self, data):
         # Call parent harmonization — applies MAPPINGS, enforces GPS schema,
         # creates geometry, and drops raw source columns
         result = super().harmonize_data(data)

src/gps_logger_parser/gps/mixin.py

@@ -69,10 +69,8 @@ def _create_geometry_column(self, data):
                 data["latitude"].values,
                 crs="EPSG:4326",
             )
-            # Convert to pandas ArrowExtensionArray
-            data["geometry"] = pd.array(
-                points.to_pylist(), dtype=pd.ArrowDtype(points.type)
-            )
+            # Convert to pandas ArrowExtensionArray directly from the Arrow array
+            data["geometry"] = pd.array(points, dtype=pd.ArrowDtype(points.type))
         else:
             # Create empty geometry column if lat/lon don't exist or are all null
             empty_points = ga.make_point(
@@ -81,7 +79,7 @@ def _create_geometry_column(self, data):
                 crs="EPSG:4326",
             )
             data["geometry"] = pd.array(
-                empty_points.to_pylist(), dtype=pd.ArrowDtype(empty_points.type)
+                empty_points, dtype=pd.ArrowDtype(empty_points.type)
             )
 
         return data

src/gps_logger_parser/gps/pathtrack.py

@@ -53,6 +53,17 @@ class PathtrackParser(GPSHarmonizationMixin, Parser):
         GPSHarmonizedColumn.TRIP_NR: None,
     }
 
+    @classmethod
+    def can_parse(cls, parsable):
+        """Check if the file starts with the expected HEAD bytes."""
+        try:
+            with parsable.get_stream(binary=False) as stream:
+                if not stream.seekable():
+                    return False
+                return stream_starts_with(stream, cls.HEAD)
+        except (UnicodeDecodeError, OSError):
+            return False
+
     def harmonize_data(self, data):
         # Combine date and time fields into timestamp
         data["timestamp"] = pd.to_datetime(

src/gps_logger_parser/gps/unknown.py

@@ -78,6 +78,22 @@ class GPSUnknownFormatParserWithEmptyColumns(GPSHarmonizationMixin, Parser):
 
     MAPPINGS = MAPPINGS
 
+    @classmethod
+    def can_parse(cls, parsable):
+        """Check if the header (with empty columns filtered) matches FIELDS."""
+        try:
+            with parsable.get_stream(binary=False) as stream:
+                if not stream.seekable():
+                    return False
+                reader = csv.reader(
+                    stream, delimiter=cls.SEPARATOR, skipinitialspace=True
+                )
+                header = next(reader)
+                header = [c for c in header if c != ""]
+                return header == cls.FIELDS
+        except (StopIteration, UnicodeDecodeError):
+            return False
+
     def harmonize_data(self, data):
         # Combine Date and Time columns into timestamp
         data["timestamp"] = pd.to_datetime(

src/gps_logger_parser/parser.py

@@ -21,6 +21,9 @@ def detect_file(path: UPath, *args, logger=logger, **kwargs):
 
     for parser in available_parsers:
         try:
+            if not parser.can_parse(parsable):
+                logger.debug(f"Skipped {parser.__name__}: can_parse returned False")
+                continue
             result = parser(parsable)
             logger.info(f"Parsed with {parser}")
             return result