Parsing AIS NMEA Sentences with Python

Automated coastal monitoring, vessel traffic density mapping, and marine spatial analysis pipelines require deterministic ingestion of raw telemetry. This workflow sits within the Marine Spatial Data Fundamentals & Architecture framework as the mandatory normalization layer that converts raw radio telemetry into structured, query-ready geospatial records. Raw Automatic Identification System (AIS) streams arrive as fragmented NMEA 0183 sentences over TCP/UDP sockets, satellite downlinks, or terrestrial base-station logs. Production environments must handle high-throughput ingestion, validate checksums, reconstruct multi-part payloads, and output spatially normalised records — without memory bloat or coordinate drift. Everything emitted by this parser feeds directly into CRS alignment and format-specific archival pipelines downstream.

Reference Configuration and Specification #

NMEA 0183 Stream Architecture and Payload Constraints #

AIS telemetry is transmitted via AIVDM (received from other vessels) and AIVDO (own-ship) sentence types, both conforming to ITU-R M.1371. Correct handling of the following constraints is non-negotiable in production:

Fragmentation. Type 1, 2, 3, 5, 18, and 19 messages frequently exceed the 82-character NMEA limit and split across two or more consecutive sentences. The <total> and <index> fields must be matched — keyed on (sequence_number, channel, total) — to reconstruct the complete binary payload. Discarding a single fragment corrupts the entire message.

6-bit encoding. The <payload> field uses a custom ASCII-to-6-bit mapping. Each character maps to a 6-bit integer by subtracting 48 from the ASCII ordinal; for values above 39, subtract a further 8. The <fill_bits> field indicates how many trailing zero bits were appended to align the final byte boundary and must be removed before field extraction.

Checksum validation. Every sentence terminates with *XX, where XX is the hexadecimal XOR of all ASCII characters between ! and *. Invalid checksums indicate transmission corruption, radio interference, or parser misalignment and must be silently dropped — logging the count for downstream QA.

Memory constraints. Terrestrial receivers in busy shipping lanes generate 50 000–200 000 sentences per minute. Loading raw logs into memory triggers OOM failures. Streaming generators and bounded fragment caches are mandatory for cloud-native readiness.

The diagram below maps the full sentence lifecycle — from raw socket bytes through checksum validation, fragment assembly, and 6-bit decoding to structured output:

Memory-Constrained Streaming Parser and Multi-Part Assembly #

The implementation below processes AIS streams line-by-line using a generator pattern. It validates checksums, buffers fragmented messages in a time-bounded LRU cache, decodes 6-bit payloads, and yields structured dictionaries. Memory is bounded by capping the fragment buffer and flushing stale entries on each iteration.

import logging
import re
import time
from collections import OrderedDict
from typing import Generator, Iterator, Dict, Optional, Tuple

logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s: %(message)s")
log = logging.getLogger("ais_parser")

# ITU-R M.1371 constants
CHECKSUM_RE = re.compile(r"^![^*]+\*[0-9A-Fa-f]{2}$")
MAX_CACHE_SIZE: int = 10_000
FRAGMENT_TTL: float = 45.0  # seconds before stale incomplete messages are evicted

_stats: Dict[str, int] = {
    "received": 0,
    "checksum_fail": 0,
    "fragment_evicted": 0,
    "decoded": 0,
}


def _validate_checksum(sentence: str) -> bool:
    """Verify NMEA 0183 XOR checksum; return False on any structural anomaly."""
    if not CHECKSUM_RE.match(sentence):
        return False
    body, expected = sentence.rsplit("*", 1)
    calc = 0
    for ch in body[1:]:  # skip leading '!'
        calc ^= ord(ch)
    return f"{calc:02X}" == expected.strip().upper()


def _char_to_6bit(c: str) -> int:
    """Map one 6-bit ASCII character to its integer value (ITU-R M.1371 Table 1)."""
    v = ord(c) - 48
    if v > 39:
        v -= 8
    return v & 0x3F


