Deduplicating AIS position reports by MMSI #

A single vessel underway is almost never reported once. The same !AIVDM broadcast is heard by a coastal terrestrial receiver, relayed again by a satellite constellation, and re-emitted a third time by a commercial aggregator that already blends both — so a naive consumer counts one 12:00:03 position three or four times. Deduplicating AIS position reports by MMSI is the stage that collapses that redundant inflow into one canonical fix per vessel per second, keyed on the Maritime Mobile Service Identity plus the UTC second, before any track, geofence, or dwell calculation runs. Do it wrong and every downstream metric — speed over ground, berth-approach timing, demurrage clocks — inflates against phantom motion that never happened.

Architecture Alignment #

This deduper sits one stage inside AIS Data Stream Integration, the ingestion boundary of the wider Container Tracking & AIS Event Synchronization domain. It consumes the checksum-clean, reassembled payloads produced by connecting to public AIS feeds with Python asyncio and hands a single de-duplicated stream to the decode-and-correlate core that feeds Container Status Mapping Rules. Its contract is narrow and strict: accept N overlapping reports of the same ITU-R M.1371 message, emit exactly one, keep the richest copy when the copies disagree, and do all of it in bounded memory so a busy strait cannot exhaust the process. Everything after this stage may safely assume that two rows with the same MMSI and timestamp are the same real-world observation.

Prerequisites & Environment Setup #

The deduper targets Python 3.11+ for datetime.UTC, faster dict internals, and precise type-narrowing. It needs only a decoder and structured logging — no broker, no database, because deduplication runs in-process directly behind the ingestion queue:

pip install "pyais>=2.6" "structlog>=24.1"

Structured JSON logging is mandatory here: a dropped-duplicate decision and a rejected-spoof decision must both be queryable and reproducible for a traffic-image audit, so bare print() and unstructured logging strings are unacceptable. Configure structlog once at startup:

import structlog

structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso", utc=True),
        structlog.processors.JSONRenderer(),
    ],
)
log = structlog.get_logger("ais.dedup")

The tunables that bound the window and its memory footprint are read from the environment:

Step-by-step Implementation #

Step 1 — Model the decoded report and derive the dedup key #

Position reports arrive as ITU-R M.1371 message types 1, 2, and 3 (Class A) or type 18 (Class B), decoded by pyais into a flat object. Wrap it in a typed, immutable record so the dedup key is computed once and cannot drift. The key is the tuple (mmsi, message_type, epoch_second): MMSI identifies the transmitter, the message type keeps a Class A position distinct from a type 5 static report that happens to share the second, and the UTC second — truncated, never rounded — is the coincidence window in which two receivers report the same broadcast.

from __future__ import annotations

from dataclasses import dataclass
from datetime import datetime

import structlog

log = structlog.get_logger("ais.dedup")

DedupKey = tuple[int, int, int]  # (mmsi, msg_type, epoch_second)


@dataclass(frozen=True, slots=True)
class PositionReport:
    mmsi: int
    msg_type: int
    lat: float
    lon: float
    sog: float | None          # speed over ground, knots
    cog: float | None          # course over ground, degrees
    heading: int | None        # true heading, 0-359 or None (511 = n/a)
    received_utc: datetime     # arrival time at THIS process, tz-aware
    source: str                # "terrestrial" | "satellite" | "aggregator"

    @property
    def epoch_second(self) -> int:
        return int(self.received_utc.timestamp())

    @property
    def key(self) -> DedupKey:
        return (self.mmsi, self.msg_type, self.epoch_second)

    @property
    def completeness(self) -> int:
        """Count of populated optional nav fields; higher wins a merge."""
        return sum(v is not None for v in (self.sog, self.cog, self.heading))

Step 2 — Build the sliding-window deduper with bounded memory #

The core is a dictionary keyed on (mmsi, epoch_second) that remembers which observations it has already emitted. Because satellite and aggregator copies of a terrestrial broadcast arrive out of order and seconds late, the window must span several seconds rather than testing only the current tick. Every ingest first evicts keys older than the TTL, so memory is bounded by the number of distinct vessels seen in the last window, not by total traffic. A hard DEDUP_MAX_KEYS ceiling forces oldest-first eviction if a burst outruns the TTL sweep — an explicit, logged bound instead of an unbounded map that ends in an OOM kill.

