openrvdas.logger.transforms.nmea_checksum_transform

  1import logging
  2# For efficient checksum code
  3from functools import reduce
  4from operator import xor
  5
  6from logger.transforms.transform import Transform  # noqa: E402
  7
  8
  9############################
 10def get_message_str(source):
 11    """ Returns message_str, which is everything between the '$' and '*' in the source string """
 12
 13    if ((source.find('$') == -1) or (source.find('*') == -1)):
 14        return None
 15
 16    start = source.index('$')+1
 17    end = source.index('*')
 18    message_str = source[start:end]
 19    return message_str
 20
 21
 22def get_checksum_value(source):
 23    """ Returns checksum_value, which is the parsed checksum value (after '*')
 24    from the source string.
 25    """
 26
 27    if (source.find('*') == -1):
 28        return None
 29
 30    start = source.index('*')+1
 31    checksum_value = source[start:]
 32    return checksum_value
 33
 34
 35def compute_checksum(source):
 36    """Return hex checksum for source string."""
 37
 38    return '%02X' % reduce(xor, (ord(c) for c in source))
 39
 40
 41################################################################################
 42class NMEAChecksumTransform(Transform):
 43    """
 44    NMEAChecksumTransform checks the integrity/completeness of a record by confirming
 45    whether or not the checksum matches. If the checksum matches, it returns the record,
 46    and otherwise it sends an error message.
 47    """
 48
 49    DEFAULT_ERROR_MESSAGE = 'Bad checksum for record: '
 50
 51    def __init__(self, checksum_optional=False, error_message=DEFAULT_ERROR_MESSAGE,
 52                 writer=None, **kwargs):
 53        """
 54        checksum_optional — If True, then pass record along even if checksum is missing
 55        error_message     — Optional custom error message; if None then use DEFAULT_ERROR_MESSAGE
 56        writer            — Optional error writer; if None, log to stderr
 57        """
 58        super().__init__(**kwargs)  # processes 'quiet' and type hints
 59
 60        self.checksum_optional = checksum_optional
 61        self.error_message = error_message
 62        self.writer = writer
 63
 64        # Tries to utilize the write() method of writer, which it
 65        # would only have if the object is a Writer. Send it a 'None'
 66        # record, which all writers should ignore. If this fails,
 67        # writer is set to None.
 68        if writer:
 69            try:
 70                writer.write(None)
 71            except AttributeError:
 72                logging.error('Writer passed to NMEAChecksumTransform has no '
 73                              'write() method!')
 74                self.writer = None
 75
 76    def transform(self, record: str):
 77        """
 78        Checks if computed checksum matches parsed checksum. If True, it returns the record,
 79        otherwise it calls send_error_message().
 80
 81        record - the record in question
 82        """
 83        # See if it's something we can process, and if not, try digesting
 84        if not self.can_process_record(record):  # inherited from BaseModule()
 85            return self.digest_record(record)  # inherited from BaseModule()
 86
 87        if not type(record) is str:
 88            logging.warning('NMEAChecksumTransform passed non-string record '
 89                            '(type %s): %s', type(record), record)
 90            return None
 91
 92        checksum_value = get_checksum_value(record)
 93
 94        if checksum_value is None:
 95            if self.checksum_optional:
 96                return record
 97
 98            # If here, checksum is not optional and does not exist
 99            self.send_error_message(record, 'No checksum found in record ')
100            return None
101
102        message_str = get_message_str(record)
103        computed_checksum = compute_checksum(message_str)
104
105        # If here, and we are about to see if it matches
106        if computed_checksum == checksum_value:
107            return record
108
109        # If here, then checksum exists but didn't match
110        self.send_error_message(record)
111        return None
112
113    def send_error_message(self, record, message=None):
114        """
115            Send error to writer if one exists, otherwise send it to stderr
116
117            record - the record with the error
118            message - optional custom message. If None, then use self.error_message
119        """
120
121        error_message = message or self.error_message
122        error_message += record
123
124        if self.writer:
125            self.writer.write(error_message)
126        else:
127            logging.warning(error_message)