def _decode_6bit(payload: str, fill_bits: int) -> bytes:
    """
    Convert a 6-bit ASCII payload string to raw bytes.

    Each of the payload's characters encodes 6 bits.  The total usable bit
    count is (len(payload) * 6 - fill_bits).  Fill bits occupy the least-
    significant positions and must be stripped before field extraction.
    """
    if not payload:
        raise ValueError("Empty AIS payload")
    total_bits = len(payload) * 6 - fill_bits
    if total_bits <= 0:
        raise ValueError(f"Payload too short after fill_bits={fill_bits} removal")
    value = 0
    for ch in payload:
        value = (value << 6) | _char_to_6bit(ch)
    value >>= fill_bits  # remove trailing fill bits
    byte_len = (total_bits + 7) // 8
    return value.to_bytes(byte_len, byteorder="big")


def _assemble_fragments(
    cache: OrderedDict,
    msg_key: str,
    total: int,
    index: int,
    payload: str,
    fill_bits: int,
) -> Optional[Tuple[str, int]]:
    """
    Buffer fragments and return (combined_payload, fill_bits) once all parts arrive.

    Keys are (sequence_number, channel, total_count) tuples serialised as a string
    so that interleaved multi-part messages on different channels do not collide.
    """
    if msg_key not in cache:
        cache[msg_key] = {"parts": [None] * total, "ts": time.monotonic()}
    cache[msg_key]["parts"][index - 1] = (payload, fill_bits)

    if all(p is not None for p in cache[msg_key]["parts"]):
        combined_payload = "".join(p[0] for p in cache[msg_key]["parts"])  # type: ignore[index]
        combined_fill = cache[msg_key]["parts"][-1][1]  # type: ignore[index]
        del cache[msg_key]
        return combined_payload, combined_fill
    return None


def _purge_stale(cache: OrderedDict, now: float) -> None:
    """Evict entries that exceed FRAGMENT_TTL or push the cache above MAX_CACHE_SIZE."""
    while cache:
        oldest_key, oldest_val = next(iter(cache.items()))
        if len(cache) > MAX_CACHE_SIZE or (now - oldest_val["ts"]) > FRAGMENT_TTL:
            cache.popitem(last=False)
            _stats["fragment_evicted"] += 1
            log.debug("Evicted stale fragment key=%s", oldest_key)
        else:
            break


def parse_ais_stream(lines: Iterator[str]) -> Generator[Dict, None, None]:
    """
    Streaming AIS NMEA parser.

    Accepts any iterable of raw NMEA sentence strings (file handle, socket
    line reader, Kafka consumer, etc.) and yields structured dicts for each
    successfully decoded message.  Memory is bounded: the fragment cache
    never exceeds MAX_CACHE_SIZE entries, and stale fragments are purged on
    every multi-part sentence arrival.

    Yielded dict keys:
        msg_type (int)        — ITU-R M.1371 message type 1–27
        channel (str)         — AIS VHF channel ('A' or 'B')
        raw_bytes (bytes)     — decoded payload, ready for field extraction
        parsed_at (float)     — Unix timestamp of successful decode
        fragment_count (int)  — 1 for single-part; >1 for reassembled
    """
    frag_cache: OrderedDict = OrderedDict()

    for raw in lines:
        line = raw.strip()
        _stats["received"] += 1

        if not line:
            continue
        if not _validate_checksum(line):
            _stats["checksum_fail"] += 1
            log.debug("Checksum failure: %s", line[:40])
            continue

        fields = line.split(",")
        if len(fields) < 7 or not fields[0].startswith("!AIVD"):
            continue

        try:
            total = int(fields[1])
            index = int(fields[2])
            seq = fields[3]
            channel = fields[4]
            payload = fields[5]
            fill_bits = int(fields[6].split("*")[0])
        except (ValueError, IndexError) as exc:
            log.debug("Malformed sentence fields: %s — %s", exc, line[:60])
            continue

        msg_key = f"{seq}_{channel}_{total}"

        if total > 1:
            _purge_stale(frag_cache, time.monotonic())
            result = _assemble_fragments(
                frag_cache, msg_key, total, index, payload, fill_bits
            )
            if result is None:
                continue  # waiting for remaining parts
            payload, fill_bits = result

        try:
            raw_bytes = _decode_6bit(payload, fill_bits)
            # Message type occupies bits 0–5 (the top 6 bits of the first byte)
            msg_type = raw_bytes[0] >> 2
            _stats["decoded"] += 1
            yield {
                "msg_type": msg_type,
                "channel": channel,
                "raw_bytes": raw_bytes,
                "parsed_at": time.time(),
                "fragment_count": total,
            }
        except (ValueError, OverflowError) as exc:
            log.warning("Decode error for payload %.20s…: %s", payload, exc)
            continue