import os
from collections import OrderedDict

WINDOW_SECONDS: int = int(os.environ.get("DEDUP_WINDOW_SECONDS", "10"))
TTL_SECONDS: int = int(os.environ.get("DEDUP_TTL_SECONDS", "15"))
MAX_KEYS: int = int(os.environ.get("DEDUP_MAX_KEYS", "250000"))


class PositionDeduper:
    """Sliding-window dedup keyed on (mmsi, epoch_second) with TTL eviction."""

    def __init__(self, window: int = WINDOW_SECONDS, ttl: int = TTL_SECONDS,
                 max_keys: int = MAX_KEYS) -> None:
        self._window = window
        self._ttl = ttl
        self._max_keys = max_keys
        # insertion-ordered so eviction is O(1) from the front
        self._seen: OrderedDict[DedupKey, PositionReport] = OrderedDict()

    def _evict(self, now_epoch: int) -> None:
        cutoff = now_epoch - self._ttl
        while self._seen:
            key, report = next(iter(self._seen.items()))
            if report.epoch_second >= cutoff:
                break
            self._seen.popitem(last=False)
        while len(self._seen) > self._max_keys:
            evicted, _ = self._seen.popitem(last=False)
            log.warning("dedup_forced_eviction", key=evicted, live_keys=len(self._seen))

    def ingest(self, report: PositionReport) -> PositionReport | None:
        """Return the canonical report to forward, or None if it is a duplicate."""
        now = report.epoch_second
        self._evict(now)
        key = report.key
        prior = self._seen.get(key)
        if prior is None:
            self._seen[key] = report
            self._seen.move_to_end(key)
            return report
        # duplicate inside the window — resolved in Step 3
        merged = self._resolve_conflict(prior, report)
        self._seen[key] = merged
        log.info("dedup_duplicate_dropped", mmsi=report.mmsi, msg_type=report.msg_type,
                 epoch_second=now, source=report.source, kept_source=merged.source)
        return None

Step 3 — Keep the most complete report when fields differ #

Overlapping copies are rarely byte-identical. A terrestrial receiver often carries a full nav solution — SOG, COG, and true heading — while a satellite copy of the same instant may drop heading to the 511 “not available” sentinel, and an aggregator may round the position. When two reports share a key but disagree, dropping the second blindly can discard the better observation. Resolve on a completeness score, breaking ties by source trust (terrestrial over satellite over aggregator), and log when the later copy wins so the merge is auditable.

_SOURCE_RANK: dict[str, int] = {"terrestrial": 3, "satellite": 2, "aggregator": 1}
MERGE_ON_CONFLICT: bool = os.environ.get("DEDUP_MERGE_ON_CONFLICT", "1") == "1"


def _resolve_conflict(self: "PositionDeduper", prior: PositionReport,
                      incoming: PositionReport) -> PositionReport:
    if not MERGE_ON_CONFLICT:
        return prior
    prior_score = (prior.completeness, _SOURCE_RANK.get(prior.source, 0))
    incoming_score = (incoming.completeness, _SOURCE_RANK.get(incoming.source, 0))
    if incoming_score > prior_score:
        log.info("dedup_merge_upgraded", mmsi=incoming.mmsi,
                 from_source=prior.source, to_source=incoming.source,
                 completeness=incoming.completeness)
        return incoming
    return prior


PositionDeduper._resolve_conflict = _resolve_conflict

Because ingest mutates the stored value with the winning copy, the canonical record grows toward its most complete form as duplicates land, and only the first arrival for a key is ever forwarded downstream — later copies refine the retained record without re-emitting.

Step 4 — Guard against MMSI collisions and spoofing #

An MMSI is not a guaranteed unique key. Vessels are re-flagged and inherit a reused number, cheap transponders ship mis-programmed, and deliberate spoofing clones a legitimate identity. The tell is physics: two reports with the same MMSI in the same window at positions implausibly far apart imply an impossible speed. Run a sanity gate before dedup so a spoofed clone is quarantined rather than silently merged into a real vessel’s track. Also reject identifiers that are not ship stations — the MMSI MID (the first three digits) must fall in the 201–775 maritime range; base stations (00…), aids-to-navigation (99…), and SART beacons (970…) are structurally valid AIS but are not vessels.

from math import radians, sin, cos, asin, sqrt

