Implementing ISPS security zones in routing APIs #

This guide solves one precise task: turning International Ship and Port Facility Security (ISPS) Code perimeters into active routing constraints inside a Python routing API — normalising heterogeneous position payloads at ingress, evaluating vessel proximity against indexed security-zone polygons, and returning graduated compliance actions (advisory, recalculate, block) with deterministic, audit-ready fallbacks.

Architecture Alignment #

This task is the enforcement leaf of the Maritime Security Boundary Setup discipline, which sits under the Core Maritime Architecture & Taxonomy framework. That parent layer defines every geofence, restricted berth, and MARSEC level as a versioned policy object; this page owns only the runtime function that answers one question per request — may this vessel’s projected track proceed, and if not, what is the alternative? Everything upstream (zone versioning, credential resolution, MARSEC escalation) has already been decided by the time a request reaches this endpoint, so the routing decision is a pure function of a normalised position and the indexed zone set. Its output propagates downstream: a BLOCK or RECALCULATE must reach the Port Call Workflow Design state machine so berth windows and pilot dispatch stay consistent, and the resulting stowage impact flows into Bill of Lading Schema Mapping and Container Hierarchy Data Models.

Modelling the perimeter as a routing constraint rather than a map overlay is what keeps the endpoint compliant with SOLAS Chapter XI-2: every crossing decision is computed, logged, and reproducible from the request payload plus the pinned zone version.

Prerequisites & Environment Setup #

The routing evaluator is CPU-bound and pins a version-controlled zone registry so a perimeter’s geometry is diffable across releases and reproducible in an audit.

• Python 3.11+ — for X | None union syntax and modern dataclass semantics used below.
• pydantic>=2.6 — strict ingress schema contracts with field and model validators.
• structlog>=24.1 — structured JSON logging; a bare print() is unacceptable for a security audit trail.
• shapely>=2.0 — geometry operations and make_valid polygon repair.
• rtree>=1.2 — libspatialindex R-tree bindings for candidate pruning.
• pytest>=8.0 — for the fixture-driven verification in the final section.
• ISPS_ZONE_REGISTRY (env var) — path to the checked-in GeoJSON holding zone polygons and their MARSEC level, loaded once at process start rather than per request.
• Reference standards — the IMO ISPS Code Part A (MARSEC levels 1–3, Section 10 access control) and SOLAS Chapter XI-2. Position inputs are assumed to have already passed the AIS Data Stream Integration normalisation layer.

python -m venv .venv && source .venv/bin/activate
pip install "pydantic>=2.6" "structlog>=24.1" "shapely>=2.0" "rtree>=1.2" "pytest>=8.0"
export ISPS_ZONE_REGISTRY=/etc/maritime/isps_zones.geojson

Step-by-step Implementation #

Step 1 — Enforce a strict schema contract at ingress #

Format drift is the most persistent failure mode: AIS transponders, port authority shapefiles, and vessel manifest APIs serialize coordinates in mixed CRS, apply inconsistent timestamp formats, and mutate zone schemas between port calls. Reject malformed payloads before they reach the geospatial engine, and coerce everything survivable into a single WGS-84 / UTC contract. MARSEC levels are constrained to 1, 2, or 3 per ISPS Code Part A.

import logging
from datetime import datetime, timezone
from pydantic import BaseModel, field_validator, model_validator
from pydantic_core import PydanticCustomError
import structlog

logger = structlog.get_logger("isps_routing.ingress")

class RoutingIngressPayload(BaseModel):
    vessel_mmsi: int
    latitude: float
    longitude: float
    timestamp_utc: datetime
    marsec_level: int  # 1, 2, or 3 per ISPS Code Part A

    @field_validator("latitude", "longitude")
    @classmethod
    def clamp_wgs84(cls, v: float, info) -> float:
        # Handle real-world quirk: AIS feeds occasionally drift past ±90/±180.
        # The validation context tells us which field is being processed.
        if info.field_name == "latitude":
            return max(-90.0, min(90.0, v))
        return max(-180.0, min(180.0, v))

    @field_validator("timestamp_utc", mode="before")
    @classmethod
    def coerce_iso8601(cls, v: "str | datetime | int | float") -> datetime:
        if isinstance(v, datetime):
            return v.astimezone(timezone.utc) if v.tzinfo else v.replace(tzinfo=timezone.utc)
        if isinstance(v, (int, float)):  # epoch milliseconds
            return datetime.fromtimestamp(v / 1000, tz=timezone.utc)
        # Handle mixed strings: ISO 8601, "YYYY/MM/DD HH:MM[:SS]", or epoch-ms text
        text = str(v).strip()
        if text.isdigit():
            return datetime.fromtimestamp(int(text) / 1000, tz=timezone.utc)
        try:
            dt = datetime.fromisoformat(text.replace("Z", "+00:00"))
        except ValueError:
            for fmt in ("%Y/%m/%d %H:%M:%S", "%Y/%m/%d %H:%M"):
                try:
                    dt = datetime.strptime(text, fmt).replace(tzinfo=timezone.utc)
                    break
                except ValueError:
                    continue
            else:
                raise PydanticCustomError("timestamp_format", "Non-ISO8601 timestamp rejected per SOLAS audit requirements")
        return dt.astimezone(timezone.utc)

    @model_validator(mode="after")
    def validate_marsec_and_log(self):
        if self.marsec_level not in (1, 2, 3):
            logger.warning("invalid_marsec", mmsi=self.vessel_mmsi, level=self.marsec_level)
            raise PydanticCustomError("marsec_range", "MARSEC must be 1, 2, or 3 per ISPS Code")
        logger.info("payload_normalized", mmsi=self.vessel_mmsi, ts=self.timestamp_utc.isoformat())
        return self

