Validating BAPLIE against ISO 9735 in Python #

A BAPLIE — the SMDG bayplan message that tells a terminal exactly which stowage cell every container on a vessel occupies — arrives as a UN/EDIFACT interchange, and one malformed envelope can hand a stowage planner a phantom slot or a mis-weighted box on a live sailing. Validating BAPLIE against ISO 9735 in Python means enforcing the interchange syntax rules deterministically before any cell is trusted: the service-string advice, the UNB/UNZ envelope balance, the UNH/UNT message counters, and the mandatory LOC/EQD/MEA/DIM segment groups that describe each container. This page builds a typed, two-tier validator — structural syntax first, container semantics second — so a bad stow position is quarantined at the gate instead of surfacing on the bay plan mid-loadout.

Architecture Alignment #

This task is a concrete instance of the rule engine described by Schema Validation Frameworks, the topic area inside the Document Ingestion & EDI Parsing Workflows domain that decides whether a payload is structurally sound before it consumes downstream compute. BAPLIE sits at the same ingestion boundary as the aggregated manifests handled by IFCSUM EDI Message Parsing — both are UN/EDIFACT interchanges governed by the ISO 9735 syntax rules — but BAPLIE’s payload is spatial rather than commercial: every message body is a list of occupied and empty stow cells, and each EQD equipment identifier must resolve into the Container Hierarchy Data Models topology the rest of the platform tracks against. The validator here owns one gate: prove the interchange is syntactically legal per ISO 9735, then prove each container reference is semantically valid, and route anything that fails to quarantine rather than the bay plan.

Prerequisites & Environment Setup #

The validator targets Python 3.11+ for its improved dataclass and typing ergonomics and exception groups. It has no heavy dependencies — parsing an EDIFACT interchange is string work — so the install footprint is just the structured-logging stack plus pytest for the test suite:

pip install structlog==24.* pytest

Structured JSON logging is non-negotiable here: every rejection must be replayable during a terminal dispute or a port-state-control audit, so a UNT counter mismatch has to be queryable by control reference, not buried in a formatted string. Configure structlog once, at process start, to emit JSON:

import structlog

structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.JSONRenderer(),
    ],
)
log = structlog.get_logger("maritime.baplie")

The validator reads a few environment variables so the same code runs strict in production and tolerant during carrier onboarding:

Step-by-step Implementation #

Step 1 — Read the UNA service string and tokenize the interchange #

