Modeling SOLAS VGM declarations as Pydantic schemas #

A packed container cannot be loaded aboard a vessel until its Verified Gross Mass is declared, and a mistyped or physically impossible weight is a stability hazard, not a paperwork nuisance. Modeling SOLAS VGM declarations as Pydantic schemas turns the SOLAS Chapter VI, Regulation 2 obligation into a typed, self-validating contract: one VgmDeclaration model that pins the container identity to its ISO 6346 check digit, records which weighing method produced the figure, normalizes the reported unit to kilograms, and rejects any mass that falls below tare or above the CSC-plate maximum before the number ever reaches a stowage planner. This page specifies that model — the field validators, the cross-field consistency rules, and the mapping back to the EDIFACT VERMAS message the terminal actually receives.

Architecture Alignment #

This model lives in the Regulatory Compliance Reference area of the Core Maritime Architecture & Taxonomy domain, where regulatory obligations are expressed as enforceable data contracts rather than prose checklists. A VGM figure is meaningless without the equipment it describes, so the container identity here reuses the same ISO 6346 rules that govern Container Hierarchy Data Models; a VGM that fails to resolve to a known box is as invalid as one that fails a plausibility bound. The declaration is also a payload that arrives over EDI, so the structural and semantic gating it performs is the same discipline documented in Schema Validation Frameworks — the difference is that VGM adds a hard regulatory gate: no verified mass, no load. Keeping the rule inside a Pydantic model rather than scattered if statements is what makes the obligation testable and the rejection auditable.

Prerequisites & Environment Setup #

The model targets Python 3.11+ and Pydantic v2, whose Rust-backed core and model_validator API make cross-field regulatory rules both fast and declarative. Install the typed-validation and structured-logging stack:

pip install "pydantic>=2.6" structlog

Structured JSON logging is not optional in this niche: a VGM rejection can hold a container off a sailing, so every accept and every reject must be a queryable, timestamped event that a port-state control inspector or a carrier dispute can replay. Configure structlog once, at import time, to emit JSON:

import structlog

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

The validator reads a few operational bounds from the environment so plausibility floors can be tuned per trade lane without a redeploy:

SOLAS Chapter VI, Regulation 2 makes the shipper — the party named on the bill of lading — legally responsible for the figure, so the model treats the responsible-party signature as a required field, never a default. See the IMO summary of the SOLAS VGM requirement for the regulatory text these rules encode.

Step-by-step Implementation #

Step 1 — Validate the container identity against its ISO 6346 check digit #

Every VGM is anchored to one piece of equipment, and a transposed digit in the container number will silently attach a real weight to the wrong box. The ISO 6346 BIC container code embeds a self-check: three owner letters, an equipment category (U, J, or Z), a six-digit serial, and a trailing check digit computed from a weighted modulo-11 sum. A field_validator recomputes that digit and rejects a mismatch before any weight logic runs.

from __future__ import annotations

import re

import structlog
from pydantic import field_validator

log = structlog.get_logger("maritime.vgm")

_ISO6346_RE = re.compile(r"^[A-Z]{3}[UJZ]\d{6}\d$")
# ISO 6346 letter values skip every multiple of 11.
_LETTER_VALUES: dict[str, int] = {
    ch: v for ch, v in zip(
        "ABCDEFGHIJKLMNOPQRSTUVWXYZ",
        [10, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 23, 24,
         25, 26, 27, 28, 29, 30, 31, 32, 34, 35, 36, 37, 38],
    )
}


def iso6346_check_digit(container_id: str) -> int:
    total = 0
    for position, char in enumerate(container_id[:10]):
        value = _LETTER_VALUES[char] if char.isalpha() else int(char)
        total += value * (2 ** position)
    remainder = total % 11
    return 0 if remainder == 10 else remainder


class _ContainerIdMixin:
    @field_validator("container_id", mode="before")
    @classmethod
    def _normalize_and_check(cls, raw: str) -> str:
        candidate = raw.strip().upper().replace(" ", "")
        if not _ISO6346_RE.match(candidate):
            raise ValueError(f"container_id {candidate!r} is not ISO 6346 shaped")
        expected = iso6346_check_digit(candidate)
        if int(candidate[-1]) != expected:
            log.warning("iso6346_check_digit_mismatch", container_id=candidate, expected=expected)
            raise ValueError(f"container_id {candidate!r} fails ISO 6346 check digit")
        return candidate

Step 2 — Declare the fields and normalize the reported unit to kilograms #