Step 2 — Index ISPS zones with an R-tree and repair invalid geometry #

Loading high-resolution boundary polygons into memory for every request exhausts heap space and triggers garbage-collection pauses that breach port SLAs during peak vessel windows. Precompute R-tree bounds once, repair self-intersecting geometry on load, and defer exact intersection to candidate matches only — this cuts the memory footprint by 60–80% while keeping query latency deterministic.

import structlog
from shapely.geometry import Point, Polygon, shape
from shapely.validation import make_valid
from rtree import index
from typing import Dict, List, Tuple

logger = structlog.get_logger("isps_routing.spatial")

class ISPSZoneIndex:
    def __init__(self):
        self._idx = index.Index()
        self._polygons: Dict[int, Polygon] = {}
        self._marsec_cache: Dict[int, int] = {}

    def load_zones(self, zone_id: int, geojson_feat: dict, marsec: int) -> None:
        """Lazy-load zones with self-intersection repair and R-tree insertion."""
        try:
            raw_poly = shape(geojson_feat)
            # Real-world quirk: port shapefiles often contain bowties or duplicate vertices
            valid_poly = make_valid(raw_poly)
            self._polygons[zone_id] = valid_poly
            self._marsec_cache[zone_id] = marsec
            self._idx.insert(zone_id, valid_poly.bounds)
            vertex_count = len(valid_poly.exterior.coords) if valid_poly.geom_type == "Polygon" else None
            logger.debug("zone_loaded", zone_id=zone_id, marsec=marsec, vertices=vertex_count)
        except Exception as e:
            logger.error("zone_load_failed", zone_id=zone_id, error=str(e))

Step 3 — Evaluate proximity against R-tree candidates only #

The R-tree bounding-box query returns a small set of candidate zones; only those get the expensive exact-topology check. Zones whose MARSEC level exceeds the vessel’s clearance are skipped, and anything beyond a 1 km routing horizon is dropped so the response stays scoped to zones that can actually constrain the track.

    def evaluate_proximity(self, lat: float, lon: float, marsec_filter: int) -> List[Tuple[int, float]]:
        """Return candidate zones within R-tree bounds, then run exact intersection."""
        point = Point(lon, lat)
        candidates = []
        for zone_id in self._idx.intersection(point.bounds):
            if self._marsec_cache.get(zone_id, 0) > marsec_filter:
                continue
            # Exact topology check only on R-tree candidates to preserve memory
            if self._polygons[zone_id].intersects(point):
                candidates.append((zone_id, 0.0))  # 0.0m = inside
            else:
                dist = point.distance(self._polygons[zone_id])
                if dist < 1000.0:  # Only track zones within 1km for routing context
                    candidates.append((zone_id, dist))
        candidates.sort(key=lambda x: x[1])
        logger.info("spatial_eval", lat=lat, lon=lon, candidates=len(candidates))
        return candidates

Step 4 — Gate the route with graduated MARSEC thresholds #

ISPS zones expand and contract with threat intelligence and port authority directives, so proximity thresholds are configurable and trigger graduated actions: advisory at 500 m, mandatory recalculation at 200 m, hard block at 50 m. When a track intersects a restricted perimeter the API returns an actionable fallback route rather than a generic error, preserving operational continuity while satisfying SOLAS Chapter XI-2. The fallback selection respects MARSEC escalation so a MARSEC-3 diversion is never silently downgraded.

import structlog
from dataclasses import dataclass
from typing import Optional, List

logger = structlog.get_logger("isps_routing.compliance")

@dataclass
class ComplianceAction:
    severity: str  # CLEAR, ADVISORY, RECALCULATE, BLOCK
    message: str
    fallback_route_id: Optional[str] = None

