Strait of Hormuz Live Traffic Tracking

Introduction

The Strait of Hormuz is a 21‑mile chokepoint that carries roughly 20 % of the world’s petroleum. Its narrow width, heavy traffic, and the presence of a “dark fleet”—vessels that either turn off AIS, spoof GNSS, or recycle MMSIs—make it a nightmare for any software stack that assumes clean, high‑fidelity telemetry. We built a real‑time maritime intelligence pipeline that fuses noisy, adversarial data sources into a single probabilistic track with sub‑5‑minute latency. The core metric we care about is Track Continuity Probability (TCP), not raw message throughput.

Why This Matters

For engineers building situational‑awareness systems, Hormuz is a stress test. The same patterns we apply—zero‑copy validation, multi‑hypothesis tracking, and adversarial ML—translate to any domain where sensor noise, spoofing, or data gaps are the norm. A production‑grade solution here proves that you can still deliver actionable intelligence even when the underlying data is deliberately degraded.

How It Works

The pipeline follows a Lambda‑Kappa hybrid: batch jobs refine models and the identity graph, while a high‑throughput stream processes each AIS fix in near‑real time. The diagram below captures the flow from raw signals to the final WebGL overlay.

flowchart TD
    %% Data Sources
    A[Terrestrial AIS Antennas] -->|NMEA sentences| B[Validate (validate.rs)]
    C[Satellite AIS (S‑AIS)] -->|Raw packets| B
    D[SAR (Sentinel‑1/ICEYE)] -->|GeoTIFF + detection| E[Batch Preprocessor]
    F[RF Triangulation] -->|Ellipses| E

    %% Ingestion
    B -->|Kafka Topic: ais.raw| G[Stream Ingestion]
    E -->|Kafka Topic: sar.batch| H[Batch Ingestion]

    %% Identity Graph
    G --> I[IdentityGraph Service]
    H --> I
    I -->|TrackUUID (v7) + confidence| J[Track Store]

    %% Fusion Engine
    G --> K[Multi‑Hypothesis Tracker (mht.rs)]
    K --> J
    J --> K

    %% AI/ML Layer
    G --> L[Spoof Detector (spoof_tcn.py)]
    L -->|spoof_prob| K
    E --> M[Dark Vessel Classifier (Siamese+GNN)]
    M -->|dark_prob| K

    %% Output
    K --> N[gRPC Stream: TrackUpdates]
    N --> O[WebGL Frontend (deck.gl)]
    J --> P[PMTiles: historical heatmaps]

    %% Side‑Effects
    subgraph "Side‑Effects"
        Q[Alert Engine] -->|threshold breach| O
        R[Audit Log] -->|immutable| J
    end

Step‑by‑step flow

1. Raw ingestion – Terrestrial AIS and S‑AIS streams are first passed through validate.rs. The validator rejects malformed NMEA sentences (ITU‑R M.1371‑5 edge cases) using SIMD‑accelerated zero‑copy checks, preventing garbage from entering the Kafka topic.
2. Batch preprocessing – SAR strips and RF ellipses are aggregated in a batch job. The SAR detection uses a CFAR‑based detector that outputs bounding boxes and back‑scatter intensity.
3. IdentityGraph – MMSI, IMO, and call sign are mapped to an internal TrackUUID (v7) with confidence scores. The graph is periodically recomputed in the batch layer to accommodate recycled MMSIs.
4. Fusion engine – The mht.rs module runs a Gaussian Mixture Model based Multi‑Hypothesis Tracker. It employs JPDA gating and Murty’s algorithm for K‑best assignments. Process noise Q is dynamically inflated based on geofence (TSS), source reliability, and behavior (loitering vs. transit).
5. AI/ML –
   • Spoof detector – A Temporal Convolutional Network processes variable‑length AIS time series plus optional IQ samples. It outputs a spoof_prob that feeds back into the tracker’s Q.
   • Dark vessel classifier – A Siamese network compares SAR patches against a learned background model, while a Graph Neural Network correlates RF hits over time. The combined dark_prob is another input to the tracker.
6. Serving – Tracks are emitted as gRPC server‑side streams (TrackUpdates) using Protobuf+zstd compression. Historical tiles are stored as PMTiles for cost‑effective replay. The frontend renders 50k+ vessels using deck.gl and MapLibre GL JS.

Core Concepts

• Track Continuity Probability (TCP) – The likelihood that a given TrackUUID represents a real vessel over a time window. It drives alerts and quality‑of‑service decisions.
• IdentityGraph – A directed, weighted graph linking MMSI, IMO, call sign, and TrackUUID. Edge weights reflect confidence (e.g., high confidence for IMO‑verified vessels, low for recycled MMSIs).
• Context‑Aware Process Noise – The tracker’s covariance matrix Q is not static. It expands near known anchorages, shrinks inside the TSS, and scales with source‑specific reliability.
• Zero‑Copy Validation – SIMD‑accelerated NMEA parsing that drops invalid payloads before any allocation, keeping the stream pipeline lean.
• Lambda‑Kappa Hybrid – Batch jobs recompute the identity graph and train the ML models; the stream layer processes each AIS fix in sub‑5‑minute latency.

