Mapping ISO container status codes to internal states #

This guide solves one precise task: deterministically translating the external status qualifiers that arrive on ISO 6346 equipment — SMDG movement codes, UN/EDIFACT CODECO/COARRI status qualifiers, and carrier-specific strings — into a single closed vocabulary of internal lifecycle states, with a defined precedence order and an audit trail for every resolution.

Architecture Alignment #

This task is the innermost resolver inside the Container Status Mapping Rules discipline, which itself sits under the Container Tracking & AIS Event Synchronization domain. That parent layer owns the end-to-end correlation graph that fuses vessel telemetry with landside events; this page owns only the leaf function that turns one raw status token into one enum value. Everything upstream — protocol de-enveloping, timestamp normalisation, (mmsi, timestamp) de-duplication — has already run by the time a code reaches this resolver, so the function is deliberately stateless and pure: same input token, same internal state, every time. Its output feeds the confidence-scored fallback chain that the correlation layer hands to the Port Call Workflow Design state machine.

The resolver applies three tiers in strict precedence — exact match, then vendor alias, then regex fallback — and only drops to UNKNOWN when all three miss. Precedence matters: an exact SMDG code must never be overridden by a looser regex that happens to also match, or a CUSTOMS_HOLD box could be silently reclassified as available.

Prerequisites & Environment Setup #

The resolver is pure-Python and needs no network credentials, but production deployments pin a version-controlled code map so that a qualifier’s meaning is diffable across releases.

• Python 3.11+ — for enum.StrEnum and X | None union syntax used below.
• structlog>=24.1 — structured JSON logging; a bare print() is unacceptable for an audit trail.
• pytest>=8.0 — for the fixture-driven verification in the final section.
• CODE_MAP_REGISTRY (env var) — path to the checked-in JSON/YAML file holding the exact-match matrix, loaded once at process start rather than per event.
• Reference standards — ISO 6346:2022 for the equipment identifier, the SMDG movement/status code list, and UN/EDIFACT data element 8249 (status of an event) for the CODECO/COARRI qualifiers.

python -m venv .venv && source .venv/bin/activate
pip install "structlog>=24.1" "pytest>=8.0"
export CODE_MAP_REGISTRY=/etc/maritime/status_code_map.json

The canonical inbound qualifiers this resolver must cover are stable across most carriers. The matrix below is the minimum contract before any alias or regex tier runs:

Step-by-step Implementation #

Each step is runnable in isolation; together they compose the tiered resolver drawn above.

Step 1 — Define the closed internal-state vocabulary. Use a StrEnum so the resolved value serialises directly into log lines and downstream events without a .value dance.

from __future__ import annotations

from enum import StrEnum


class InternalState(StrEnum):
    AVAILABLE = "AVAILABLE"
    IN_TRANSIT = "IN_TRANSIT"
    HELD_CUSTOMS = "HELD_CUSTOMS"
    MAINTENANCE = "MAINTENANCE"
    UNKNOWN = "UNKNOWN"

Step 2 — Build the version-controlled exact-match matrix. Keep SMDG movement codes and EDIFACT 8249 status qualifiers in one dictionary loaded from CODE_MAP_REGISTRY. This is the highest-precedence tier and the only one that should carry standardised codes.

EXACT_MAP: dict[str, InternalState] = {
    "44": InternalState.IN_TRANSIT,    # COARRI load
    "45": InternalState.IN_TRANSIT,    # COARRI discharge + rehandle
    "AE": InternalState.AVAILABLE,     # SMDG empty yard move
    "CH": InternalState.HELD_CUSTOMS,  # SMDG customs hold
    "MT": InternalState.AVAILABLE,
    "FULL": InternalState.IN_TRANSIT,
}

Step 3 — Layer a vendor alias dictionary. Carriers ship non-standard tokens (YD, GATE_IN, IMPND) that are not in any published list but are stable per partner. Aliases resolve after exact codes so a partner override can never mask a canonical qualifier.

ALIAS_MAP: dict[str, InternalState] = {
    "YD": InternalState.AVAILABLE,      # carrier "yard, empty"
    "GATE_IN": InternalState.IN_TRANSIT,
    "IMPND": InternalState.HELD_CUSTOMS,
    "SVC": InternalState.MAINTENANCE,
}

Step 4 — Precompile a regex fallback for format drift. Truncation, locale variants, and typos never reach the exact tier. Precompile the patterns once at construction time so the per-event cost is a scan, not a recompile. Anchor short tokens like MT to boundaries so they cannot match a substring of FORMAT or GMT.

import re

FALLBACK_PATTERNS: list[tuple[re.Pattern[str], InternalState]] = [
    (re.compile(r"(?i)cust.*hold|cstm.*det|impound"), InternalState.HELD_CUSTOMS),
    (re.compile(r"(?i)repair|maint|svc|damaged"), InternalState.MAINTENANCE),
    (re.compile(r"(?i)full|laden|loaded|stow|onboard"), InternalState.IN_TRANSIT),
    (re.compile(r"(?i)empty|vacant|stripped|(?:^|_)mt(?:_|$)"), InternalState.AVAILABLE),
]

Step 5 — Resolve statelessly with strict precedence and audit logging. Normalise, then walk the three tiers in order. Every non-exact resolution and every miss is logged with the raw token and source system so the decision is reconstructable during a customs audit.

import structlog

log = structlog.get_logger()


