Validating IATA SSIM Files with Pydantic

The exact task this guide solves is narrow and unforgiving: given a raw IATA Standard Schedules Information Manual (SSIM) file — a delimiter-free, fixed-width text dump where every field is defined by its byte position — turn it into a stream of typed, validated flight-leg records, and quarantine every malformed line with enough field-level context to remediate it. SSIM remains the foundational exchange format for airline schedule distribution, yet its rigid column architecture and legacy encoding conventions routinely introduce silent corruption at ingestion. When an operations team consumes an unvalidated SSIM payload, downstream systems inherit mismatched aircraft type codes, empty frequency bitmaps, and non-compliant time offsets, and those defects cascade into crew-pairing violations and audit findings. This page builds a deterministic gate that maps each fixed-width leg record to a Pydantic model, streams the file so a multi-gigabyte manifest never loads entirely into RAM, and emits a structured verdict per line.

This how-to applies the data schema validation rules patterns from the broader flight data ingestion architecture. It assumes SSIM records will later feed a duty time validation rule engine, so field names are anchored to the shared crew duty time taxonomy mapping and every timestamp is preserved with its zone rather than flattened to a naive local string.

Prerequisites

Before building the validator, confirm the following are in place:

The SSIM record layout you are validating

SSIM files use no delimiters; each field is recovered by absolute character position, so parsing must precede validation. For a Section 3 flight-leg record in the 7th-edition specification, the fields this guide validates sit at these 1-indexed positions: the record type indicator at position 1, the airline designator at 3–5, the flight number at 6–9, the itinerary variation identifier at 10–11, the operating-period start date at 15–21, the period end date at 22–28, the days-of-operation frequency bitmap at 29–35, the departure station at 37–39, the scheduled departure time (passenger, local) at 40–43, the UTC/local time variation at 44–48, the arrival station at 55–57, and the aircraft type at 73–75. Because a single mis-counted offset shifts every subsequent field, the column map is defined once, as data, and every slice is derived from it.

Step 1 — Define the column map and a streaming slicer

Hardcoding line[2:5] throughout the parser makes an edition change a code rewrite. Instead, express the layout as a table of (name, start, end) 1-indexed spans and derive Python slices from it. A generator reads the file line by line so memory stays flat regardless of file size.

from typing import Iterator

# 1-indexed inclusive spans from the SSIM 7th-edition Section 3 leg record.
LEG_COLUMN_MAP: dict[str, tuple[int, int]] = {
    "record_type": (1, 1),
    "airline_designator": (3, 5),
    "flight_number": (6, 9),
    "period_start": (15, 21),
    "period_end": (22, 28),
    "frequency_bitmap": (29, 35),
    "departure_station": (37, 39),
    "departure_time_local": (40, 43),
    "utc_variation": (44, 48),
    "arrival_station": (55, 57),
    "aircraft_type": (73, 75),
}


def slice_leg(line: str) -> dict[str, str]:
    """Extract one leg record's fields by position, trimming pad spaces."""
    return {
        name: line[start - 1:end].strip()
        for name, (start, end) in LEG_COLUMN_MAP.items()
    }


def stream_leg_lines(path: str) -> Iterator[dict[str, str]]:
    """Yield sliced Section 3 records without loading the whole file."""
    with open(path, encoding="utf-8", errors="strict") as handle:
        for line in handle:
            if line[:1] == "3":  # Section 3 = flight leg
                yield slice_leg(line)

Verify: feeding a single well-formed leg line to slice_leg returns a dict whose airline_designator is a two-or-three character code and whose frequency_bitmap is a seven-character field; stream_leg_lines yields only records beginning with 3, skipping header (1), carrier (2), and trailer (5) records.

Step 2 — Model the leg record with Pydantic v2

The sliced dict is still all strings. The Pydantic model is the syntactic gate: it coerces types, enforces field lengths, parses the operating dates, and rejects any record that deviates from the specification. Using pydantic compiles the schema into an optimised validator, keeping per-record latency low during high-throughput ingestion.

from __future__ import annotations

from datetime import date, datetime

from pydantic import BaseModel, ConfigDict, Field, field_validator


