Step-by-Step AIS Message Decoding in Python #

Raw AIS telemetry fails silently: a single truncated !AIVDM sentence can produce NaN-infected MMSI fields that propagate through spatial joins for hours before anyone notices. This guide extends the parent workflow on parsing AIS NMEA sentences with Python and addresses its exact operational bottleneck — the gap between receiving bytes off the wire and handing valid, CRS-compliant position records to downstream analytics. Five sequential steps cover stream sanitisation, multi-part reassembly, bitwise payload extraction, Arrow-native serialisation, and spatial bounds validation; each step eliminates one class of silent corruption and can be unit-tested in isolation.

Why Raw AIS Streams Corrupt Spatial Pipelines #

AIS messages arrive in one of two ways: complete single-sentence position reports (!AIVDM,1,1,,A,...) or multi-part sequences where a payload too large for 82 characters is split across two or more sentences sharing a sequence number. Terrestrial receivers, satellite aggregators, and vessel-mounted transponders all exhibit different failure patterns:

• Terrestrial VHF: high duplicate rate, occasional bit-flips due to RF interference, valid checksum on a corrupted payload (the checksum only covers the NMEA fields, not the AIS payload content).
• Satellite AIS: out-of-order fragment delivery and dropped parts are common; the reassembler must handle gaps without blocking the pipeline.
• Replay/historical feeds: mixed talker IDs (!AIVDM vs !BSVDM), Windows-style CRLF line endings, and BOM markers at file boundaries.

The pipeline must enforce strict NMEA-0183 syntax and XOR checksum verification before consuming CPU cycles on bit decoding. Feeding malformed sentences into a bitwise decoder produces out-of-bounds bit extractions that return structurally valid but geographically impossible coordinates — exactly the kind of data that breaks anomaly detection in AIS trajectories without raising an exception.

The five stages form a deterministic, stateless-where-possible chain:

Step 1 — Checksum Validation and Stream Sanitisation #

Eliminate corrupted or malformed NMEA sentences before they consume decoding resources or trigger silent data drift in downstream spatial joins.

Raw feeds frequently contain truncated payloads, invalid checksums, or non-standard sentence prefixes. The pipeline must enforce strict NMEA-0183 syntax compliance and XOR checksum verification. Every character between ! and * is XOR’d; the two-hex-digit result must match the suffix.

import re
from typing import Optional, Tuple

NMEA_PATTERN = re.compile(
    r"^\!(AIVDM|AIVDO|BSVDM),(\d),(\d),([A-Z]?),([A-Z@!0-9]+),(\d)\*([0-9A-Fa-f]{2})$"
)

def validate_nmea_checksum(sentence: str) -> bool:
    """Verify NMEA XOR checksum against payload."""
    if "*" not in sentence:
        return False
    body, checksum_str = sentence.rsplit("*", 1)
    computed = 0
    for char in body[1:]:   # skip leading '!'
        computed ^= ord(char)
    try:
        return computed == int(checksum_str.strip()[:2], 16)
    except ValueError:
        return False

def sanitize_stream(
    raw_line: str,
) -> Optional[Tuple[str, int, int, str, str, int]]:
    """
    Parse and validate a raw AIS NMEA line.

    Returns (talker_id, total_parts, seq_num, channel, payload, fill_bits)
    or None for any line that fails structural or checksum validation.
    Invalid lines must be routed to a dead-letter queue by the caller.
    """
    stripped = raw_line.strip().lstrip("\xef\xbb\xbf")  # strip BOM
    match = NMEA_PATTERN.match(stripped)
    if not match:
        return None
    if not validate_nmea_checksum(stripped):
        return None

    talker, total, seq, channel, payload, fill_bits, _cksum = match.groups()
    return (talker, int(total), int(seq), channel, payload, int(fill_bits))

This sanitisation layer is stateless and safe to parallelise across ingestion workers. Invalid lines are routed to a dead-letter queue for audit, preserving pipeline throughput while preventing bitwise decoder failures on the steps below.

Step 2 — Multi-Part Message Reassembly #

Deterministically reconstruct fragmented AIS messages using sequence identifiers and total fragment counts while enforcing strict memory ceilings.

AIS payloads exceeding 82 characters are split across multiple NMEA sentences. Out-of-order delivery or dropped fragments are common in satellite telemetry. The reassembler buffers fragments keyed by (seq_id, channel), enforces a maximum fragment age (TTL), and discards incomplete sets after expiry rather than blocking.

from collections import OrderedDict
from typing import Dict, Tuple, Optional
import time