class ThresholdEngine:
    def __init__(self, advisory_m: float = 500.0, recalc_m: float = 200.0, block_m: float = 50.0):
        self.advisory_m = advisory_m
        self.recalc_m = recalc_m
        self.block_m = block_m

    def evaluate(self, zone_id: int, distance_m: float, marsec: int) -> ComplianceAction:
        """Apply graduated thresholds per ISPS Code Part A, Section 10."""
        if distance_m <= self.block_m:
            logger.warning("compliance_block", zone=zone_id, dist=distance_m, marsec=marsec,
                           reason="Hard restriction per port authority directive")
            return ComplianceAction("BLOCK", f"Vessel prohibited within {self.block_m}m of Zone {zone_id}")

        if distance_m <= self.recalc_m:
            # Trigger fallback routing logic to avoid deadlocks in congested terminals
            fallback = self._generate_fallback(zone_id, marsec)
            logger.warning("compliance_recalc", zone=zone_id, dist=distance_m, fallback=fallback)
            return ComplianceAction("RECALCULATE", f"Route deviation required. Approaching Zone {zone_id} boundary.", fallback)

        if distance_m <= self.advisory_m:
            logger.info("compliance_advisory", zone=zone_id, dist=distance_m)
            return ComplianceAction("ADVISORY", f"Security zone proximity alert: {zone_id}")

        return ComplianceAction("CLEAR", "No active security constraints detected")

    def _generate_fallback(self, zone_id: int, marsec: int) -> Optional[str]:
        """Deterministic fallback route selection respecting MARSEC escalation."""
        if marsec == 3:
            return f"route_{zone_id}_marsec3_diversion"
        return f"route_{zone_id}_standard_alternate"

Edge Cases & Carrier Deviations #

• WGS-84 overflow. AIS feeds intermittently emit latitude beyond ±90 or longitude beyond ±180 when a transponder resets. Clamp at ingress (Step 1) — never let a wild coordinate reach evaluate_proximity, where it would silently place the vessel outside every zone bound and return a false CLEAR.
• Self-intersecting zone polygons. Port authority shapefiles routinely contain bowties or duplicate vertices; raw shapely intersection on them raises TopologyException. make_valid on load (Step 2) converts them to valid multipolygons instead of crashing a request mid-transit.
• CRS mismatch. Some registries ship zone geometry in a projected national grid rather than EPSG:4326. Reproject at load time; a metres-vs-degrees mismatch turns a 50 m block radius into thousands of kilometres.
• Epoch-ms versus ISO-8601 timestamps. Manifest APIs and AIS gateways disagree on time encoding. The coerce_iso8601 validator accepts both and normalises to UTC so dwell and escalation windows compare correctly.
• MARSEC volatility mid-request. A port can escalate from MARSEC 1 to 2 while a vessel is inbound. Cache eviction keyed to MARSEC change (rather than a fixed TTL) ensures the next evaluate_proximity call reflects the new level immediately.
• MMSI collisions. Reused or spoofed MMSIs mean identity is not a routing key — the endpoint gates on position and clearance, not on trusting the transmitted MMSI, and logs the vessel_mmsi only for correlation.

Verification & Testing #

Assert the whole path — ingress coercion, candidate pruning, and threshold selection — against a fixed zone fixture. The block/recalculate/advisory boundaries are exact, so test the values on either side of each threshold.

import pytest
from shapely.geometry import mapping, box

@pytest.fixture
def zone_index() -> "ISPSZoneIndex":
    idx = ISPSZoneIndex()
    # A 0.001° square restricted berth (~111 m) tagged MARSEC 2
    idx.load_zones(101, mapping(box(0.0, 0.0, 0.001, 0.001)), marsec=2)
    return idx


def test_ingress_clamps_and_coerces() -> None:
    p = RoutingIngressPayload(
        vessel_mmsi=477123456, latitude=95.0, longitude=200.0,
        timestamp_utc="1719792000000", marsec_level=2,
    )
    assert p.latitude == 90.0 and p.longitude == 180.0
    assert p.timestamp_utc.tzinfo is not None


def test_higher_marsec_zone_is_filtered(zone_index) -> None:
    # Vessel cleared only to MARSEC 1 must not see the MARSEC-2 zone
    assert zone_index.evaluate_proximity(0.0005, 0.0005, marsec_filter=1) == []


def test_threshold_ladder() -> None:
    eng = ThresholdEngine()
    assert eng.evaluate(101, 40.0, marsec=2).severity == "BLOCK"
    assert eng.evaluate(101, 150.0, marsec=3).fallback_route_id == "route_101_marsec3_diversion"
    assert eng.evaluate(101, 450.0, marsec=2).severity == "ADVISORY"
    assert eng.evaluate(101, 900.0, marsec=2).severity == "CLEAR"

A RECALCULATE decision emits a structured line the compliance pipeline can index directly:

{"event": "compliance_recalc", "zone": 101, "dist": 150.0, "fallback": "route_101_marsec3_diversion", "level": "warning"}

Frequently Asked Questions #

Related topics #

• Maritime Security Boundary Setup — the parent discipline that models zones, clearances, and audit routing as typed policy objects.
• Port Call Workflow Design — the state machine that consumes block and recalculate decisions to keep berth windows consistent.
• Bill of Lading Schema Mapping — cargo documentation that must adjust when a security routing decision changes stowage.
• Container Hierarchy Data Models — the equipment model whose crane and slot assignments react to routing constraints.
• AIS Data Stream Integration — the upstream position feed that this endpoint assumes has already been normalised.

↑ Back to Maritime Security Boundary Setup.