openrvdas.logger.transforms.parse_nmea_transform

No module-level documentation available.
 1#!/usr/bin/env python3
 2
 3
 4from logger.utils.das_record import DASRecord  # noqa: E402
 5from logger.utils import nmea_parser  # noqa: E402
 6from logger.transforms.transform import Transform  # noqa: E402
 7
 8
 9################################################################################
10class ParseNMEATransform(Transform):
11    """Parse a "<data_id> <timestamp> <nmea>" record and return
12    corresponding DASRecord."""
13
14    def __init__(self, json=False,
15                 message_path=nmea_parser.DEFAULT_MESSAGE_PATH,
16                 sensor_path=nmea_parser.DEFAULT_SENSOR_PATH,
17                 sensor_model_path=nmea_parser.DEFAULT_SENSOR_MODEL_PATH,
18                 time_format=None, **kwargs):
19        """
20        ```
21        json    Return a JSON-encoded representation of the DASRecord instead
22                of DASRecord itself.
23
24        message_path, sensor_path, sensor_model_path
25                Wildcarded path matching JSON definitions for sensor messages,
26                sensors and sensor models.
27        ```
28        """
29        super().__init__(**kwargs)  # processes 'quiet' and type hints
30        self.json = json
31        self.parser = nmea_parser.NMEAParser(message_path, sensor_path,
32                                             sensor_model_path,
33                                             time_format=time_format)
34
35    ############################
36    def transform(self, record: str) -> DASRecord:
37        """Parse record and return DASRecord."""
38
39        # See if it's something we can process, and if not, try digesting
40        if not self.can_process_record(record):  # inherited from BaseModule()
41            return self.digest_record(record)  # inherited from BaseModule()
42
43        result = self.parser.parse_record(record)
44        if not result:
45            return None
46        if self.json:
47            return result.as_json()
48        return result
class ParseNMEATransform(logger.transforms.transform.Transform): 
11class ParseNMEATransform(Transform):
12    """Parse a "<data_id> <timestamp> <nmea>" record and return
13    corresponding DASRecord."""
14
15    def __init__(self, json=False,
16                 message_path=nmea_parser.DEFAULT_MESSAGE_PATH,
17                 sensor_path=nmea_parser.DEFAULT_SENSOR_PATH,
18                 sensor_model_path=nmea_parser.DEFAULT_SENSOR_MODEL_PATH,
19                 time_format=None, **kwargs):
20        """
21        ```
22        json    Return a JSON-encoded representation of the DASRecord instead
23                of DASRecord itself.
24
25        message_path, sensor_path, sensor_model_path
26                Wildcarded path matching JSON definitions for sensor messages,
27                sensors and sensor models.
28        ```
29        """
30        super().__init__(**kwargs)  # processes 'quiet' and type hints
31        self.json = json
32        self.parser = nmea_parser.NMEAParser(message_path, sensor_path,
33                                             sensor_model_path,
34                                             time_format=time_format)
35
36    ############################
37    def transform(self, record: str) -> DASRecord:
38        """Parse record and return DASRecord."""
39
40        # See if it's something we can process, and if not, try digesting
41        if not self.can_process_record(record):  # inherited from BaseModule()
42            return self.digest_record(record)  # inherited from BaseModule()
43
44        result = self.parser.parse_record(record)
45        if not result:
46            return None
47        if self.json:
48            return result.as_json()
49        return result

Parse a " " record and return corresponding DASRecord.

ParseNMEATransform( json=False, message_path='local/message/*.yaml', sensor_path='local/sensor/*.yaml', sensor_model_path='local/sensor_model/*.yaml', time_format=None, **kwargs) 
15    def __init__(self, json=False,
16                 message_path=nmea_parser.DEFAULT_MESSAGE_PATH,
17                 sensor_path=nmea_parser.DEFAULT_SENSOR_PATH,
18                 sensor_model_path=nmea_parser.DEFAULT_SENSOR_MODEL_PATH,
19                 time_format=None, **kwargs):
20        """
21        ```
22        json    Return a JSON-encoded representation of the DASRecord instead
23                of DASRecord itself.
24
25        message_path, sensor_path, sensor_model_path
26                Wildcarded path matching JSON definitions for sensor messages,
27                sensors and sensor models.
28        ```
29        """
30        super().__init__(**kwargs)  # processes 'quiet' and type hints
31        self.json = json
32        self.parser = nmea_parser.NMEAParser(message_path, sensor_path,
33                                             sensor_model_path,
34                                             time_format=time_format)

json    Return a JSON-encoded representation of the DASRecord instead
        of DASRecord itself.

message_path, sensor_path, sensor_model_path
        Wildcarded path matching JSON definitions for sensor messages,
        sensors and sensor models.
json
parser
def transform(self, record: str) -> logger.utils.das_record.DASRecord: 
37    def transform(self, record: str) -> DASRecord:
38        """Parse record and return DASRecord."""
39
40        # See if it's something we can process, and if not, try digesting
41        if not self.can_process_record(record):  # inherited from BaseModule()
42            return self.digest_record(record)  # inherited from BaseModule()
43
44        result = self.parser.parse_record(record)
45        if not result:
46            return None
47        if self.json:
48            return result.as_json()
49        return result

Parse record and return DASRecord.