class FragmentBuffer:
    """
    Bounded LRU buffer for multi-part AIS message reassembly.

    max_entries caps memory; TTL prevents stale fragments blocking
    the pipeline during satellite downlink gaps.
    """

    def __init__(self, max_entries: int = 100_000, ttl_seconds: float = 5.0) -> None:
        self._buffer: OrderedDict[Tuple[str, str], Dict] = OrderedDict()
        self.max_entries = max_entries
        self.ttl = ttl_seconds

    def add_fragment(
        self,
        seq_id: str,
        channel: str,
        index: int,
        total: int,
        payload: str,
        fill_bits: int,
    ) -> Optional[Tuple[str, int]]:
        """
        Buffer one fragment.

        Returns (combined_payload, fill_bits) when all parts have arrived,
        otherwise None.  The fill_bits of the *last* part are used.
        """
        self._evict_expired()
        key = (seq_id, channel)
        if key not in self._buffer:
            if len(self._buffer) >= self.max_entries:
                self._buffer.popitem(last=False)  # evict oldest
            self._buffer[key] = {
                "parts": [None] * total,
                "ts": time.monotonic(),
            }
        self._buffer[key]["parts"][index - 1] = (payload, fill_bits)

        parts = self._buffer[key]["parts"]
        if all(p is not None for p in parts):
            combined = "".join(p[0] for p in parts)
            last_fill = parts[-1][1]
            del self._buffer[key]
            return combined, last_fill
        return None

    def _evict_expired(self) -> None:
        now = time.monotonic()
        stale = [k for k, v in self._buffer.items() if now - v["ts"] > self.ttl]
        for k in stale:
            del self._buffer[k]

The buffer operates in-memory with bounded growth. Expired fragments are purged before each insertion, preventing memory leaks during high-throughput satellite downlink windows. For the broader ingestion context — including Kafka offset management and deduplication — see building an AIS Kafka consumer in Python.

Step 3 — Bitwise Payload Extraction and Field Mapping #

Decode 6-bit ASCII payloads into binary streams and map fields according to ITU-R M.1371 message-type schemas without intermediate string allocations.

AIS encodes data using a 6-bit ASCII subset. Each character maps to a 6-bit integer by subtracting 48 from the ASCII code and, for values ≥ 40, subtracting a further 8 (per the ITU-R M.1371-5 character table). The decoder reconstructs the bitstream as a Python integer and extracts fields using fixed-width bit offsets — no intermediate lists or strings.

The position-report field offsets differ between transponder classes, which is the single most common source of silently wrong coordinates. Always branch on msg_type (bits 0–5) before selecting an extractor:

The function below implements the Class A layout; the Class B variant reuses the same _extract_bits/_signed helpers with the offsets from the right-hand column.

from typing import Dict, Any

def _char_to_6bit(c: str) -> int:
    """Convert one AIS 6-bit ASCII character to an integer (0–63)."""
    v = ord(c) - 48
    if v > 39:
        v -= 8
    return v & 0x3F

def decode_payload_to_bits(payload: str, fill_bits: int) -> int:
    """Return the payload as a single integer bitstream (MSB first)."""
    value = 0
    for ch in payload:
        value = (value << 6) | _char_to_6bit(ch)
    value >>= fill_bits   # strip LSB padding
    return value

def _extract_bits(bitstream: int, total_bits: int, offset: int, length: int) -> int:
    """Extract `length` bits starting at `offset` from MSB."""
    shift = total_bits - offset - length
    return (bitstream >> shift) & ((1 << length) - 1)

def _signed(value: int, bits: int) -> int:
    """Two's-complement conversion for signed AIS fields."""
    if value >= (1 << (bits - 1)):
        value -= 1 << bits
    return value

def extract_type1_position_report(payload: str, fill_bits: int) -> Dict[str, Any]:
    """
    Parse AIS Type 1/2/3 Position Report per ITU-R M.1371-5.

    Bit layout (offsets):
      0–5   message type (6 bits)
      6–7   repeat indicator (2 bits)
      8–37  MMSI (30 bits)
      38–41 navigation status (4 bits)
      42–49 rate of turn — signed (8 bits)
      50–59 speed over ground × 10 knots (10 bits)
      60    position accuracy (1 bit)
      61–88 longitude × 10 000 min, signed (28 bits)
      89–115 latitude × 10 000 min, signed (27 bits)
      116–127 course over ground × 10 deg (12 bits)
    """
    total_bits = len(payload) * 6 - fill_bits
    bits = decode_payload_to_bits(payload, fill_bits)

    msg_type   = _extract_bits(bits, total_bits, 0,   6)
    mmsi       = _extract_bits(bits, total_bits, 8,  30)
    nav_status = _extract_bits(bits, total_bits, 38,  4)
    rot_raw    = _signed(_extract_bits(bits, total_bits, 42,  8),  8)
    sog_raw    = _extract_bits(bits, total_bits, 50, 10)
    lon_raw    = _signed(_extract_bits(bits, total_bits, 61, 28), 28)
    lat_raw    = _signed(_extract_bits(bits, total_bits, 89, 27), 27)
    cog_raw    = _extract_bits(bits, total_bits, 116, 12)

    return {
        "msg_type":   msg_type,
        "mmsi":       mmsi,
        "nav_status": nav_status,
        "rot":        rot_raw,
        "sog_knots":  sog_raw   / 10.0,      # 1/10-knot resolution
        "lon_deg":    lon_raw   / 600000.0,  # 1/10 000 min → decimal degrees
        "lat_deg":    lat_raw   / 600000.0,
        "cog_deg":    cog_raw   / 10.0,      # 1/10-degree resolution
    }

