GPS Data Stream Processor

Build a GPS data stream processor that reads marine navigation sentences from continuous data streams and extracts position information.

Requirements

Stream Processing

The processor should handle continuous streams of GPS data (e.g., from serial ports, network sockets, or files). The stream may contain:

• Partial lines that need buffering
• Multiple types of navigation sentences
• Invalid or malformed sentences that should be handled gracefully

Position Tracking

Extract and track GPS position data including:

• Latitude and longitude coordinates (in decimal degrees)
• Timestamp of the position fix
• GPS quality indicators (valid/invalid fixes)

Error Handling

The processor must handle errors appropriately:

• Ignore or report sentences with invalid checksums
• Handle incomplete lines in the stream
• Continue processing after encountering errors

Output

The processor should output position data in a readable format showing:

• Timestamp
• Latitude and longitude in decimal degrees
• Fix validity status

Test Cases

• When processing a stream containing valid GGA sentences with complete position data, the processor extracts all position information correctly. @test
• When processing a stream with partial lines at buffer boundaries, the processor correctly buffers and parses complete sentences. @test
• When processing a stream containing sentences with checksum errors, the processor handles errors according to the configured error mode. @test
• When processing a stream with mixed valid and invalid GPS fixes, the processor correctly identifies which positions are valid. @test

Implementation

@generates

API

class GPSStreamProcessor:
    """
    Processes continuous streams of GPS navigation data and extracts position information.

    The processor handles buffering of partial lines, parsing of navigation sentences,
    and extraction of position data with appropriate error handling.
    """

    def __init__(self, error_mode='raise'):
        """
        Initialize the GPS stream processor.

        Args:
            error_mode (str): How to handle parsing errors. Options: 'raise', 'ignore', 'yield'
        """
        pass

    def process_stream(self, stream):
        """
        Process a stream of GPS data and yield position information.

        Args:
            stream: An iterable that yields chunks of GPS data (strings or bytes)

        Yields:
            dict: Position information with keys 'timestamp', 'latitude', 'longitude', 'valid'
        """
        pass

    def process_file(self, file_path):
        """
        Process a file containing GPS data and return all positions.

        Args:
            file_path (str): Path to the file containing GPS data

        Returns:
            list: List of position dictionaries
        """
        pass

Dependencies { .dependencies }

pynmea2 { .dependency }

Provides NMEA 0183 sentence parsing and stream processing support.