UBX_Integrity_Checker.py

# Checks the integrity of u-blox binary files

# Written by: Paul Clark
# Last update: August 26th 2020

# Reads a UBX file and checks the integrity of both UBX and NMEA data
# Will rewind and re-sync if an error is found

# SparkFun code, firmware, and software is released under the MIT License (http://opensource.org/licenses/MIT)
#
# The MIT License (MIT)
#
# Copyright (c) 2020 SparkFun Electronics
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.

import sys
import os

# Add byte to checksums sum1 and sum2
def csum(byte, sum1, sum2):
    sum1 = sum1 + byte
    sum2 = sum2 + sum1
    sum1 = sum1 & 0xFF
    sum2 = sum2 & 0xFF
    return sum1,sum2

print('UBX Integrity Checker')
print()

filename = ''

if filename == '':
    # Check if the bin filename was passed in argv
    if len(sys.argv) > 1: filename = sys.argv[1]

# Find first .ubx file in the current directory
firstfile = ''
for root, dirs, files in os.walk("."):
    if len(files) > 0:
        if root == ".": # Comment this line to check sub-directories too
            for afile in files:
                if afile[-4:] == '.ubx':
                    if firstfile == '': firstfile = os.path.join(root, afile)

# Ask user for .bin filename offering firstfile as the default
if filename == '': filename = input('Enter the UBX filename (default: ' + firstfile + '): ') # Get the filename
if filename == '': filename = firstfile

# Ask user if the data contains NMEA messages
response = input('Could this file contain any NMEA messages? (Y/n): ') # Get the response
if (response == '') or (response == 'Y') or (response == 'y'):
    containsNMEA = True
else:
    containsNMEA = False

print()
print('Processing',filename)
print()
filesize = os.path.getsize(filename) # Record the file size

# Try to open file for reading
try:
    fi = open(filename,"rb")
except:
    raise Exception('Invalid file!')

processed = -1 # The nunber of bytes processed
messages = {} # The collected message types
longest = 0 # The length of the longest UBX message
keepGoing = True

# Sync 'state machine'
looking_for_B5_dollar   = 0 # Looking for either a UBX 0xB5 or an NMEA '$'
looking_for_62          = 1 # Looking for a UBX 0x62 header byte
looking_for_class       = 2 # Looking for UBX class byte
looking_for_ID          = 3 # Looking for UBX ID byte
looking_for_length_LSB  = 4 # Looking for UBX length bytes
looking_for_length_MSB  = 5
processing_payload      = 6 # Processing the payload. Keep going until length bytes have been processed
looking_for_checksum_A  = 7 # Looking for UBX checksum bytes
looking_for_checksum_B  = 8
sync_lost               = 9 # Go into this state if sync is lost (bad checksum etc.)
looking_for_asterix     = 10 # Looking for NMEA '*'
looking_for_csum1       = 11 # Looking for NMEA checksum bytes
looking_for_csum2       = 12
looking_for_term1       = 13 # Looking for NMEA terminating bytes (CR and LF)
looking_for_term2       = 14

ubx_nmea_state = sync_lost # Initialize the state machine

# Storage for UBX messages
ubx_length = 0
ubx_class = 0
ubx_ID = 0
ubx_checksum_A = 0
ubx_checksum_B = 0
ubx_expected_checksum_A = 0
ubx_expected_checksum_B = 0

# Storage for NMEA messages
nmea_length = 0
nmea_char_1 = 0 # e.g. G
nmea_char_2 = 0 # e.g. P
nmea_char_3 = 0 # e.g. G
nmea_char_4 = 0 # e.g. G
nmea_char_5 = 0 # e.g. A
nmea_csum = 0
nmea_csum1 = 0
nmea_csum2 = 0
nmea_expected_csum1 = 0
nmea_expected_csum2 = 0

max_nmea_len = 100 # Maximum length for an NMEA message: use this to detect if we have lost sync while receiving an NMEA message
sync_lost_at = -1 # Record where we lost sync
rewind_to = -1 # Keep a note of where we should rewind to if sync is lost
rewind_attempts = 0 # Keep a note of how many rewinds have been attempted
max_rewinds = 100 # Abort after this many rewinds
rewind_in_progress = False # Flag to indicate if a rewind is in progress
resyncs = 0 # Record the number of successful resyncs
resync_in_progress = False # Flag to indicate if a resync is in progress
message_start_byte = 0 # Record where the latest message started (for resync reporting)