MAX_SPEED_KN: float = float(os.environ.get("AIS_MAX_SPEED_KN", "60"))


def haversine_nm(a: PositionReport, b: PositionReport) -> float:
    r_nm = 3440.065
    dlat, dlon = radians(b.lat - a.lat), radians(b.lon - a.lon)
    h = sin(dlat / 2) ** 2 + cos(radians(a.lat)) * cos(radians(b.lat)) * sin(dlon / 2) ** 2
    return 2 * r_nm * asin(sqrt(h))


def is_ship_station(mmsi: int) -> bool:
    mid = mmsi // 1_000_000
    return 201 <= mid <= 775  # excludes base (00), AtoN (99), SART (970) prefixes


def spoof_check(prior: PositionReport | None, report: PositionReport) -> bool:
    """False => reject: non-ship MMSI or an implausible position jump."""
    if not is_ship_station(report.mmsi):
        log.warning("dedup_non_ship_mmsi", mmsi=report.mmsi, source=report.source)
        return False
    if prior is not None:
        dt_s = max(abs(report.epoch_second - prior.epoch_second), 1)
        implied_kn = haversine_nm(prior, report) / (dt_s / 3600)
        if implied_kn > MAX_SPEED_KN:
            log.warning("dedup_spoof_suspected", mmsi=report.mmsi,
                        implied_kn=round(implied_kn, 1), source=report.source)
            return False
    return True

Edge Cases & Carrier Deviations #

A specific, recurring deviation is the satellite latency skew: an S-AIS copy of a broadcast can arrive 4–8 seconds after the terrestrial copy because of store-and-forward downlink, so a window narrower than that skew lets the late copy slip past as a “new” observation and doubles the point. Size DEDUP_WINDOW_SECONDS to the worst-case aggregator delay you actually observe, and prefer the message’s embedded time tag over local arrival time so the two copies collapse onto the same epoch_second regardless of when they landed.

Verification & Testing #

Assert the two behaviours that matter — that overlapping copies collapse to one, and that the richer copy is retained — plus the spoof rejection, with pytest fixtures and structlog’s capture helper.

from datetime import datetime, timedelta, UTC

import pytest

from ais.dedup import PositionDeduper, PositionReport


def _rep(source: str, second: int, **over) -> PositionReport:
    base = dict(mmsi=257_045_000, msg_type=1, lat=59.91, lon=10.75,
                sog=12.3, cog=87.0, heading=88,
                received_utc=datetime(2026, 7, 17, 12, 0, second, tzinfo=UTC),
                source=source)
    return PositionReport(**{**base, **over})


def test_overlapping_copies_collapse_to_one() -> None:
    d = PositionDeduper()
    assert d.ingest(_rep("terrestrial", 3)) is not None   # first wins
    assert d.ingest(_rep("satellite", 3)) is None          # same second, dropped
    assert d.ingest(_rep("aggregator", 3)) is None


def test_more_complete_copy_is_retained() -> None:
    d = PositionDeduper()
    d.ingest(_rep("satellite", 3, heading=None, cog=None))   # sparse first
    d.ingest(_rep("terrestrial", 3))                         # full copy upgrades
    kept = d._seen[(257_045_000, 1, int(_rep("terrestrial", 3).epoch_second))]
    assert kept.source == "terrestrial" and kept.completeness == 3


def test_non_ship_mmsi_rejected() -> None:
    from ais.dedup import is_ship_station
    assert is_ship_station(257_045_000) is True
    assert is_ship_station(2_579_999) is False   # MID 002 base station

A passing run emits one JSON line per decision. A dropped duplicate reads {"event": "dedup_duplicate_dropped", "mmsi": 257045000, "epoch_second": 1784030403, "source": "satellite", "kept_source": "terrestrial", "level": "info", ...} — assert on the structured keys, never on a formatted string.

Frequently Asked Questions #

Related #

• Connecting to public AIS feeds with Python asyncio — the transport tier whose checksum-clean payloads feed this deduper
• AIS Data Stream Integration — the ingestion boundary and decode stage this stage sits inside
• Container Status Mapping Rules — how the de-duplicated track is fused with terminal gate events into container states
• Container Tracking & AIS Event Synchronization — the domain that depends on one canonical fix per vessel per second

Up: AIS Data Stream Integration — the parent discipline governing AIS ingestion, decoding, validation, and downstream handoff.