class StatusMapper:
    """Stateless, per-event resolver. Exact → alias → regex → UNKNOWN."""

    def __init__(
        self,
        exact_map: dict[str, InternalState],
        alias_map: dict[str, InternalState],
        patterns: list[tuple[re.Pattern[str], InternalState]],
    ) -> None:
        self.exact_map = exact_map
        self.alias_map = alias_map
        self.patterns = patterns

    def resolve(self, raw_code: str | None, source_system: str = "UNKNOWN") -> InternalState:
        if not raw_code or not raw_code.strip():
            log.warning("empty_status_payload", source=source_system)
            return InternalState.UNKNOWN

        cleaned = raw_code.strip().upper().replace("-", "_").replace(" ", "_")

        if cleaned in self.exact_map:                      # tier 1: canonical
            return self.exact_map[cleaned]
        if cleaned in self.alias_map:                      # tier 2: vendor alias
            log.info("alias_matched", raw=raw_code, source=source_system)
            return self.alias_map[cleaned]
        for pattern, state in self.patterns:               # tier 3: drift fallback
            if pattern.search(cleaned):
                log.info("regex_fallback", raw=raw_code, resolved=state, source=source_system)
                return state

        log.error("unresolvable_status_code", raw=raw_code, source=source_system)
        return InternalState.UNKNOWN

Step 6 — Gate the resolved state before it actuates operations. A resolved UNKNOWN or HELD_CUSTOMS must not advance a box to crane scheduling or a customs release. Emit an immutable audit record keyed on a deterministic SHA-256 hash and hard-block indeterminate states.

import hashlib
import json
from datetime import datetime, timezone


def gate_transition(container_id: str, raw_payload: dict, resolved: InternalState) -> bool:
    payload_hash = hashlib.sha256(
        json.dumps(raw_payload, sort_keys=True, default=str).encode()
    ).hexdigest()
    log.info(
        "state_transition_eval",
        container_id=container_id,
        resolved_state=resolved,
        payload_hash=payload_hash,
        ts_utc=datetime.now(timezone.utc).isoformat(),
    )
    if resolved in (InternalState.UNKNOWN, InternalState.HELD_CUSTOMS):
        log.error("compliance_gate_blocked", container_id=container_id, state=resolved)
        return False
    return True

Edge Cases & Carrier Deviations #

• MT substring collisions. Without the (?:^|_)mt(?:_|$) anchor, the empty-container fallback matches tokens like FORMAT or GMT and flips a laden box to AVAILABLE. Always anchor two-letter aliases to token boundaries.
• Discharge-then-available ambiguity. EDIFACT 45 means discharged and rehandled, not gated out. Mapping it to AVAILABLE releases a box that is still laden in the yard. Keep discharge in IN_TRANSIT and let the gate-out qualifier (GO) drive availability.
• Proprietary qualifier extensions. A carrier may extend CODECO/COARRI with a qualifier absent from UN/EDIFACT 8249. Never guess the nearest state — return UNKNOWN, log it, and raise a code-map update ticket. This mirrors the PENDING_VERIFICATION discipline in the parent Container Status Mapping Rules resolver.
• Locale-cased tokens. Terminals in non-English ports emit accented or lowercase variants (vacío, lleno). Normalise to upper-ASCII at the boundary and register the locale token as an explicit alias rather than widening a regex.
• Codes that arrive without an identifier. A status with a failed ISO 6346 check digit is still parseable but must be quarantined, not dropped — the correlation layer keeps tracking it by AIS join. Identifier structure is owned by Container Hierarchy Data Models; this resolver only trusts the status token.

Verification & Testing #

Assert precedence explicitly — the tests that matter are the ones proving a regex can never override an exact code, and that an unknown token fails closed.

import pytest

MAPPER = StatusMapper(EXACT_MAP, ALIAS_MAP, FALLBACK_PATTERNS)


@pytest.mark.parametrize(
    "raw,expected",
    [
        ("CH", InternalState.HELD_CUSTOMS),        # exact SMDG
        ("ch", InternalState.HELD_CUSTOMS),        # case-normalised
        ("YD", InternalState.AVAILABLE),           # vendor alias
        ("customs-hold-72h", InternalState.HELD_CUSTOMS),  # regex drift
        ("FORMAT_ERR", InternalState.UNKNOWN),     # MT anchor must NOT match
        ("", InternalState.UNKNOWN),               # fail closed
        (None, InternalState.UNKNOWN),
    ],
)
def test_resolution_precedence(raw: str | None, expected: InternalState) -> None:
    assert MAPPER.resolve(raw, source_system="TEST") == expected


def test_gate_blocks_indeterminate() -> None:
    assert gate_transition("MSKU1234565", {"code": "??"}, InternalState.UNKNOWN) is False
    assert gate_transition("MSKU1234565", {"code": "AE"}, InternalState.AVAILABLE) is True

A passing regex_fallback resolution emits a structured line the audit pipeline can index directly:

{"event": "regex_fallback", "raw": "customs-hold-72h", "resolved": "HELD_CUSTOMS", "source": "TEST", "level": "info"}

Frequently Asked Questions #

Related topics #

• Container Status Mapping Rules — the parent correlation graph and confidence-scored fallback chain that consumes this resolver.
• Terminal API Polling Strategies — where the raw status tokens are pulled from TOS REST endpoints before mapping.
• AIS Data Stream Integration — the vessel-telemetry side that the mapped state is correlated against.
• Threshold Tuning for Alerts — adaptive alerting on synchronization drift, downstream of resolution.
• Port Call Workflow Design — the state machine that acts on the resolved, gated container state.

↑ Back to Container Status Mapping Rules.