Automating Port Call Sequence Validation #

Automating port call sequence validation means turning a stream of arrival, berthing, and departure signals into a deterministic pass/fail decision — enforcing temporal tolerances and regulatory gates before any event is allowed to advance a vessel through the berth queue. This page shows how to build that validator in Python so that format drift, memory pressure, ad-hoc thresholds, and silent compliance gaps can never corrupt a terminal’s schedule.

Architecture Alignment #

Sequence validation is the guard that sits directly in front of the Port Call Workflow Design state machine: it decides whether an incoming signal is trustworthy enough to trigger a milestone transition, and it belongs to the broader Core Maritime Architecture & Taxonomy that owns how heterogeneous maritime data is normalized and gated. Upstream, it depends on the same identifiers established during Bill of Lading Schema Mapping and the equipment references resolved by Container Hierarchy Data Models; downstream, a validated PASS is what lets a berth slot be committed and a FAIL is what routes an event into an exception queue instead of the schedule. The validator never mutates state itself — it produces an auditable verdict that the state machine consumes.

The validator addresses four persistent failure modes that break naive port-call pipelines. Format drift occurs when a carrier changes a field’s shape without versioning — vessel_call_sign shifting from a plain string to a base64 payload silently breaks a regex parser. Memory bottlenecks appear when a pipeline loads an entire voyage history into memory to cross-reference high-frequency AIS pings, exhausting heap on an edge gateway. Threshold drift is the temptation to hardcode ±15 min constants that cannot flex for tide windows or terminal congestion. Compliance gaps are the quiet ones: an event that advances despite a missing SOLAS Verified Gross Mass (VGM) or an unaligned ISPS security level. A stream-oriented, typed validator closes all four.

Prerequisites & Environment Setup #

The validator targets Python 3.11+ (for datetime.UTC and the X | None type syntax) and a small, production-grade dependency set. Bare print() is unacceptable in this niche; all output is structured JSON via structlog so it lands cleanly in a SIEM or Logstash pipeline.

python -m venv .venv && source .venv/bin/activate
pip install "pydantic>=2.6" "structlog>=24.1" "pyais>=2.6" pytest

Thresholds are read from the environment rather than hardcoded so they can be tuned per terminal without a redeploy — the same discipline applied in Threshold Tuning for Alerts. Live position signals feeding the validator come from an AIS Data Stream Integration consumer, which is why MMSI sanitisation is the first gate.

Step-by-step Implementation #

Build the validator as six composable steps. Each snippet is runnable in isolation and can be unit-tested on its own fixture.

1. Configure structured logging and the event contract #

Model the event with Pydantic so a malformed payload is rejected at construction, not deep inside a scheduling calculation. Every field is typed and every constant lives in a frozen threshold object.

import os
import structlog
from datetime import datetime, timezone
from enum import Enum
from pydantic import BaseModel, Field

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


class ComplianceGate(str, Enum):
    SOLAS_VGM = "SOLAS_VGM"
    ISPS_SECURITY = "ISPS_SECURITY"
    CUSTOMS = "CUSTOMS"


class ValidationThresholds(BaseModel, frozen=True):
    eta_tolerance_min: int = 15
    etb_tolerance_min: int = 15
    etd_tolerance_min: int = 30


class PortCallEvent(BaseModel):
    mmsi: str
    unlocode: str
    event_code: str  # ETA, ETB, ATB, ETD, ATD
    scheduled_ts: datetime | None = None
    actual_ts: datetime | None = None
    vgm_declared_kg: float | None = None
    isps_level: int | None = None
    customs_cleared: bool = False

2. Sanitise identifiers at ingress #

A valid MMSI is exactly nine digits. Reject — never zero-pad — anything else, because a leading zero rewrites the Maritime Identification Digits and would silently reassign a vessel to another flag state. This is the single most common source of AIS MMSI collisions.

def sanitize_mmsi(raw: str) -> str | None:
    """Strip AIS transport padding; a valid MMSI is exactly 9 digits."""
    cleaned = raw.strip().replace(" ", "").replace("-", "")
    if not cleaned.isdigit() or len(cleaned) != 9:
        return None
    return cleaned

3. Normalise every timestamp to UTC #

Legacy EDIFACT translators emit naive local timestamps; AIS pings arrive Z-suffixed. Comparing the two without coercion produces phantom deviations of several hours. Force everything to UTC at the boundary, per ISO 8601.

def to_utc(ts: datetime | None, fallback: timezone = timezone.utc) -> datetime | None:
    if ts is None:
        return None
    return ts.replace(tzinfo=fallback) if ts.tzinfo is None else ts.astimezone(timezone.utc)

4. Apply the temporal tolerance check #

Select the tolerance by event code and compare the actual against the scheduled time. Because the thresholds live in ValidationThresholds, a congested terminal can widen its ETB window without touching this logic.

def temporal_deviation(
    event: PortCallEvent, thresholds: ValidationThresholds
) -> tuple[bool, float | None]:
    sched, actual = to_utc(event.scheduled_ts), to_utc(event.actual_ts)
    if not (sched and actual):
        return True, None  # nothing to compare yet
    delta_min = (actual - sched).total_seconds() / 60.0
    tolerance = {
        "ETB": thresholds.etb_tolerance_min, "ATB": thresholds.etb_tolerance_min,
        "ETA": thresholds.eta_tolerance_min, "ATA": thresholds.eta_tolerance_min,
    }.get(event.event_code, thresholds.etd_tolerance_min)
    return abs(delta_min) <= tolerance, round(delta_min, 2)