A VERMAS message can carry the mass in kilograms (KGM) or pounds (LBR), and stowage software works in a single canonical unit. The model captures the raw value and unit exactly as submitted for audit, then derives a normalized vgm_kg so no downstream consumer ever divides by the wrong factor. Enumerating the weighing method and unit turns two of the most common data-entry mistakes — a blank method flag, a free-text unit — into validation errors at the boundary.

import os
from datetime import datetime
from decimal import Decimal
from enum import Enum

from pydantic import BaseModel, Field, computed_field

LBR_TO_KG = Decimal(os.environ.get("LBR_TO_KG", "0.45359237"))


class WeighingMethod(str, Enum):
    METHOD_1 = "1"  # weigh the packed, sealed container as one unit
    METHOD_2 = "2"  # weigh contents and packaging, add the certified tare


class MassUnit(str, Enum):
    KGM = "KGM"
    LBR = "LBR"


class VgmDeclaration(_ContainerIdMixin, BaseModel):
    container_id: str
    vgm_value: Decimal = Field(gt=0, description="mass as declared, in vgm_unit")
    vgm_unit: MassUnit
    weighing_method: WeighingMethod
    weighing_party: str = Field(min_length=2)
    authorization_ref: str = Field(min_length=1, description="weighing-party certificate id")
    weighbridge_id: str | None = None
    weighed_at: datetime
    responsible_party_signature: str = Field(min_length=2)
    tare_kg: Decimal = Field(gt=0, description="CSC-plate tare weight")
    max_gross_kg: Decimal = Field(gt=0, description="CSC-plate maximum gross mass")
    cargo_kg: Decimal | None = Field(default=None, description="Method 2 contents + packaging")

    @computed_field  # type: ignore[prop-decorator]
    @property
    def vgm_kg(self) -> Decimal:
        if self.vgm_unit is MassUnit.LBR:
            return (self.vgm_value * LBR_TO_KG).quantize(Decimal("0.1"))
        return self.vgm_value.quantize(Decimal("0.1"))

Step 3 — Enforce method consistency and CSC-plate plausibility bounds #

The hard regulatory logic is cross-field, so it belongs in a model_validator that runs after every field is individually valid. Method 1 requires a weighbridge identifier because the whole container was physically weighed; Method 2 requires the cargo weight because the VGM is a sum, and that sum must reconcile with the declared figure inside a tolerance. Regardless of method, a verified mass below the container’s own tare is physically impossible, and one above the CSC-plate maximum gross means an overloaded box that must never be stowed — the CSC safety-approval plate fixes that ceiling.

from pydantic import model_validator

VGM_ABS_MIN_KG = Decimal(os.environ.get("VGM_ABS_MIN_KG", "1500"))
METHOD2_TOLERANCE_KG = Decimal(os.environ.get("VGM_METHOD2_TOLERANCE_KG", "50"))


class VgmDeclaration(VgmDeclaration):  # extend the Step 2 model
    @model_validator(mode="after")
    def _check_method_and_bounds(self) -> "VgmDeclaration":
        bound = structlog.get_logger("maritime.vgm").bind(
            container_id=self.container_id, method=self.weighing_method.value
        )

        if self.weighing_method is WeighingMethod.METHOD_1 and not self.weighbridge_id:
            raise ValueError("Method 1 requires a weighbridge_id (the whole box was weighed)")

        if self.weighing_method is WeighingMethod.METHOD_2:
            if self.cargo_kg is None:
                raise ValueError("Method 2 requires cargo_kg to sum against the tare")
            summed = (self.cargo_kg + self.tare_kg).quantize(Decimal("0.1"))
            if abs(summed - self.vgm_kg) > METHOD2_TOLERANCE_KG:
                raise ValueError(
                    f"Method 2 sum {summed} disagrees with declared VGM {self.vgm_kg}"
                )

        if self.vgm_kg < self.tare_kg:
            raise ValueError(f"VGM {self.vgm_kg} kg is below tare {self.tare_kg} kg")
        if self.vgm_kg < VGM_ABS_MIN_KG:
            raise ValueError(f"VGM {self.vgm_kg} kg is below the absolute plausibility floor")
        if self.vgm_kg > self.max_gross_kg:
            raise ValueError(f"VGM {self.vgm_kg} kg exceeds CSC max gross {self.max_gross_kg} kg")

        bound.info("vgm_validated", vgm_kg=float(self.vgm_kg), unit=self.vgm_unit.value)
        return self

Step 4 — Map the validated model to the EDIFACT VERMAS / MEA+VGM segment #

