Generating IMO FAL forms as structured data
Generating IMO FAL forms as structured data replaces the fax-and-PDF port-call clearance packet with a typed, machine-validated dataset that a Maritime Single Window accepts on first submission. The IMO FAL Convention (Convention on Facilitation of International Maritime Traffic, 1965) standardises seven arrival and departure declarations — FAL 1 through FAL 7 — and since January 2024 requires them to be exchanged electronically through a national Single Window. This page specifies how to derive those forms from one canonical port-call model, validate each with pydantic, and emit both a submission payload and a human-review artifact for the ship’s agent, focusing the code on FAL 1 (General Declaration), FAL 5 (Crew List), and FAL 7 (Dangerous Goods).
Architecture Alignment
This task lives in the Regulatory Compliance Reference discipline of the Core Maritime Architecture & Taxonomy domain, where every regulatory artifact must trace back to a single typed source of truth rather than being re-keyed per form. The seven FAL forms are not independent documents: they are seven projections of the same vessel, voyage, crew, and cargo facts, so the correct design derives them from the canonical port-call record produced by Port Call Workflow Design and never lets an operator edit a form field directly. Where inbound EDIFACT feeds the same facts — a consolidator’s aggregated manifest, for instance — the mapping discipline mirrors IFCSUM EDI Message Parsing: parse once into typed models, validate against the standard, then generate outbound messages from those models. The FAL generators here are the outbound counterpart, turning the port-call model into PAXLST, IFTDGN, and arrival-report messages plus a validated JSON dataset.
Prerequisites & Environment Setup
The generators target Python 3.11+ (for StrEnum, datetime.UTC, and precise typing) and pydantic v2 for constraint validation and JSON emission. Install the typed-validation and structured-logging stack:
pip install "pydantic>=2.6" structlog
Structured JSON logging is mandatory here: every form generation is a regulatory event that must be reproducible for a port-authority or customs audit, so bare print() and unstructured strings are unacceptable. Configure structlog once at process boot 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.fal")
Environment variables the generators expect:
| Variable | Purpose | Example |
|---|---|---|
MSW_ENDPOINT |
Maritime Single Window submission URL | https://msw.example-port.gov/fal/v3 |
PORT_UNLOCODE |
UN/LOCODE of the arrival port | NLRTM |
FAL_DATASET_VERSION |
IMO Compendium harmonized data set version | 2023.1 |
AUDIT_COPY_DIR |
Directory for human-review artifacts | /var/lib/fal/audit |
CREW_ID_MIN_VALIDITY_DAYS |
Passport / seafarer ID validity floor at ETA | 30 |
The IMO Compendium on Facilitation and Electronic Business is the authority for the harmonized data set: pin FAL_DATASET_VERSION so a generated payload can always be replayed against the exact element definitions a Single Window accepted it under.
Step-by-step Implementation
Step 1 — Model the canonical port-call as the single source of truth
Every FAL form is a projection of the same underlying facts, so define one immutable port-call record and forbid per-form editing. This model carries the vessel identity (IMO number, call sign, flag), the voyage particulars, and the crew and cargo collections that the individual generators slice. The IMO ship identification number is a seven-digit value with a check digit, and refusing anything else here stops a malformed identifier from propagating into all seven forms.
from __future__ import annotations
from datetime import datetime
from pydantic import BaseModel, Field, field_validator
class Vessel(BaseModel):
imo_number: str = Field(pattern=r"^\d{7}$")
name: str = Field(min_length=1, max_length=35)
call_sign: str = Field(min_length=1, max_length=8)
flag_state: str = Field(pattern=r"^[A-Z]{2}$") # ISO 3166-1 alpha-2
@field_validator("imo_number")
@classmethod
def check_imo_check_digit(cls, v: str) -> str:
weights = (7, 6, 5, 4, 3, 2)
total = sum(int(d) * w for d, w in zip(v[:6], weights))
if total % 10 != int(v[6]):
raise ValueError(f"IMO check digit invalid for {v}")
return v
class PortCall(BaseModel):
vessel: Vessel
port_unlocode: str = Field(pattern=r"^[A-Z]{5}$")
eta: datetime
last_port_unlocode: str = Field(pattern=r"^[A-Z]{5}$")
voyage_number: str
crew: list["CrewMember"] = Field(default_factory=list)
dangerous_goods: list["DangerousGoodsItem"] = Field(default_factory=list)
Step 2 — Generate FAL 1 (General Declaration) as validated structured data
FAL 1 is the master arrival declaration: it summarises the ship, the voyage, and headline counts of crew, passengers, and cargo. Rather than accept free-text counts, derive them from the canonical collections so FAL 1 can never disagree with FAL 5 or the cargo manifest. The generator below builds a typed Fal1GeneralDeclaration, and pydantic’s model_dump(mode="json") emits the Single Window payload with ISO 8601 timestamps.
import structlog
from pydantic import BaseModel, Field, computed_field
log = structlog.get_logger("maritime.fal.form1")
class Fal1GeneralDeclaration(BaseModel):
form: str = Field(default="FAL1", frozen=True)
imo_number: str
ship_name: str
call_sign: str
flag_state: str
port_of_arrival: str
eta_utc: str
last_port: str
voyage_number: str
number_of_crew: int = Field(ge=1)
number_of_dg_items: int = Field(ge=0)
@computed_field # brief particulars of voyage, per FAL Convention Annex
def voyage_summary(self) -> str:
return f"{self.last_port} -> {self.port_of_arrival} voy {self.voyage_number}"
def generate_fal1(call: "PortCall") -> Fal1GeneralDeclaration:
form = Fal1GeneralDeclaration(
imo_number=call.vessel.imo_number,
ship_name=call.vessel.name,
call_sign=call.vessel.call_sign,
flag_state=call.vessel.flag_state,
port_of_arrival=call.port_unlocode,
eta_utc=call.eta.isoformat(),
last_port=call.last_port_unlocode,
voyage_number=call.voyage_number,
number_of_crew=len(call.crew),
number_of_dg_items=len(call.dangerous_goods),
)
log.info("fal1_generated", imo=form.imo_number, crew=form.number_of_crew)
return form
Step 3 — Generate FAL 5 (Crew List) rows and emit JSON plus a human-review artifact
FAL 5 is a repeating-row form: one row per crew member carrying name, rank, nationality, date and place of birth, and the nature and number of the identity document (passport or seafarer identity document). Model the row explicitly so each field is validated, then emit two outputs — a compact JSON payload for the Single Window and a flattened, human-readable audit copy the ship’s agent can eyeball before submission. Identity-document expiry is checked against the ETA plus a validity floor, because a port state will reject a crew member whose passport lapses during the call.
import json
import os
from datetime import UTC, datetime, timedelta
from enum import StrEnum
import structlog
from pydantic import BaseModel, Field
log = structlog.get_logger("maritime.fal.form5")
MIN_VALIDITY_DAYS = int(os.environ.get("CREW_ID_MIN_VALIDITY_DAYS", "30"))
class IdDocType(StrEnum):
PASSPORT = "PASSPORT"
SEAFARER_ID = "SID" # ILO Seafarers' Identity Documents Convention
class CrewMember(BaseModel):
family_name: str = Field(min_length=1, max_length=35)
given_names: str = Field(min_length=1, max_length=35)
rank: str = Field(min_length=1, max_length=35)
nationality: str = Field(pattern=r"^[A-Z]{2}$")
date_of_birth: datetime
place_of_birth: str
id_doc_type: IdDocType
id_doc_number: str = Field(min_length=1, max_length=20)
id_doc_issuing_state: str = Field(pattern=r"^[A-Z]{2}$")
id_doc_expiry: datetime
def generate_fal5(call: "PortCall") -> dict[str, object]:
cutoff = call.eta + timedelta(days=MIN_VALIDITY_DAYS)
rows: list[dict[str, object]] = []
review_lines: list[str] = []
for seq, member in enumerate(call.crew, start=1):
if member.id_doc_expiry < cutoff:
log.warning(
"crew_id_expiring",
seq=seq,
family_name=member.family_name,
expiry=member.id_doc_expiry.date().isoformat(),
)
rows.append({"seq": seq, **member.model_dump(mode="json")})
review_lines.append(
f"{seq:>3} {member.family_name.upper()}, {member.given_names} "
f"({member.rank}) {member.nationality} "
f"{member.id_doc_type}:{member.id_doc_number}"
)
payload = {
"form": "FAL5",
"imo_number": call.vessel.imo_number,
"port_of_arrival": call.port_unlocode,
"generated_utc": datetime.now(UTC).isoformat(),
"crew_count": len(rows),
"rows": rows,
}
audit_dir = os.environ.get("AUDIT_COPY_DIR", ".")
audit_path = f"{audit_dir}/fal5_{call.vessel.imo_number}_{call.voyage_number}.txt"
with open(audit_path, "w", encoding="utf-8") as fh:
fh.write("FAL 5 CREW LIST — review before submission\n")
fh.write("\n".join(review_lines) + "\n")
log.info("fal5_generated", crew_count=len(rows), audit_copy=audit_path)
return payload
The JSON payload is what the Single Window ingests; the .txt artifact is the human-review copy. Keeping them from the same model_dump guarantees the reviewed document and the submitted document are byte-for-byte the same facts.
Step 4 — Map typed forms onto the EDIFACT reporting messages
A Single Window still exchanges many of these declarations as UN/EDIFACT messages, so the typed forms must map cleanly onto the standard reporting messages: FAL 5 crew and FAL 6 passenger lists onto PAXLST, FAL 7 dangerous goods onto IFTDGN, and the FAL 1 arrival facts onto CUSREP and BERMAN for customs and berth management. Model FAL 7 as typed items keyed on the IMDG UN number and IMO hazard class, then serialise each crew row into a PAXLST-style segment group. Generating segments from validated models — rather than string-concatenating raw EDIFACT — is what keeps the outbound message free of the qualifier drift that plagues inbound parsing.
import structlog
from pydantic import BaseModel, Field
log = structlog.get_logger("maritime.fal.edifact")
class DangerousGoodsItem(BaseModel):
un_number: str = Field(pattern=r"^\d{4}$") # IMDG UN number
imo_hazard_class: str = Field(pattern=r"^\d(\.\d)?$") # e.g. 3 or 5.1
proper_shipping_name: str = Field(min_length=1)
packing_group: str | None = Field(default=None, pattern=r"^(I|II|III)$")
net_weight_kg: float = Field(gt=0)
def crew_to_paxlst(payload: dict[str, object]) -> list[str]:
"""Emit PAXLST NAD/ATT/DOC segment groups from a FAL 5 payload."""
segments: list[str] = ["UNH+1+PAXLST:D:05B:UN:IATA'"]
for row in payload["rows"]: # type: ignore[index]
segments.append(f"NAD+FL+++{row['family_name']}:{row['given_names']}'")
segments.append(f"ATT+2++{row['rank']}'")
segments.append(f"DOC+{row['id_doc_type']}+{row['id_doc_number']}'")
log.info("paxlst_built", segment_count=len(segments))
return segments
def dg_to_iftdgn(items: list[DangerousGoodsItem]) -> list[str]:
"""Emit IFTDGN DGS segments for FAL 7 dangerous goods."""
segments: list[str] = ["UNH+1+IFTDGN:D:05B:UN'"]
for item in items:
segments.append(
f"DGS+IMD+{item.imo_hazard_class}+{item.un_number}"
f"+{item.packing_group or ''}'"
)
log.info("iftdgn_built", dg_items=len(items))
return segments
Edge Cases & Carrier Deviations
| Symptom | Root cause | Fix |
|---|---|---|
| FAL 1 crew count ≠ FAL 5 row count | Counts entered by hand instead of derived | Compute number_of_crew from len(call.crew); never accept a literal |
| Single Window rejects FAL 5 row | Passport expiry falls inside the call window | Gate on eta + CREW_ID_MIN_VALIDITY_DAYS; flag before submission, not after |
IFTDGN DGS refused |
UN number sent as 3 digits or hazard class free-text | Constrain un_number to ^\d{4}$ and class to the IMDG pattern |
| PAXLST names truncated | Family/given names exceed 35-char EDIFACT limit | Enforce max_length=35 on the model so truncation is caught at generation |
| Payload accepted but wrong dataset | FAL_DATASET_VERSION drifted from the Single Window’s |
Pin and stamp the Compendium version into every payload for replay |
A common deviation is the seafarer identity document versus passport question: some flag states issue an ILO seafarer’s identity document (SID) that a given port state does not accept in lieu of a passport for shore leave. Model the document type as an explicit IdDocType enum rather than a free-text string, so a policy layer can reject an unacceptable document class per port state instead of discovering it at the gangway. Treat an unrecognised document type as a review flag, never a silent pass — the crew member is legally aboard and a human can reconcile the paperwork.
Verification & Testing
Assert the two behaviours most likely to cause a Single Window rejection — the IMO check-digit gate and crew-count consistency between FAL 1 and FAL 5 — with fixtures and pytest. Use structlog’s capture helper to confirm the audit event fires.
import pytest
import structlog
from maritime.fal.model import CrewMember, PortCall, Vessel
from maritime.fal.form1 import generate_fal1
from maritime.fal.form5 import generate_fal5
@pytest.fixture
def call() -> PortCall:
return PortCall(
vessel=Vessel(imo_number="9074729", name="EVER GIVEN",
call_sign="H3RC", flag_state="PA"),
port_unlocode="NLRTM", last_port_unlocode="DEHAM",
eta="2026-08-01T06:00:00Z", voyage_number="084W",
crew=[CrewMember(family_name="Rossi", given_names="Marco", rank="Master",
nationality="IT", date_of_birth="1979-04-11T00:00:00Z",
place_of_birth="Genoa", id_doc_type="PASSPORT",
id_doc_number="YA1234567", id_doc_issuing_state="IT",
id_doc_expiry="2030-01-01T00:00:00Z")],
)
def test_invalid_imo_check_digit_rejected() -> None:
with pytest.raises(ValueError, match="check digit"):
Vessel(imo_number="9074720", name="X", call_sign="AB", flag_state="PA")
def test_fal1_crew_count_matches_fal5(call: PortCall) -> None:
fal1 = generate_fal1(call)
fal5 = generate_fal5(call)
assert fal1.number_of_crew == fal5["crew_count"] == len(call.crew)
def test_audit_event_shape(call: PortCall) -> None:
cap = structlog.testing.LogCapture()
structlog.configure(processors=[cap])
generate_fal1(call)
assert cap.entries[-1]["event"] == "fal1_generated"
A passing run emits one JSON line per generated form; the FAL 1 event looks like {"event": "fal1_generated", "imo": "9074729", "crew": 1, "log_level": "info", "timestamp": "..."}. Assert on the structured keys, never on a formatted string.
Frequently Asked Questions
Do I generate all seven FAL forms from one model or seven separate inputs?
One model. FAL 1 through FAL 7 are seven projections of the same vessel, voyage, crew, stores, and cargo facts, and the fastest way to get a Single Window rejection is to let two forms disagree — a FAL 1 crew count that does not equal the FAL 5 row count, for example. Build one canonical PortCall, derive every count and every row from it, and treat each generator as a read-only view. That is also what makes the audit copy trustworthy: the reviewed document and the submitted document come from the same model_dump.
Why still map to EDIFACT if the Single Window takes JSON?
Because many national Single Windows and their downstream customs and border systems still exchange these declarations as UN/EDIFACT — PAXLST for crew and passenger lists, IFTDGN for dangerous goods, CUSREP and BERMAN for arrival and berth reporting. Generating those messages from validated pydantic models rather than string concatenation keeps the outbound segments free of qualifier drift and length overruns, and it lets you emit JSON and EDIFACT from the same typed source so the two representations can never diverge.
What belongs in the human-review artifact versus the submission payload?
The submission payload is the machine dataset the Single Window ingests; the human-review artifact is a flattened, readable rendering the ship’s agent checks before release. Both must be generated from the identical validated model so they are the same facts in two encodings. The review copy is where you surface soft warnings — an identity document expiring inside the call window, an unusual seafarer document type — that are legal but merit a human decision rather than an automatic block.
Related
- Regulatory Compliance Reference — the discipline defining how regulatory artifacts derive from typed source-of-truth models.
- Port Call Workflow Design — the state machine that produces the canonical port-call record these forms project from.
- IFCSUM EDI Message Parsing — the inbound counterpart: parse EDIFACT into typed models before generating outbound messages.
- Core Maritime Architecture & Taxonomy — the domain governing shared data models, validation gates, and audit trails.
Up: Regulatory Compliance Reference — the parent discipline governing maritime regulatory data modelling and compliance auditing.