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/.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 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) diff --git a/bin/raw_analysis.py b/bin/raw_analysis.py index d424301..90070ec 100755 --- a/bin/raw_analysis.py +++ b/bin/raw_analysis.py @@ -1,5 +1,11 @@ #!/usr/bin/env python3 +from __future__ import annotations + +from typing import BinaryIO, Dict, List, Optional, Tuple + +import argparse +import math import io import os import sys @@ -31,11 +37,13 @@ EOF_FORMAT = 'eof' -def is_msm_id(msg_id): - return msg_id == 1005 or msg_id == 1006 or (msg_id >= 1071 and msg_id <= 1227) +def is_rtcm_with_station_id(msg_id): + 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, 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,8 +51,10 @@ 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): - if input_path.endswith('.p1bin'): +def get_fd(input_path: str, options: argparse.Namespace) -> BinaryIO: + 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) @@ -52,131 +62,282 @@ 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=4096, 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 - - 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 +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 + nmea = NMEAFramer(return_offset=True) if 'nmea' in options.format else None + return rtcm, fe, nmea - # 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 _get_index_path(input_path: str, options: argparse.Namespace) -> str: 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) + postfix = '.index.' + '_'.join(sorted(options.format)) + '.csv' else: - index_file = index_file_full + postfix = '.index.csv' + return get_output_file_path(input_path, postfix, output_dir=options.output_dir, + prefix=None if options.prefix == '-' else options.prefix) - 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) - # 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 +def _open_output_files(input_path: str, options: argparse.Namespace, text_nmea: bool = True) -> Dict[str, BinaryIO]: + """! + @brief Open output files for extraction. + + @return An `output_map` dict keyed by protocol name. + """ + write_to_stdout = options.prefix == '-' + output_map = {} + + if write_to_stdout: + # 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: + # 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') + + 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_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: + 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, + 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. + + 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 - _logger.info(f"Indexing raw input.") + # 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, + math.nan if elapsed_sec == 0.0 else 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: + 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'] + 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: - 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) + + # 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() - for entry in sorted(entries, key=lambda e: e[2]): - timestamp_fd.write( - f'{",".join([str(elem) for elem in entry])}\n') + # Print final status after the loop exits. + elapsed_sec = (datetime.now() - start_time).total_seconds() + if elapsed_sec > 0: + _print_status(elapsed_sec) - timestamp_fd.write(f'{EOF_FORMAT},0,{bytes_to_process},0,\n') + return sorted(index, key=lambda e: e[2]), total_bytes_read + + +def index_messages(input_path: str, options: argparse.Namespace): + 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) - 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)) + # 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 load_index(index_file), file_size + 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): +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(): 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 -def find_gaps(indexes): +def find_gaps(indexes: List[Tuple]): next_offset = 0 has_gaps = False for index in indexes: @@ -191,43 +352,70 @@ def find_gaps(indexes): _logger.info(f"No gaps found.") -def generate_separated_logs(input_path, indexes, options): - output_map = {} +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 - 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 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] - 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) +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 + 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 + 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): + 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 for extraction if requested. + 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) + + rtcm_framer, fe_framer, nmea_framer = _create_framers(options, return_bytes=options.extract) + + # 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. @@ -245,10 +433,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 " @@ -274,41 +464,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 @@ -319,12 +537,15 @@ 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}.") + # 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: @@ -344,14 +565,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} |' 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 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", 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()