Bitwise operations execute directly on Python integers, avoiding string concatenation overhead and ensuring deterministic latency under high-volume satellite downlink conditions.

Step 4 — Memory-Constrained Serialisation and CRS Normalisation #

Convert raw decoded dicts to columnar Arrow format, apply explicit WGS84 metadata, and stage for spatial indexing.

AIS coordinates arrive as signed integers in units of 1/10 000 of a minute. The conversion to decimal degrees divides by 600 000 (10 000 minute-fractions × 60 minutes per degree):

$$\lambda_{\deg} = \frac{\text{lon}_{\text{raw}}}{10\,000 \times 60} = \frac{\text{lon}_{\text{raw}}}{600\,000}, \qquad \varphi_{\deg} = \frac{\text{lat}_{\text{raw}}}{600\,000}$$

where $\text{lon}_{\text{raw}}$ and $\text{lat}_{\text{raw}}$ are the two’s-complement integers extracted in Step 3. The sentinel “not available” values — longitude $181°$ ($\text{lon}_{\text{raw}} = 108\,600\,000$) and latitude $91°$ — survive this arithmetic intact and are caught by the bounds check in Step 5. Using Apache Arrow avoids Python object allocation per row and enables zero-copy memory sharing between ingestion workers and downstream spatial engines.

import pyarrow as pa
from typing import List

def normalize_and_serialize(records: List[dict]) -> pa.Table:
    """
    Convert a batch of decoded AIS dicts to an Arrow table.

    CRS metadata is embedded in the schema so downstream GeoPandas,
    PostGIS loaders, and GeoParquet writers receive projection context
    without an out-of-band sidecar.
    """
    schema = pa.schema([
        pa.field("mmsi",       pa.uint32()),
        pa.field("timestamp",  pa.timestamp("us", tz="UTC")),
        pa.field("lat",        pa.float64()),
        pa.field("lon",        pa.float64()),
        pa.field("sog_knots",  pa.float32()),
        pa.field("cog_deg",    pa.float32()),
        pa.field("nav_status", pa.uint8()),
    ])

    table = pa.table(
        {
            "mmsi":       pa.array([r["mmsi"]       for r in records], type=pa.uint32()),
            "timestamp":  pa.array([r["ts"]         for r in records], type=pa.timestamp("us", tz="UTC")),
            "lat":        pa.array([r["lat_deg"]    for r in records], type=pa.float64()),
            "lon":        pa.array([r["lon_deg"]    for r in records], type=pa.float64()),
            "sog_knots":  pa.array([r["sog_knots"]  for r in records], type=pa.float32()),
            "cog_deg":    pa.array([r["cog_deg"]    for r in records], type=pa.float32()),
            "nav_status": pa.array([r["nav_status"] for r in records], type=pa.uint8()),
        },
        schema=schema,
    )

    # Explicit CRS metadata prevents silent projection mismatches
    # during spatial joins with bathymetric or jurisdictional layers.
    return table.replace_schema_metadata({
        "crs":            "EPSG:4326",
        "encoding":       "WGS84",
        "pipeline_stage": "ais_decoded",
    })

The explicit EPSG:4326 metadata prevents silent projection mismatches when these positions later enter a CRS alignment pipeline and are joined against bathymetric grids or jurisdictional boundary layers in a projected datum. For speed and heading analytics that consume this table, see speed and heading profiling for maritime analytics.

Step 5 — Spatial Validation and Pipeline Handoff #

Filter geometrically impossible coordinates, validate kinematic bounds, and route valid records to trajectory segmentation while quarantining invalid ones.

Decoded positions frequently contain land-locked artifacts, coordinate wrap errors, or kinematically impossible SOG values. Validation occurs on the Arrow table via pandas boolean masks, keeping the operation vectorised rather than row-by-row.

import pandas as pd
import pyarrow as pa
from typing import Tuple