try:
    while keepGoing:

        # Read one byte from the file
        fileBytes = fi.read(1)
        if (len(fileBytes) == 0):
            print('ERROR: Read zero bytes. End of file?! Or zero file size?!')
            raise Exception('End of file?! Or zero file size?!')
        c = fileBytes[0]

        processed = processed + 1 # Keep a record of how many bytes have been read and processed

        # Process data bytes according to ubx_nmea_state
        # For UBX messages:
        # Sync Char 1: 0xB5
        # Sync Char 2: 0x62
        # Class byte
        # ID byte
        # Length: two bytes, little endian
        # Payload: length bytes
        # Checksum: two bytes
        # For NMEA messages:
        # Starts with a '$'
        # The next five characters indicate the message type (stored in nmea_char_1 to nmea_char_5)
        # Message fields are comma-separated
        # Followed by an '*'
        # Then a two character checksum (the logical exclusive-OR of all characters between the $ and the * as ASCII hex)
        # Ends with CR LF
        # Only allow a new file to be opened when a complete packet has been processed and ubx_nmea_state has returned to "looking_for_B5_dollar"
        # Or when a data error is detected (sync_lost)

        # RXM_RAWX is class 0x02 ID 0x15
        # RXM_SFRBF is class 0x02 ID 0x13
        # TIM_TM2 is class 0x0d ID 0x03
        # NAV_POSLLH is class 0x01 ID 0x02
        # NAV_PVT is class 0x01 ID 0x07
        # NAV-STATUS is class 0x01 ID 0x03

        if (ubx_nmea_state == looking_for_B5_dollar) or (ubx_nmea_state == sync_lost):
            if (c == 0xB5): # Have we found Sync Char 1 (0xB5) if we were expecting one?
                if (ubx_nmea_state == sync_lost):
                    print("UBX Sync Char 1 (0xB5) found at byte "+str(processed)+". Checking for Sync Char 2")
                ubx_nmea_state = looking_for_62 # Now look for Sync Char 2 (0x62)
                message_start_byte = processed # Record the message start byte for resync reporting
            elif (c == 0x24) and (containsNMEA == True): # Have we found an NMEA '$' if we were expecting one?
                if (ubx_nmea_state == sync_lost):
                    print("NMEA $ found at byte "+str(processed)+". Attempting to process the message")
                ubx_nmea_state = looking_for_asterix # Now keep going until we receive an asterix
                nmea_length = 0 # Reset nmea_length then use it to check for excessive message length
                nmea_csum = 0 # Reset the nmea_csum. Update it as each character arrives
                nmea_char_1 = 0x30 # Reset the first five NMEA chars to something invalid
                nmea_char_2 = 0x30
                nmea_char_3 = 0x30
                nmea_char_4 = 0x30
                nmea_char_5 = 0x30
                message_start_byte = processed # Record the message start byte for resync reporting
            else:
                #print("Was expecting Sync Char 0xB5 or an NMEA $ but did not receive one!")
                if (c == 0x24):
                    print("Warning: * found at byte "+str(processed)+"! Are you sure this file does not contain NMEA messages?")
                sync_lost_at = processed
                ubx_nmea_state = sync_lost
        elif (ubx_nmea_state == looking_for_62):
            if (c == 0x62): # Have we found Sync Char 2 (0x62) when we were expecting one?
                ubx_expected_checksum_A = 0 # Reset the expected checksum
                ubx_expected_checksum_B = 0
                ubx_nmea_state = looking_for_class # Now look for Class byte
            else:
                print("Panic!! Was expecting Sync Char 2 (0x62) but did not receive one!")
                print("Sync lost at byte "+str(processed)+". Attemting to re-sync")
                sync_lost_at = processed
                resync_in_progress = True
                ubx_nmea_state = sync_lost
        elif (ubx_nmea_state == looking_for_class):
            ubx_class = c
            ubx_expected_checksum_A = ubx_expected_checksum_A + c # Update the expected checksum
            ubx_expected_checksum_B = ubx_expected_checksum_B + ubx_expected_checksum_A
            ubx_nmea_state = looking_for_ID # Now look for ID byte
        elif (ubx_nmea_state == looking_for_ID):
            ubx_ID = c
            ubx_expected_checksum_A = ubx_expected_checksum_A + c # Update the expected checksum
            ubx_expected_checksum_B = ubx_expected_checksum_B + ubx_expected_checksum_A
            message_type = '0x%02X 0x%02X'%(ubx_class,ubx_ID) # Record the message type
            ubx_nmea_state = looking_for_length_LSB # Now look for length LSB
        elif (ubx_nmea_state == looking_for_length_LSB):
            ubx_length = c # Store the length LSB
            ubx_expected_checksum_A = ubx_expected_checksum_A + c # Update the expected checksum
            ubx_expected_checksum_B = ubx_expected_checksum_B + ubx_expected_checksum_A
            ubx_nmea_state = looking_for_length_MSB # Now look for length MSB
        elif (ubx_nmea_state == looking_for_length_MSB):
            ubx_length = ubx_length + (c * 256) # Add the length MSB
            ubx_expected_checksum_A = ubx_expected_checksum_A + c # Update the expected checksum
            ubx_expected_checksum_B = ubx_expected_checksum_B + ubx_expected_checksum_A
            if (ubx_length > longest): # Update the longest UBX message length    
                longest = ubx_length
            rewind_to = processed # If we lose sync due to dropped bytes then rewind to here
            ubx_nmea_state = processing_payload # Now look for payload bytes (length: ubx_length)
        elif (ubx_nmea_state == processing_payload):
            ubx_length = ubx_length - 1 # Decrement length by one
            ubx_expected_checksum_A = ubx_expected_checksum_A + c # Update the expected checksum
            ubx_expected_checksum_B = ubx_expected_checksum_B + ubx_expected_checksum_A
            if (ubx_length == 0):
                ubx_expected_checksum_A = ubx_expected_checksum_A & 0xff # Limit checksums to 8-bits
                ubx_expected_checksum_B = ubx_expected_checksum_B & 0xff
                ubx_nmea_state = looking_for_checksum_A # If we have received length payload bytes, look for checksum bytes
        elif (ubx_nmea_state == looking_for_checksum_A):
            ubx_checksum_A = c
            ubx_nmea_state = looking_for_checksum_B
        elif (ubx_nmea_state == looking_for_checksum_B):
            ubx_checksum_B = c
            ubx_nmea_state = looking_for_B5_dollar # All bytes received so go back to looking for a new Sync Char 1 unless there is a checksum error
            if ((ubx_expected_checksum_A != ubx_checksum_A) or (ubx_expected_checksum_B != ubx_checksum_B)):
                print("Panic!! UBX checksum error!")
                print("Sync lost at byte "+str(processed)+". Attemting to re-sync.")
                sync_lost_at = processed
                resync_in_progress = True
                ubx_nmea_state = sync_lost
            else:
                # Valid UBX message was received. Check if we have seen this message type before
                if message_type in messages:
                    messages[message_type] += 1 # if we have, increment its count
                else:
                    messages[message_type] = 1 # if we have not, set its count to 1
                rewind_in_progress = False # Clear rewind_in_progress
                rewind_to = -1
                if (resync_in_progress == True): # Check if we are resyncing
                    resync_in_progress = False # Clear the flag now that a valid message has been received
                    resyncs += 1 # Increment the number of successful resyncs
                    print("Sync successfully re-established at byte "+str(processed)+". The UBX message started at byte "+str(message_start_byte))
                    print()
        # NMEA messages
        elif (ubx_nmea_state == looking_for_asterix):
            nmea_length = nmea_length + 1 # Increase the message length count
            if (nmea_length > max_nmea_len): # If the length is greater than max_nmea_len, something bad must have happened (sync_lost)
                print("Panic!! Excessive NMEA message length!")
                print("Sync lost at byte "+str(processed)+". Attemting to re-sync")
                sync_lost_at = processed
                resync_in_progress = True
                ubx_nmea_state = sync_lost
                continue
            # If this is one of the first five characters, store it
            if (nmea_length <= 5):
                if (nmea_length == 1):
                    nmea_char_1 = c
                    rewind_to = processed # If we lose sync due to dropped bytes then rewind to here
                elif (nmea_length == 2):
                    nmea_char_2 = c
                elif (nmea_length == 3):
                    nmea_char_3 = c
                elif (nmea_length == 4):
                    nmea_char_4 = c
                else: # ubx_length == 5
                    nmea_char_5 = c
                    message_type = chr(nmea_char_1) + chr(nmea_char_2) + chr(nmea_char_3) + chr(nmea_char_4) + chr(nmea_char_5) # Record the message type
            # Now check if this is an '*'
            if (c == 0x2A):
                # Asterix received
                # Don't exOR it into the checksum
                # Instead calculate what the expected checksum should be (nmea_csum in ASCII hex)
                nmea_expected_csum1 = ((nmea_csum & 0xf0) >> 4) + 0x30 # Convert MS nibble to ASCII hex
                if (nmea_expected_csum1 >= 0x3A): # : follows 9 so add 7 to convert to A-F
                    nmea_expected_csum1 += 7
                nmea_expected_csum2 = (nmea_csum & 0x0f) + 0x30 # Convert LS nibble to ASCII hex
                if (nmea_expected_csum2 >= 0x3A): # : follows 9 so add 7 to convert to A-F
                    nmea_expected_csum2 += 7
                # Next, look for the first csum character
                ubx_nmea_state = looking_for_csum1
                continue # Don't include the * in the checksum
            # Now update the checksum
            # The checksum is the exclusive-OR of all characters between the $ and the *
            nmea_csum = nmea_csum ^ c
        elif (ubx_nmea_state == looking_for_csum1):
            # Store the first NMEA checksum character
            nmea_csum1 = c
            ubx_nmea_state = looking_for_csum2
        elif (ubx_nmea_state == looking_for_csum2):
            # Store the second NMEA checksum character
            nmea_csum2 = c
            # Now check if the checksum is correct
            if ((nmea_csum1 != nmea_expected_csum1) or (nmea_csum2 != nmea_expected_csum2)):
                # The checksum does not match so sync_lost
                print("Panic!! NMEA checksum error!")
                print("Sync lost at byte "+str(processed)+". Attemting to re-sync")
                sync_lost_at = processed
                resync_in_progress = True
                ubx_nmea_state = sync_lost
            else:
                # Checksum was valid so wait for the terminators
                ubx_nmea_state = looking_for_term1
        elif (ubx_nmea_state == looking_for_term1):
            # Check if this is CR
            if (c != 0x0D):
                print("Panic!! NMEA CR not found!")
                print("Sync lost at byte "+str(processed)+". Attemting to re-sync")
                sync_lost_at = processed
                resync_in_progress = True
                ubx_nmea_state = sync_lost
            else:
                ubx_nmea_state = looking_for_term2
        elif (ubx_nmea_state == looking_for_term2):
            # Check if this is LF
            if (c != 0x0A):
                print("Panic!! NMEA LF not found!")
                print("Sync lost at byte "+str(processed)+". Attemting to re-sync")
                sync_lost_at = processed
                resync_in_progress = True
                ubx_nmea_state = sync_lost
            else:
                # Valid NMEA message was received. Check if we have seen this message type before
                if message_type in messages:
                    messages[message_type] += 1 # if we have, increment its count
                else:
                    messages[message_type] = 1 # if we have not, set its count to 1
                # LF was received so go back to looking for B5 or a $
                ubx_nmea_state = looking_for_B5_dollar
                rewind_in_progress = False # Clear rewind_in_progress
                rewind_to = -1
                if (resync_in_progress == True): # Check if we are resyncing
                    resync_in_progress = False # Clear the flag now that a valid message has been received
                    resyncs += 1 # Increment the number of successful resyncs
                    print("Sync successfully re-established at byte "+str(processed)+". The NMEA message started at byte "+str(message_start_byte))
                    print()

        # Check if the end of the file has been reached
        if (processed >= filesize - 1): keepGoing = False

        # Check if we should attempt to rewind
        # Don't rewind if we have not yet seen a valid message
        # Don't rewind if a rewind is already in progress
        if (ubx_nmea_state == sync_lost) and (len(messages) > 0) and (rewind_in_progress == False) and (rewind_to >= 0):
            rewind_attempts += 1 # Increment the number of rewind attempts
            if (rewind_attempts > max_rewinds): # Only rewind up to max_rewind times
                print("Panic! Maximum rewind attempts reached! Aborting...")
                keepGoing = False
            else:
                print("Sync has been lost. Currently processing byte "+str(processed)+". Rewinding to byte "+str(rewind_to))
                fi.seek(rewind_to) # Rewind the file
                processed = rewind_to - 1 # Rewind processed too! (-1 is needed as processed is incremented at the start of the loop)
                rewind_in_progress = True # Flag that a rewind is in progress
            

finally:
    fi.close() # Close the file

    # Print the file statistics
    print()
    processed += 1
    print('Processed',processed,'bytes')
    print('File size was',filesize)
    if (processed != filesize):
        print('FILE SIZE MISMATCH!!')
    print('Longest UBX message was %i data bytes'%longest)
    if len(messages) > 0:
        print('Message types and totals were:')
        for key in messages.keys():
            print('Message type:',key,'  Total:',messages[key])
    if (resyncs > 0):
        print('Number of successful resyncs:',resyncs)
    print()
    print('Bye!')