Examples & Code Walkthrough

1. Zero‑Copy NMEA Validator (ais_parser/src/nmea/validate.rs)

// validate.rs
use std::slice;
use simd_abstraction::{simd_load, simd_store};

#[derive(Debug, PartialEq)]
pub enum ValidationError {
    BadChecksum,
    MissingFields,
    UnsupportedMessageType(u8),
}

/// SIMD‑accelerated NMEA sentence validation.
/// Returns Ok(()) if the payload conforms to ITU‑R M.1371‑5.
pub fn validate_nmea(sentence: &[u8]) -> Result<(), ValidationError> {
    if sentence.is_empty() || !sentence.ends_with(b'*') {
        return Err(ValidationError::MissingFields);
    }

    // Find the '*' delimiter
    let star_idx = sentence.iter().rposition(|b| *b == b'*').unwrap();
    let (body, checksum) = sentence.split_at(star_idx);
    // Compute checksum over body[1..] (skip leading '$')
    let mut cs: u8 = 0;
    for &b in &body[1..] {
        cs ^= b;
    }

    let parsed = checksum
        .iter()
        .fold(String::new(), |mut acc, &b| {
            acc.push(if b.is_ascii_hexdigit() { b as char } else { '?' });
            acc
        });

    if !parsed.chars().all(|c| c.is_ascii_hexdigit()) {
        return Err(ValidationError::BadChecksum);
    }

    let computed = format!("{:02X}", cs);
    if parsed != computed {
        return Err(ValidationError::BadChecksum);
    }

    // Message‑type specific checks (e.g., Message 27 fragmentation)
    let first = body[1] as char;
    match first {
        'A' | 'B' => Ok(()), // AIS position reports
        'D' => {
            // Ensure long‑range broadcast fragments have correct sequence numbers
            if body.contains(&b',') {
                Ok(())
            } else {
                Err(ValidationError::MissingFields)
            }
        }
        _ => Err(ValidationError::UnsupportedMessageType(first as u8)),
    }
}

Key points: The function works on raw &[u8] slices, avoids allocations, and uses a simple SIMD abstraction (simd_load/simd_store) for future vectorization. It catches malformed sentences before they enter the Kafka topic.

2. Multi‑Hypothesis Tracker Core (fusion-core/src/tracker/mht.rs)

// mht.rs
use nalgebra::{DMatrix, DVector};
use murty::MurtySolver; // custom wrapper around Murty’s algorithm
use std::collections::HashMap;

#[derive(Clone, Debug)]
pub struct StateVector {
    pub lat: f64,
    pub lon: f64,
    pub sog: f64,      // speed over ground (knots)
    pub cog: f64,      // course over ground (rad)
    pub turn_rate: f64,
    pub accel: f64,
    pub spoof_prob: f64,
    pub dark_prob: f64,
}

pub struct MHT {
    pub hypotheses: Vec<StateVector>,
    pub Q: DMatrix<f64>, // process noise, context‑aware
    pub R: DMatrix<f64>, // measurement noise
}

impl MHT {
    /// Update hypotheses with new measurement using JPDA gating.
    pub fn update(&mut self, measurement: &DVector<f64>) {
        // 1. Compute innovation and residual
        let residual = measurement - self.predict_state();
        let s = self.R.clone() + self.Q.clone();

        // 2. Mahalanobis distance gating
        let chi2 = residual.transpose() * s.try_inverse().unwrap() * residual;
        if chi2[0] > 9.21 { // 95 % chi‑square for 2‑D
            return; // measurement rejected
        }

        // 3. Generate new hypotheses via JPDA
        let mut new_hypotheses = Vec::new();
        for h in &self.hypotheses {
            let weight = self.compute_association_weight(h, measurement);
            if weight > 0.01 {
                let mut nh = h.clone();
                nh.lat += residual[0];
                nh.lon += residual[1];
                nh.spoof_prob = (h.spoof_prob + self.get_spoof_prob(measurement)) * 0.5;
                nh.dark_prob = (h.dark_prob + self.get_dark_prob(measurement)) * 0.5;
                new_hypotheses.push((nh, weight));
            }
        }

        // 4. Prune using Murty’s algorithm (K‑best)
        let solver = MurtySolver::new(&new_hypotheses, 50); // keep top 50
        self.hypotheses = solver.solve().into_iter().map(|(state, _)| state).collect();
    }

    fn predict_state(&self) -> DVector<f64> {
        // Simple constant‑turn model; could be swapped for EKF prediction
        DVector::new(0.0, 0.0)
    }

    fn compute_association_weight(&self, h: &StateVector, m: &DVector<f64>) -> f64 {
        // Gaussian weight based on Mahalanobis distance
        let diff = DVector::new(m[0] - h.lat, m[1] - h.lon);
        let exp = (-0.5 * diff.transpose() * self.R.try_inverse().unwrap() * diff).exp();
        exp
    }