Validation Gates and Quality Control #

Correctness at this stage determines the fidelity of every downstream spatial model. Run these checks against any batch before committing records to persistent storage. The bit-level field semantics referenced throughout these gates — sentinel encodings, signed-field widths, and per-type field offsets — are enumerated in the step-by-step AIS message decoding guide, which is the companion debug reference for any gate that fails.

Checksum rejection rate #

A healthy terrestrial feed rejects fewer than 0.1 % of sentences. Radio interference, TCP re-framing errors, or injected malformed data push this above 1 %. Log _stats["checksum_fail"] / _stats["received"] after each batch; alert if it exceeds 0.005 (0.5 %).

def report_stats() -> None:
    received = _stats["received"] or 1  # guard divide-by-zero
    ck_rate = _stats["checksum_fail"] / received
    ev_rate = _stats["fragment_evicted"] / received
    log.info(
        "Batch stats — received=%d decoded=%d checksum_fail=%.4f evicted=%.4f",
        received, _stats["decoded"], ck_rate, ev_rate,
    )
    if ck_rate > 0.005:
        log.warning("Checksum failure rate %.2f%% exceeds threshold", ck_rate * 100)

Coordinate range assertion #

After decoding, validate latitude (−90 to 90°) and longitude (−180 to 180°). Values outside these ranges indicate a fill-bit miscalculation or a type/subtype mismatch. The sentinel 91.0 and 181.0 (decimal-degree equivalents of the 0x6791AC0 and 0xA50DE80 integer sentinels in M.1371) signal “not available” and must be filtered rather than stored as positions.

LAT_SENTINEL  = 91.0   # M.1371 § 3.3.4 — "not available"
LON_SENTINEL  = 181.0

def valid_position(lat: float, lon: float) -> bool:
    return abs(lat) < LAT_SENTINEL and abs(lon) < LON_SENTINEL

Fragment eviction rate #

An eviction rate above 0.1 % on a clean feed means fragments arrive out of order across a TCP session boundary, a channel switch is happening mid-message, or the TTL is set too low for a satellite AIS source (where burst delivery latency can reach 20–30 s). Increase FRAGMENT_TTL to 90 s for satellite feeds and add the S-AIS talker-ID !S to the sentence filter.

Message-type distribution sanity check #

Types 1–3 (position reports, class A) should constitute the bulk of traffic in port environments. A sudden spike in type 21 (aid-to-navigation) or type 24 (class B static) without corresponding positional messages usually indicates a replay or injected synthetic stream. Validate the distribution against a known-good baseline before ingestion.

Common Failure Modes and Diagnosis #

1. Fill-bit subtraction on concatenated fragments #

Symptom: OverflowError: int too big to convert or wrong message type extracted from multi-part messages.

Root cause: fill_bits applies only to the final fragment’s payload. Applying it to each fragment’s bit count before concatenation corrupts the combined bit stream.

Fix: Concatenate the payload strings from all fragments first, then decode _decode_6bit(combined_payload, last_fragment_fill_bits). The implementation above captures combined_fill = cache[msg_key]["parts"][-1][1] specifically for this reason.

2. Axis-order drift in coordinate conversion #

Symptom: Vessel positions appear in the South Atlantic instead of the North Sea; spatial joins return zero matches.

Root cause: Applying / 600_000.0 to the unsigned integer field. Latitude and longitude in M.1371 are signed twos-complement 28-bit (latitude) and 29-bit (longitude) integers. Treating the raw bit pattern as unsigned shifts negative positions to large positive values.

Fix: Extract the signed integer using Python’s int.from_bytes with twos-complement interpretation, or apply a manual sign-extension mask before dividing. See the coordinate normalisation section below.