Terminals and carriers exchange VGM over the UN/EDIFACT VERMAS (Verified Gross Mass) message, where the figure rides in a MEA measurement segment qualified VGM. Once the model has accepted a declaration, serializing it back to that segment is a deterministic projection — and doing it from the validated object, never from raw input, guarantees the wire message can only carry a plausibility-checked mass. The submission-cutoff rule lives here too: a VGM received after the terminal’s load-list cutoff is late by definition and must be flagged, because a container without a timely verified mass cannot be assigned a stow position.

from datetime import timedelta

VGM_SUBMISSION_LEAD_MIN = int(os.environ.get("VGM_SUBMISSION_LEAD_MIN", "120"))


def to_mea_vgm_segment(decl: VgmDeclaration) -> str:
    """Project a validated declaration to a MEA+VGM EDIFACT segment (always in KGM)."""
    kg = decl.vgm_kg.quantize(Decimal("1"))
    return f"MEA+AAE+VGM+KGM:{kg}"


def is_before_cutoff(decl: VgmDeclaration, load_list_cutoff: datetime) -> bool:
    deadline = load_list_cutoff - timedelta(minutes=VGM_SUBMISSION_LEAD_MIN)
    on_time = decl.weighed_at <= deadline
    log.info(
        "vgm_cutoff_check",
        container_id=decl.container_id,
        weighed_at=decl.weighed_at.isoformat(),
        deadline=deadline.isoformat(),
        on_time=on_time,
    )
    return on_time

Edge Cases & Carrier Deviations #

A recurring carrier deviation is the missing method flag: some EDI gateways emit a VERMAS with the mass but leave the method qualifier blank, defaulting silently to Method 1. That is not a safe default — a Method 1 figure asserts the whole container crossed a calibrated weighbridge, a claim a blank flag cannot support. The WeighingMethod enum makes the omission a hard validation error, so the declaration bounces back to the shipper rather than entering the stow plan on an unverified assumption.

Verification & Testing #

Assert the two behaviours that carry the most regulatory weight — the ISO 6346 gate and the tare/CSC plausibility bounds — with pytest fixtures, and confirm the accept event fires as structured JSON.

import pytest
from decimal import Decimal
from datetime import datetime

from maritime.vgm import VgmDeclaration, WeighingMethod, MassUnit


def _base(**overrides) -> dict:
    data = dict(
        container_id="MSKU1234565",  # valid ISO 6346 check digit
        vgm_value=Decimal("21000"), vgm_unit=MassUnit.KGM,
        weighing_method=WeighingMethod.METHOD_1, weighing_party="ACME Terminals",
        authorization_ref="WB-CERT-88", weighbridge_id="WB-07",
        weighed_at=datetime(2026, 7, 17, 6, 0), responsible_party_signature="J. Shipper",
        tare_kg=Decimal("2200"), max_gross_kg=Decimal("30480"),
    )
    data.update(overrides)
    return data


def test_valid_method1_declaration_normalizes_to_kg() -> None:
    decl = VgmDeclaration(**_base())
    assert decl.vgm_kg == Decimal("21000.0")


def test_bad_check_digit_rejected() -> None:
    with pytest.raises(ValueError, match="check digit"):
        VgmDeclaration(**_base(container_id="MSKU1234567"))


def test_vgm_above_csc_max_rejected() -> None:
    with pytest.raises(ValueError, match="exceeds CSC max"):
        VgmDeclaration(**_base(vgm_value=Decimal("31000")))


def test_method2_sum_must_reconcile() -> None:
    with pytest.raises(ValueError, match="disagrees"):
        VgmDeclaration(**_base(
            weighing_method=WeighingMethod.METHOD_2, weighbridge_id=None,
            cargo_kg=Decimal("15000"),  # 15000 + 2200 = 17200, not 21000
        ))

A passing accept emits one JSON line — {"event": "vgm_validated", "container_id": "MSKU1234565", "method": "1", "vgm_kg": 21000.0, "unit": "KGM", "level": "info", "timestamp": "..."} — so assert on the structured keys, never on a formatted string.

Frequently Asked Questions #

Related #

• Regulatory Compliance Reference — the compliance-contract area this VGM model belongs to.
• Container Hierarchy Data Models — the ISO 6346 equipment identity a VGM must resolve against.
• Schema Validation Frameworks — the structural, semantic, and regulatory gating pattern this model applies.
• Core Maritime Architecture & Taxonomy — the domain governing shared data models and boundary controls.

Up: Regulatory Compliance Reference — the area governing how maritime regulations become enforceable schemas.