ISO 9735 lets an interchange redefine its own delimiters through the optional UNA service string advice — six characters after the UNA tag that set the component separator, element separator, decimal notation, release character, a reserved position, and the segment terminator. A validator that hardcodes the UNOA defaults (:, +, ., ?, space, ') will mis-split any carrier that ships a non-standard UNA, fusing the whole message into one segment. The first stage extracts the delimiters, then splits the raw bytes into segments and elements against them.

from __future__ import annotations

from dataclasses import dataclass

import structlog

log = structlog.get_logger("maritime.baplie.syntax")


@dataclass(frozen=True, slots=True)
class Delimiters:
    component: str = ":"
    element: str = "+"
    decimal: str = "."
    release: str = "?"
    segment: str = "'"


def read_service_string(raw: str) -> Delimiters:
    """Parse the optional UNA service string advice per ISO 9735."""
    if raw.startswith("UNA") and len(raw) >= 9:
        s = raw[3:9]
        log.info("una_detected", advice=s)
        return Delimiters(
            component=s[0], element=s[1], decimal=s[2], release=s[3], segment=s[5]
        )
    log.info("una_absent", note="defaulting to UNOA delimiters")
    return Delimiters()


def tokenize(raw: str, delim: Delimiters) -> list[list[str]]:
    body = raw[9:] if raw.startswith("UNA") else raw
    segments = [s.strip() for s in body.split(delim.segment) if s.strip()]
    return [seg.split(delim.element) for seg in segments]

Step 2 — Reconcile the UNB/UNZ envelope and UNH/UNT counters #

The load-bearing ISO 9735 rule is envelope balance. The interchange trailer UNZ carries an interchange control count (UNZ0401) that must equal the number of messages inside, and an interchange control reference (UNZ0402) that must match the one opened in UNB0405. Inside each message, the UNT trailer carries a segment count (UNT0401) covering every segment from UNH through UNT inclusive, and a message reference (UNT0402) that must match UNH0101. A drifting counter is the single most common corruption in transmitted BAPLIE, and it is fatal: it means segments were dropped or duplicated in transit, so the stow cell list cannot be trusted.

from dataclasses import dataclass, field


@dataclass(slots=True)
class SyntaxError_:
    code: str
    detail: str


def reconcile_envelope(segments: list[list[str]]) -> list[SyntaxError_]:
    errors: list[SyntaxError_] = []
    tags = [seg[0] for seg in segments]

    unb = next((s for s in segments if s[0] == "UNB"), None)
    unz = next((s for s in segments if s[0] == "UNZ"), None)
    if unb is None or unz is None:
        return [SyntaxError_("ENVELOPE_MISSING", "UNB or UNZ segment absent")]

    # Control reference: UNB element 5, UNZ element 2.
    unb_ref = unb[5] if len(unb) > 5 else ""
    unz_ref = unz[2] if len(unz) > 2 else ""
    if unb_ref != unz_ref:
        errors.append(SyntaxError_("CONTROL_REF_MISMATCH", f"{unb_ref!r} != {unz_ref!r}"))

    # Interchange control count vs. number of UNH messages present.
    declared = int(unz[1]) if len(unz) > 1 and unz[1].isdigit() else -1
    actual = tags.count("UNH")
    if declared != actual:
        errors.append(SyntaxError_("MSG_COUNT_MISMATCH", f"UNZ={declared} UNH={actual}"))

    errors.extend(_reconcile_messages(segments))
    if errors:
        log.error("envelope_reject", control_ref=unb_ref, errors=[e.code for e in errors])
    return errors


def _reconcile_messages(segments: list[list[str]]) -> list[SyntaxError_]:
    """Check each UNH..UNT block: segment count and message reference match."""
    errors: list[SyntaxError_] = []
    start: int | None = None
    for i, seg in enumerate(segments):
        if seg[0] == "UNH":
            start = i
        elif seg[0] == "UNT" and start is not None:
            counted = i - start + 1  # inclusive of UNH and UNT
            declared = int(seg[1]) if len(seg) > 1 and seg[1].isdigit() else -1
            if declared != counted:
                errors.append(SyntaxError_("UNT_COUNT_MISMATCH", f"declared={declared} counted={counted}"))
            unh_ref = segments[start][1] if len(segments[start]) > 1 else ""
            unt_ref = seg[2] if len(seg) > 2 else ""
            if unh_ref != unt_ref:
                errors.append(SyntaxError_("MSG_REF_MISMATCH", f"{unh_ref!r} != {unt_ref!r}"))
            start = None
    return errors

Step 3 — Validate the mandatory per-container segment group #

A BAPLIE message body is a repeating group anchored on a LOC+147 segment carrying the stowage position, followed by the equipment (EQD), the SOLAS Verified Gross Mass (MEA), and the container dimensions (DIM). The stow cell in LOC+147 is a seven-digit ISO bay/row/tier code: three digits of bay, two of row, two of tier — for example 0100882 is bay 010, row 08, tier 82. This stage walks each group, requires the mandatory members to be present, and parses the stow cell into a typed structure so an impossible position (tier above the deck ceiling, missing weight) is caught here rather than by the planner.

from dataclasses import dataclass


@dataclass(frozen=True, slots=True)
class StowCell:
    bay: int
    row: int
    tier: int


def parse_stow_cell(code: str) -> StowCell:
    """Parse an ISO bay/row/tier stow cell from a LOC+147 element."""
    if len(code) != 7 or not code.isdigit():
        raise ValueError(f"malformed stow cell {code!r}")
    return StowCell(bay=int(code[:3]), row=int(code[3:5]), tier=int(code[5:7]))


@dataclass(slots=True)
class ContainerSlot:
    stow: StowCell
    equipment_id: str
    vgm_kg: float | None
    dim_present: bool


def extract_slots(segments: list[list[str]], *, max_tier: int, vgm_min: float) -> tuple[list[ContainerSlot], list[str]]:
    slots: list[ContainerSlot] = []
    problems: list[str] = []
    i = 0
    while i < len(segments):
        seg = segments[i]
        if seg[0] == "LOC" and len(seg) > 2 and seg[1] == "147":
            cell = parse_stow_cell(seg[2].split(":")[0])
            if not 0 < cell.tier <= max_tier:
                problems.append(f"tier_out_of_range:{cell.tier}")
            eqd = _find_after(segments, i, "EQD")
            mea = _find_after(segments, i, "MEA")
            dim = _find_after(segments, i, "DIM")
            if eqd is None:
                problems.append(f"missing_EQD:bay{cell.bay}")
                i += 1
                continue
            equipment_id = eqd[2].split(":")[0] if len(eqd) > 2 else ""
            vgm = _read_vgm(mea, floor=vgm_min, problems=problems) if mea else None
            slots.append(ContainerSlot(cell, equipment_id, vgm, dim is not None))
        i += 1
    log.info("slots_extracted", count=len(slots), problems=len(problems))
    return slots, problems


def _find_after(segments: list[list[str]], start: int, tag: str) -> list[str] | None:
    for seg in segments[start + 1 : start + 6]:
        if seg[0] == "LOC":
            break  # next container group begins
        if seg[0] == tag:
            return seg
    return None


def _read_vgm(mea: list[str], *, floor: float, problems: list[str]) -> float | None:
    # MEA+AAE+VGM+KGM:24500 -> qualifier VGM, unit KGM, value 24500
    if len(mea) < 4 or mea[2] != "VGM":
        return None
    try:
        value = float(mea[3].split(":")[-1])
    except ValueError:
        problems.append("vgm_unparseable")
        return None
    if value < floor:
        problems.append(f"vgm_below_floor:{value}")
    return value

Step 4 — Run the semantic tier: ISO 6346 check digit and stow legality #

Structural validity per ISO 9735 does not make a container reference real. The semantic tier proves each EQD equipment identifier is a legal ISO 6346 container number by recomputing its weighted modulo-11 check digit — four owner/category letters and six serial digits weighted by ascending powers of two, summed, taken modulo 11, with a remainder of 10 folded to 0. A drifting check digit almost always means a transposition upstream, and it is a quarantine reason, not a hard drop: the box exists, someone just fat-fingered the serial. The check-digit routine is the same one the semantic tier of the parent Schema Validation Frameworks applies across the platform.

_ISO6346_LETTERS: dict[str, int] = {
    "A": 10, "B": 12, "C": 13, "D": 14, "E": 15, "F": 16, "G": 17,
    "H": 18, "I": 19, "J": 20, "K": 21, "L": 23, "M": 24, "N": 25,
    "O": 26, "P": 27, "Q": 28, "R": 29, "S": 30, "T": 31, "U": 32,
    "V": 34, "W": 35, "X": 36, "Y": 37, "Z": 38,
}


def iso6346_check_digit(container_id: str) -> int:
    """Compute the ISO 6346 check digit from the 10-character body."""
    body = container_id[:10]
    total = sum(
        (_ISO6346_LETTERS[ch] if ch.isalpha() else int(ch)) * (2 ** i)
        for i, ch in enumerate(body)
    )
    remainder = total % 11
    return 0 if remainder == 10 else remainder


def check_container_id(container_id: str) -> bool:
    if len(container_id) != 11:
        log.warning("iso6346_length", container_id=container_id)
        return False
    expected = iso6346_check_digit(container_id)
    actual = int(container_id[10])
    if expected != actual:
        log.warning("iso6346_drift", container_id=container_id, expected=expected, actual=actual)
        return False
    return True

Chaining the four stages gives the two-tier verdict: Steps 1–2 assert ISO 9735 syntax, Steps 3–4 assert container semantics, and any failure at any stage routes the whole interchange to quarantine with the failing codes attached. Version tokens and directory versions in the UNH (for example BAPLIE:D:95B:UN:SMDG20) follow the UN/EDIFACT directories the parent domain tracks.

Edge Cases & Carrier Deviations #

A specific deviation worth pre-empting: some carriers emit the stowage LOC under qualifier 147 while others historically used ZZZ or an unqualified LOC for the same cell. Treat an unexpected LOC qualifier inside a container group as a quarantine reason, not a hard reject — the position is legally expressed, and a human can map the qualifier before replay. The same tolerance applies to a UNA that redefines the decimal notation to a comma: honour it for MEA value parsing rather than misreading a European-formatted weight.

Verification & Testing #

Assert both tiers with pytest: a balanced interchange passes, a doctored counter fails structurally, and a transposed serial fails semantically. Use raw-string fixtures so the tokenizer and UNA handling are genuinely exercised.

import pytest

from maritime_baplie.validator import (
    Delimiters, check_container_id, iso6346_check_digit,
    parse_stow_cell, read_service_string, reconcile_envelope, tokenize,
)


@pytest.fixture
def interchange() -> str:
    return (
        "UNB+UNOA:2+MAEU+TERMINAL+240617:0900+IC001'"
        "UNH+1+BAPLIE:D:95B:UN:SMDG20'"
        "LOC+147+0100882::5'EQD+CN+MSKU0000481+22G1'MEA+AAE+VGM+KGM:24500'"
        "UNT+5+1'UNZ+1+IC001'"
    )


def test_una_defaults_when_absent(interchange: str) -> None:
    assert read_service_string(interchange) == Delimiters()


def test_balanced_envelope_passes(interchange: str) -> None:
    segments = tokenize(interchange, Delimiters())
    assert reconcile_envelope(segments) == []


def test_counter_drift_rejects(interchange: str) -> None:
    broken = interchange.replace("UNT+5+1", "UNT+9+1")
    segments = tokenize(broken, Delimiters())
    codes = {e.code for e in reconcile_envelope(segments)}
    assert "UNT_COUNT_MISMATCH" in codes


def test_stow_cell_and_check_digit() -> None:
    assert parse_stow_cell("0100882").tier == 82
    assert iso6346_check_digit("CSQU305438") == 3
    assert check_container_id("CSQU3054383") is True
    assert check_container_id("CSQU3054384") is False

A passing run emits one JSON line per decision; a clean interchange logs {"event": "slots_extracted", "count": 1, "problems": 0, "log_level": "info", ...}, and a rejection logs envelope_reject with the control_ref and a list of failing codes. Assert on those structured keys, never on a rendered message string.

Frequently Asked Questions #

Related #

• Schema Validation Frameworks — the structural, semantic, and regulatory tiers this BAPLIE gate implements
• IFCSUM EDI Message Parsing — sibling UN/EDIFACT parsing governed by the same ISO 9735 envelope rules
• Container Hierarchy Data Models — the topology each validated EQD equipment identifier resolves into
• pydantic envelope validation for UN/EDIFACT interchanges — the same envelope checks expressed as typed models
• Routing failed payloads to quarantine and DLQ — where a rejected interchange goes after this gate

Up: Schema Validation Frameworks — the topic area governing structural, semantic, and regulatory validation at the ingestion boundary.