def validate_spatial_bounds(
    table: pa.Table,
) -> Tuple[pa.Table, pa.Table]:
    """
    Split a decoded AIS table into (valid, invalid) partitions.

    Rejection criteria:
      - Coordinates outside WGS84 extremes
      - SOG > 150 knots (physically impossible for any surface vessel)
      - COG outside [0, 360)
      - Null-island fill values (lat ≈ 0, lon ≈ 0 within ±0.001°)
    """
    df = table.to_pandas()

    lat_ok    = (df["lat"] >= -90.0)  & (df["lat"] <= 90.0)
    lon_ok    = (df["lon"] >= -180.0) & (df["lon"] <= 180.0)
    sog_ok    = (df["sog_knots"] >= 0.0) & (df["sog_knots"] <= 150.0)
    cog_ok    = (df["cog_deg"] >= 0.0)   & (df["cog_deg"] <= 360.0)
    not_null  = ~((df["lat"].abs() < 0.001) & (df["lon"].abs() < 0.001))

    mask = lat_ok & lon_ok & sog_ok & cog_ok & not_null
    valid_tbl   = table.filter(pa.array(mask.values))
    invalid_tbl = table.filter(pa.array((~mask).values))
    return valid_tbl, invalid_tbl

Invalid records are routed to a quarantine dataset for manual review or automated flagging. Valid tables hand off directly to behaviour segmentation — see segmenting vessel routes by behaviour for the next stage.

Verification — Confirming the Pipeline End-to-End #

Run this acceptance test against a known-good sentence before deploying to a production Kafka consumer:

import pytest

KNOWN_GOOD = "!AIVDM,1,1,,A,15M67N0000G?Uf6E`FepT@3n00Sa,0*73"

def test_decode_known_good_sentence() -> None:
    parsed = sanitize_stream(KNOWN_GOOD)
    assert parsed is not None, "sanitise_stream rejected a valid sentence"

    _talker, total, _seq, _channel, payload, fill_bits = parsed
    assert total == 1, "single-part message should report total=1"

    record = extract_type1_position_report(payload, fill_bits)
    assert record["msg_type"] == 1
    assert isinstance(record["mmsi"], int) and len(str(record["mmsi"])) == 9
    assert -90.0 <= record["lat_deg"] <= 90.0
    assert -180.0 <= record["lon_deg"] <= 180.0
    assert 0.0 <= record["sog_knots"] <= 150.0

    table = normalize_and_serialize([{**record, "ts": pd.Timestamp.utcnow()}])
    valid, invalid = validate_spatial_bounds(table)
    assert len(valid) == 1
    assert len(invalid) == 0

if __name__ == "__main__":
    import pandas as pd
    test_decode_known_good_sentence()
    print("All assertions passed.")

A passing run confirms: NMEA regex acceptance, XOR checksum, bit-field offsets, Arrow schema construction, and spatial rejection logic.

Edge Cases and Gotchas #

• Talker-ID variants: Some receivers emit !BSVDM (base station) or !SAVDM (simplex) instead of !AIVDM. The regex in Step 1 accepts only AIVDM/AIVDO unless widened. For mixed feeds, replace the alternation with [A-Z]{5} and log the unexpected talker IDs for later triage.

• Sequence-number collision across channels: The FragmentBuffer keys on (seq_id, channel). If a data source mixes channel A and channel B with the same sequence number — common when multiplexing two antenna feeds into one socket — parts will land in separate buffer slots correctly, but if the channel field is left blank (empty string for single-part messages), both parts map to the same key. Guard by substituting "_" for an empty channel before insertion.

• Type 18 Class B transponders: Class B position reports (message type 18) use a 168-bit layout whose field offsets differ from the Type 1/2/3 layout (see the offset table in Step 3). Feeding a Type 18 payload into extract_type1_position_report will silently produce wrong coordinates because the bit windows no longer line up — always branch on msg_type before choosing the field extractor.

• Static and voyage data (Type 5) spans many fragments: Type 5 reports carry the ship name, call sign, and destination across a 424-bit payload, which always arrives as a multi-part sequence. If the FragmentBuffer TTL in Step 2 is set too low for a slow satellite downlink, these messages expire before reassembly and the vessel never acquires a static profile — symptomised by position tracks with a valid MMSI but a perpetually empty name field.

---

Related #

• Parsing AIS NMEA Sentences with Python — parent page covering ingestion architecture and sentence grammar
• Building an AIS Kafka Consumer in Python — offset management and deduplication for high-throughput AIS feeds
• Segmenting Vessel Routes by Behaviour — trajectory segmentation that consumes the validated position tables produced here
• Anomaly Detection in AIS Trajectories — downstream stage where corrupt coordinates become statistical outliers if not caught here
• CRS Alignment for Coastal GIS Projects — reprojecting the WGS84 positions emitted here onto projected datums for spatial joins

Up: Parsing AIS NMEA Sentences with Python