Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 10 additions & 1 deletion src/gps_logger_parser/accelerometer/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)

Expand All @@ -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
Expand Down
6 changes: 4 additions & 2 deletions src/gps_logger_parser/gps/axytrek.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 = [
Expand Down
29 changes: 20 additions & 9 deletions src/gps_logger_parser/gps/catlog.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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(
Expand All @@ -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

Expand Down Expand Up @@ -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

Expand Down
25 changes: 20 additions & 5 deletions src/gps_logger_parser/gps/ecotone.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,5 @@
import csv

import geoarrow.pyarrow as ga
import numpy as np
import pandas as pd
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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(
Expand All @@ -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
Expand Down
13 changes: 12 additions & 1 deletion src/gps_logger_parser/gps/gpx.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand All @@ -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)
21 changes: 20 additions & 1 deletion src/gps_logger_parser/gps/jm.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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)
Expand Down
8 changes: 3 additions & 5 deletions src/gps_logger_parser/gps/mixin.py
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand All @@ -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
Expand Down
11 changes: 11 additions & 0 deletions src/gps_logger_parser/gps/pathtrack.py
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down
16 changes: 16 additions & 0 deletions src/gps_logger_parser/gps/unknown.py
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down
3 changes: 3 additions & 0 deletions src/gps_logger_parser/parser.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
53 changes: 44 additions & 9 deletions src/gps_logger_parser/parser_base.py
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,7 @@ def get_stream(self, binary=False, errors="strict"):
def _detect_encoding(self):
detector = UniversalDetector()
with self.get_stream(binary=True) as stream:
for line in stream.readlines():
for line in stream:
detector.feed(line)
if detector.done:
break
Expand All @@ -56,7 +56,19 @@ class Parser:
def __init__(self, parsable: Parsable):
self.file = parsable
self.data = []
self.harmonized_data = None

@classmethod
def can_parse(cls, parsable: Parsable) -> bool:
"""Lightweight detection: check if this parser can handle the file.

Subclasses should override this to perform cheap checks (magic bytes,
header line) without reading or parsing the entire file. Returns True
if the file appears to match, False otherwise.

The default implementation returns False — every parser must explicitly
implement its own detection logic.
"""
return False

def _raise_not_supported(self, text):
raise ParserNotSupported(f"{self.__class__.__name__}: {text}")
Expand Down Expand Up @@ -113,20 +125,26 @@ def get_harmonization_schema(self) -> dict:
)

def as_table(self) -> pa.Table:
# Capture raw source data as JSON before any harmonization
original_records = self.data.to_dict(orient="records")
original_json = [json.dumps(row, default=str) for row in original_records]
# Build JSON array directly from row iteration to avoid materializing
# both a list-of-dicts and a list-of-JSON-strings simultaneously
original_json = pa.array(
(
json.dumps(row, default=str)
for row in self.data.to_dict(orient="records")
),
type=pa.json_(pa.large_utf8()),
)

self.harmonized_data = self.harmonize_data(self.data.copy())
harmonized_data = self.harmonize_data(self.data)

if len(self.harmonized_data) == 0:
if len(harmonized_data) == 0:
raise ValueError("Harmonized data is empty, cannot create table")

table = pa.Table.from_pandas(self.harmonized_data, preserve_index=False)
table = pa.Table.from_pandas(harmonized_data, preserve_index=False)

table = table.append_column(
"_original_data",
pa.array(original_json, type=pa.json_(pa.large_utf8())),
original_json,
)
table = table.append_column(
"_datatype", pa.array([self.DATATYPE] * len(table), pa.string())
Expand Down Expand Up @@ -166,6 +184,23 @@ def _check_headers(self, header):
f"{header} != {self.FIELDS}"
)

@classmethod
def can_parse(cls, parsable: Parsable) -> bool:
"""Check if the CSV header matches FIELDS without reading the full file."""
try:
with parsable.get_stream(binary=False) as stream:
if not stream.seekable():
return False
reader = csv.reader(
stream,
delimiter=cls.SEPARATOR,
skipinitialspace=cls.SKIP_INITIAL_SPACE,
)
header = next(reader)
return header == cls.FIELDS
except (StopIteration, UnicodeDecodeError):
return False

def __init__(self, parsable: Parsable):
super().__init__(parsable)

Expand Down
Loading
Loading