3. Fragment cache key collision #

Symptom: Garbled or merged payloads; wrong MMSI numbers in output records.

Root cause: Using only seq (the sequence number field) as the cache key. The sequence number field is a single digit (0–9) and wraps frequently. Two simultaneous multi-part messages on different channels with the same sequence digit will merge their payloads.

Fix: Key the cache on f"{seq}_{channel}_{total}" as implemented above, incorporating both the VHF channel identifier and the declared total fragment count to ensure uniqueness within a realistic collision window.

4. Stale cache memory growth under satellite feeds #

Symptom: RSS memory grows unboundedly; cache contains tens of thousands of keys after a long satellite replay.

Root cause: Satellite S-AIS sources burst-deliver messages with inter-fragment gaps of 20–60 s, exceeding a 45 s TTL configured for terrestrial feeds. Fragments arrive incomplete and are never assembled; the eviction sweep runs but finds entries still within TTL, so the cache fills.

Fix: Set FRAGMENT_TTL = 90.0 for satellite feeds. Also enforce MAX_CACHE_SIZE strictly — the _purge_stale implementation above evicts the oldest entry whenever the cap is exceeded, regardless of TTL, preventing unbounded growth.

Geospatial Normalisation and Coordinate Extraction #

Parsed payloads require immediate coordinate normalisation before entering analytical workflows. In ITU-R M.1371-compliant messages, latitude and longitude are encoded as signed integers in units of 1/10 000 of a minute. The conversion to WGS 84 decimal degrees is a single fixed scale factor:

$$\lambda_{\text{deg}} = \frac{I_{\text{signed}}}{10\,000 \times 60} = \frac{I_{\text{signed}}}{600\,000}$$

where $I_{\text{signed}}$ is the twos-complement integer field and the divisor decomposes as 10 000 (minute fractions per minute) × 60 (minutes per degree). Applying any other scale factor introduces systematic coordinate drift that compounds during spatial indexing or rasterisation. Misalignment at this stage propagates into downstream CRS alignment for coastal GIS projects workflows, causing trajectory artifacts and false proximity alerts in high-density shipping corridors. Where positions must be reconciled against a vertical reference such as chart datum, the same records flow into tidal datum transformations in Python once the horizontal normalisation here is locked.

The 28-bit latitude field spans bits 89–116 of a type-1 message payload; the 28-bit longitude field spans bits 61–88. Both are encoded as twos-complement signed integers. The diagram below traces a single ASCII payload character through the 6-bit mapping into the contiguous bit-stream, then shows where the signed longitude and latitude fields are sliced out and how the trailing fill bits are discarded:

The implementation slices these fields directly from the decoded byte string:

def extract_position(raw: bytes) -> Tuple[float, float]:
    """
    Extract signed latitude and longitude from a decoded type-1/2/3 payload.

    Bit offsets (M.1371-5):
        longitude — bits 61–88  (28 bits, signed, 1/10 000 min)
        latitude  — bits 89–116 (27 bits, signed, 1/10 000 min)
    """
    # Convert bytes to a big-endian integer for bit-level slicing
    bits = int.from_bytes(raw, byteorder="big")
    total_bits = len(raw) * 8

    def signed_field(offset: int, width: int) -> int:
        shift = total_bits - offset - width
        raw_val = (bits >> shift) & ((1 << width) - 1)
        # Twos-complement sign extension
        if raw_val >= (1 << (width - 1)):
            raw_val -= 1 << width
        return raw_val

    lon_int = signed_field(61, 28)
    lat_int = signed_field(89, 27)

    lat = lat_int / 600_000.0
    lon = lon_int / 600_000.0
    return lat, lon

Coordinate precision should be locked to 6 decimal places (≈ 0.11 m at the equator) before writing to persistent storage. This balances storage efficiency against navigational accuracy requirements and avoids false precision from floating-point representation.

