gps-logger-parser/src/gps_logger_parser/gps/unknown.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
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
import csv

import numpy as np
import pandas as pd

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

FIELDS = [
    "DataID",
    "ID",
    "Ring_nr",
    "Date",
    "Time",
    "Altitude",
    "Speed",
    "Course",
    "HDOP",
    "Latitude",
    "Longitude",
    "TripNr",
]

MAPPINGS = {
    GPSHarmonizedColumn.ID: "DataID",
    GPSHarmonizedColumn.TIMESTAMP: None,
    GPSHarmonizedColumn.LATITUDE: "Latitude",
    GPSHarmonizedColumn.LONGITUDE: "Longitude",
    GPSHarmonizedColumn.ALTITUDE: "Altitude",
    GPSHarmonizedColumn.SPEED_KM_H: "Speed",
    GPSHarmonizedColumn.TYPE: None,
    GPSHarmonizedColumn.DISTANCE: None,
    GPSHarmonizedColumn.COURSE: "Course",
    GPSHarmonizedColumn.HDOP: "HDOP",
    GPSHarmonizedColumn.PDOP: None,
    GPSHarmonizedColumn.SATELLITES_COUNT: None,
    GPSHarmonizedColumn.TEMPERATURE: None,
    GPSHarmonizedColumn.SOLAR_I_MA: None,
    GPSHarmonizedColumn.BAT_SOC_PCT: None,
    GPSHarmonizedColumn.RING_NR: "Ring_nr",
    GPSHarmonizedColumn.TRIP_NR: "TripNr",
}


class GPSUnknownFormatParser(GPSHarmonizationMixin, CSVParser):
    """
    Parser for a format, its a GPS CSV like format
    with the following fields
    """

    DATATYPE = "gps_unknown"
    SEPARATOR = "\t"
    FIELDS = FIELDS

    MAPPINGS = MAPPINGS

    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",
        )
        return super().harmonize_data(data)


class GPSUnknownFormatParserWithEmptyColumns(GPSHarmonizationMixin, Parser):
    """
    Parser for a format, its a GPS CSV like format
    with the following fields
    """

    DATATYPE = "gps_unknown"
    SEPARATOR = "\t"
    FIELDS = FIELDS
    SKIP_INITIAL_SPACE = True

    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(
            data["Date"] + " " + data["Time"],
            errors="raise",
            format="%d.%m.%Y %H:%M:%S",
        )
        # this file seem to have been manipulated in excel, using some formulas
        data = data.replace("#VALUE!", np.nan)
        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)

            # Filter empty columns
            header = [c for c in header if c != ""]

            if header != self.FIELDS:
                self._raise_not_supported(
                    f"Stream have a header different than expected, "
                    f"{header} != {self.FIELDS}"
                )

            stream.seek(0)
            self.data = pd.read_csv(
                stream,
                header=1,
                names=self.FIELDS,
                sep=self.SEPARATOR,
                index_col=False,
                usecols=list(range(len(self.FIELDS))),
            )


PARSERS = [
    GPSUnknownFormatParser,
    GPSUnknownFormatParserWithEmptyColumns,
]