class SSIMLeg(BaseModel):
    model_config = ConfigDict(extra="forbid", str_strip_whitespace=True)

    airline_designator: str = Field(..., min_length=2, max_length=3)
    flight_number: int = Field(..., ge=1, le=9999)
    period_start: date
    period_end: date
    frequency_bitmap: str = Field(..., min_length=7, max_length=7)
    departure_station: str = Field(..., min_length=3, max_length=3)
    arrival_station: str = Field(..., min_length=3, max_length=3)
    departure_time_local: str = Field(..., min_length=4, max_length=4)
    aircraft_type: str = Field(..., min_length=3, max_length=3)

    @field_validator("period_start", "period_end", mode="before")
    @classmethod
    def parse_ssim_date(cls, value: str) -> date:
        # SSIM dates are DDMMMYY, e.g. 06JAN26.
        return datetime.strptime(value, "%d%b%y").date()

    @field_validator("departure_time_local")
    @classmethod
    def check_hhmm(cls, value: str) -> str:
        hours, minutes = int(value[:2]), int(value[2:])
        if not (0 <= hours <= 23 and 0 <= minutes <= 59):
            raise ValueError(f"invalid 24-hour local time: {value!r}")
        return value

Verify: SSIMLeg.model_validate(slice_leg(good_line)) returns a typed instance with period_start as a date object, while a line carrying 2461 as the departure time raises ValidationError on the departure_time_local field — the framework reports the exact field and message rather than failing silently.

Step 3 — Add cross-field and registry checks

Field-shape validation cannot know whether a leg names a live aerodrome, flies on any day, or uses a type in the certified fleet. A model_validator catches the frequency-bitmap and date-ordering rules that span fields; a separate pure function resolves the semantic checks against live registries, keeping every verdict reproducible from the audit log.

from pydantic import model_validator


class SSIMLeg(SSIMLeg):  # illustrative extension of the model above
    @model_validator(mode="after")
    def check_period_and_frequency(self) -> "SSIMLeg":
        if self.period_end < self.period_start:
            raise ValueError("period_end precedes period_start")
        # A day is 1-7 (Mon-Sun) at its position, or "0" for no service.
        if set(self.frequency_bitmap) == {"0"}:
            raise ValueError("zero-frequency leg: no operating days")
        for pos, char in enumerate(self.frequency_bitmap, start=1):
            if char not in {"0", str(pos)}:
                raise ValueError(f"frequency day {pos} malformed: {char!r}")
        return self

A zero-frequency bitmap — a leg that operates on no day — must trigger an explicit quarantine rather than defaulting to a daily assumption, because a silent default here fabricates flights the airline never files. The same discipline that keeps the EASA FTL compliance frameworks rules auditable applies to schedule data: every rejection is explained, never guessed around.

SSIM validation: parse, two gates, and quarantine branch A fixed-width SSIM file is read line by line by a streaming generator that keeps only Section 3 records, sliced by absolute character position from the LEG_COLUMN_MAP, then run through a syntactic gate (the Pydantic model coercing types, enforcing field lengths and parsing DDMMMYY dates) and a semantic gate (a model_validator plus airport and aircraft registries checking the frequency bitmap, date ordering and live codes). Records passing both gates flow on as typed records to the scheduling engine; records failing either gate — an invalid shape, a zero-frequency bitmap, or an unknown code — branch down to a quarantine sink carrying the line number, the raw sliced fields and the structured ValidationError list. TWO-GATE VALIDATION SSIM file fixed-width, no delimiters Line generator streams one line; keeps record_type 3 Positional slice by absolute column LEG_COLUMN_MAP Syntactic gate Pydantic model types · lengths · dates Semantic gate model_validator frequency · registries valid Typed record → scheduling engine invalid shape zero-freq · bad code Quarantine sink field-level diagnostics per line: line_no · raw fields · errors()

Figure: SSIM validation — a generator slices fixed-width records by the column map and feeds Pydantic models; invalid or zero-frequency records are quarantined with field diagnostics.

Step 4 — Drive the stream and route failures to quarantine

The final step wires the pieces together: iterate the sliced records, attempt validation, and split the stream into typed records and structured quarantine entries. Pydantic’s ValidationError.errors() yields machine-readable diagnostics — the failing field, the input value, and the message — which become the remediation payload a data steward acts on.

from pydantic import ValidationError