Once normalised, vessel positions can be aggregated into spatiotemporal grids for density mapping, emission modelling, or habitat impact assessment. The choice of output format depends on analytical requirements: dense time-series trajectories benefit from chunked, compressed arrays with temporal indexing, while static density rasters require tiled geospatial formats optimised for spatial joins. Refer to Understanding NetCDF vs GeoTIFF for Marine Data for format-specific ingestion strategies and archival compression trade-offs.

Pipeline Integration and Downstream Handoff #

The streaming parser is designed to be composed with upstream sources and downstream sinks without holding state beyond the bounded fragment cache. A typical integration pattern wires a socket reader, the parser generator, the coordinate extractor, and a batch writer:

import socket
from itertools import islice

def socket_line_reader(host: str, port: int) -> Iterator[str]:
    """Yield raw NMEA lines from a TCP AIS aggregator socket."""
    with socket.create_connection((host, port), timeout=30) as sock:
        buffer = b""
        while True:
            chunk = sock.recv(4096)
            if not chunk:
                break
            buffer += chunk
            while b"\n" in buffer:
                line, buffer = buffer.split(b"\n", 1)
                yield line.decode("ascii", errors="replace")

def ingest_to_duckdb(host: str, port: int, db_path: str, batch_size: int = 5000) -> None:
    """
    Stream AIS records from a TCP source into DuckDB for spatial analytics.

    Records are staged in memory up to batch_size then flushed as a single
    INSERT to minimise transaction overhead.  Coordinate precision is locked
    at 6 decimal places before persistence.
    """
    import duckdb

    con = duckdb.connect(db_path)
    con.execute("""
        CREATE TABLE IF NOT EXISTS ais_positions (
            msg_type     INTEGER,
            mmsi         INTEGER,
            lat          DOUBLE,
            lon          DOUBLE,
            channel      VARCHAR,
            parsed_at    DOUBLE,
            fragment_count INTEGER
        )
    """)

    source = socket_line_reader(host, port)
    batch: list = []

    for record in parse_ais_stream(source):
        raw = record["raw_bytes"]
        msg_type = record["msg_type"]
        if msg_type not in (1, 2, 3, 18):
            continue  # skip non-positional types at this stage

        try:
            lat, lon = extract_position(raw)
        except (ValueError, IndexError):
            continue

        if not valid_position(lat, lon):
            continue  # sentinel value — position not available

        # Extract MMSI: bits 8–37 (30 bits, unsigned)
        bits = int.from_bytes(raw, byteorder="big")
        total_bits = len(raw) * 8
        mmsi = (bits >> (total_bits - 38)) & 0x3FFFFFFF

        batch.append((
            msg_type,
            mmsi,
            round(lat, 6),
            round(lon, 6),
            record["channel"],
            record["parsed_at"],
            record["fragment_count"],
        ))

        if len(batch) >= batch_size:
            con.executemany(
                "INSERT INTO ais_positions VALUES (?,?,?,?,?,?,?)", batch
            )
            log.info("Flushed %d records to DuckDB", len(batch))
            batch.clear()

    if batch:
        con.executemany("INSERT INTO ais_positions VALUES (?,?,?,?,?,?,?)", batch)
    con.close()
    report_stats()

For long-term retention, parsed AIS records should be partitioned by MMSI and day-of-year timestamp. A metadata manifest accompanying each partition should record: source feed URL or file path, parse timestamp, total sentences received, checksum failure rate, and fragment eviction count. This provenance information is required for reproducible spatial analysis and regulatory audit trails.

Bit-level field extraction for speed over ground, course, heading, navigational status, and vessel dimensions follows the same signed-integer unpacking pattern and is covered in full in the step-by-step AIS message decoding guide. For real-time ingestion at scale — including Kafka consumer configuration, message deduplication, and timestamp drift correction — see real-time AIS stream ingestion pipelines.

---

Related

• Step-by-Step AIS Message Decoding in Python — bit-field extraction for type 1–27 messages
• Real-Time AIS Stream Ingestion Pipelines — Kafka consumers, deduplication, and timestamp drift
• CRS Alignment for Coastal GIS Projects — normalising vessel positions into a unified spatial reference
• Understanding NetCDF vs GeoTIFF for Marine Data — archival format selection for parsed AIS trajectories

Up: Marine Spatial Data Fundamentals & Architecture