gps-logger-parser/src/gps_logger_parser/gps/axytrek.py at b6253a808fc19316a454fe6c218fb4a1b954bf76 · NINAnor/gps-logger-parser · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
import csv

import pandas as pd
import pyarrow.csv as pacsv

from ..parser_base import CSVParser, Parsable
from .columns import GPSHarmonizedColumn
from .mixin import GPSHarmonizationMixin


def skip(row):
    if row.text == "Power off command received.":
        return "skip"

    return "error"


class AXYTREKParser(GPSHarmonizationMixin, CSVParser):
    DATATYPE = "gps_axytrek"
    FIELDS = [
        "TagID",
        "Date",
        "Time",
        "X",
        "Y",
        "Z",
        "Activity",
        "Depth",
        "Temp. (?C)",
        "location-lat",
        "location-lon",
        "height-above-msl",
        "ground-speed",
        "satellite-count",
        "hdop",
        "maximum-signal-strength",
        "Sensor Raw",
        "Battery Voltage (V)",
    ]

    MAPPINGS = {
        GPSHarmonizedColumn.ID: "TagID",
        GPSHarmonizedColumn.TIMESTAMP: None,
        GPSHarmonizedColumn.LATITUDE: "location-lat",
        GPSHarmonizedColumn.LONGITUDE: "location-lon",
        GPSHarmonizedColumn.ALTITUDE: "height-above-msl",
        GPSHarmonizedColumn.SPEED_KM_H: "ground-speed",
        GPSHarmonizedColumn.TYPE: None,
        GPSHarmonizedColumn.DISTANCE: None,
        GPSHarmonizedColumn.COURSE: None,
        GPSHarmonizedColumn.HDOP: "hdop",
        GPSHarmonizedColumn.PDOP: None,
        GPSHarmonizedColumn.SATELLITES_COUNT: "satellite-count",
        GPSHarmonizedColumn.TEMPERATURE: "Temp. (?C)",
        GPSHarmonizedColumn.SOLAR_I_MA: None,
        GPSHarmonizedColumn.BAT_SOC_PCT: None,
        GPSHarmonizedColumn.RING_NR: None,
        GPSHarmonizedColumn.TRIP_NR: None,
    }

    def harmonize_data(self, data):
        # Combine Date and Time columns into timestamp
        data["timestamp"] = pd.to_datetime(
            data["Date"] + " " + data["Time"],
            errors="raise",
            format="%d.%m.%Y %H:%M:%S.%f",
        )
        return super().harmonize_data(data)

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

        with self.file.get_stream(binary=False) as stream:
            if not stream.seekable():
                self._raise_not_supported("Stream not seekable")

            reader = csv.reader(
                stream,
                delimiter=self.SEPARATOR,
                skipinitialspace=self.SKIP_INITIAL_SPACE,
            )
            header = next(reader)
            if header != self.FIELDS:
                self._raise_not_supported(
                    f"Stream have a header different than expected, "
                    f"{header} != {self.FIELDS}"
                )

        parse_options = pacsv.ParseOptions(
            delimiter=self.SEPARATOR, invalid_row_handler=skip
        )
        with self.file.get_stream(binary=True) as binary_stream:
            self.data = pacsv.read_csv(
                binary_stream, parse_options=parse_options
            ).to_pandas()


PARSERS = [
    AXYTREKParser,
]