def validate_ssim(path: str) -> tuple[list[SSIMLeg], list[dict]]:
    valid: list[SSIMLeg] = []
    quarantined: list[dict] = []
    for line_no, record in enumerate(stream_leg_lines(path), start=1):
        try:
            valid.append(SSIMLeg.model_validate(record))
        except ValidationError as exc:
            quarantined.append(
                {
                    "line_no": line_no,
                    "raw": record,
                    "errors": exc.errors(include_url=False),
                }
            )
    return valid, quarantined

This immediate, per-line feedback loop lets a compliance team isolate malformed records without halting the whole ingestion run — a single bad leg is quarantined, not fatal. Records that clear both gates flow on as legally coherent input for the pairing engines, exactly as the parent data schema validation rules layer requires.

Verification queries and assertions

After wiring the steps together, pin the behaviour at the boundaries that break real schedule feeds:

def test_ssim_validation_boundaries():
    good = {
        "airline_designator": "BA", "flight_number": "112",
        "period_start": "06JAN26", "period_end": "28MAR26",
        "frequency_bitmap": "1234500", "departure_station": "LHR",
        "arrival_station": "JFK", "departure_time_local": "0855",
        "aircraft_type": "777", "utc_variation": "", "record_type": "3",
    }
    leg = SSIMLeg.model_validate(good)
    assert leg.flight_number == 112
    assert leg.period_start.isoformat() == "2026-01-06"

    zero = {**good, "frequency_bitmap": "0000000"}
    try:
        SSIMLeg.model_validate(zero)
        assert False, "zero-frequency leg should not validate"
    except ValidationError as exc:
        assert "zero-frequency" in str(exc)

    bad_time = {**good, "departure_time_local": "2461"}
    try:
        SSIMLeg.model_validate(bad_time)
        assert False, "invalid time should not validate"
    except ValidationError as exc:
        assert exc.errors()[0]["loc"] == ("departure_time_local",)

The assertions fix the exact contract: a valid leg parses its DDMMMYY period to a date, a zero-frequency bitmap is rejected by name, and an out-of-range clock time reports its offending field for the scheduler to fix before the schedule is published. The date and time semantics used here are documented in the Python datetime documentation, and the model API in the Pydantic documentation.

Failure modes and troubleshooting

Frequently Asked Questions

Why validate SSIM at ingestion instead of in the scheduling engine?

Because a malformed leg that reaches the scheduling engine has already contaminated pairing legality, crew notifications, and cumulative-duty arithmetic before anyone notices. Validating at the boundary means a duty time validation rule engine can assume every field it reads is present, typed, and range-checked, and the audit trail records precisely which records were rejected and why — the same gatekeeping contract the data schema validation rules layer defines for every feed.

How should I parse the SSIM days-of-operation frequency field?

The seven-character bitmap encodes Monday through Sunday by position: each position holds either its day number (1 for Monday in position 1, through 7 for Sunday) or 0 when the leg does not operate that day. Validate that every character is either 0 or equal to its 1-indexed position, and treat an all-zero field as an explicit error rather than a valid “never operates” leg.

Does Pydantic v2 change how I write SSIM validators versus v1?

Yes. Replace v1’s @validator with @field_validator (adding mode="before" when you need to transform the raw string, such as parsing a DDMMMYY date), and replace @root_validator with @model_validator(mode="after") for cross-field rules. Use model_validate instead of parse_obj, and read structured errors from ValidationError.errors(). The ConfigDict(extra="forbid") setting rejects any unexpected key, which catches column-map drift early.

How do I handle the local departure time and UTC variation together?

Keep them as separate fields through validation and combine them only when you need an absolute instant. The departure_time_local is the published passenger local time; the UTC/local time variation field gives the offset to apply. Storing the resolved UTC instant alongside the retained local time and offset preserves both the audit-relevant wall clock and the unambiguous instant the duty engine needs.

What belongs in the quarantine record for a rejected leg?

Enough context to remediate without re-deriving the failure: the source line number, the raw sliced fields, and the structured ValidationError.errors() list naming each failing field, its input value, and the violated constraint. That payload lets a data steward correct the feed or file a defect against the sending carrier, and it gives an audit an exact account of what was excluded and why.

Back to Data Schema Validation Rules.