From 1079efb689c7d915bfd11bfc8ba317b48f882f41 Mon Sep 17 00:00:00 2001 From: Adam Shapiro Date: Fri, 24 Apr 2026 11:04:53 -0400 Subject: [PATCH 01/17] Fixed bad __del__() call in device_bridge.py. --- bin/device_bridge.py | 2 -- 1 file changed, 2 deletions(-) diff --git a/bin/device_bridge.py b/bin/device_bridge.py index 170ed43..9a4b914 100755 --- a/bin/device_bridge.py +++ b/bin/device_bridge.py @@ -83,8 +83,6 @@ def __del__(self): self.virtual_serial.close() self.virtual_serial = None - super().__del__() - def start(self, other_device): if other_device is not None: self.handler = SerialDataHandler(source=self, destination=other_device) From f6dc2944858935e737190ebb68feaecea8213a72 Mon Sep 17 00:00:00 2001 From: Adam Shapiro Date: Fri, 24 Apr 2026 11:05:17 -0400 Subject: [PATCH 02/17] Increased max FE payload length in extract/analysis tool. --- bin/raw_analysis.py | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/bin/raw_analysis.py b/bin/raw_analysis.py index d424301..a130dd5 100755 --- a/bin/raw_analysis.py +++ b/bin/raw_analysis.py @@ -54,10 +54,8 @@ def get_fd(input_path: str, options): def index_messages(input_path, options): rtcm_framer = RTCMFramer() if 'rtcm' in options.format else None - fe_framer = FusionEngineDecoder( - max_payload_len_bytes=4096, return_offset=True) if 'fe' in options.format else None - nmea_framer = NMEAFramer( - return_offset=True) if 'nmea' in options.format else None + fe_framer = FusionEngineDecoder(max_payload_len_bytes=16536, return_offset=True) if 'fe' in options.format else None + nmea_framer = NMEAFramer(return_offset=True) if 'nmea' in options.format else None in_fd = get_fd(input_path, options) skip_bytes = options.skip_bytes From f80cf4fde3d60c199bf07ec05a507aa7d17d2c5d Mon Sep 17 00:00:00 2001 From: Adam Shapiro Date: Fri, 24 Apr 2026 11:05:36 -0400 Subject: [PATCH 03/17] Fixed >= condition when finished reading. --- bin/raw_analysis.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/bin/raw_analysis.py b/bin/raw_analysis.py index a130dd5..757b8bb 100755 --- a/bin/raw_analysis.py +++ b/bin/raw_analysis.py @@ -121,11 +121,10 @@ def index_messages(input_path, options): (total_bytes_read, bytes_to_process, 100.0 * float(total_bytes_read) / bytes_to_process, elapsed_sec, total_bytes_read / elapsed_sec / 1e6)) - if total_bytes_read > bytes_to_process: + if total_bytes_read >= bytes_to_process: break data = in_fd.read(READ_SIZE) - if len(data) == 0: break From 49093ddb3ce436a4e91c45105d38408fa970dbba Mon Sep 17 00:00:00 2001 From: Adam Shapiro Date: Fri, 24 Apr 2026 11:06:01 -0400 Subject: [PATCH 04/17] Added streaming stdin/stdout support for extract_nmea.py, etc. --- bin/raw_analysis.py | 298 +++++++++++++++++++++++++++++++++++++++----- 1 file changed, 267 insertions(+), 31 deletions(-) diff --git a/bin/raw_analysis.py b/bin/raw_analysis.py index 757b8bb..e311e7c 100755 --- a/bin/raw_analysis.py +++ b/bin/raw_analysis.py @@ -44,7 +44,9 @@ def get_output_file_path(input_path, postfix, output_dir=None, prefix=None): def get_fd(input_path: str, options): - if input_path.endswith('.p1bin'): + if input_path == '-': + return sys.stdin.buffer + elif input_path.endswith('.p1bin'): _logger.info(f"Reading raw data from p1bin {options.p1bin_type}.") return P1BinFileStream(input_path, options.p1bin_type, ignore_index=options.ignore_index, show_read_progress=True) @@ -225,6 +227,203 @@ def generate_separated_logs(input_path, indexes, options): output_map[index[0]].write(data) +def separate_and_index(input_path, options): + # Setup. + rtcm_framer = RTCMFramer() if 'rtcm' in options.format else None + fe_framer = FusionEngineDecoder(max_payload_len_bytes=16536, return_bytes=True, return_offset=True) \ + if 'fe' in options.format else None + nmea_framer = NMEAFramer(return_offset=True) if 'nmea' in options.format else None + + index = [] + total_bytes_read = 0 + + # Open the file and skip the first N bytes if requested. + in_fd = get_fd(input_path, options) + read_from_stdin = in_fd is sys.stdin.buffer + write_to_stdout = options.prefix == '-' + + skip_bytes = options.skip_bytes + if read_from_stdin: + file_size = 0 + if options.bytes_to_process is None: + bytes_to_process = sys.maxsize + else: + bytes_to_process = options.bytes_to_process + + try: + in_fd.read(skip_bytes) + except (BrokenPipeError, KeyboardInterrupt): + # User hit Ctrl-C -- done processing. + return index, total_bytes_read + else: + in_fd.seek(0, io.SEEK_END) + file_size = in_fd.tell() + in_fd.seek(skip_bytes, 0) + + bytes_to_process = file_size - skip_bytes + if options.bytes_to_process is not None and options.bytes_to_process < bytes_to_process: + bytes_to_process = options.bytes_to_process + + # Open output files if extracting data. + output_map = {} + current_base_id = -1 + rtcm_file_idx = 0 + if options.extract: + if write_to_stdout: + if len(options.format) > 1: + _logger.error('Only one data type may be written to stdout.') + sys.exit(1) + elif 'nmea' in options.format: + output_map['nmea'] = sys.stdout + elif 'rtcm' in options.format: + output_map['rtcm'] = sys.stdout.buffer + elif 'fe' in options.format: + output_map['fe'] = sys.stdout.buffer + else: + if 'nmea' in options.format: + # Note need the write binary to avoid needing to decode the ascii in the for loop. + output_map['nmea'] = open(get_output_file_path(input_path, '.nmea', + output_dir=options.output_dir, prefix=options.prefix), + 'wt') + if 'rtcm' in options.format: + suffix = '_0.rtcm3' if options.split_rtcm_base_id else '.rtcm3' + output_map['rtcm'] = open(get_output_file_path(input_path, suffix, + output_dir=options.output_dir, prefix=options.prefix), + 'wb') + if 'fe' in options.format: + output_map['fe'] = open(get_output_file_path(input_path, '.p1log', + output_dir=options.output_dir, prefix=options.prefix), + 'wb') + + # Open an index file only if writing to disk. + if write_to_stdout: + index_path = None + elif read_from_stdin: + if len(options.format) < len(FORMAT_STRS): + index_file = get_output_file_path( + input_path=None if read_from_stdin else input_path, + postfix='_' + '_'.join(options.format) + '_index.csv', + output_dir=options.output_dir, + prefix=options.prefix) + else: + index_file = get_output_file_path( + input_path=None if read_from_stdin else input_path, + postfix='_index.csv', + output_dir=options.output_dir, + prefix=options.prefix) + else: + index_path = get_output_file_path(input_path, '_index.csv', output_dir=options.output_dir) + + if index_path is None: + timestamp_fd = None + else: + timestamp_fd = open(index_path, 'wt') + timestamp_fd.write( + 'Protocol, ID, Offset (Bytes), Length (Bytes), P1 Time\n') + + # Status printout helper. + def _print_status(elapsed_sec): + if file_size == 0: + _logger.log(logging.INFO, + 'Processed %d bytes. [elapsed=%.1f sec, rate=%.1f MB/s]' % + (total_bytes_read, elapsed_sec, total_bytes_read / elapsed_sec / 1e6)) + else: + _logger.log(logging.INFO, + 'Processed %d/%d bytes (%.1f%%). [elapsed=%.1f sec, rate=%.1f MB/s]' % + (total_bytes_read, bytes_to_process, 100.0 * float(total_bytes_read) / bytes_to_process, + elapsed_sec, total_bytes_read / elapsed_sec / 1e6)) + + # Now read from the file. + start_time = datetime.now() + next_update_time = 0 + try: + while True: + # Update status. + elapsed_sec = (datetime.now() - start_time).total_seconds() + if elapsed_sec > next_update_time: + _print_status(elapsed_sec) + next_update_time = elapsed_sec + 5 + + # If we've hit the limit, we're done. + if total_bytes_read >= bytes_to_process: + break + + # Otherwise, read more data. If reading from stdin, this will block unless interrupted. + data = in_fd.read(READ_SIZE) + if len(data) == 0: + break + + total_bytes_read += len(data) + + # Index the data: + # Protocol, Message ID, Stream/File Offset, Size, P1 Time (if applicable) + current_block_entries = [] + if rtcm_framer is not None: + for msg in rtcm_framer.on_data(data, return_size=True, return_offset=True, + return_bytes=options.extract): + message_id = msg["message"].message_id + current_block_entries.append( + ('rtcm', message_id, skip_bytes + msg["offset"], msg["size"], '')) + if options.extract: + raw_bytes = msg['bytes'] + + # If requested, open a new file if the base station changes. + if options.split_rtcm_base_id and is_msm_id(message_id): + # offset 36 bits, length 12 bits. + base_id = ((raw_bytes[4] & 0xF) << 8) + raw_bytes[5] + if base_id != current_base_id: + if current_base_id != -1: + output_map['rtcm'].close() + rtcm_file_idx += 1 + path = get_output_file_path(input_path, f'_{rtcm_file_idx}.rtcm3', + output_dir=options.output_dir, prefix=options.prefix) + output_map['rtcm'] = open(path, 'wb') + _logger.info(f"Writing for base station id: {base_id}") + current_base_id = base_id + + output_map['rtcm'].write(raw_bytes) + if fe_framer is not None: + for header, payload, raw_data, offset_bytes in fe_framer.on_data(data): + p1_time = payload.get_p1_time() if isinstance(payload, MessagePayload) else None + current_block_entries.append( + ('fe', int(header.message_type), skip_bytes + offset_bytes, + header.get_message_size(), + '%.3f' % float(p1_time) if p1_time is not None else '')) + if options.extract: + output_map['fe'].write(raw_data) + if nmea_framer is not None: + for msg in nmea_framer.on_data(data): + current_block_entries.append( + ('nmea', msg[0].split(',')[0][1:], skip_bytes + msg[1], len(msg[0]), '')) + if options.extract: + output_map['nmea'].write(msg[0]) + + # The returned index omits P1 time. + index.extend(current_block_entries[:4]) + + if timestamp_fd is not None: + for entry in sorted(current_block_entries, key=lambda e: e[2]): + timestamp_fd.write( + f'{",".join([str(elem) for elem in entry])}\n') + except (BrokenPipeError, KeyboardInterrupt): + # User hit Ctrl-C -- done processing. + pass + + # Write index file EOF. + if timestamp_fd is not None: + timestamp_fd.write(f'{EOF_FORMAT},0,{bytes_to_process},0,\n') + + # Sort the data. + index = sorted(index, key=lambda e: e[2]) + + # Print status one last time. + elapsed_sec = (datetime.now() - start_time).total_seconds() + _print_status(elapsed_sec) + + return index, total_bytes_read + + + parser = ArgumentParser(description="""\ Analyze contents of a input.raw or input.p1bin and create csv with offset and length of each NMEA, RTCM, and FE message. Print out locations of data gaps. @@ -242,10 +441,12 @@ def generate_separated_logs(input_path, indexes, options): help="If set, re-run index generation.") parser.add_argument('-o', '--output-dir', type=str, metavar='DIR', help="The directory where output will be stored. Defaults to the parent directory of the input" - "file, or to the log directory if reading from a log.") + "file, or to the log directory if reading from a log. When reading from stdin, defaults to the " + "current working directory.") parser.add_argument('-p', '--prefix', type=str, - help="Use the specified prefix for the output file: `.p1log`. Otherwise, use the " - "filename of the input data file.") + help="Use the specified prefix for the output file: .p1log, .nmea, etc. Otherwise, " + "use the filename of the input data file. Set to '-' to write to stdout. If not specified and " + "reading from stdin, output will be written to stdout.") parser.add_argument( '-t', '--p1bin-type', type=str, action='append', help="An optional list message types to analyse from a p1bin file. Defaults to 'EXTERNAL_UNFRAMED_GNSS'. Only used " @@ -271,41 +472,69 @@ def generate_separated_logs(input_path, indexes, options): parser.add_argument('--check-gaps', action=ExtendedBooleanAction, default=True, help="If set, search for unframed bytes that do not belong to a complete message from any " "protocol, indicating the existence of a gap in the data stream.") -parser.add_argument('log', +parser.add_argument('log', nargs='?', default='-', help="The log to be read. May be one of:\n" "- The path to a binary log file\n" "- The path to a FusionEngine log directory\n" "- A pattern matching a FusionEngine log directory under the specified base directory " - "(see find_fusion_engine_log() and --log-base-dir)") + "(see find_fusion_engine_log() and --log-base-dir)\n" + "- '-' or omit to read from stdin") def raw_analysis(options): + # If we're reading from stdin, we have some different behaviors below: + # - If the user does not specify --prefix so we cannot set an output filename, we will write output to stdout + # - If the user does specify --prefix but doesn't set --output-dir, we'll write to CWD + # - When writing to stdout, we'll redirect logger prints to stderr + read_from_stdin = options.log == '-' + if read_from_stdin: + if options.prefix is None: + options.prefix = '-' + + if options.output_dir is None: + options.output_dir = os.getcwd() + + write_to_stdout = options.prefix == '-' + + # When writing to stdout, we cannot split RTCM data into multiple files (there is only one stdout). + if write_to_stdout: + options.split_rtcm_base_id = False + # Configure logging. + if write_to_stdout: + logging_stream = sys.stderr + else: + logging_stream = sys.stdout + logger = logging.getLogger('point_one') if options.verbose >= 1: logging.basicConfig(level=logging.INFO, format='%(levelname)s - %(name)s:%(lineno)d - %(message)s', - stream=sys.stdout) + stream=logging_stream) if options.verbose == 1: logger.setLevel(logging.DEBUG) else: - logging.basicConfig(level=logging.INFO, - format='%(message)s', stream=sys.stdout) + logging.basicConfig(level=logging.INFO, format='%(message)s', stream=logging_stream) # Locate the input file and set the output directory. - try: - input_path, output_dir, log_id = find_log_file(options.log, candidate_files=['input.raw', 'input.p1bin'], - return_output_dir=True, return_log_id=True, - log_base_dir=options.log_base_dir) - - if log_id is None: - logger.info('Loading %s.' % os.path.basename(input_path)) - else: - logger.info('Loading %s from log %s.' % - (os.path.basename(input_path), log_id)) - - except FileNotFoundError as e: - logger.error(str(e)) - sys.exit(1) + if read_from_stdin: + input_path = options.log + output_dir = options.output_dir + log_id = None + else: + try: + input_path, output_dir, log_id = find_log_file(options.log, candidate_files=['input.raw', 'input.p1bin'], + return_output_dir=True, return_log_id=True, + log_base_dir=options.log_base_dir) + + if log_id is None: + logger.info('Loading %s.' % os.path.basename(input_path)) + else: + logger.info('Loading %s from log %s.' % + (os.path.basename(input_path), log_id)) + + except FileNotFoundError as e: + logger.error(str(e)) + sys.exit(1) if options.format is not None: # If the user specified a set of formats, lookup their type values. Below, we will limit the processing to only @@ -316,9 +545,7 @@ def raw_analysis(options): logger.error(f'Invalid format "{f}".') sys.exit(1) options.format = format - format_str = '_' + '_'.join(format) else: - format_str = '' options.format = FORMAT_STRS logger.info(f"Processing {options.format}.") @@ -341,14 +568,23 @@ def raw_analysis(options): else: options.p1bin_type = [P1BinType.EXTERNAL_UNFRAMED_GNSS] - logger.info(f"Output index stored in '{output_dir}'.") - index, file_size_bytes = index_messages(input_path, options) + # If reading from stdin, we can't preemptively index the data. Build the index as we go. + if read_from_stdin: + if options.extract: + logger.info(f"Output stored in '{output_dir}'.") + index, file_size_bytes = separate_and_index(input_path, options) + if options.check_gaps: + find_gaps(index) + # If reading from a file, index the file and then perform the requested operation. + else: + logger.info(f"Output stored in '{output_dir}'.") + index, file_size_bytes = index_messages(input_path, options) - if options.check_gaps: - find_gaps(index) + if options.check_gaps: + find_gaps(index) - if options.extract: - generate_separated_logs(input_path, index, options) + if options.extract: + generate_separated_logs(input_path, index, options) _logger.info("") format_string = '| {:<10} | {:>10} | {:>10} |' From 580a5a7bc673d9c7e54d570868f7f96a1da69668 Mon Sep 17 00:00:00 2001 From: Adam Shapiro Date: Fri, 24 Apr 2026 11:22:49 -0400 Subject: [PATCH 05/17] Write to stdout when reading from a file. --- bin/raw_analysis.py | 38 ++++++++++++++++++++++++++------------ 1 file changed, 26 insertions(+), 12 deletions(-) diff --git a/bin/raw_analysis.py b/bin/raw_analysis.py index e311e7c..27738e8 100755 --- a/bin/raw_analysis.py +++ b/bin/raw_analysis.py @@ -191,26 +191,40 @@ def find_gaps(indexes): def generate_separated_logs(input_path, indexes, options): + write_to_stdout = options.prefix == '-' output_map = {} current_base_id = -1 rtcm_file_idx = 0 - if 'nmea' in options.format: - # Note need the write binary to avoid needing to decode the ascii in the for loop. - output_map['nmea'] = open(get_output_file_path(input_path, '.nmea', - output_dir=options.output_dir, prefix=options.prefix), 'wb') - if 'rtcm' in options.format: - suffix = '_0.rtcm3' if options.split_rtcm_base_id else '.rtcm3' - output_map['rtcm'] = open(get_output_file_path(input_path, suffix, - output_dir=options.output_dir, prefix=options.prefix), 'wb') - if 'fe' in options.format: - output_map['fe'] = open(get_output_file_path(input_path, '.p1log', - output_dir=options.output_dir, prefix=options.prefix), 'wb') + if write_to_stdout: + if len(options.format) > 1: + _logger.error('Only one data type may be written to stdout.') + sys.exit(1) + elif 'nmea' in options.format: + output_map['nmea'] = sys.stdout.buffer + elif 'rtcm' in options.format: + output_map['rtcm'] = sys.stdout.buffer + elif 'fe' in options.format: + output_map['fe'] = sys.stdout.buffer + else: + if 'nmea' in options.format: + # Note need the write binary to avoid needing to decode the ascii in the for loop. + output_map['nmea'] = open(get_output_file_path(input_path, '.nmea', + output_dir=options.output_dir, prefix=options.prefix), 'wb') + if 'rtcm' in options.format: + suffix = '_0.rtcm3' if options.split_rtcm_base_id else '.rtcm3' + output_map['rtcm'] = open(get_output_file_path(input_path, suffix, + output_dir=options.output_dir, prefix=options.prefix), 'wb') + if 'fe' in options.format: + output_map['fe'] = open(get_output_file_path(input_path, '.p1log', + output_dir=options.output_dir, prefix=options.prefix), 'wb') in_fd = get_fd(input_path, options) for index in indexes: if index[0] in output_map: in_fd.seek(index[2], io.SEEK_SET) data = in_fd.read(index[3]) + + # If requested, open a new file if the base station changes. if options.split_rtcm_base_id and index[0] == 'rtcm' and is_msm_id(int(index[1])): # offset 36 bits, length 12 bits. base_id = ((data[4] & 0xF) << 8) + data[5] @@ -222,8 +236,8 @@ def generate_separated_logs(input_path, indexes, options): input_path, f'_{rtcm_file_idx}.rtcm3', output_dir=options.output_dir, prefix=options.prefix), 'wb') _logger.info(f"Writing for base station id: {base_id}") + current_base_id = base_id - current_base_id = base_id output_map[index[0]].write(data) From ad78cfa47df0f29b0342938ac4cd7e4f6bbdc1e2 Mon Sep 17 00:00:00 2001 From: Adam Shapiro Date: Fri, 24 Apr 2026 12:42:49 -0400 Subject: [PATCH 06/17] Reorganized raw_analysis.py to eliminate duplicate code. --- bin/raw_analysis.py | 496 ++++++++++++++++++++------------------------ 1 file changed, 229 insertions(+), 267 deletions(-) diff --git a/bin/raw_analysis.py b/bin/raw_analysis.py index 27738e8..ef592d4 100755 --- a/bin/raw_analysis.py +++ b/bin/raw_analysis.py @@ -54,123 +54,261 @@ def get_fd(input_path: str, options): return open(input_path, 'rb') -def index_messages(input_path, options): - rtcm_framer = RTCMFramer() if 'rtcm' in options.format else None - fe_framer = FusionEngineDecoder(max_payload_len_bytes=16536, return_offset=True) if 'fe' in options.format else None - nmea_framer = NMEAFramer(return_offset=True) if 'nmea' in options.format else None +def _create_framers(options, return_bytes=False): + rtcm = RTCMFramer() if 'rtcm' in options.format else None + fe = FusionEngineDecoder(max_payload_len_bytes=16536, return_offset=True, + return_bytes=return_bytes) if 'fe' in options.format else None + nmea = NMEAFramer(return_offset=True) if 'nmea' in options.format else None + return rtcm, fe, nmea - in_fd = get_fd(input_path, options) - skip_bytes = options.skip_bytes - in_fd.seek(0, io.SEEK_END) - file_size = in_fd.tell() - in_fd.seek(skip_bytes, 0) +def _get_index_path(input_path, options): + if len(options.format) < len(FORMAT_STRS): + postfix = '_' + '_'.join(sorted(options.format)) + '_index.csv' + else: + postfix = '_index.csv' + return get_output_file_path(input_path, postfix, output_dir=options.output_dir, prefix=options.prefix) - bytes_to_process = file_size - skip_bytes - if options.bytes_to_process is not None and options.bytes_to_process < bytes_to_process: - bytes_to_process = options.bytes_to_process - # File without prefix indicates all parsers used. - index_file_full = get_output_file_path( - input_path, '_index.csv', output_dir=options.output_dir, prefix=options.prefix) +def _open_output_files(input_path, options, text_nmea=False): + """! + @brief Open output files for extraction. - if len(options.format) < len(FORMAT_STRS): - index_file = get_output_file_path( - input_path, '_' + '_'.join(options.format) + '_index.csv', output_dir=options.output_dir, - prefix=options.prefix) + @return An `output_map` dict keyed by protocol name. + """ + write_to_stdout = options.prefix == '-' + output_map = {} + + if write_to_stdout: + # Only one format may be written to stdout - error if multiple are requested. + if len(options.format) > 1: + _logger.error('Only one data type may be written to stdout.') + sys.exit(1) + + # Map the single requested format to stdout. NMEA is text when the framer produces strings, RTCM and FE are + # always binary. + for fmt in ('nmea', 'rtcm', 'fe'): + if fmt in options.format: + output_map[fmt] = sys.stdout if (fmt == 'nmea' and text_nmea) else sys.stdout.buffer + break else: - index_file = index_file_full + # Open a dedicated output file for each requested protocol. + if 'nmea' in options.format: + output_map['nmea'] = open( + get_output_file_path(input_path, '.nmea', output_dir=options.output_dir, prefix=options.prefix), + 'wt' if text_nmea else 'wb') + if 'rtcm' in options.format: + # When splitting by base station, start with file index 0. New files are opened as the base ID changes. + suffix = '_0.rtcm3' if options.split_rtcm_base_id else '.rtcm3' + output_map['rtcm'] = open( + get_output_file_path(input_path, suffix, output_dir=options.output_dir, prefix=options.prefix), 'wb') + if 'fe' in options.format: + output_map['fe'] = open( + get_output_file_path(input_path, '.p1log', output_dir=options.output_dir, prefix=options.prefix), 'wb') - if not options.ignore_index: - # Try using the full index when processing a subset of formats. - index_files_to_load = set([index_file, index_file_full]) - for index_file_to_load in index_files_to_load: - if os.path.exists(index_file_to_load): - index = load_index(index_file_to_load) + return output_map - # Check if index was generated for same datafile. - # - # The index file should always have an EOF marker. If it does not, we might have run into an error while - # generating it. - eof_index = index[-1] if len(index) > 0 else None - if eof_index is None or eof_index[0] != EOF_FORMAT: - _logger.warning( - f'Index file "{index_file_to_load}" missing EOF entry, skipping load.') - else: - if eof_index[2] != bytes_to_process: - _logger.info( - f'Index file "{index_file_to_load}" was generated for different input data, skipping load.') - else: - _logger.info( - f'Using existing index "{index_file_to_load}".') - return index, file_size - _logger.info(f"Indexing raw input.") +def _stream_and_index(input_path, in_fd, options, + rtcm_framer, fe_framer, nmea_framer, + skip_bytes, bytes_to_process, file_size, + output_map, index_path): + """ + @brief Core read loop: parse messages, write index CSV, and optionally extract to output files. + + Pass `output_map=None` for index-only (no extraction). `file_size=0` means stdin (progress shown as raw bytes). + + @return Returns a tuple: `(index, total_bytes_read)`. + """ + extract = output_map is not None + index = [] + total_bytes_read = 0 + current_base_id = -1 + rtcm_file_idx = 0 + + # Open the index file if requested. + timestamp_fd = None + if index_path is not None: + timestamp_fd = open(index_path, 'wt') + timestamp_fd.write('Protocol, ID, Offset (Bytes), Length (Bytes), P1 Time\n') start_time = datetime.now() next_update_time = 0 - with open(index_file, 'w') as timestamp_fd: - timestamp_fd.write( - 'Protocol, ID, Offset (Bytes), Length (Bytes), P1 Time\n') + # Status print helper function. + def _print_status(elapsed_sec): + if file_size == 0: + _logger.info('Processed %d bytes. [elapsed=%.1f sec, rate=%.1f MB/s]' % + (total_bytes_read, elapsed_sec, total_bytes_read / elapsed_sec / 1e6)) + else: + _logger.info('Processed %d/%d bytes (%.1f%%). [elapsed=%.1f sec, rate=%.1f MB/s]' % + (total_bytes_read, bytes_to_process, + 100.0 * float(total_bytes_read) / bytes_to_process, + elapsed_sec, total_bytes_read / elapsed_sec / 1e6)) + + # Read all incoming data until EOF or Ctrl-C. + try: while True: - total_bytes_read = in_fd.tell() - skip_bytes + # Print a progress update every 5 seconds. elapsed_sec = (datetime.now() - start_time).total_seconds() if elapsed_sec > next_update_time: + _print_status(elapsed_sec) next_update_time = elapsed_sec + 5 - _logger.log(logging.INFO, - 'Processed %d/%d bytes (%.1f%%). [elapsed=%.1f sec, rate=%.1f MB/s]' % - (total_bytes_read, bytes_to_process, 100.0 * float(total_bytes_read) / bytes_to_process, - elapsed_sec, total_bytes_read / elapsed_sec / 1e6)) if total_bytes_read >= bytes_to_process: break data = in_fd.read(READ_SIZE) - if len(data) == 0: + if not data: break + total_bytes_read += len(data) + + # Parse all three protocols from the current chunk. Each entry is a + # (protocol, id, stream_offset, size, p1_time) tuple. + # + # Note that we pass the entire chunk to each framer in sequence, not one byte at a time, so the framers may + # output messages out of order. entries = [] if rtcm_framer is not None: - for msg in rtcm_framer.on_data(data, return_size=True, return_offset=True): - entries.append( - ('rtcm', msg["message"].message_id, skip_bytes + msg["offset"], msg["size"], '')) + for msg in rtcm_framer.on_data(data, return_size=True, return_offset=True, return_bytes=extract): + message_id = msg["message"].message_id + offset_bytes = msg["offset"] + size_bytes = msg["size"] + entries.append(('rtcm', message_id, skip_bytes + offset_bytes, size_bytes, '')) + + if extract: + raw_data = msg['bytes'] + + # If splitting by base station ID, open a new output file each time the base changes. + if options.split_rtcm_base_id and is_msm_id(message_id): + # Base station ID is encoded at bit offset 36, length 12 bits. + base_id = ((raw_data[4] & 0xF) << 8) + raw_data[5] + if base_id != current_base_id: + if current_base_id != -1: + output_map['rtcm'].close() + rtcm_file_idx += 1 + output_map['rtcm'] = open(get_output_file_path( + input_path, f'_{rtcm_file_idx}.rtcm3', + output_dir=options.output_dir, prefix=options.prefix), 'wb') + _logger.info(f"Writing for base station id: {base_id}") + current_base_id = base_id + + output_map['rtcm'].write(raw_data) + if fe_framer is not None: - for header, payload, offset_bytes in fe_framer.on_data(data): + for result in fe_framer.on_data(data): + # The framer returns raw bytes as a third element only when constructed with return_bytes=True. + if extract: + header, payload, raw_data, offset_bytes = result + else: + header, payload, offset_bytes = result + raw_data = None + + message_id = int(header.message_type) + size_bytes = header.get_message_size() p1_time = payload.get_p1_time() if isinstance(payload, MessagePayload) else None - entries.append(('fe', int(header.message_type), skip_bytes + offset_bytes, - header.get_message_size(), + entries.append(('fe', message_id, skip_bytes + offset_bytes, size_bytes, '%.3f' % float(p1_time) if p1_time is not None else '')) + + if extract: + output_map['fe'].write(raw_data) + if nmea_framer is not None: for msg in nmea_framer.on_data(data): - entries.append(('nmea', msg[0].split( - ',')[0][1:], skip_bytes + msg[1], len(msg[0]), '')) + # Note: NMEA messages are strings, not binary. + raw_data = msg[0] + message_id = raw_data.split(',')[0][1:] + offset_bytes = msg[1] + size_bytes = len(raw_data) + entries.append(('nmea', message_id, skip_bytes + offset_bytes, size_bytes, '')) + + if extract: + output_map['nmea'].write(raw_data) - for entry in sorted(entries, key=lambda e: e[2]): - timestamp_fd.write( - f'{",".join([str(elem) for elem in entry])}\n') + # Accumulate index entries, dropping the P1 time field which is only written to the CSV. + index.extend(e[:4] for e in entries) + # Write entries to the index CSV sorted by byte offset within this chunk. + if timestamp_fd is not None: + for entry in sorted(entries, key=lambda e: e[2]): + timestamp_fd.write(f'{",".join([str(elem) for elem in entry])}\n') + except (BrokenPipeError, KeyboardInterrupt): + # User hit Ctrl-C - done processing. + pass + + # Close the index file. + if timestamp_fd is not None: + # Write the EOF sentinel so future loads can verify the index covers the full byte range. timestamp_fd.write(f'{EOF_FORMAT},0,{bytes_to_process},0,\n') + timestamp_fd.close() - total_bytes_read = bytes_to_process - _logger.log(logging.INFO, - 'Processed %d/%d bytes (%.1f%%). [elapsed=%.1f sec, rate=%.1f MB/s]' % - (total_bytes_read, bytes_to_process, 100.0 * float(total_bytes_read) / bytes_to_process, - elapsed_sec, total_bytes_read / elapsed_sec / 1e6)) + # Print final status after the loop exits. + elapsed_sec = (datetime.now() - start_time).total_seconds() + if elapsed_sec > 0: + _print_status(elapsed_sec) - return load_index(index_file), file_size + return sorted(index, key=lambda e: e[2]), total_bytes_read + + +def index_messages(input_path, options): + in_fd = get_fd(input_path, options) + skip_bytes = options.skip_bytes + + # Determine the range of bytes to process. + in_fd.seek(0, io.SEEK_END) + file_size = in_fd.tell() + in_fd.seek(skip_bytes, 0) + + bytes_to_process = file_size - skip_bytes + if options.bytes_to_process is not None and options.bytes_to_process < bytes_to_process: + bytes_to_process = options.bytes_to_process + + # Determine the path to the index file. If the file exists already, read it and return. If not, generate it. + index_file_full = get_output_file_path(input_path, '_index.csv', output_dir=options.output_dir, + prefix=options.prefix) + index_file = _get_index_path(input_path, options) + + if not options.ignore_index: + # Try to reuse a previously generated index if one exists and covers the same byte range. + # When processing a subset of formats, also try the full-format index as a fallback. + for index_file_to_load in {index_file, index_file_full}: + if os.path.exists(index_file_to_load): + index = load_index(index_file_to_load) + + # The index file should always end with an EOF marker. A missing marker means the file + # was incomplete, likely due to an error during a previous indexing run. + eof_index = index[-1] if index else None + if eof_index is None or eof_index[0] != EOF_FORMAT: + _logger.warning(f'Index file "{index_file_to_load}" missing EOF entry, skipping load.') + elif eof_index[2] != bytes_to_process: + _logger.info( + f'Index file "{index_file_to_load}" was generated for different input data, skipping load.') + else: + _logger.info(f'Using existing index "{index_file_to_load}".') + return index, file_size + + # No usable existing index found; generate a new one by parsing the input file. + _logger.info(f"Generating index file {index_file}.") + rtcm_framer, fe_framer, nmea_framer = _create_framers(options) + + return _stream_and_index(input_path=input_path, in_fd=in_fd, options=options, + rtcm_framer=rtcm_framer, fe_framer=fe_framer, nmea_framer=nmea_framer, + skip_bytes=skip_bytes, bytes_to_process=bytes_to_process, file_size=file_size, + output_map=None, index_path=index_file) def load_index(index_file): indexes = [] with open(index_file, 'r') as index_fd: + # Skip the header line. index_fd.readline() for line in index_fd.readlines(): fields = line.split(',') indexes.append( (fields[0], fields[1], int(fields[2]), int(fields[3]))) - # If the data has dropouts, the messages might not be in order due to how long the framers take to detect the error. + # Sort by offset to handle any out-of-order entries caused by framer latency during data dropouts. indexes = sorted(indexes, key=lambda x: x[2]) return indexes @@ -191,42 +329,21 @@ def find_gaps(indexes): def generate_separated_logs(input_path, indexes, options): - write_to_stdout = options.prefix == '-' - output_map = {} + # Open output files for each requested protocol. + output_map = _open_output_files(input_path, options, text_nmea=False) current_base_id = -1 rtcm_file_idx = 0 - if write_to_stdout: - if len(options.format) > 1: - _logger.error('Only one data type may be written to stdout.') - sys.exit(1) - elif 'nmea' in options.format: - output_map['nmea'] = sys.stdout.buffer - elif 'rtcm' in options.format: - output_map['rtcm'] = sys.stdout.buffer - elif 'fe' in options.format: - output_map['fe'] = sys.stdout.buffer - else: - if 'nmea' in options.format: - # Note need the write binary to avoid needing to decode the ascii in the for loop. - output_map['nmea'] = open(get_output_file_path(input_path, '.nmea', - output_dir=options.output_dir, prefix=options.prefix), 'wb') - if 'rtcm' in options.format: - suffix = '_0.rtcm3' if options.split_rtcm_base_id else '.rtcm3' - output_map['rtcm'] = open(get_output_file_path(input_path, suffix, - output_dir=options.output_dir, prefix=options.prefix), 'wb') - if 'fe' in options.format: - output_map['fe'] = open(get_output_file_path(input_path, '.p1log', - output_dir=options.output_dir, prefix=options.prefix), 'wb') + # Seek to each message's offset and copy its bytes to the appropriate output file. in_fd = get_fd(input_path, options) for index in indexes: if index[0] in output_map: in_fd.seek(index[2], io.SEEK_SET) data = in_fd.read(index[3]) - # If requested, open a new file if the base station changes. + # If splitting by base station, open a new file each time the base station ID changes. if options.split_rtcm_base_id and index[0] == 'rtcm' and is_msm_id(int(index[1])): - # offset 36 bits, length 12 bits. + # Base station ID is encoded at bit offset 36, length 12 bits. base_id = ((data[4] & 0xF) << 8) + data[5] if base_id != current_base_id: if current_base_id != -1: @@ -242,202 +359,47 @@ def generate_separated_logs(input_path, indexes, options): def separate_and_index(input_path, options): - # Setup. - rtcm_framer = RTCMFramer() if 'rtcm' in options.format else None - fe_framer = FusionEngineDecoder(max_payload_len_bytes=16536, return_bytes=True, return_offset=True) \ - if 'fe' in options.format else None - nmea_framer = NMEAFramer(return_offset=True) if 'nmea' in options.format else None - - index = [] - total_bytes_read = 0 - - # Open the file and skip the first N bytes if requested. + # Open the input file (or stdin). in_fd = get_fd(input_path, options) read_from_stdin = in_fd is sys.stdin.buffer write_to_stdout = options.prefix == '-' + # Determine byte range to process. For stdin the file size is unknown, so process all incoming data unless the user + # specifies --bytes-to-process. skip_bytes = options.skip_bytes if read_from_stdin: file_size = 0 - if options.bytes_to_process is None: - bytes_to_process = sys.maxsize - else: - bytes_to_process = options.bytes_to_process - + bytes_to_process = options.bytes_to_process if options.bytes_to_process is not None else sys.maxsize try: in_fd.read(skip_bytes) except (BrokenPipeError, KeyboardInterrupt): - # User hit Ctrl-C -- done processing. - return index, total_bytes_read + return [], 0 else: in_fd.seek(0, io.SEEK_END) file_size = in_fd.tell() in_fd.seek(skip_bytes, 0) - bytes_to_process = file_size - skip_bytes if options.bytes_to_process is not None and options.bytes_to_process < bytes_to_process: bytes_to_process = options.bytes_to_process - # Open output files if extracting data. - output_map = {} - current_base_id = -1 - rtcm_file_idx = 0 - if options.extract: - if write_to_stdout: - if len(options.format) > 1: - _logger.error('Only one data type may be written to stdout.') - sys.exit(1) - elif 'nmea' in options.format: - output_map['nmea'] = sys.stdout - elif 'rtcm' in options.format: - output_map['rtcm'] = sys.stdout.buffer - elif 'fe' in options.format: - output_map['fe'] = sys.stdout.buffer - else: - if 'nmea' in options.format: - # Note need the write binary to avoid needing to decode the ascii in the for loop. - output_map['nmea'] = open(get_output_file_path(input_path, '.nmea', - output_dir=options.output_dir, prefix=options.prefix), - 'wt') - if 'rtcm' in options.format: - suffix = '_0.rtcm3' if options.split_rtcm_base_id else '.rtcm3' - output_map['rtcm'] = open(get_output_file_path(input_path, suffix, - output_dir=options.output_dir, prefix=options.prefix), - 'wb') - if 'fe' in options.format: - output_map['fe'] = open(get_output_file_path(input_path, '.p1log', - output_dir=options.output_dir, prefix=options.prefix), - 'wb') - - # Open an index file only if writing to disk. - if write_to_stdout: - index_path = None - elif read_from_stdin: - if len(options.format) < len(FORMAT_STRS): - index_file = get_output_file_path( - input_path=None if read_from_stdin else input_path, - postfix='_' + '_'.join(options.format) + '_index.csv', - output_dir=options.output_dir, - prefix=options.prefix) - else: - index_file = get_output_file_path( - input_path=None if read_from_stdin else input_path, - postfix='_index.csv', - output_dir=options.output_dir, - prefix=options.prefix) - else: - index_path = get_output_file_path(input_path, '_index.csv', output_dir=options.output_dir) + # Open output files for extraction if requested. + output_map = _open_output_files(input_path, options, text_nmea=True) if options.extract else None - if index_path is None: - timestamp_fd = None - else: - timestamp_fd = open(index_path, 'wt') - timestamp_fd.write( - 'Protocol, ID, Offset (Bytes), Length (Bytes), P1 Time\n') + # Open an index CSV only when writing output to disk, skip it when streaming to stdout. + index_path = None if write_to_stdout else _get_index_path(input_path, options) - # Status printout helper. - def _print_status(elapsed_sec): - if file_size == 0: - _logger.log(logging.INFO, - 'Processed %d bytes. [elapsed=%.1f sec, rate=%.1f MB/s]' % - (total_bytes_read, elapsed_sec, total_bytes_read / elapsed_sec / 1e6)) - else: - _logger.log(logging.INFO, - 'Processed %d/%d bytes (%.1f%%). [elapsed=%.1f sec, rate=%.1f MB/s]' % - (total_bytes_read, bytes_to_process, 100.0 * float(total_bytes_read) / bytes_to_process, - elapsed_sec, total_bytes_read / elapsed_sec / 1e6)) + rtcm_framer, fe_framer, nmea_framer = _create_framers(options, return_bytes=options.extract) - # Now read from the file. - start_time = datetime.now() - next_update_time = 0 - try: - while True: - # Update status. - elapsed_sec = (datetime.now() - start_time).total_seconds() - if elapsed_sec > next_update_time: - _print_status(elapsed_sec) - next_update_time = elapsed_sec + 5 - - # If we've hit the limit, we're done. - if total_bytes_read >= bytes_to_process: - break - - # Otherwise, read more data. If reading from stdin, this will block unless interrupted. - data = in_fd.read(READ_SIZE) - if len(data) == 0: - break - - total_bytes_read += len(data) - - # Index the data: - # Protocol, Message ID, Stream/File Offset, Size, P1 Time (if applicable) - current_block_entries = [] - if rtcm_framer is not None: - for msg in rtcm_framer.on_data(data, return_size=True, return_offset=True, - return_bytes=options.extract): - message_id = msg["message"].message_id - current_block_entries.append( - ('rtcm', message_id, skip_bytes + msg["offset"], msg["size"], '')) - if options.extract: - raw_bytes = msg['bytes'] - - # If requested, open a new file if the base station changes. - if options.split_rtcm_base_id and is_msm_id(message_id): - # offset 36 bits, length 12 bits. - base_id = ((raw_bytes[4] & 0xF) << 8) + raw_bytes[5] - if base_id != current_base_id: - if current_base_id != -1: - output_map['rtcm'].close() - rtcm_file_idx += 1 - path = get_output_file_path(input_path, f'_{rtcm_file_idx}.rtcm3', - output_dir=options.output_dir, prefix=options.prefix) - output_map['rtcm'] = open(path, 'wb') - _logger.info(f"Writing for base station id: {base_id}") - current_base_id = base_id - - output_map['rtcm'].write(raw_bytes) - if fe_framer is not None: - for header, payload, raw_data, offset_bytes in fe_framer.on_data(data): - p1_time = payload.get_p1_time() if isinstance(payload, MessagePayload) else None - current_block_entries.append( - ('fe', int(header.message_type), skip_bytes + offset_bytes, - header.get_message_size(), - '%.3f' % float(p1_time) if p1_time is not None else '')) - if options.extract: - output_map['fe'].write(raw_data) - if nmea_framer is not None: - for msg in nmea_framer.on_data(data): - current_block_entries.append( - ('nmea', msg[0].split(',')[0][1:], skip_bytes + msg[1], len(msg[0]), '')) - if options.extract: - output_map['nmea'].write(msg[0]) - - # The returned index omits P1 time. - index.extend(current_block_entries[:4]) - - if timestamp_fd is not None: - for entry in sorted(current_block_entries, key=lambda e: e[2]): - timestamp_fd.write( - f'{",".join([str(elem) for elem in entry])}\n') - except (BrokenPipeError, KeyboardInterrupt): - # User hit Ctrl-C -- done processing. - pass - - # Write index file EOF. - if timestamp_fd is not None: - timestamp_fd.write(f'{EOF_FORMAT},0,{bytes_to_process},0,\n') - - # Sort the data. - index = sorted(index, key=lambda e: e[2]) - - # Print status one last time. - elapsed_sec = (datetime.now() - start_time).total_seconds() - _print_status(elapsed_sec) + # Run the streaming read loop until stopped or reaching EOF. + index, total_bytes_read = _stream_and_index( + input_path=input_path, in_fd=in_fd, options=options, + rtcm_framer=rtcm_framer, fe_framer=fe_framer, nmea_framer=nmea_framer, + skip_bytes=skip_bytes, bytes_to_process=bytes_to_process, file_size=file_size, + output_map=output_map, index_path=index_path) return index, total_bytes_read - parser = ArgumentParser(description="""\ Analyze contents of a input.raw or input.p1bin and create csv with offset and length of each NMEA, RTCM, and FE message. Print out locations of data gaps. From f8ddbfa6184333a7dbcb9582efd3ba39d9bb95dd Mon Sep 17 00:00:00 2001 From: Adam Shapiro Date: Fri, 24 Apr 2026 12:53:22 -0400 Subject: [PATCH 07/17] Fixed index file issues. --- bin/raw_analysis.py | 23 ++++++++++++----------- 1 file changed, 12 insertions(+), 11 deletions(-) diff --git a/bin/raw_analysis.py b/bin/raw_analysis.py index ef592d4..271ed72 100755 --- a/bin/raw_analysis.py +++ b/bin/raw_analysis.py @@ -64,10 +64,11 @@ def _create_framers(options, return_bytes=False): def _get_index_path(input_path, options): if len(options.format) < len(FORMAT_STRS): - postfix = '_' + '_'.join(sorted(options.format)) + '_index.csv' + postfix = '.index.' + '_'.join(sorted(options.format)) + '.csv' else: - postfix = '_index.csv' - return get_output_file_path(input_path, postfix, output_dir=options.output_dir, prefix=options.prefix) + postfix = '.index.csv' + return get_output_file_path(input_path, postfix, output_dir=options.output_dir, + prefix=None if options.prefix == '-' else options.prefix) def _open_output_files(input_path, options, text_nmea=False): @@ -80,11 +81,6 @@ def _open_output_files(input_path, options, text_nmea=False): output_map = {} if write_to_stdout: - # Only one format may be written to stdout - error if multiple are requested. - if len(options.format) > 1: - _logger.error('Only one data type may be written to stdout.') - sys.exit(1) - # Map the single requested format to stdout. NMEA is text when the framer produces strings, RTCM and FE are # always binary. for fmt in ('nmea', 'rtcm', 'fe'): @@ -265,7 +261,7 @@ def index_messages(input_path, options): bytes_to_process = options.bytes_to_process # Determine the path to the index file. If the file exists already, read it and return. If not, generate it. - index_file_full = get_output_file_path(input_path, '_index.csv', output_dir=options.output_dir, + index_file_full = get_output_file_path(input_path, '.index.csv', output_dir=options.output_dir, prefix=options.prefix) index_file = _get_index_path(input_path, options) @@ -385,8 +381,8 @@ def separate_and_index(input_path, options): # Open output files for extraction if requested. output_map = _open_output_files(input_path, options, text_nmea=True) if options.extract else None - # Open an index CSV only when writing output to disk, skip it when streaming to stdout. - index_path = None if write_to_stdout else _get_index_path(input_path, options) + # Open an index CSV only when reading from to disk, skip it when reading from stdin. + index_path = None if read_from_stdin else _get_index_path(input_path, options) rtcm_framer, fe_framer, nmea_framer = _create_framers(options, return_bytes=options.extract) @@ -525,6 +521,11 @@ def raw_analysis(options): options.format = FORMAT_STRS logger.info(f"Processing {options.format}.") + # Only one format may be written to stdout - error if multiple are requested. + if write_to_stdout and len(options.format) > 1: + _logger.error('Only one data type may be written to stdout.') + sys.exit(1) + # Use the EXTERNAL_UNFRAMED_GNSS unless the user explicitly specified different P1BinType. if options.p1bin_type is not None: # Pattern match to any of: From cb09ec85fa8abaae46f8bd475f92123134cefc8e Mon Sep 17 00:00:00 2001 From: Adam Shapiro Date: Fri, 24 Apr 2026 15:16:21 -0400 Subject: [PATCH 08/17] Added type hints. --- bin/raw_analysis.py | 44 +++++++++++++++++++++++++++++--------------- 1 file changed, 29 insertions(+), 15 deletions(-) diff --git a/bin/raw_analysis.py b/bin/raw_analysis.py index 271ed72..12b30c7 100755 --- a/bin/raw_analysis.py +++ b/bin/raw_analysis.py @@ -1,5 +1,10 @@ #!/usr/bin/env python3 +from __future__ import annotations + +from typing import BinaryIO, Dict, List, Optional, Tuple + +import argparse import io import os import sys @@ -35,7 +40,8 @@ def is_msm_id(msg_id): return msg_id == 1005 or msg_id == 1006 or (msg_id >= 1071 and msg_id <= 1227) -def get_output_file_path(input_path, postfix, output_dir=None, prefix=None): +def get_output_file_path(input_path: str, postfix: str, output_dir: Optional[str] = None, + prefix: Optional[str] = None) -> str: if output_dir is None: output_dir = os.path.dirname(input_path) if prefix is None: @@ -43,7 +49,7 @@ def get_output_file_path(input_path, postfix, output_dir=None, prefix=None): return os.path.join(output_dir, prefix + postfix) -def get_fd(input_path: str, options): +def get_fd(input_path: str, options: argparse.Namespace) -> BinaryIO: if input_path == '-': return sys.stdin.buffer elif input_path.endswith('.p1bin'): @@ -54,7 +60,8 @@ def get_fd(input_path: str, options): return open(input_path, 'rb') -def _create_framers(options, return_bytes=False): +def _create_framers(options: argparse.Namespace, return_bytes: bool = False) -> \ + Tuple[Optional[RTCMFramer], Optional[FusionEngineDecoder], Optional[NMEAFramer]]: rtcm = RTCMFramer() if 'rtcm' in options.format else None fe = FusionEngineDecoder(max_payload_len_bytes=16536, return_offset=True, return_bytes=return_bytes) if 'fe' in options.format else None @@ -62,7 +69,7 @@ def _create_framers(options, return_bytes=False): return rtcm, fe, nmea -def _get_index_path(input_path, options): +def _get_index_path(input_path: str, options: argparse.Namespace) -> str: if len(options.format) < len(FORMAT_STRS): postfix = '.index.' + '_'.join(sorted(options.format)) + '.csv' else: @@ -71,7 +78,7 @@ def _get_index_path(input_path, options): prefix=None if options.prefix == '-' else options.prefix) -def _open_output_files(input_path, options, text_nmea=False): +def _open_output_files(input_path: str, options: argparse.Namespace, text_nmea: bool = False) -> Dict[str, BinaryIO]: """! @brief Open output files for extraction. @@ -105,10 +112,17 @@ def _open_output_files(input_path, options, text_nmea=False): return output_map -def _stream_and_index(input_path, in_fd, options, - rtcm_framer, fe_framer, nmea_framer, - skip_bytes, bytes_to_process, file_size, - output_map, index_path): +def _stream_and_index(input_path: str, + in_fd: BinaryIO, + options: argparse.Namespace, + rtcm_framer: Optional[RTCMFramer], + fe_framer: Optional[FusionEngineDecoder], + nmea_framer: Optional[NMEAFramer], + skip_bytes: int, + bytes_to_process: int, + file_size: int, + output_map: Dict[str, BinaryIO], + index_path: Optional[str]): """ @brief Core read loop: parse messages, write index CSV, and optionally extract to output files. @@ -247,7 +261,7 @@ def _print_status(elapsed_sec): return sorted(index, key=lambda e: e[2]), total_bytes_read -def index_messages(input_path, options): +def index_messages(input_path: str, options: argparse.Namespace): in_fd = get_fd(input_path, options) skip_bytes = options.skip_bytes @@ -294,9 +308,9 @@ def index_messages(input_path, options): output_map=None, index_path=index_file) -def load_index(index_file): +def load_index(index_path: str): indexes = [] - with open(index_file, 'r') as index_fd: + with open(index_path, 'r') as index_fd: # Skip the header line. index_fd.readline() for line in index_fd.readlines(): @@ -309,7 +323,7 @@ def load_index(index_file): return indexes -def find_gaps(indexes): +def find_gaps(indexes: List[Tuple]): next_offset = 0 has_gaps = False for index in indexes: @@ -324,7 +338,7 @@ def find_gaps(indexes): _logger.info(f"No gaps found.") -def generate_separated_logs(input_path, indexes, options): +def generate_separated_logs(input_path: str, indexes: List[Tuple], options: argparse.Namespace): # Open output files for each requested protocol. output_map = _open_output_files(input_path, options, text_nmea=False) current_base_id = -1 @@ -354,7 +368,7 @@ def generate_separated_logs(input_path, indexes, options): output_map[index[0]].write(data) -def separate_and_index(input_path, options): +def separate_and_index(input_path: str, options: argparse.Namespace): # Open the input file (or stdin). in_fd = get_fd(input_path, options) read_from_stdin = in_fd is sys.stdin.buffer From 72d4cc32d290679880bdfa57ed6b3cbd488ac3d4 Mon Sep 17 00:00:00 2001 From: Adam Shapiro Date: Fri, 24 Apr 2026 15:51:48 -0400 Subject: [PATCH 09/17] Pulled out common base station switch check. --- bin/raw_analysis.py | 60 ++++++++++++++++++++++----------------------- 1 file changed, 30 insertions(+), 30 deletions(-) diff --git a/bin/raw_analysis.py b/bin/raw_analysis.py index 12b30c7..713a351 100755 --- a/bin/raw_analysis.py +++ b/bin/raw_analysis.py @@ -112,6 +112,29 @@ def _open_output_files(input_path: str, options: argparse.Namespace, text_nmea: return output_map +def _check_rtcm_base_station(input_path: str, + options: argparse.Namespace, + message_id: int, + raw_data: bytes, + output_map: Dict[str, BinaryIO], + current_base_id: int, + rtcm_file_idx: int) -> Tuple[int, int]: + # If splitting by base station ID, open a new output file each time the base changes. + if options.split_rtcm_base_id and is_msm_id(message_id): + # Base station ID is encoded at bit offset 36, length 12 bits. + base_id = ((raw_data[4] & 0xF) << 8) + raw_data[5] + if base_id != current_base_id: + if current_base_id != -1: + output_map['rtcm'].close() + rtcm_file_idx += 1 + output_map['rtcm'] = open(get_output_file_path( + input_path, f'_{rtcm_file_idx}.rtcm3', + output_dir=options.output_dir, prefix=options.prefix), 'wb') + _logger.info(f"Writing for base station id: {base_id}") + current_base_id = base_id + return current_base_id, rtcm_file_idx + + def _stream_and_index(input_path: str, in_fd: BinaryIO, options: argparse.Namespace, @@ -189,21 +212,9 @@ def _print_status(elapsed_sec): if extract: raw_data = msg['bytes'] - - # If splitting by base station ID, open a new output file each time the base changes. - if options.split_rtcm_base_id and is_msm_id(message_id): - # Base station ID is encoded at bit offset 36, length 12 bits. - base_id = ((raw_data[4] & 0xF) << 8) + raw_data[5] - if base_id != current_base_id: - if current_base_id != -1: - output_map['rtcm'].close() - rtcm_file_idx += 1 - output_map['rtcm'] = open(get_output_file_path( - input_path, f'_{rtcm_file_idx}.rtcm3', - output_dir=options.output_dir, prefix=options.prefix), 'wb') - _logger.info(f"Writing for base station id: {base_id}") - current_base_id = base_id - + current_base_id, rtcm_file_idx = _check_rtcm_base_station( + input_path=input_path, options=options, message_id=message_id, raw_data=raw_data, + output_map=output_map, current_base_id=current_base_id, rtcm_file_idx=rtcm_file_idx) output_map['rtcm'].write(raw_data) if fe_framer is not None: @@ -350,21 +361,10 @@ def generate_separated_logs(input_path: str, indexes: List[Tuple], options: argp if index[0] in output_map: in_fd.seek(index[2], io.SEEK_SET) data = in_fd.read(index[3]) - - # If splitting by base station, open a new file each time the base station ID changes. - if options.split_rtcm_base_id and index[0] == 'rtcm' and is_msm_id(int(index[1])): - # Base station ID is encoded at bit offset 36, length 12 bits. - base_id = ((data[4] & 0xF) << 8) + data[5] - if base_id != current_base_id: - if current_base_id != -1: - output_map['rtcm'].close() - rtcm_file_idx += 1 - output_map['rtcm'] = open(get_output_file_path( - input_path, f'_{rtcm_file_idx}.rtcm3', - output_dir=options.output_dir, prefix=options.prefix), 'wb') - _logger.info(f"Writing for base station id: {base_id}") - current_base_id = base_id - + if index[0] == 'rtcm': + current_base_id, rtcm_file_idx = _check_rtcm_base_station( + input_path=input_path, options=options, message_id=int(index[1]), raw_data=data, + output_map=output_map, current_base_id=current_base_id, rtcm_file_idx=rtcm_file_idx) output_map[index[0]].write(data) From a18870808f97b876fcdff1a374c0520e54bc6c09 Mon Sep 17 00:00:00 2001 From: Adam Shapiro Date: Fri, 24 Apr 2026 17:34:40 -0400 Subject: [PATCH 10/17] Corrected MSM vs RTCM in general. --- bin/raw_analysis.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/bin/raw_analysis.py b/bin/raw_analysis.py index 713a351..533947e 100755 --- a/bin/raw_analysis.py +++ b/bin/raw_analysis.py @@ -36,7 +36,7 @@ EOF_FORMAT = 'eof' -def is_msm_id(msg_id): +def is_rtcm_with_station_id(msg_id): return msg_id == 1005 or msg_id == 1006 or (msg_id >= 1071 and msg_id <= 1227) @@ -120,7 +120,7 @@ def _check_rtcm_base_station(input_path: str, current_base_id: int, rtcm_file_idx: int) -> Tuple[int, int]: # If splitting by base station ID, open a new output file each time the base changes. - if options.split_rtcm_base_id and is_msm_id(message_id): + if options.split_rtcm_base_id and is_rtcm_with_station_id(message_id): # Base station ID is encoded at bit offset 36, length 12 bits. base_id = ((raw_data[4] & 0xF) << 8) + raw_data[5] if base_id != current_base_id: From 3896a9a06afb013cc002766aa062949a810ae810 Mon Sep 17 00:00:00 2001 From: Adam Shapiro Date: Fri, 24 Apr 2026 17:34:53 -0400 Subject: [PATCH 11/17] Don't include invalid IDs in MSM grouping. --- bin/raw_analysis.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/bin/raw_analysis.py b/bin/raw_analysis.py index 533947e..0c6b8fd 100755 --- a/bin/raw_analysis.py +++ b/bin/raw_analysis.py @@ -37,7 +37,8 @@ def is_rtcm_with_station_id(msg_id): - return msg_id == 1005 or msg_id == 1006 or (msg_id >= 1071 and msg_id <= 1227) + return (msg_id == 1005 or msg_id == 1006 or + (msg_id >= 1071 and msg_id <= 1227 and (msg_id % 10) not in (0, 8, 9))) def get_output_file_path(input_path: str, postfix: str, output_dir: Optional[str] = None, From 57ebcb5972afb9ff74b5ed92c205b20b15b1cf2f Mon Sep 17 00:00:00 2001 From: Adam Shapiro Date: Fri, 24 Apr 2026 17:35:05 -0400 Subject: [PATCH 12/17] Added raw_analysis.py unit tests. --- tests/test_raw_analysis.py | 635 +++++++++++++++++++++++++++++++++++++ 1 file changed, 635 insertions(+) create mode 100644 tests/test_raw_analysis.py diff --git a/tests/test_raw_analysis.py b/tests/test_raw_analysis.py new file mode 100644 index 0000000..cdc5063 --- /dev/null +++ b/tests/test_raw_analysis.py @@ -0,0 +1,635 @@ +"""Unit and integration tests for bin/raw_analysis.py.""" + +import argparse +import io +import os +import sys +from unittest.mock import patch + +import pytest + +# Add bin/ to the path so raw_analysis can be imported as a plain module. +sys.path.insert(0, os.path.normpath(os.path.join(os.path.dirname(__file__), '..', 'bin'))) +from raw_analysis import ( # noqa: E402 + EOF_FORMAT, + FORMAT_STRS, + _create_framers, + _get_index_path, + _stream_and_index, + find_gaps, + generate_separated_logs, + get_output_file_path, + index_messages, + is_rtcm_with_station_id, + load_index, + raw_analysis, + separate_and_index, +) + +# Two real NMEA sentences used across all integration tests. +NMEA_GGA = "$GPGGA,000000.000,3746.37327400,N,12224.26599800,W,2,13,2.1,3.260,M,34.210,M,11.1,0234*5B\r\n" +NMEA_RMC = "$GPRMC,000000.000,A,3746.37327400,N,12224.26599800,W,0.00,0.00,010101,,,D*76\r\n" + +# Valid RTCM frames used in multi-protocol tests. Payloads are zero-padded to the standard field widths; only the 12-bit +# message number and CRC matter here. +RTCM_MSG_1005 = b'\xd3\x00\x13>\xd0\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\xf2K\xf4' +RTCM_MSG_1006 = b'\xd3\x00\x15>\xe0\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x02\xd2\x1f' + + +class FakeStdin: + """Wraps a BytesIO to stand in for sys.stdin when sys.stdin.buffer is accessed.""" + def __init__(self, data: bytes): + self.buffer = io.BytesIO(data) + + +class FakeTextStdout: + """Stands in for sys.stdout, recording text writes and exposing .buffer for binary writes.""" + def __init__(self): + self._text = io.StringIO() + self.buffer = io.BytesIO() + + def write(self, s: str): + self._text.write(s) + + def flush(self): + pass + + def getvalue(self) -> str: + return self._text.getvalue() + + +def make_options(**kwargs): + """Return an options Namespace with defaults suitable for unit testing.""" + ns = argparse.Namespace( + format=FORMAT_STRS.copy(), + output_dir=None, + prefix=None, + skip_bytes=0, + bytes_to_process=None, + ignore_index=False, + extract=False, + split_rtcm_base_id=False, + ) + for key, value in kwargs.items(): + setattr(ns, key, value) + return ns + + +@pytest.fixture +def nmea_file(tmp_path): + """Write two NMEA sentences to a temp file and return its path as a string.""" + path = tmp_path / 'input.bin' + path.write_bytes((NMEA_GGA + NMEA_RMC).encode()) + return path + + +@pytest.fixture +def multi_protocol_file(tmp_path): + """Write interleaved NMEA and RTCM messages to a temp file and return its path.""" + data = NMEA_GGA.encode() + RTCM_MSG_1005 + NMEA_RMC.encode() + RTCM_MSG_1006 + path = tmp_path / 'input.bin' + path.write_bytes(data) + return path + + +# --------------------------------------------------------------------------- +# is_rtcm_with_station_id +# --------------------------------------------------------------------------- + +class TestIsRTCMID: + def test_1005_1006(self): + assert is_rtcm_with_station_id(1005) + assert is_rtcm_with_station_id(1006) + + def test_msm(self): + for i in range(1070, 1230, 10): + for j in range(1, 8): + assert is_rtcm_with_station_id(i + j) + + def test_non_matching_ids(self): + assert not is_rtcm_with_station_id(1007) + assert not is_rtcm_with_station_id(1070) + assert not is_rtcm_with_station_id(1078) + assert not is_rtcm_with_station_id(1300) + assert not is_rtcm_with_station_id(999) + + +# --------------------------------------------------------------------------- +# get_output_file_path +# --------------------------------------------------------------------------- + +class TestGetOutputFilePath: + def test_derives_prefix_and_dir_from_input_path(self): + assert get_output_file_path('/data/input.raw', '.nmea') == '/data/input.nmea' + + def test_explicit_output_dir_overrides_input_dir(self): + assert get_output_file_path('/data/input.raw', '.nmea', output_dir='/out') == '/out/input.nmea' + + def test_explicit_prefix_overrides_input_stem(self): + assert get_output_file_path('/data/input.raw', '.nmea', prefix='mylog') == '/data/mylog.nmea' + + def test_explicit_prefix_and_output_dir(self): + result = get_output_file_path('/data/input.raw', '.nmea', output_dir='/out', prefix='mylog') + assert result == '/out/mylog.nmea' + + +# --------------------------------------------------------------------------- +# _get_index_path +# --------------------------------------------------------------------------- + +class TestGetIndexPath: + def test_all_formats_produces_plain_index_name(self): + options = make_options(format=FORMAT_STRS.copy(), prefix=None) + assert _get_index_path('/data/input.raw', options) == '/data/input.index.csv' + + def test_single_format_appended_to_name(self): + options = make_options(format={'nmea'}, prefix=None) + assert _get_index_path('/data/input.raw', options) == '/data/input.index.nmea.csv' + + def test_two_formats_appear_in_sorted_order(self): + # Set iteration order is non-deterministic; the filename must be stable. + options = make_options(format={'rtcm', 'fe'}, prefix=None) + assert _get_index_path('/data/input.raw', options) == '/data/input.index.fe_rtcm.csv' + + def test_respects_output_dir(self, tmp_path): + options = make_options(format=FORMAT_STRS.copy(), output_dir=str(tmp_path), prefix=None) + assert _get_index_path('/data/input.raw', options) == str(tmp_path / 'input.index.csv') + + def test_stdout_prefix_derives_name_from_input_file(self): + # When prefix='-' (stdout mode), the index is named after the input file, not '-'. + options = make_options(format=FORMAT_STRS.copy(), prefix='-') + assert _get_index_path('/data/input.raw', options) == '/data/input.index.csv' + + +# --------------------------------------------------------------------------- +# load_index +# --------------------------------------------------------------------------- + +class TestLoadIndex: + def test_parses_protocol_id_offset_and_size(self, tmp_path): + csv = tmp_path / 'test.index.csv' + csv.write_text( + 'Protocol, ID, Offset (Bytes), Length (Bytes), P1 Time\n' + 'nmea,GPGGA,0,92,\n' + 'nmea,GPRMC,92,77,\n' + f'{EOF_FORMAT},0,169,0,\n' + ) + index = load_index(str(csv)) + assert index[0] == ('nmea', 'GPGGA', 0, 92) + assert index[1] == ('nmea', 'GPRMC', 92, 77) + + def test_sorts_out_of_order_entries_by_offset(self, tmp_path): + # Framers can emit messages out of offset order during data dropouts. + csv = tmp_path / 'test.index.csv' + csv.write_text( + 'Protocol, ID, Offset (Bytes), Length (Bytes), P1 Time\n' + 'nmea,GPRMC,92,77,\n' + 'nmea,GPGGA,0,92,\n' + f'{EOF_FORMAT},0,169,0,\n' + ) + index = load_index(str(csv)) + assert index[0][2] == 0 # GPGGA should sort first. + assert index[1][2] == 92 # GPRMC should sort second. + + def test_eof_sentinel_is_included(self, tmp_path): + csv = tmp_path / 'test.index.csv' + csv.write_text( + 'Protocol, ID, Offset (Bytes), Length (Bytes), P1 Time\n' + f'{EOF_FORMAT},0,169,0,\n' + ) + index = load_index(str(csv)) + assert index[-1][0] == EOF_FORMAT + assert index[-1][2] == 169 + + +# --------------------------------------------------------------------------- +# find_gaps +# --------------------------------------------------------------------------- + +class TestFindGaps: + def test_contiguous_entries_do_not_raise(self): + index = [ + ('nmea', 'GPGGA', 0, 10), + ('nmea', 'GPRMC', 10, 20), + (EOF_FORMAT, '0', 30, 0), + ] + find_gaps(index) # Must not raise. + + def test_gap_between_entries_does_not_raise(self): + # Verify that gap detection completes without exception. + index = [ + ('nmea', 'GPGGA', 0, 10), + ('nmea', 'GPRMC', 15, 20), # 5-byte gap before this entry. + ] + find_gaps(index) + + def test_empty_index(self): + find_gaps([]) + + +# --------------------------------------------------------------------------- +# _create_framers +# --------------------------------------------------------------------------- + +class TestCreateFramers: + def test_all_formats_produces_all_three_framers(self): + options = make_options(format=FORMAT_STRS.copy()) + rtcm, fe, nmea = _create_framers(options) + assert rtcm is not None + assert fe is not None + assert nmea is not None + + def test_nmea_only_returns_none_for_rtcm_and_fe(self): + options = make_options(format={'nmea'}) + rtcm, fe, nmea = _create_framers(options) + assert rtcm is None + assert fe is None + assert nmea is not None + + def test_rtcm_only_returns_none_for_fe_and_nmea(self): + options = make_options(format={'rtcm'}) + rtcm, fe, nmea = _create_framers(options) + assert rtcm is not None + assert fe is None + assert nmea is None + + +# --------------------------------------------------------------------------- +# _stream_and_index (integration tests using real NMEA data) +# --------------------------------------------------------------------------- + +class TestStreamAndIndex: + def _run(self, nmea_file, tmp_path, **kwargs): + """Helper that runs _stream_and_index with NMEA-only framers and returns (index, total_bytes).""" + content = nmea_file.read_bytes() + options = make_options(format={'nmea'}, output_dir=str(tmp_path), **kwargs) + rtcm_framer, fe_framer, nmea_framer = _create_framers(options) + with open(nmea_file, 'rb') as in_fd: + return _stream_and_index( + input_path=str(nmea_file), in_fd=in_fd, options=options, + rtcm_framer=rtcm_framer, fe_framer=fe_framer, nmea_framer=nmea_framer, + skip_bytes=0, bytes_to_process=len(content), file_size=len(content), + output_map=None, index_path=None, + ), content + + def test_returns_one_entry_per_message(self, nmea_file, tmp_path): + (index, total_bytes), content = self._run(nmea_file, tmp_path) + assert total_bytes == len(content) + assert len(index) == 2 + + def test_entry_contains_correct_protocol_id_offset_and_size(self, nmea_file, tmp_path): + (index, _), _ = self._run(nmea_file, tmp_path) + assert index[0] == ('nmea', 'GPGGA', 0, len(NMEA_GGA)) + assert index[1] == ('nmea', 'GPRMC', len(NMEA_GGA), len(NMEA_RMC)) + + def test_writes_valid_index_csv(self, nmea_file, tmp_path): + content = nmea_file.read_bytes() + index_path = str(tmp_path / 'out.index.csv') + options = make_options(format={'nmea'}, output_dir=str(tmp_path)) + rtcm_framer, fe_framer, nmea_framer = _create_framers(options) + + with open(nmea_file, 'rb') as in_fd: + _stream_and_index( + input_path=str(nmea_file), in_fd=in_fd, options=options, + rtcm_framer=rtcm_framer, fe_framer=fe_framer, nmea_framer=nmea_framer, + skip_bytes=0, bytes_to_process=len(content), file_size=len(content), + output_map=None, index_path=index_path, + ) + + loaded = load_index(index_path) + data_entries = [e for e in loaded if e[0] != EOF_FORMAT] + assert len(data_entries) == 2 + assert data_entries[0] == ('nmea', 'GPGGA', 0, len(NMEA_GGA)) + assert data_entries[1] == ('nmea', 'GPRMC', len(NMEA_GGA), len(NMEA_RMC)) + + def test_csv_includes_eof_sentinel(self, nmea_file, tmp_path): + content = nmea_file.read_bytes() + index_path = str(tmp_path / 'out.index.csv') + options = make_options(format={'nmea'}, output_dir=str(tmp_path)) + rtcm_framer, fe_framer, nmea_framer = _create_framers(options) + + with open(nmea_file, 'rb') as in_fd: + _stream_and_index( + input_path=str(nmea_file), in_fd=in_fd, options=options, + rtcm_framer=rtcm_framer, fe_framer=fe_framer, nmea_framer=nmea_framer, + skip_bytes=0, bytes_to_process=len(content), file_size=len(content), + output_map=None, index_path=index_path, + ) + + loaded = load_index(index_path) + assert loaded[-1][0] == EOF_FORMAT + assert loaded[-1][2] == len(content) + + def test_extracts_nmea_messages_to_output_file(self, nmea_file, tmp_path): + content = nmea_file.read_bytes() + out_file = tmp_path / 'output.nmea' + options = make_options(format={'nmea'}, output_dir=str(tmp_path), + extract=True, split_rtcm_base_id=False) + rtcm_framer, fe_framer, nmea_framer = _create_framers(options, return_bytes=True) + output_map = {'nmea': open(str(out_file), 'wt')} + + with open(nmea_file, 'rb') as in_fd: + _stream_and_index( + input_path=str(nmea_file), in_fd=in_fd, options=options, + rtcm_framer=rtcm_framer, fe_framer=fe_framer, nmea_framer=nmea_framer, + skip_bytes=0, bytes_to_process=len(content), file_size=len(content), + output_map=output_map, index_path=None, + ) + output_map['nmea'].close() + + # Use read_bytes().decode() rather than read_text() to avoid universal-newline + # translation stripping the \r from NMEA's \r\n terminators. + assert out_file.read_bytes().decode() == NMEA_GGA + NMEA_RMC + + def test_skip_bytes_shifts_absolute_offset_in_index(self, nmea_file, tmp_path): + content = nmea_file.read_bytes() + skip = len(NMEA_GGA) + options = make_options(format={'nmea'}, output_dir=str(tmp_path)) + rtcm_framer, fe_framer, nmea_framer = _create_framers(options) + + with open(nmea_file, 'rb') as in_fd: + in_fd.seek(skip) + index, total_bytes = _stream_and_index( + input_path=str(nmea_file), in_fd=in_fd, options=options, + rtcm_framer=rtcm_framer, fe_framer=fe_framer, nmea_framer=nmea_framer, + skip_bytes=skip, bytes_to_process=len(NMEA_RMC), file_size=len(content), + output_map=None, index_path=None, + ) + + # Only the second message should be indexed, but its offset is absolute within the file. + assert len(index) == 1 + assert index[0] == ('nmea', 'GPRMC', len(NMEA_GGA), len(NMEA_RMC)) + + +# --------------------------------------------------------------------------- +# index_messages (integration tests) +# --------------------------------------------------------------------------- + +class TestIndexMessages: + def test_generates_index_file_and_returns_entries(self, nmea_file, tmp_path): + options = make_options(format={'nmea'}, output_dir=str(tmp_path), ignore_index=True) + index, _ = index_messages(str(nmea_file), options) + + data_entries = [e for e in index if e[0] != EOF_FORMAT] + assert len(data_entries) == 2 + assert (tmp_path / 'input.index.nmea.csv').exists() + + def test_reuses_valid_existing_index_without_reparsing(self, nmea_file, tmp_path): + options = make_options(format={'nmea'}, output_dir=str(tmp_path)) + + # First call generates the index file. + index_messages(str(nmea_file), options) + index_file = tmp_path / 'input.index.nmea.csv' + mtime_before = index_file.stat().st_mtime + + # Second call should return without modifying the index file. + index_messages(str(nmea_file), options) + assert index_file.stat().st_mtime == mtime_before + + def test_ignores_index_whose_byte_count_does_not_match(self, nmea_file, tmp_path): + options = make_options(format={'nmea'}, output_dir=str(tmp_path)) + index_messages(str(nmea_file), options) + + # Shorten the input so that bytes_to_process changes on the next call. + nmea_file.write_bytes(NMEA_GGA.encode()) + + index, _ = index_messages(str(nmea_file), options) + data_entries = [e for e in index if e[0] != EOF_FORMAT] + assert len(data_entries) == 1 + + def test_ignores_index_missing_eof_sentinel(self, nmea_file, tmp_path): + options = make_options(format={'nmea'}, output_dir=str(tmp_path)) + + # Write an index file that was truncated before the EOF sentinel was written. + index_file = tmp_path / 'input.index.nmea.csv' + index_file.write_text( + 'Protocol, ID, Offset (Bytes), Length (Bytes), P1 Time\n' + 'nmea,GPGGA,0,92,\n' + ) + + # The incomplete index should be discarded and a fresh one generated. + index, _ = index_messages(str(nmea_file), options) + data_entries = [e for e in index if e[0] != EOF_FORMAT] + assert len(data_entries) == 2 + + def test_ignore_index_flag_forces_regeneration(self, nmea_file, tmp_path): + options = make_options(format={'nmea'}, output_dir=str(tmp_path)) + + # First call creates the index. + index_messages(str(nmea_file), options) + index_file = tmp_path / 'input.index.nmea.csv' + mtime_before = index_file.stat().st_mtime + + # With ignore_index=True the file should be regenerated. + options.ignore_index = True + index_messages(str(nmea_file), options) + assert index_file.stat().st_mtime >= mtime_before + + +# --------------------------------------------------------------------------- +# generate_separated_logs (integration tests) +# --------------------------------------------------------------------------- + +class TestGenerateSeparatedLogs: + def test_writes_all_messages_to_nmea_file(self, nmea_file, tmp_path): + content = nmea_file.read_bytes() + index = [ + ('nmea', 'GPGGA', 0, len(NMEA_GGA)), + ('nmea', 'GPRMC', len(NMEA_GGA), len(NMEA_RMC)), + (EOF_FORMAT, '0', len(content), 0), + ] + options = make_options(format={'nmea'}, output_dir=str(tmp_path), + prefix=nmea_file.stem, split_rtcm_base_id=False) + + generate_separated_logs(str(nmea_file), index, options) + + out_file = tmp_path / 'input.nmea' + assert out_file.exists() + assert out_file.read_bytes() == content + + def test_skips_protocols_absent_from_format(self, nmea_file, tmp_path): + # An 'fe' index entry should be silently ignored when format contains only 'nmea'. + index = [ + ('fe', '10001', 0, len(NMEA_GGA)), + ('nmea', 'GPRMC', len(NMEA_GGA), len(NMEA_RMC)), + ] + options = make_options(format={'nmea'}, output_dir=str(tmp_path), + prefix=nmea_file.stem, split_rtcm_base_id=False) + + generate_separated_logs(str(nmea_file), index, options) + + out_file = tmp_path / 'input.nmea' + assert out_file.read_bytes() == NMEA_RMC.encode() + + def test_eof_sentinel_is_skipped(self, nmea_file, tmp_path): + content = nmea_file.read_bytes() + index = [ + ('nmea', 'GPGGA', 0, len(NMEA_GGA)), + (EOF_FORMAT, '0', len(content), 0), + ] + options = make_options(format={'nmea'}, output_dir=str(tmp_path), + prefix=nmea_file.stem, split_rtcm_base_id=False) + + generate_separated_logs(str(nmea_file), index, options) + + out_file = tmp_path / 'input.nmea' + assert out_file.read_bytes() == NMEA_GGA.encode() + + +# --------------------------------------------------------------------------- +# separate_and_index (integration tests) +# --------------------------------------------------------------------------- + +class TestSeparateAndIndex: + def test_indexes_all_messages_in_file(self, nmea_file, tmp_path): + options = make_options(format={'nmea'}, output_dir=str(tmp_path), prefix=nmea_file.stem) + index, total_bytes = separate_and_index(str(nmea_file), options) + + data_entries = [e for e in index if e[0] != EOF_FORMAT] + assert len(data_entries) == 2 + assert total_bytes == len(nmea_file.read_bytes()) + + def test_extracts_nmea_to_output_file(self, nmea_file, tmp_path): + options = make_options(format={'nmea'}, output_dir=str(tmp_path), + prefix=nmea_file.stem, extract=True, split_rtcm_base_id=False) + separate_and_index(str(nmea_file), options) + + out_file = tmp_path / (nmea_file.stem + '.nmea') + assert out_file.exists() + # Use read_bytes().decode() rather than read_text() to avoid universal-newline + # translation stripping the \r from NMEA's \r\n terminators. + assert out_file.read_bytes().decode() == NMEA_GGA + NMEA_RMC + + def test_indexes_interleaved_nmea_and_rtcm(self, multi_protocol_file, tmp_path): + options = make_options(format={'nmea', 'rtcm'}, output_dir=str(tmp_path), + prefix=multi_protocol_file.stem) + index, _ = separate_and_index(str(multi_protocol_file), options) + + data_entries = [e for e in index if e[0] != EOF_FORMAT] + nmea_entries = [e for e in data_entries if e[0] == 'nmea'] + rtcm_entries = [e for e in data_entries if e[0] == 'rtcm'] + assert len(nmea_entries) == 2 + assert len(rtcm_entries) == 2 + + def test_extracts_interleaved_protocols_to_separate_files(self, multi_protocol_file, tmp_path): + options = make_options(format={'nmea', 'rtcm'}, output_dir=str(tmp_path), + prefix=multi_protocol_file.stem, extract=True, split_rtcm_base_id=False) + separate_and_index(str(multi_protocol_file), options) + + nmea_out = tmp_path / 'input.nmea' + rtcm_out = tmp_path / 'input.rtcm3' + assert nmea_out.exists() + assert rtcm_out.exists() + assert nmea_out.read_bytes().decode() == NMEA_GGA + NMEA_RMC + assert rtcm_out.read_bytes() == RTCM_MSG_1005 + RTCM_MSG_1006 + + +def make_raw_analysis_options(**kwargs): + """Return an options Namespace with defaults suitable for passing to raw_analysis().""" + ns = argparse.Namespace( + log='-', + format=None, + verbose=0, + bytes_to_process=None, + skip_bytes=0, + ignore_index=False, + extract=False, + split_rtcm_base_id=False, + check_gaps=False, + output_dir=None, + prefix=None, + log_base_dir='/tmp', + p1bin_type=None, + ) + for key, value in kwargs.items(): + setattr(ns, key, value) + return ns + + +# --------------------------------------------------------------------------- +# separate_and_index (stdin/stdout tests) +# --------------------------------------------------------------------------- + +class TestSeparateAndIndexStdin: + def test_indexes_nmea_from_stdin(self, tmp_path): + data = (NMEA_GGA + NMEA_RMC).encode() + options = make_options(format={'nmea'}, output_dir=str(tmp_path)) + + with patch('sys.stdin', FakeStdin(data)): + index, total_bytes = separate_and_index('-', options) + + data_entries = [e for e in index if e[0] != EOF_FORMAT] + assert len(data_entries) == 2 + assert total_bytes == len(data) + + def test_extracts_nmea_from_stdin_to_file(self, tmp_path): + data = (NMEA_GGA + NMEA_RMC).encode() + options = make_options(format={'nmea'}, output_dir=str(tmp_path), prefix='output', + extract=True, split_rtcm_base_id=False) + + with patch('sys.stdin', FakeStdin(data)): + separate_and_index('-', options) + + out_file = tmp_path / 'output.nmea' + assert out_file.exists() + assert out_file.read_bytes().decode() == NMEA_GGA + NMEA_RMC + + def test_extracts_nmea_from_stdin_to_stdout(self): + data = (NMEA_GGA + NMEA_RMC).encode() + options = make_options(format={'nmea'}, prefix='-', extract=True, split_rtcm_base_id=False) + fake_stdout = FakeTextStdout() + + with patch('sys.stdin', FakeStdin(data)), patch('sys.stdout', fake_stdout): + separate_and_index('-', options) + + assert fake_stdout.getvalue() == NMEA_GGA + NMEA_RMC + + +# --------------------------------------------------------------------------- +# raw_analysis (application-level tests) +# --------------------------------------------------------------------------- + +class TestRawAnalysis: + def test_stdin_indexes_nmea_without_error(self, tmp_path): + data = (NMEA_GGA + NMEA_RMC).encode() + # Use an explicit prefix so output goes to a file rather than stdout. + options = make_raw_analysis_options( + log='-', format=['nmea'], prefix='out', output_dir=str(tmp_path), check_gaps=False) + + with patch('sys.stdin', FakeStdin(data)): + raw_analysis(options) # Must not raise or call sys.exit. + + def test_stdin_multiple_formats_to_stdout_exits(self): + # Multiple formats cannot be multiplexed onto a single stdout stream; raw_analysis must + # exit before attempting to read stdin. + options = make_raw_analysis_options(log='-', format=['nmea', 'rtcm']) + + with pytest.raises(SystemExit): + raw_analysis(options) + + def test_invalid_format_exits(self, tmp_path): + options = make_raw_analysis_options( + log='-', format=['not_a_format'], prefix='out', output_dir=str(tmp_path)) + + with patch('sys.stdin', FakeStdin(b'')): + with pytest.raises(SystemExit): + raw_analysis(options) + + def test_file_input_creates_index(self, nmea_file, tmp_path): + options = make_raw_analysis_options( + log='some_log', format=['nmea'], output_dir=str(tmp_path), check_gaps=False) + + with patch('raw_analysis.find_log_file', return_value=(str(nmea_file), str(tmp_path), None)): + raw_analysis(options) + + assert (tmp_path / 'input.index.nmea.csv').exists() + + def test_file_input_with_extract_creates_output_files(self, nmea_file, tmp_path): + options = make_raw_analysis_options( + log='some_log', format=['nmea'], output_dir=str(tmp_path), + extract=True, check_gaps=False, split_rtcm_base_id=False) + + with patch('raw_analysis.find_log_file', return_value=(str(nmea_file), str(tmp_path), None)): + raw_analysis(options) + + assert (tmp_path / 'input.nmea').exists() From 843dbc2135c61675a14d08cfe0779e5223fc0e6c Mon Sep 17 00:00:00 2001 From: Adam Shapiro Date: Fri, 24 Apr 2026 17:39:22 -0400 Subject: [PATCH 13/17] Check for elapsed time divide-by-zero. --- bin/raw_analysis.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/bin/raw_analysis.py b/bin/raw_analysis.py index 0c6b8fd..17141f5 100755 --- a/bin/raw_analysis.py +++ b/bin/raw_analysis.py @@ -5,6 +5,7 @@ from typing import BinaryIO, Dict, List, Optional, Tuple import argparse +import math import io import os import sys @@ -178,7 +179,8 @@ def _print_status(elapsed_sec): _logger.info('Processed %d/%d bytes (%.1f%%). [elapsed=%.1f sec, rate=%.1f MB/s]' % (total_bytes_read, bytes_to_process, 100.0 * float(total_bytes_read) / bytes_to_process, - elapsed_sec, total_bytes_read / elapsed_sec / 1e6)) + elapsed_sec, + math.nan if elapsed_sec == 0.0 else total_bytes_read / elapsed_sec / 1e6)) # Read all incoming data until EOF or Ctrl-C. try: From 5e6a7b467d7b314a0b37b33e3da5bdf54dbd5913 Mon Sep 17 00:00:00 2001 From: Adam Shapiro Date: Fri, 24 Apr 2026 17:39:34 -0400 Subject: [PATCH 14/17] Open NMEA file as text by default. --- bin/raw_analysis.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/bin/raw_analysis.py b/bin/raw_analysis.py index 17141f5..90070ec 100755 --- a/bin/raw_analysis.py +++ b/bin/raw_analysis.py @@ -80,7 +80,7 @@ def _get_index_path(input_path: str, options: argparse.Namespace) -> str: prefix=None if options.prefix == '-' else options.prefix) -def _open_output_files(input_path: str, options: argparse.Namespace, text_nmea: bool = False) -> Dict[str, BinaryIO]: +def _open_output_files(input_path: str, options: argparse.Namespace, text_nmea: bool = True) -> Dict[str, BinaryIO]: """! @brief Open output files for extraction. @@ -354,6 +354,9 @@ def find_gaps(indexes: List[Tuple]): def generate_separated_logs(input_path: str, indexes: List[Tuple], options: argparse.Namespace): # Open output files for each requested protocol. + # + # Note: We're reading the input files as binary here, even the NMEA file, so we'll open the NMEA output stream as + # binary too so we can call write() below without having to convert to ASCII. output_map = _open_output_files(input_path, options, text_nmea=False) current_base_id = -1 rtcm_file_idx = 0 @@ -396,7 +399,7 @@ def separate_and_index(input_path: str, options: argparse.Namespace): bytes_to_process = options.bytes_to_process # Open output files for extraction if requested. - output_map = _open_output_files(input_path, options, text_nmea=True) if options.extract else None + output_map = _open_output_files(input_path, options) if options.extract else None # Open an index CSV only when reading from to disk, skip it when reading from stdin. index_path = None if read_from_stdin else _get_index_path(input_path, options) From 6d78c719360d4c743ae43f8777ed3c22e1f6b8bb Mon Sep 17 00:00:00 2001 From: Adam Shapiro Date: Fri, 24 Apr 2026 18:22:12 -0400 Subject: [PATCH 15/17] Added Python unit test CI workflow. --- .github/workflows/python_tests.yml | 45 ++++++++++++++++++++++++++++++ requirements-dev.txt | 1 + 2 files changed, 46 insertions(+) create mode 100644 .github/workflows/python_tests.yml create mode 100644 requirements-dev.txt diff --git a/.github/workflows/python_tests.yml b/.github/workflows/python_tests.yml new file mode 100644 index 0000000..779f640 --- /dev/null +++ b/.github/workflows/python_tests.yml @@ -0,0 +1,45 @@ +name: Python Tests + +on: + pull_request: + paths: + - '.github/workflows/python_tests.yml' + - 'tools/logunitas/**' + push: + branches: + - 'st-develop' + paths: + - '.github/workflows/python_tests.yml' + - 'tools/logunitas/**' + +jobs: + test: + name: Python ${{ matrix.python-version }} + runs-on: ubuntu-latest + + strategy: + fail-fast: false + matrix: + python-version: ['3.8', '3.9', '3.10', '3.11', '3.12', '3.13'] + + steps: + - uses: actions/checkout@v4 + + - name: Setup Python + uses: actions/setup-python@v4 + with: + python-version: ${{ matrix.python-version }} + cache: 'pip' + cache-dependency-path: | + requirements.txt + requirements-dev.txt + + - name: Configure git + run: | + git config --global url."https://${{ secrets.GH_ACCESS_TOKEN_NAUTILUS }}@github.com/".insteadOf "ssh://git@github.com/" + + - name: Install dependencies + run: pip install -r requirements.txt -r requirements-dev.txt + + - name: Run tests + run: python -m pytest tests/ -v diff --git a/requirements-dev.txt b/requirements-dev.txt new file mode 100644 index 0000000..e079f8a --- /dev/null +++ b/requirements-dev.txt @@ -0,0 +1 @@ +pytest From 4d3c95e2dee3c647fda44ffde4685b147640bb04 Mon Sep 17 00:00:00 2001 From: Adam Shapiro Date: Fri, 24 Apr 2026 18:22:19 -0400 Subject: [PATCH 16/17] Added .claude to .gitignore. --- .gitignore | 3 +++ 1 file changed, 3 insertions(+) diff --git a/.gitignore b/.gitignore index cf39fb4..03cccd3 100644 --- a/.gitignore +++ b/.gitignore @@ -112,6 +112,9 @@ ENV/ env.bak/ venv.bak/ +# AI +.claude/ + # Spyder project settings .spyderproject .spyproject From ffade1a684c54e2d0c2114c7d9ac9dc7f69a55a7 Mon Sep 17 00:00:00 2001 From: Adam Shapiro Date: Fri, 24 Apr 2026 18:39:28 -0400 Subject: [PATCH 17/17] Updated to FE client 1.27.0 for Python <3.10 support. --- requirements.txt | 2 +- setup.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/requirements.txt b/requirements.txt index 59425fc..5520c0b 100644 --- a/requirements.txt +++ b/requirements.txt @@ -2,7 +2,7 @@ argparse-formatter>=1.4 colorama>=0.4.4 construct~=2.10.67 deepdiff>=8.0.1 -fusion-engine-client==1.24.4 +fusion-engine-client==1.27.0 pynmea2~=1.18.0 pyserial~=3.5 pyvirtualserialports~=2.0.0 diff --git a/setup.py b/setup.py index 34743f5..bbad43d 100644 --- a/setup.py +++ b/setup.py @@ -10,7 +10,7 @@ "colorama>=0.4.4", "construct~=2.10.67", "deepdiff>=8.0.1", - "fusion-engine-client==1.24.3", + "fusion-engine-client==1.27.0", "psutil>=5.9.4", "pynmea2~=1.18.0", "pyserial~=3.5",