Parsing IFCSUM 311 messages with Python
In high-throughput terminal environments, the IFCSUM 311 EDIFACT message serves as the definitive manifest handshake between freight forwarders, vessel operators, and port authorities. When discharge windows compress and TEU volumes exceed 10,000 per vessel call, legacy XML/CSV converters and monolithic EDI parsers routinely collapse under memory pressure or fail to meet strict Terminal Operating System (TOS) ingestion windows. The operational reality demands deterministic extraction, linear memory scaling, and explicit compliance gating. This guide details a production-grade methodology for parsing IFCSUM 311 messages with Python, engineered for shipping operations teams, port IT departments, and maritime automation developers who must guarantee sub-15-minute processing SLAs under peak load.
The architecture required to handle these payloads aligns directly with modern Document Ingestion & EDI Parsing Workflows, where fault-tolerant routing must coexist with strict schema validation and audit-ready logging. Below, we dissect the exact failure modes encountered in live port environments and provide a runnable, memory-safe parser that enforces regulatory compliance without sacrificing throughput.
Memory Bottlenecks & Streaming Architecture
Naive EDI parsers load entire .edi or .txt payloads into memory before tokenizing. A single IFCSUM 311 file containing 15,000 CNI (Container Information) loops and nested GID (Goods Item Details) segments can easily exceed 800 MB of raw text. When combined with recursive object instantiation, this triggers garbage collection thrashing and OOM kills during concurrent vessel processing.
The production solution is a generator-driven, character-aware tokenizer that yields segments on-the-fly. The following implementation processes payloads in a single pass, tracks memory pressure, and enforces explicit timeout thresholds aligned with port SLAs.
import re
import logging
import time
import sys
import json
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Iterator, Generator, Dict, Optional, List, Tuple
from enum import Enum
# Structured logging configuration for observability pipelines
class StructuredFormatter(logging.Formatter):
def format(self, record):
# time.strftime (used by Formatter.formatTime) does not support %f,
# so derive a millisecond UTC timestamp directly from datetime.
log_entry = {
"ts": datetime.fromtimestamp(record.created, tz=timezone.utc)
.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z",
"level": record.levelname,
"msg": record.getMessage(),
"module": record.module,
"func": record.funcName
}
return json.dumps(log_entry)
logger = logging.getLogger("ifcsum_parser")
logger.setLevel(logging.INFO)
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(StructuredFormatter())
logger.addHandler(handler)
class ErrorCategory(Enum):
STRUCTURAL = "STRUCTURAL"
COMPLIANCE = "COMPLIANCE"
REGULATORY = "REGULATORY"
TIMEOUT = "TIMEOUT"
@dataclass
class IfcsumManifest:
message_ref: str
vessel_name: str
voyage_number: str
container_count: int = 0
compliance_status: str = "PENDING"
processing_time_sec: float = 0.0
error_log: List[Dict[str, str]] = field(default_factory=list)
class Ifcsum311StreamParser:
# Default EDIFACT structural delimiters
SEG_TERM = "'"
ELEM_SEP = "+"
COMP_SEP = ":"
REP_SEP = "*"
ESC_CHAR = "?"
# TOS & Customs mandatory segments for compliance gating
MANDATORY_SEGMENTS = {"UNB", "UNH", "BGM", "TDT", "CNI", "UNT"}
def __init__(self, sla_timeout_sec: float = 900.0):
self.sla_timeout_sec = sla_timeout_sec
self._errors: List[Dict[str, str]] = []
self._start_time = 0.0
def _check_timeout(self) -> bool:
if time.monotonic() - self._start_time > self.sla_timeout_sec:
self._log_error(ErrorCategory.TIMEOUT, "SLA timeout exceeded")
return True
return False
def _log_error(self, category: ErrorCategory, detail: str, segment: Optional[str] = None):
entry = {"category": category.value, "detail": detail}
if segment:
entry["segment_preview"] = segment[:60]
self._errors.append(entry)
logger.warning(json.dumps(entry))
def _parse_una(self, service_string: str) -> None:
"""Apply UNA service-string advice: 'UNA' followed by six service chars."""
if service_string.startswith("UNA") and len(service_string) >= 9:
self.COMP_SEP = service_string[3] # component data element separator
self.ELEM_SEP = service_string[4] # data element separator
# service_string[5] is the decimal notation character (not stored)
self.ESC_CHAR = service_string[6] # release (escape) character
self.REP_SEP = service_string[7] # repetition separator (space in older syntax)
self.SEG_TERM = service_string[8] # segment terminator
def _split_buffer(self, buffer: str) -> Tuple[Optional[str], str]:
"""Return (segment, remainder) at the first unescaped segment terminator."""
i = 0
while i < len(buffer):
if buffer[i] == self.ESC_CHAR:
i += 2 # skip the released (escaped) character
continue
if buffer[i] == self.SEG_TERM:
return buffer[:i], buffer[i + 1:]
i += 1
return None, buffer
def _split_segment(self, raw: str) -> List[str]:
"""Split segment respecting EDIFACT escape sequences."""
elements = []
current = []
i = 0
while i < len(raw):
if raw[i] == self.ESC_CHAR and i + 1 < len(raw):
current.append(raw[i+1])
i += 2
continue
if raw[i] == self.ELEM_SEP:
elements.append("".join(current))
current = []
else:
current.append(raw[i])
i += 1
elements.append("".join(current))
return elements
def parse_stream(self, file_path: str) -> Generator[Tuple[str, List[str]], None, None]:
"""Memory-efficient generator yielding (segment_tag, elements)."""
self._start_time = time.monotonic()
self._errors.clear()
logger.info(f"Initializing stream parser for {file_path}")
try:
with open(file_path, "r", encoding="utf-8-sig") as fh:
buffer = ""
una_checked = False
for chunk in iter(lambda: fh.read(65536), ""):
buffer += chunk
# Consume the optional UNA service string before tokenizing.
if not una_checked and len(buffer) >= 9:
if buffer.startswith("UNA"):
self._parse_una(buffer[:9])
buffer = buffer[9:]
una_checked = True
while True:
segment, rest = self._split_buffer(buffer)
if segment is None:
break # wait for more data
buffer = rest
segment = segment.strip()
if not segment:
continue
if self._check_timeout():
break
tag = segment[:3]
elements = self._split_segment(segment[3:])
yield tag, elements
except Exception as e:
self._log_error(ErrorCategory.STRUCTURAL, f"Stream read failure: {str(e)}")
raise
def extract_manifest(self, file_path: str) -> IfcsumManifest:
"""Single-pass extraction with compliance validation."""
manifest = IfcsumManifest(
message_ref="UNKNOWN",
vessel_name="UNKNOWN",
voyage_number="UNKNOWN"
)
encountered_segments = set()
for tag, elements in self.parse_stream(file_path):
encountered_segments.add(tag)
if tag == "UNH":
# UNH DE 0062 (first element) is the message reference number.
manifest.message_ref = elements[0] if elements else "MISSING"
elif tag == "BGM":
pass # Document type validation
elif tag == "TDT":
# TDT+20++VSL:VesselName:166:IMO:IMO1234567+Voyage
if len(elements) > 2:
vessel_comp = elements[2].split(self.COMP_SEP)
manifest.vessel_name = vessel_comp[1] if len(vessel_comp) > 1 else "UNKNOWN"
if len(elements) > 3:
manifest.voyage_number = elements[3].split(self.COMP_SEP)[0]
elif tag == "CNI":
manifest.container_count += 1
elif tag == "UNT":
pass # Trailer validation
if self._check_timeout():
manifest.compliance_status = "TIMEOUT_FAILED"
break
# Compliance gating
missing = self.MANDATORY_SEGMENTS - encountered_segments
if missing:
self._log_error(ErrorCategory.COMPLIANCE, f"Missing mandatory segments: {missing}")
manifest.compliance_status = "NON_COMPLIANT"
else:
manifest.compliance_status = "COMPLIANT"
manifest.error_log = self._errors
manifest.processing_time_sec = time.monotonic() - self._start_time
logger.info(f"Extraction complete. Status: {manifest.compliance_status} | TEUs: {manifest.container_count}")
return manifestReal-World EDIFACT Quirks & Format Drift
flowchart LR A["Read 64 KB chunks"] --> B["Consume UNA
service string"] B --> C["Split at unescaped
segment terminator"] C --> D["Tokenise elements
escape-aware"] D --> E{"Mandatory segments
present?"} E -->|no| NC["NON_COMPLIANT"] E -->|yes| M["IfcsumManifest"]
Port environments rarely receive pristine EDIFACT. Terminal gate systems frequently inject unescaped apostrophes in FTX (Free Text) segments, legacy mainframe gateways append Windows \r\n line endings mid-segment, and some forwarders omit the UNA header entirely. The parser above mitigates these through explicit escape-aware tokenization and utf-8-sig decoding to strip Byte Order Marks. When integrating with broader document format drift handling strategies, it is critical to normalize segment terminators before ingestion and apply strict regex boundaries to composite element extraction. This prevents cascading parse failures that typically corrupt downstream TOS allocation tables.
Compliance Gating & Regulatory Constraints
Maritime manifests operate under strict regulatory frameworks. The IFCSUM 311 must satisfy IMO FAL Convention pre-arrival filing windows, customs AMS/ISF data requirements, and hazardous material (IMDG) declaration rules. The MANDATORY_SEGMENTS set in the parser acts as a structural compliance gate. In production deployments, this should be extended with a dedicated Schema Validation Frameworks layer that cross-references CNI loops against port authority tariff codes and verifies GID commodity descriptions against HS code registries. Regulatory holds (e.g., customs seals, quarantine flags) are typically embedded in FTX or RFF segments; the parser’s structured error log ensures these are flagged without halting the entire manifest stream, enabling targeted quarantine routing rather than full vessel rejection.
Pipeline Integration & Error Categorization
The generator-based design allows seamless integration into Async Batch Processing Pipelines where multiple vessel manifests are processed concurrently. By categorizing failures via ErrorCategory, operations teams can implement deterministic retry logic: structural errors trigger immediate pipeline halt, compliance gaps route to manual review queues, and timeout events trigger graceful degradation to partial ingestion. When paired with IFCSUM EDI Message Parsing or PDF Bill of Lading Extraction, this architecture maintains a unified audit trail, ensuring that every TEU movement is traceable from forwarder submission to quay crane assignment.