5. Enforce the compliance gates #

Compliance gating is a hard stop, not a warning. SOLAS Chapter VI regulation 2 mandates a verified gross mass before loading, and the ISPS Code requires the ship’s declared security level to be one of 1, 2, or 3. Each failed gate is recorded by name.

def failed_gates(event: PortCallEvent, required: list[ComplianceGate]) -> list[str]:
    failures: list[str] = []
    if ComplianceGate.SOLAS_VGM in required and not (event.vgm_declared_kg and event.vgm_declared_kg > 0):
        failures.append("VGM_MISSING")
    if ComplianceGate.ISPS_SECURITY in required and event.isps_level not in (1, 2, 3):
        failures.append("ISPS_INVALID")
    if ComplianceGate.CUSTOMS in required and not event.customs_cleared:
        failures.append("CUSTOMS_PENDING")
    return failures

6. Emit the audit verdict as a stream #

Tie the steps together in a generator so the validator processes one event at a time — bounded memory regardless of voyage-history size. Every event yields an immutable audit record and a structured log line; a FAIL is excluded from the berth queue rather than silently dropped.

from collections.abc import Iterator


def validate_sequence(
    events: Iterator[PortCallEvent],
    thresholds: ValidationThresholds,
    required_gates: list[ComplianceGate],
) -> Iterator[dict]:
    for event in events:
        mmsi = sanitize_mmsi(event.mmsi)
        if mmsi is None:
            log.warning("mmsi_invalid", mmsi=event.mmsi, unlocode=event.unlocode)
            continue
        temporal_ok, deviation_min = temporal_deviation(event, thresholds)
        gate_failures = failed_gates(event, required_gates)
        status = "PASS" if temporal_ok and not gate_failures else "FAIL"
        record = {
            "mmsi": mmsi,
            "sequence_id": f"{event.unlocode}_{event.event_code}",
            "status": status,
            "deviation_min": deviation_min,
            "gates_failed": gate_failures,
            "processed_utc": datetime.now(timezone.utc).isoformat(),
        }
        log.info("sequence_validated", **record) if status == "PASS" \
            else log.error("sequence_failed", **record)
        yield record

Edge Cases & Carrier Deviations #

Real port-call traffic breaks textbook assumptions. The failures below are the ones that recur across terminals and carriers.

An event that fails a compliance gate is a recoverable business exception — route it to a quarantine topic operations staff can triage. An event whose payload cannot be constructed at all (missing required fields, unparseable timestamp) is a permanent defect and belongs in a dead-letter queue. Mixing the two buries genuine corruption under recoverable noise. A FAIL that touches security posture must also be reconciled against the boundary rules defined in Maritime Security Boundary Setup.

Verification & Testing #

Assert correctness against fixtures that exercise each gate independently. The following pytest cases confirm a clean event passes, a missing VGM fails, and an out-of-tolerance berthing is caught.

from datetime import datetime, timedelta, timezone

BASE = datetime(2026, 7, 3, 8, 0, tzinfo=timezone.utc)
GATES = [ComplianceGate.SOLAS_VGM, ComplianceGate.ISPS_SECURITY, ComplianceGate.CUSTOMS]
TH = ValidationThresholds()


def _event(**over) -> PortCallEvent:
    base = dict(mmsi="219001234", unlocode="NLRTM", event_code="ETB",
                scheduled_ts=BASE, actual_ts=BASE + timedelta(minutes=5),
                vgm_declared_kg=24000.0, isps_level=1, customs_cleared=True)
    return PortCallEvent(**{**base, **over})


def test_clean_event_passes():
    (result,) = list(validate_sequence(iter([_event()]), TH, GATES))
    assert result["status"] == "PASS"
    assert result["gates_failed"] == []


def test_missing_vgm_fails():
    (result,) = list(validate_sequence(iter([_event(vgm_declared_kg=None)]), TH, GATES))
    assert result["status"] == "FAIL"
    assert "VGM_MISSING" in result["gates_failed"]


def test_late_berthing_fails():
    late = _event(actual_ts=BASE + timedelta(minutes=40))  # ETB tolerance is 15
    (result,) = list(validate_sequence(iter([late]), TH, GATES))
    assert result["status"] == "FAIL"
    assert result["deviation_min"] == 40.0


def test_bad_mmsi_is_dropped():
    assert list(validate_sequence(iter([_event(mmsi="12345")]), TH, GATES)) == []

A passing run emits one JSON line per validated event. The test_missing_vgm_fails case produces an error-level record shaped like this, ready for a SIEM query on gates_failed:

{"event": "sequence_failed", "level": "error", "timestamp": "2026-07-03T08:00:00Z",
 "mmsi": "219001234", "sequence_id": "NLRTM_ETB", "status": "FAIL",
 "deviation_min": 5.0, "gates_failed": ["VGM_MISSING"]}

Frequently Asked Questions #

Related #

• Port Call Workflow Design — the state machine that consumes this validator’s PASS/FAIL verdicts
• AIS Data Stream Integration — the live position feed whose MMSI signals seed validation
• Threshold Tuning for Alerts — how to set the tolerance windows this validator enforces
• Maritime Security Boundary Setup — the ISPS boundary rules a security-related FAIL reconciles against
• Bill of Lading Schema Mapping — the upstream layer that establishes the identifiers validation trusts

Up: Port Call Workflow Design — the parent workflow this validation task belongs to.