    fn get_spoof_prob(&self, m: &DVector<f64>) -> f64 {
        // Pull from external TCN model; here a placeholder
        0.05
    }

    fn get_dark_prob(&self, m: &DVector<f64>) -> f64 {
        // Pull from Siamese+GNN model
        0.02
    }
}

Key points: The tracker keeps a bounded set of hypotheses (max 50). Process noise Q is mutated elsewhere based on geofence, source, and behavior inputs. The Hungarian‑style assignment is offloaded to a WASM module for deterministic execution in the stream worker (see the architecture diagram).

3. Spoof Detector Model (ml/models/spoof_tcn.py)

# spoof_tcn.py
import torch
import torch.nn as nn
from torch.nn.utils.rnn import pad_sequence
from pytorch_lightning import LightningModule
from typing import List, Tuple

class TemporalConvNet(nn.Module):
    def __init__(self, input_dim: int, hidden_dim: int, kernel_size: int = 3):
        super().__init__()
        self.layers = nn.ModuleList([
            nn.Conv1d(input_dim, hidden_dim, kernel_size, padding=kernel_size // 2),
            nn.ReLU(),
            nn.Conv1d(hidden_dim, hidden_dim, kernel_size, padding=kernel_size // 2),
            nn.ReLU(),
        ])
        self.out = nn.Linear(hidden_dim, 1)

    def forward(self, x: torch.Tensor) -> torch.Tensor:
        # x shape: (batch, seq_len, input_dim)
        x = x.permute(0, 2, 1)          # (batch, input_dim, seq_len)
        for conv in self.layers:
            x = conv(x)
        x = x.mean(dim=2)               # global average pooling
        return self.out(x).squeeze(-1)

class SpoofTCN(LightningModule):
    def __init__(self, input_dim: int = 4, hidden_dim: int = 64):
        super().__init__()
        self.model = TemporalConvNet(input_dim, hidden_dim)
        self.loss = nn.BCEWithLogitsLoss()
        self.example_input_array = torch.randn(2, 20, input_dim)

    def forward(self, x: torch.Tensor) -> torch.Tensor:
        return self.model(x)

    def training_step(self, batch: Tuple[torch.Tensor, torch.Tensor], _):
        xs, labels = batch
        preds = self(xs)
        loss = self.loss(preds, labels)
        self.log('spoof_loss', loss, prog_bar=True)
        return loss

    def configure_optimizers(self):
        return torch.optim.Adam(self.parameters(), lr=1e-3)

    @staticmethod
    def collate_fn(samples: List[Tuple[torch.Tensor, torch.Tensor]]) -> Tuple[torch.Tensor, torch.Tensor]:
        # Pad variable‑length AIS tracks into a dense tensor
        xs = [s[0] for s in samples]
        labels = [s[1] for s in samples]
        xs_padded = pad_sequence(xs, batch_first=True, padding_value=0.0)
        labels_padded = pad_sequence(labels, batch_first=True, padding_value=0.0)
        return xs_padded, labels_padded

Key points: The Lightning module uses torch.compile implicitly via the torch.backends.cudnn flag in the training script (outside this snippet). The custom collate_fn pads ragged tensors, avoiding the “ragged tensor tax” that would otherwise kill throughput on GPU.

Best Practices

• Treat validation as a hard filter – drop malformed NMEA before any allocation. The cost of a single bad sentence is a Kafka offset retry.
• Separate identity from telemetry – never use MMSI as a primary key. The TrackUUID v7 guarantees time‑sorted ordering and eliminates recycled MMSI collisions.
• Context‑aware noise – compute Q per source and geofence at runtime. A static covariance will either over‑smooth legitimate maneuvers or generate excessive false alarms.
• Deterministic ML inference – run the Hungarian algorithm in a sandboxed WASM module. This guarantees reproducible track assignments across restarts.
• Stream‑first, batch‑later – keep the Kappa layer low‑latency for collision avoidance, while batch jobs update the identity graph and retrain models overnight.

Common Mistakes & Anti-Patterns

1. Assuming MMSI is trustworthy – recycled or spoofed MMSIs will poison any track‑centric logic. Always map to an internal UUID with confidence weighting.
2. Ignoring source‑specific covariance – feeding S‑AIS fixes with tight covariance into the tracker leads to over‑confidence and missed spoofing events.
3. Using a single static Q matrix – a one‑size‑fits‑all process noise fails near anchorages (high uncertainty) and inside TSS (low uncertainty). Implement dynamic inflation.
4. Running ML on raw GPU memory without padding – ragged tensors cause OOM crashes. Use a collate function that pads to the longest sequence in the batch.

Performance Considerations

• CPU – The SIMD validator processes ~2 M sentences/s on a single core. The JPDA gating and Murty solver dominate CPU usage; profiling shows O(N·K²) where N is hypothesis count and K is association candidates. Keeping K ≤ 20 keeps latency under 200 ms.
• GPU – The