Add OARMP routing engine, dashboard and documentation

- Complete routing engine: ingest, optimizer (CG+B&B), maintenance
  monitor, metrics, pipeline, quality checks
- Streamlit dashboard with Input/Output tab structure, editable data
  editors, interactive Folium map with satellite layer and maintenance
  base highlights, FH stacked bar chart with TTM availability
- CSV data files: AERONAVES, CHECKS, AIRPORTS, ESCALA DE VOO
- README, CONTEXTO and CHANGELOG added
- Remove legacy pre_process scripts and raw binary files (PDFs/xlsx)
- Update .gitignore to exclude outputs/, data/, raw/*.pdf, raw/*.xlsx

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

src/routing_engine/__init__.py

@@ -0,0 +1,6 @@
+"""Aircraft Routing Engine – Set Partitioning / Column Generation / Branch & Bound."""
+
+from .config import RoutingConfig, DEFAULT_CONFIG
+from .pipeline import RoutingPipeline
+
+__all__ = ["RoutingConfig", "DEFAULT_CONFIG", "RoutingPipeline"]

src/routing_engine/config.py

@@ -0,0 +1,74 @@
+"""Centralised configuration for the routing engine."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from pathlib import Path
+from datetime import datetime
+
+PROJECT_ROOT = Path(__file__).resolve().parents[2]
+
+
+@dataclass
+class RoutingConfig:
+    # ── Paths ──────────────────────────────────────────────────────────────────
+    project_root: Path = field(default_factory=lambda: PROJECT_ROOT)
+
+    @property
+    def raw_dir(self) -> Path:
+        return self.project_root / "raw"
+
+    @property
+    def raw_index_dir(self) -> Path:
+        return self.project_root / "data" / "raw_index"
+
+    @property
+    def processed_dir(self) -> Path:
+        return self.project_root / "data" / "processed"
+
+    @property
+    def reference_dir(self) -> Path:
+        return self.project_root / "data" / "reference"
+
+    @property
+    def quality_dir(self) -> Path:
+        return self.project_root / "data" / "quality"
+
+    @property
+    def schedules_dir(self) -> Path:
+        return self.project_root / "outputs" / "schedules"
+
+    @property
+    def figures_dir(self) -> Path:
+        return self.project_root / "outputs" / "figures"
+
+    @property
+    def exports_dir(self) -> Path:
+        return self.project_root / "outputs" / "exports"
+
+    # ── File names (auto-detected if left as empty string) ─────────────────────
+    flight_schedule_file: str = "ESCALA DE VOO MODELO 1.csv"
+    aircraft_file: str = "AERONAVES.csv"
+    checks_file: str = "CHECKS.csv"
+    airports_file: str = "AIRPORTS.csv"
+
+    # ── Calendar / planning ────────────────────────────────────────────────────
+    planning_year: int = field(default_factory=lambda: datetime.now().year)
+    maintenance_base_code: str = "SBMN"
+
+    # ── Operational constraints ────────────────────────────────────────────────
+    tat_minutes: int = 60          # minimum turnaround time between OFRAGs
+    aircraft_availability_offset_hours: int = 0  # hours before first OFRAG aircraft is available
+
+    # ── Optimiser parameters ───────────────────────────────────────────────────
+    big_m: float = 1e6             # penalty for uncovered OFRAGs (artificial variable)
+    cg_tolerance: float = 1e-6     # column-generation stopping threshold on reduced cost
+    max_cg_iterations: int = 200
+    mip_time_limit_seconds: int = 300
+    mip_gap: float = 0.0           # optimality gap (0 = exact)
+
+    # ── Logging ────────────────────────────────────────────────────────────────
+    log_level: str = "INFO"
+
+
+DEFAULT_CONFIG = RoutingConfig()

src/routing_engine/ingest.py

@@ -0,0 +1,322 @@
+"""
+Data ingestion layer.
+
+Reads raw files, parses dates/times, aggregates legs into OFRAG profiles,
+and computes TTM (Time to Maintenance) per aircraft.
+"""
+
+from __future__ import annotations
+
+import logging
+from datetime import datetime, timedelta
+from pathlib import Path
+from typing import Dict, List, Optional, Tuple
+
+import pandas as pd
+
+from .config import RoutingConfig
+from .inspect_files import read_aircraft, read_checks, read_airports, read_schedule
+
+logger = logging.getLogger(__name__)
+
+_BR_MONTHS = {
+    "jan": 1, "fev": 2, "mar": 3, "abr": 4,
+    "mai": 5, "jun": 6, "jul": 7, "ago": 8,
+    "set": 9, "out": 10, "nov": 11, "dez": 12,
+}
+
+
+# ── Low-level parsers ─────────────────────────────────────────────────────────
+
+def _parse_br_date(date_str: str, year: int) -> Optional[datetime]:
+    """Parse 'DD/MM/AAAA' or legacy 'DD/mon' → datetime."""
+    try:
+        parts = date_str.strip().split("/")
+        day = int(parts[0])
+        if len(parts) == 3:  # DD/MM/AAAA
+            return datetime(int(parts[2]), int(parts[1]), day)
+        # Legacy: DD/mon  (uses fallback year from config)
+        mon = _BR_MONTHS.get(parts[1].lower(), 0)
+        return datetime(year, mon, day) if mon else None
+    except Exception:
+        return None
+
+
+def _parse_time(time_str: str) -> Optional[Tuple[int, int]]:
+    """Parse 'HH:MM' or 'HH:MM:SS' → (hour, minute).  Returns None on failure."""
+    try:
+        parts = time_str.strip().split(":")
+        return int(parts[0]), int(parts[1])
+    except Exception:
+        return None
+
+
+def _parse_flight_hours(tempo_str: str) -> float:
+    """Convert 'HH:MM' flight-time string → decimal hours."""
+    try:
+        parts = tempo_str.strip().split(":")
+        return int(parts[0]) + int(parts[1]) / 60.0
+    except Exception:
+        return 0.0
+
+
+# ── Schedule parsing ──────────────────────────────────────────────────────────
+
+def _parse_schedule_datetimes(df: pd.DataFrame, year: int) -> pd.DataFrame:
+    """
+    Add DATETIME_DEP and DATETIME_ARR columns to the schedule DataFrame.
+
+    The DATA column holds the departure date; arrival is the same date unless
+    the arrival time string contains a second ':' colon (e.g. '00:30:00'),
+    which signals midnight crossing → arrival is departure date + 1 day.
+    """
+    dep_dts, arr_dts, fh_vals = [], [], []
+
+    for _, row in df.iterrows():
+        base_date = _parse_br_date(str(row["DATA"]), year)
+        if base_date is None:
+            dep_dts.append(pd.NaT)
+            arr_dts.append(pd.NaT)
+            fh_vals.append(0.0)
+            continue
+
+        dep_t = _parse_time(str(row["HORA_DEP"]))
+        arr_str = str(row["HORA_ARR"])
+        arr_t = _parse_time(arr_str)
+
+        if dep_t is None or arr_t is None:
+            dep_dts.append(pd.NaT)
+            arr_dts.append(pd.NaT)
+            fh_vals.append(0.0)
+            continue
+
+        dep_dt = base_date.replace(hour=dep_t[0], minute=dep_t[1])
+        arr_dt = base_date.replace(hour=arr_t[0], minute=arr_t[1])
+
+        # Midnight crossing: arrival string has 3 colon-separated parts OR arr <= dep
+        midnight_flag = arr_str.count(":") >= 2 or arr_dt <= dep_dt
+        if midnight_flag:
+            arr_dt += timedelta(days=1)
+
+        dep_dts.append(dep_dt)
+        arr_dts.append(arr_dt)
+        fh_vals.append(_parse_flight_hours(str(row["TEMPO_VOO"])))
+
+    df = df.copy()
+    df["DATETIME_DEP"] = dep_dts
+    df["DATETIME_ARR"] = arr_dts
+    df["FH_LEG"] = fh_vals
+    return df.dropna(subset=["DATETIME_DEP", "DATETIME_ARR"]).reset_index(drop=True)
+
+
+# ── OFRAG aggregation ─────────────────────────────────────────────────────────
+
+def build_ofrags(schedule_df: pd.DataFrame, base_codes: List[str]) -> pd.DataFrame:
+    """
+    Group schedule legs by OFRAG number and produce one row per OFRAG with:
+    ofrag_id, departure (first leg dep at base), arrival (last leg arr at base),
+    flight_hours (sum), origin, destination, starts_at_base, ends_at_base.
+    """
+    records = []
+    for ofrag_num, group in schedule_df.groupby("OFRAG", sort=False):
+        group = group.sort_values("DATETIME_DEP").reset_index(drop=True)
+
+        first = group.iloc[0]
+        last = group.iloc[-1]
+
+        total_fh = group["FH_LEG"].sum()
+        origin = str(first["DEP"]).strip().upper()
+        destination = str(last["ARR"]).strip().upper()
+
+        starts_base = origin in base_codes
+        ends_base = destination in base_codes
+
+        records.append(
+            {
+                "ofrag_id": f"OFRAG{int(ofrag_num):03d}",
+                "ofrag_num": int(ofrag_num),
+                "departure": first["DATETIME_DEP"],
+                "arrival": last["DATETIME_ARR"],
+                "flight_hours": round(total_fh, 3),
+                "origin": origin,
+                "destination": destination,
+                "starts_at_base": starts_base,
+                "ends_at_base": ends_base,
+                "n_legs": len(group),
+                "missions": ",".join(group["MISSAO"].dropna().unique()),
+            }
+        )
+
+    df = pd.DataFrame(records).sort_values("departure").reset_index(drop=True)
+    logger.info(
+        "Built %d OFRAGs  (%d start+end at base)",
+        len(df),
+        df["starts_at_base"].sum() & df["ends_at_base"].sum(),
+    )
+    return df
+
+
+# ── TTM computation ────────────────────────────────────────────────────────────
+
+def _check_cycles(current_fh: float, thresholds: List[float], durations_days: List[float]) -> List[Dict]:
+    """
+    Return the ordered sequence of upcoming maintenance checks for an aircraft
+    currently at *current_fh* total flight hours.
+
+    Each entry: {'fh_threshold': …, 'ttm': …, 'duration_hours': …}
+    The first entry is the immediately upcoming check; subsequent entries follow.
+    """
+    # Sort checks by threshold
+    paired = sorted(zip(thresholds, durations_days), key=lambda x: x[0])
+    cycles = []
+    prev_threshold = current_fh
+    for threshold, days in paired:
+        if threshold > current_fh:
+            ttm = threshold - prev_threshold if cycles else threshold - current_fh
+            cycles.append(
+                {
+                    "fh_threshold": threshold,
+                    "ttm": ttm,
+                    "duration_hours": days * 24.0,
+                }
+            )
+            prev_threshold = threshold
+
+    # Add a synthetic final cycle using the last interval (extrapolation)
+    if paired:
+        last_t, last_d = paired[-1]
+        second_last_t = paired[-2][0] if len(paired) >= 2 else 0.0
+        extra_ttm = last_t - second_last_t
+        cycles.append(
+            {
+                "fh_threshold": last_t + extra_ttm,
+                "ttm": extra_ttm,
+                "duration_hours": last_d * 24.0,
+            }
+        )
+
+    return cycles
+
+
+def build_fleet(
+    aircraft_df: pd.DataFrame,
+    checks_df: pd.DataFrame,
+    planning_start: datetime,
+) -> pd.DataFrame:
+    """
+    Combine aircraft and checks tables to produce the fleet reference DataFrame.
+
+    Columns: tail_number, model, fh_total, checks (list of cycle dicts),
+             ttm_hours (first upcoming TTM), available_from.
+    """
+    thresholds = checks_df["fh_threshold"].astype(float).tolist()
+    durations = checks_df["duration_days"].astype(float).tolist()
+
+    records = []
+    for _, row in aircraft_df.iterrows():
+        fh = float(row["fh_total"])
+        cycles = _check_cycles(fh, thresholds, durations)
+        records.append(
+            {
+                "tail_number": str(row["tail_number"]).strip(),
+                "model": str(row.get("model", "")).strip(),
+                "fh_total": fh,
+                "checks": cycles,
+                "ttm_hours": cycles[0]["ttm"] if cycles else 0.0,
+                "available_from": planning_start,
+            }
+        )
+
+    return pd.DataFrame(records)
+
+
+# ── Public entry points ───────────────────────────────────────────────────────
+
+def load_from_dfs(dfs: Dict[str, pd.DataFrame], cfg: RoutingConfig) -> Dict[str, pd.DataFrame]:
+    """
+    Same processing as load_all but accepts pre-loaded DataFrames.
+
+    dfs keys: 'aircraft', 'checks', 'airports', 'schedule'
+    The DataFrames may use the original CSV column names (synonym mapping is applied).
+    'schedule' must already have columns: DATA, ETAPA, DEP, ARR, HORA_DEP,
+    HORA_ARR, TEMPO_VOO, SEGMTO, MISSAO, OFRAG.
+    """
+    from .inspect_files import _map_columns, _AIRCRAFT_SYNONYMS, _CHECK_SYNONYMS, _AIRPORT_SYNONYMS
+
+    def _norm(df: pd.DataFrame, synonyms: Dict) -> pd.DataFrame:
+        df = df.copy().dropna(how="all")
+        mapping = _map_columns(list(df.columns), synonyms)
+        return df.rename(columns={v: k for k, v in mapping.items()})
+
+    aircraft_df = _norm(dfs["aircraft"], _AIRCRAFT_SYNONYMS)
+    checks_df = _norm(dfs["checks"], _CHECK_SYNONYMS)
+    airports_df = _norm(dfs["airports"], _AIRPORT_SYNONYMS)
+    schedule_raw = dfs["schedule"].copy()
+
+    # Drop obviously empty rows
+    for col in ("tail_number",):
+        if col in aircraft_df.columns:
+            aircraft_df = aircraft_df.dropna(subset=[col])
+    for col in ("fh_threshold",):
+        if col in checks_df.columns:
+            checks_df = checks_df.dropna(subset=[col])
+
+    # Maintenance bases – accept "1", "True", "true", 1
+    base_col = airports_df.get("is_maintenance_base", pd.Series(dtype=object))
+    base_mask = base_col.astype(str).str.strip().isin(["1", "True", "true"])
+    base_codes = airports_df.loc[base_mask, "airport_code"].str.upper().tolist() if "airport_code" in airports_df.columns else []
+    if not base_codes:
+        base_codes = [cfg.maintenance_base_code]
+    logger.info("Maintenance base(s): %s", base_codes)
+
+    schedule_df = _parse_schedule_datetimes(schedule_raw, cfg.planning_year)
+    ofrags_all = build_ofrags(schedule_df, base_codes)
+    ofrags = ofrags_all[ofrags_all["starts_at_base"] & ofrags_all["ends_at_base"]].copy().reset_index(drop=True)
+    logger.info("OFRAGs after base filter: %d / %d", len(ofrags), len(ofrags_all))
+
+    first_dep = ofrags["departure"].min() if not ofrags.empty else datetime.now()
+    planning_start = first_dep.replace(hour=0, minute=0, second=0, microsecond=0)
+    fleet = build_fleet(aircraft_df, checks_df, planning_start)
+
+    return {"ofrags": ofrags, "fleet": fleet, "airports": airports_df, "schedule": schedule_df}
+
+
+def load_all(cfg: RoutingConfig) -> Dict[str, pd.DataFrame]:
+    """
+    Read all raw files and return a dict with keys:
+        'ofrags'  – OFRAG profiles (only those that start and end at the maintenance base)
+        'fleet'   – fleet reference with TTM cycles
+        'airports'– airport table
+    """
+    raw = cfg.raw_dir
+
+    aircraft_df = read_aircraft(raw / cfg.aircraft_file)
+    checks_df = read_checks(raw / cfg.checks_file)
+    airports_df = read_airports(raw / cfg.airports_file)
+    schedule_raw = read_schedule(raw / cfg.flight_schedule_file)
+
+    # Maintenance-base ICAO codes
+    base_mask = airports_df.get("is_maintenance_base", pd.Series(dtype=int)).astype(int) == 1
+    base_codes = airports_df.loc[base_mask, "airport_code"].str.upper().tolist()
+    if not base_codes:
+        base_codes = [cfg.maintenance_base_code]
+    logger.info("Maintenance base(s): %s", base_codes)
+
+    # Parse schedule
+    schedule_df = _parse_schedule_datetimes(schedule_raw, cfg.planning_year)
+
+    # Build OFRAG table
+    ofrags_all = build_ofrags(schedule_df, base_codes)
+    ofrags = ofrags_all[ofrags_all["starts_at_base"] & ofrags_all["ends_at_base"]].copy()
+    ofrags = ofrags.reset_index(drop=True)
+    logger.info("OFRAGs after base filter: %d / %d", len(ofrags), len(ofrags_all))
+
+    # Planning horizon start = midnight of the earliest OFRAG departure day
+    # (aircraft are available from start-of-day, not the exact departure time)
+    first_dep = ofrags["departure"].min() if not ofrags.empty else datetime.now()
+    planning_start = first_dep.replace(hour=0, minute=0, second=0, microsecond=0)
+
+    # Build fleet
+    fleet = build_fleet(aircraft_df, checks_df, planning_start)
+
+    return {"ofrags": ofrags, "fleet": fleet, "airports": airports_df, "schedule": schedule_df}

src/routing_engine/inspect_files.py

@@ -0,0 +1,160 @@
+"""Auto-inspection of raw input files: detect encodings, separators, and column mappings."""
+
+from __future__ import annotations
+
+import logging
+from pathlib import Path
+from typing import Dict, List, Optional
+
+import pandas as pd
+
+logger = logging.getLogger(__name__)
+
+# ── Column-name synonym dictionaries ─────────────────────────────────────────
+
+_AIRCRAFT_SYNONYMS: Dict[str, List[str]] = {
+    "tail_number":  ["matricula", "tail", "registration", "aeronave", "ac"],
+    "model":        ["modelo", "model", "type", "tipo"],
+    "fh_total":     ["fh totais", "fh_total", "total fh", "horas totais", "flight hours"],
+}
+
+_CHECK_SYNONYMS: Dict[str, List[str]] = {
+    "check_name":        ["checks", "check", "nome", "name", "descricao"],
+    "fh_threshold":      ["fh", "threshold", "limite fh", "horas", "hours"],
+    "duration_days":     ["tempo de execucao", "duracao", "duration", "dias", "days"],
+    "location":          ["local de execucao", "local", "location", "base"],
+}
+
+_AIRPORT_SYNONYMS: Dict[str, List[str]] = {
+    "airport_code":      ["airport_code", "icao", "code", "codigo"],
+    "airport_name":      ["airport_name", "name", "nome"],
+    "is_maintenance_base": ["is_maintenance_base", "base", "manutencao"],
+}
+
+_SCHEDULE_COL_NAMES = [
+    "DATA", "ETAPA", "DEP", "ARR",
+    "HORA_DEP", "HORA_ARR", "TEMPO_VOO",
+    "SEGMTO", "MISSAO", "OFRAG",
+]
+
+
+def _normalise(text: str) -> str:
+    """Lowercase + strip accents (ASCII fold)."""
+    import unicodedata
+    nfkd = unicodedata.normalize("NFKD", str(text))
+    return "".join(c for c in nfkd if not unicodedata.combining(c)).lower().strip()
+
+
+def _detect_encoding(filepath: Path) -> str:
+    for enc in ("utf-8-sig", "utf-8", "latin-1", "cp1252"):
+        try:
+            with open(filepath, encoding=enc) as f:
+                f.read(2048)
+            return enc
+        except UnicodeDecodeError:
+            continue
+    return "latin-1"
+
+
+def _detect_separator(filepath: Path, encoding: str) -> str:
+    with open(filepath, encoding=encoding) as f:
+        sample = f.read(512)
+    counts = {sep: sample.count(sep) for sep in (";", ",", "\t", "|")}
+    return max(counts, key=counts.get)
+
+
+def _map_columns(df_columns: List[str], synonyms: Dict[str, List[str]]) -> Dict[str, str]:
+    """Return {canonical_name: actual_column_name} for columns that can be matched."""
+    mapping: Dict[str, str] = {}
+    norm_cols = {_normalise(c): c for c in df_columns}
+
+    for canonical, candidates in synonyms.items():
+        for cand in candidates:
+            if _normalise(cand) in norm_cols:
+                mapping[canonical] = norm_cols[_normalise(cand)]
+                break
+        if canonical not in mapping:
+            # Partial match fallback
+            for norm_col, orig_col in norm_cols.items():
+                if any(_normalise(cand) in norm_col for cand in candidates):
+                    mapping[canonical] = orig_col
+                    break
+
+    return mapping
+
+
+def inspect_file(filepath: Path) -> Dict:
+    """Return metadata dict for a single raw file."""
+    enc = _detect_encoding(filepath)
+    sep = _detect_separator(filepath, enc)
+    try:
+        df = pd.read_csv(filepath, sep=sep, encoding=enc, nrows=5)
+        return {
+            "path": str(filepath),
+            "encoding": enc,
+            "separator": sep,
+            "columns": list(df.columns),
+            "n_cols": len(df.columns),
+            "sample_rows": df.to_dict(orient="records"),
+        }
+    except Exception as exc:
+        logger.warning("Could not read %s: %s", filepath, exc)
+        return {"path": str(filepath), "error": str(exc)}
+
+
+def inspect_all(raw_dir: Path) -> Dict[str, Dict]:
+    """Inspect every CSV in raw_dir and return a metadata dict keyed by stem."""
+    results: Dict[str, Dict] = {}
+    for p in sorted(raw_dir.glob("*.csv")):
+        results[p.stem] = inspect_file(p)
+        logger.info("Inspected %s  (%d cols)", p.name, results[p.stem].get("n_cols", 0))
+    return results
+
+
+def read_aircraft(filepath: Path) -> pd.DataFrame:
+    enc = _detect_encoding(filepath)
+    sep = _detect_separator(filepath, enc)
+    df = pd.read_csv(filepath, sep=sep, encoding=enc)
+    mapping = _map_columns(list(df.columns), _AIRCRAFT_SYNONYMS)
+    logger.info("Aircraft column map: %s", mapping)
+    df = df.rename(columns={v: k for k, v in mapping.items()})
+    return df
+
+
+def read_checks(filepath: Path) -> pd.DataFrame:
+    enc = _detect_encoding(filepath)
+    sep = _detect_separator(filepath, enc)
+    df = pd.read_csv(filepath, sep=sep, encoding=enc)
+    mapping = _map_columns(list(df.columns), _CHECK_SYNONYMS)
+    logger.info("Checks column map: %s", mapping)
+    df = df.rename(columns={v: k for k, v in mapping.items()})
+    return df
+
+
+def read_airports(filepath: Path) -> pd.DataFrame:
+    enc = _detect_encoding(filepath)
+    sep = _detect_separator(filepath, enc)
+    df = pd.read_csv(filepath, sep=sep, encoding=enc)
+    mapping = _map_columns(list(df.columns), _AIRPORT_SYNONYMS)
+    logger.info("Airports column map: %s", mapping)
+    df = df.rename(columns={v: k for k, v in mapping.items()})
+    return df
+
+
+def read_schedule(filepath: Path) -> pd.DataFrame:
+    """Read the two-row merged-header flight schedule."""
+    enc = _detect_encoding(filepath)
+    sep = _detect_separator(filepath, enc)
+    df = pd.read_csv(
+        filepath,
+        sep=sep,
+        encoding=enc,
+        header=None,
+        skiprows=2,
+        names=_SCHEDULE_COL_NAMES,
+        dtype=str,
+    )
+    df = df.dropna(subset=["DATA", "OFRAG"])
+    df = df[df["DATA"].str.strip() != ""]
+    df = df[df["OFRAG"].str.strip() != ""]
+    return df.reset_index(drop=True)

src/routing_engine/maintenance_monitor.py

@@ -0,0 +1,173 @@
+"""
+Maintenance monitor: validate route TTM compliance and compute maintenance events.
+
+A Route is TTM-compliant if at every point in the route the accumulated flight
+hours since the last maintenance check do not exceed the current check cycle's TTM.
+Maintenance is forced when adding the next OFRAG would exceed that TTM.
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass, field
+from datetime import datetime, timedelta
+from typing import Dict, List, Optional, Tuple
+
+import pandas as pd
+
+from .config import RoutingConfig
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class MaintenanceEvent:
+    after_ofrag_id: Optional[str]   # None = before first OFRAG (proactive)
+    check_cycle_index: int           # which check cycle is being performed
+    fh_threshold: float
+    accum_fh_at_check: float         # accumulated FH at the moment of the check
+    ttm_loss: float                  # = cycle_ttm - accum_fh_at_check
+    calendar_start: datetime
+    calendar_end: datetime
+
+
+@dataclass
+class RouteValidationResult:
+    feasible: bool
+    total_flight_hours: float
+    total_ttm_loss: float
+    maintenance_events: List[MaintenanceEvent] = field(default_factory=list)
+    infeasibility_reason: str = ""
+
+
+def validate_route(
+    ofrag_sequence: List[str],
+    ofrags_df: pd.DataFrame,
+    aircraft_checks: List[Dict],   # ordered list of {'fh_threshold','ttm','duration_hours'}
+    aircraft_available_from: datetime,
+    cfg: RoutingConfig,
+) -> RouteValidationResult:
+    """
+    Simulate flying a route and return whether it is TTM-feasible.
+
+    aircraft_checks is the list produced by ingest._check_cycles(); it encodes
+    the sequence of upcoming check cycles for this specific aircraft.
+    """
+    ofrag_index = {row["ofrag_id"]: row for _, row in ofrags_df.iterrows()}
+    tat = timedelta(minutes=cfg.tat_minutes)
+
+    check_idx = 0           # which cycle we are currently in
+    accum_fh = 0.0          # FH accumulated in the current cycle
+    current_time = aircraft_available_from
+    total_fh = 0.0
+    total_loss = 0.0
+    events: List[MaintenanceEvent] = []
+
+    for ofrag_id in ofrag_sequence:
+        if ofrag_id not in ofrag_index:
+            return RouteValidationResult(
+                feasible=False, total_flight_hours=total_fh, total_ttm_loss=total_loss,
+                infeasibility_reason=f"OFRAG {ofrag_id} not found",
+            )
+
+        ofrag = ofrag_index[ofrag_id]
+        h = float(ofrag["flight_hours"])
+        dep = ofrag["departure"]
+        arr = ofrag["arrival"]
+
+        if check_idx >= len(aircraft_checks):
+            return RouteValidationResult(
+                feasible=False, total_flight_hours=total_fh, total_ttm_loss=total_loss,
+                infeasibility_reason="No remaining check cycles – route exceeds maintenance plan",
+            )
+
+        cycle = aircraft_checks[check_idx]
+
+        # Can the current cycle accommodate this OFRAG?
+        if accum_fh + h > cycle["ttm"]:
+            # Maintenance required before this OFRAG
+            dur = timedelta(hours=cycle["duration_hours"])
+            maint_end = current_time + dur
+
+            if maint_end + tat > dep:
+                return RouteValidationResult(
+                    feasible=False, total_flight_hours=total_fh, total_ttm_loss=total_loss,
+                    infeasibility_reason=(
+                        f"No time for maintenance before {ofrag_id}: "
+                        f"maint ends {maint_end}, OFRAG departs {dep}"
+                    ),
+                )
+
+            loss = cycle["ttm"] - accum_fh
+            ofrag_pos = ofrag_sequence.index(ofrag_id)
+            event = MaintenanceEvent(
+                after_ofrag_id=None if ofrag_pos == 0 else ofrag_sequence[ofrag_pos - 1],
+                check_cycle_index=check_idx,
+                fh_threshold=cycle["fh_threshold"],
+                accum_fh_at_check=accum_fh,
+                ttm_loss=loss,
+                calendar_start=current_time,
+                calendar_end=maint_end,
+            )
+            events.append(event)
+            total_loss += loss
+            accum_fh = 0.0
+            check_idx += 1
+            current_time = maint_end
+
+            # Verify new cycle can accommodate the OFRAG
+            if check_idx >= len(aircraft_checks):
+                return RouteValidationResult(
+                    feasible=False, total_flight_hours=total_fh, total_ttm_loss=total_loss,
+                    infeasibility_reason="No remaining check cycles after maintenance",
+                )
+            cycle = aircraft_checks[check_idx]
+            if h > cycle["ttm"]:
+                return RouteValidationResult(
+                    feasible=False, total_flight_hours=total_fh, total_ttm_loss=total_loss,
+                    infeasibility_reason=f"OFRAG {ofrag_id} ({h:.1f} FH) exceeds cycle TTM ({cycle['ttm']:.1f} FH)",
+                )
+
+        # Time feasibility: aircraft must be available before OFRAG departs
+        if current_time + tat > dep:
+            return RouteValidationResult(
+                feasible=False, total_flight_hours=total_fh, total_ttm_loss=total_loss,
+                infeasibility_reason=(
+                    f"Time conflict: aircraft ready at {current_time + tat}, "
+                    f"but {ofrag_id} departs at {dep}"
+                ),
+            )
+
+        accum_fh += h
+        total_fh += h
+        current_time = arr
+
+    return RouteValidationResult(
+        feasible=True,
+        total_flight_hours=total_fh,
+        total_ttm_loss=total_loss,
+        maintenance_events=events,
+    )
+
+
+def summarise_fleet_maintenance(solution_routes: List[Dict], ofrags_df: pd.DataFrame) -> pd.DataFrame:
+    """
+    Build a human-readable maintenance summary for the complete solution.
+    Each row = one maintenance event.
+    """
+    rows = []
+    for r in solution_routes:
+        for evt in r.get("maintenance_events", []):
+            rows.append(
+                {
+                    "tail_number": r["aircraft_id"],
+                    "after_ofrag": evt.after_ofrag_id,
+                    "check_cycle_index": evt.check_cycle_index,
+                    "fh_threshold": evt.fh_threshold,
+                    "accum_fh_at_check": round(evt.accum_fh_at_check, 2),
+                    "ttm_loss_hours": round(evt.ttm_loss, 2),
+                    "maint_start": evt.calendar_start,
+                    "maint_end": evt.calendar_end,
+                }
+            )
+    return pd.DataFrame(rows)

src/routing_engine/metrics.py

@@ -0,0 +1,117 @@
+"""Compute and format output metrics from the optimiser solution."""
+
+from __future__ import annotations
+
+from typing import Dict, List
+
+import pandas as pd
+
+
+def build_schedule_table(routes: List[Dict], ofrags_df: pd.DataFrame) -> pd.DataFrame:
+    """One row per OFRAG in the solution, showing which aircraft serves it."""
+    ofrag_lookup = ofrags_df.set_index("ofrag_id")
+    rows = []
+    for r in routes:
+        prev_arr = r.get("aircraft_available_from", None)
+        for pos, oid in enumerate(r["ofrag_ids"]):
+            maint_before = pos in r["maint_before_index"]
+            ofrag_row = ofrag_lookup.loc[oid] if oid in ofrag_lookup.index else {}
+            rows.append(
+                {
+                    "aircraft": r["aircraft_id"],
+                    "position_in_route": pos + 1,
+                    "maintenance_before": maint_before,
+                    "ofrag_id": oid,
+                    "missions": ofrag_row.get("missions", ""),
+                    "departure": ofrag_row.get("departure", pd.NaT),
+                    "arrival": ofrag_row.get("arrival", pd.NaT),
+                    "flight_hours": ofrag_row.get("flight_hours", 0.0),
+                    "n_legs": ofrag_row.get("n_legs", 0),
+                }
+            )
+    return pd.DataFrame(rows).sort_values(["aircraft", "departure"]).reset_index(drop=True)
+
+
+def build_maintenance_table(routes: List[Dict]) -> pd.DataFrame:
+    """One row per maintenance event in the solution."""
+    rows = []
+    for r in routes:
+        for evt in r.get("maintenance_events", []):
+            rows.append(
+                {
+                    "aircraft": r["aircraft_id"],
+                    "after_ofrag": evt.after_ofrag_id,
+                    "check_cycle_index": evt.check_cycle_index,
+                    "fh_threshold": evt.fh_threshold,
+                    "accum_fh_at_check": round(evt.accum_fh_at_check, 2),
+                    "ttm_loss_hours": round(evt.ttm_loss, 2),
+                    "maint_start": evt.calendar_start,
+                    "maint_end": evt.calendar_end,
+                }
+            )
+    return pd.DataFrame(rows)
+
+
+def build_fleet_summary(routes: List[Dict], fleet_df: pd.DataFrame) -> pd.DataFrame:
+    """One row per aircraft with utilisation statistics."""
+    utilisation: Dict[str, Dict] = {
+        aid: {
+            "flight_hours": 0.0,
+            "n_ofrags": 0,
+            "n_maint_events": 0,
+            "total_ttm_loss": 0.0,
+        }
+        for aid in fleet_df["tail_number"].tolist()
+    }
+
+    for r in routes:
+        aid = r["aircraft_id"]
+        if aid in utilisation:
+            utilisation[aid]["flight_hours"] += r["flight_hours"]
+            utilisation[aid]["n_ofrags"] += len(r["ofrag_ids"])
+            utilisation[aid]["n_maint_events"] += len(r.get("maintenance_events", []))
+            utilisation[aid]["total_ttm_loss"] += r["ttm_loss"]
+
+    rows = []
+    for _, ac in fleet_df.iterrows():
+        aid = ac["tail_number"]
+        u = utilisation.get(aid, {})
+        ttm0 = ac["ttm_hours"]
+        fh_done = u.get("flight_hours", 0.0)
+        rows.append(
+            {
+                "aircraft": aid,
+                "model": ac.get("model", ""),
+                "initial_ttm_h": round(ttm0, 2),
+                "flight_hours_scheduled": round(fh_done, 2),
+                "ttm_utilisation_pct": round(fh_done / ttm0 * 100, 1) if ttm0 > 0 else 0.0,
+                "n_ofrags_assigned": u.get("n_ofrags", 0),
+                "n_maintenance_events": u.get("n_maint_events", 0),
+                "total_ttm_loss_h": round(u.get("total_ttm_loss", 0.0), 2),
+            }
+        )
+
+    df = pd.DataFrame(rows)
+    df["idle"] = df["n_ofrags_assigned"] == 0
+    return df
+
+
+def solution_summary(result: Dict, ofrags_df: pd.DataFrame, fleet_df: pd.DataFrame) -> Dict:
+    """Return a compact summary dict of the optimisation result."""
+    routes = result.get("routes", [])
+    total_ofrags = len(ofrags_df)
+    covered = sum(len(r["ofrag_ids"]) for r in routes)
+    uncovered = result.get("uncovered_ofrags", [])
+    total_ttm_loss = sum(r["ttm_loss"] for r in routes)
+    n_maint = sum(len(r.get("maintenance_events", [])) for r in routes)
+
+    return {
+        "status": result.get("status", "?"),
+        "objective": round(result.get("objective", 0.0), 4),
+        "total_ofrags": total_ofrags,
+        "covered_ofrags": covered,
+        "uncovered_ofrags": uncovered,
+        "total_ttm_loss_hours": round(total_ttm_loss, 2),
+        "n_maintenance_events": n_maint,
+        "columns_generated": result.get("total_columns_generated", 0),
+    }

src/routing_engine/network_generator.py

@@ -0,0 +1,84 @@
+"""
+Time-space network for the OFRAG routing problem.
+
+Builds the precedence / adjacency structure used by the pricing subproblem:
+  - can_follow[i][j]             : OFRAG j can start directly after OFRAG i (no maintenance)
+  - can_follow_with_check[k][i][j]: OFRAG j can start after OFRAG i + check-cycle k maintenance
+"""
+
+from __future__ import annotations
+
+import logging
+from datetime import timedelta
+from typing import Dict, List
+
+import pandas as pd
+
+from .config import RoutingConfig
+
+logger = logging.getLogger(__name__)
+
+
+def build_adjacency(
+    ofrags: pd.DataFrame,
+    checks_unique_durations: List[float],
+    cfg: RoutingConfig,
+) -> Dict:
+    """
+    Pre-compute adjacency matrices.
+
+    Parameters
+    ----------
+    ofrags : sorted by departure, indexed 0..n-1
+    checks_unique_durations : list of check-cycle durations in hours (one per cycle)
+    cfg : RoutingConfig
+
+    Returns
+    -------
+    dict with keys:
+        'can_follow'            : List[List[bool]]  (n × n)
+        'can_follow_with_check' : List[List[List[bool]]]  (n_checks × n × n)
+        'n_ofrags'              : int
+        'n_checks'              : int
+    """
+    n = len(ofrags)
+    tat = timedelta(minutes=cfg.tat_minutes)
+
+    arrivals = ofrags["arrival"].tolist()
+    departures = ofrags["departure"].tolist()
+
+    # Direct adjacency (no maintenance between i and j)
+    can_follow = [[False] * n for _ in range(n)]
+    for i in range(n):
+        for j in range(n):
+            if i != j and arrivals[i] + tat <= departures[j]:
+                can_follow[i][j] = True
+
+    # Adjacency after check cycle k (maintenance inserted between i and j)
+    n_checks = len(checks_unique_durations)
+    can_follow_with_check = [
+        [[False] * n for _ in range(n)] for _ in range(n_checks)
+    ]
+    for k, dur_hours in enumerate(checks_unique_durations):
+        dur = timedelta(hours=dur_hours)
+        for i in range(n):
+            for j in range(n):
+                if i != j and arrivals[i] + dur + tat <= departures[j]:
+                    can_follow_with_check[k][i][j] = True
+
+    stats = {
+        "direct_edges": sum(sum(row) for row in can_follow),
+        "check_edges_per_cycle": [
+            sum(sum(row) for row in can_follow_with_check[k])
+            for k in range(n_checks)
+        ],
+    }
+    logger.info("Network  n_ofrags=%d  direct_edges=%d", n, stats["direct_edges"])
+
+    return {
+        "can_follow": can_follow,
+        "can_follow_with_check": can_follow_with_check,
+        "n_ofrags": n,
+        "n_checks": n_checks,
+        "stats": stats,
+    }

src/routing_engine/optimizer.py

@@ -0,0 +1,550 @@
+"""
+Aircraft Routing Optimizer
+==========================
+Solves the Aircraft Routing problem via:
+
+  1. Column Generation (CG)  – builds the LP relaxation of the Set Partitioning
+     formulation iteratively by pricing new columns with a label-setting DP.
+  2. Branch and Bound (B&B)  – applied by PuLP/CBC on the full column pool once
+     CG has converged, producing the optimal integer schedule.
+
+Mathematical formulation
+------------------------
+  Minimise   Σ_r  c_r · x_r                         (total TTM loss)
+  s.t.       Σ_{r: j∈r}  x_r  = 1   ∀ j ∈ OFRAGs  (each OFRAG covered once)
+             Σ_{r: a(r)=a} x_r ≤ 1  ∀ a ∈ Aircraft  (one route per aircraft)
+             x_r ∈ {0, 1}
+
+Column-generation pricing subproblem
+-------------------------------------
+For each aircraft a, find the feasible route r* that minimises:
+
+    c_r  –  Σ_{j ∈ r} π_j  –  μ_a
+
+where π_j = dual variable of coverage constraint for OFRAG j,
+      μ_a = dual variable of aircraft-usage constraint for aircraft a.
+
+This is solved by label-setting DP on the DAG of OFRAGs ordered by departure.
+
+Label at (node j, check_cycle k): (reduced_cost, accum_fh, route_trace)
+  - reduced_cost : running objective (negative → route is profitable to add)
+  - accum_fh     : flight hours accumulated since last maintenance
+  - route_trace  : list of (ofrag_idx, maint_before_flag)
+
+Dominance rule  (for labels at the same node and same check_cycle k)
+---------------------------------------------------------------------
+Label 1 (rc1, h1) dominates Label 2 (rc2, h2) iff:
+    rc1 ≤ rc2   AND   (h1 – rc1) ≥ (h2 – rc2)
+
+This ensures Label 1 is never worse than Label 2 on any future extension,
+whether that extension requires maintenance or not.
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass, field
+from datetime import timedelta
+from typing import Dict, List, Optional, Tuple
+
+import pandas as pd
+import pulp
+
+from .config import RoutingConfig
+from .maintenance_monitor import validate_route, MaintenanceEvent
+
+logger = logging.getLogger(__name__)
+
+_EPS = 1e-9   # float comparison tolerance
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Data structures
+# ─────────────────────────────────────────────────────────────────────────────
+
+@dataclass
+class Route:
+    """A feasible schedule for one aircraft (= one column in the master problem)."""
+    route_id: int
+    aircraft_id: str
+    ofrag_ids: List[str]              # OFRAGs served, in order
+    maint_before_index: List[int]     # positions in ofrag_ids where maint precedes
+    flight_hours: float
+    ttm_loss: float                   # objective cost
+    coverage: frozenset               # frozenset of ofrag_ids covered
+
+    @property
+    def cost(self) -> float:
+        return self.ttm_loss
+
+    def reduced_cost(self, dual_ofrags: Dict[str, float], dual_aircraft: float) -> float:
+        return self.ttm_loss - sum(dual_ofrags.get(j, 0.0) for j in self.ofrag_ids) - dual_aircraft
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Helpers
+# ─────────────────────────────────────────────────────────────────────────────
+
+def _prune_labels(
+    labels: List[Tuple[float, float, float, List]],
+) -> List[Tuple[float, float, float, List]]:
+    """
+    Remove dominated labels.  Each label is (rc, h, cost, trace).
+      rc   = running reduced cost (includes dual subtractions)
+      h    = accumulated FH since last maintenance event
+      cost = TTM loss accumulated so far (no dual subtractions)
+      trace = list of (ofrag_idx, maint_before_flag)
+
+    Dominance: Label 1 dominates Label 2 iff rc1 ≤ rc2 AND (h1 - rc1) ≥ (h2 - rc2).
+    Pareto front: sorted by rc asc, keep only those with strictly increasing (h - rc).
+    """
+    if len(labels) <= 1:
+        return labels
+    labels_sorted = sorted(labels, key=lambda t: t[0])
+    pareto: List[Tuple[float, float, float, List]] = []
+    best_value = -float("inf")
+    for rc, h, cost, trace in labels_sorted:
+        v = h - rc
+        if v > best_value - _EPS:
+            pareto.append((rc, h, cost, trace))
+            best_value = max(best_value, v)
+    return pareto
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Main optimizer
+# ─────────────────────────────────────────────────────────────────────────────
+
+class AircraftRoutingOptimizer:
+    """
+    Orchestrates Column Generation + Branch and Bound.
+
+    Parameters
+    ----------
+    ofrags_df : DataFrame with columns [ofrag_id, departure, arrival, flight_hours]
+    fleet_df  : DataFrame with columns [tail_number, fh_total, checks, ttm_hours, available_from]
+                  where checks = list of {'fh_threshold','ttm','duration_hours'}
+    adjacency : dict returned by network_generator.build_adjacency()
+    cfg       : RoutingConfig
+    """
+
+    def __init__(
+        self,
+        ofrags_df: pd.DataFrame,
+        fleet_df: pd.DataFrame,
+        adjacency: Dict,
+        cfg: RoutingConfig,
+    ):
+        self.ofrags = ofrags_df.sort_values("departure").reset_index(drop=True)
+        self.fleet = fleet_df.reset_index(drop=True)
+        self.adj = adjacency
+        self.cfg = cfg
+
+        self._ofrag_ids: List[str] = self.ofrags["ofrag_id"].tolist()
+        self._ofrag_fh: Dict[str, float] = dict(
+            zip(self.ofrags["ofrag_id"], self.ofrags["flight_hours"])
+        )
+        self._ofrag_idx: Dict[str, int] = {oid: i for i, oid in enumerate(self._ofrag_ids)}
+        self._aircraft_ids: List[str] = self.fleet["tail_number"].tolist()
+
+        self._columns: List[Route] = []
+        self._route_counter = 0
+
+    # ── Public API ────────────────────────────────────────────────────────────
+
+    def solve(self) -> Dict:
+        """Run Column Generation → B&B and return the solution dict."""
+        logger.info("=== Aircraft Routing Optimizer ===")
+        logger.info("OFRAGs: %d   Aircraft: %d", len(self._ofrag_ids), len(self._aircraft_ids))
+
+        self._initialise_columns()
+        logger.info("Initial column pool: %d", len(self._columns))
+
+        self._column_generation()
+        logger.info("After CG: %d columns", len(self._columns))
+
+        result = self._solve_mip()
+
+        # Post-process: attach maintenance events to selected routes
+        result["routes"] = self._attach_maintenance_events(result["selected_routes"])
+        return result
+
+    # ── Initialisation ────────────────────────────────────────────────────────
+
+    def _next_id(self) -> int:
+        self._route_counter += 1
+        return self._route_counter
+
+    def _make_route(
+        self,
+        aircraft_id: str,
+        ofrag_ids: List[str],
+        maint_before_index: List[int],
+        aircraft_checks: List[Dict],
+        starting_check_idx: int = 0,
+    ) -> Route:
+        """Build a Route object from its components and compute its cost."""
+        fh_total = sum(self._ofrag_fh[o] for o in ofrag_ids)
+
+        # Compute TTM loss: for each maintenance event, loss = cycle.ttm - accumulated_at_that_point
+        ttm_loss = 0.0
+        check_idx = starting_check_idx
+        accum = 0.0
+        for pos, oid in enumerate(ofrag_ids):
+            if pos in maint_before_index and check_idx < len(aircraft_checks):
+                ttm_loss += aircraft_checks[check_idx]["ttm"] - accum
+                accum = 0.0
+                check_idx += 1
+            accum += self._ofrag_fh[oid]
+
+        return Route(
+            route_id=self._next_id(),
+            aircraft_id=aircraft_id,
+            ofrag_ids=ofrag_ids,
+            maint_before_index=maint_before_index,
+            flight_hours=fh_total,
+            ttm_loss=round(ttm_loss, 6),
+            coverage=frozenset(ofrag_ids),
+        )
+
+    def _initialise_columns(self):
+        """Seed the column pool with one single-OFRAG route per aircraft × OFRAG pair."""
+        tat = timedelta(minutes=self.cfg.tat_minutes)
+        n = len(self._ofrag_ids)
+
+        for _, ac in self.fleet.iterrows():
+            checks = ac["checks"]
+            avail = ac["available_from"]
+            ttm0 = checks[0]["ttm"] if checks else 0.0
+
+            added = False
+            for i in range(n):
+                dep = self.ofrags.iloc[i]["departure"]
+                fh = self.ofrags.iloc[i]["flight_hours"]
+                if avail + tat <= dep and fh <= ttm0:
+                    route = self._make_route(
+                        ac["tail_number"], [self._ofrag_ids[i]], [], checks
+                    )
+                    self._columns.append(route)
+                    added = True
+                    break  # one seed per aircraft is enough
+
+            if not added:
+                # Aircraft needs maintenance before first OFRAG; try with maint flag
+                for i in range(n):
+                    fh = self.ofrags.iloc[i]["flight_hours"]
+                    if len(checks) > 1 and fh <= checks[1]["ttm"]:
+                        route = self._make_route(
+                            ac["tail_number"], [self._ofrag_ids[i]], [0], checks
+                        )
+                        self._columns.append(route)
+                        break
+
+    # ── Column Generation ─────────────────────────────────────────────────────
+
+    def _column_generation(self):
+        for iteration in range(self.cfg.max_cg_iterations):
+            lp = self._solve_rmp_lp()
+            if lp is None:
+                logger.warning("RMP infeasible at iteration %d – stopping CG", iteration)
+                break
+
+            pi = lp["dual_ofrags"]     # coverage duals
+            mu = lp["dual_aircraft"]   # aircraft-usage duals
+
+            added_any = False
+            for _, ac in self.fleet.iterrows():
+                new_routes = self._price(ac, pi, mu.get(ac["tail_number"], 0.0))
+                for r in new_routes:
+                    self._columns.append(r)
+                    added_any = True
+
+            logger.debug(
+                "CG iter %d  obj=%.4f  cols=%d  added=%s",
+                iteration, lp["objective"], len(self._columns), added_any,
+            )
+
+            if not added_any:
+                logger.info("CG converged at iteration %d", iteration)
+                break
+
+    def _solve_rmp_lp(self) -> Optional[Dict]:
+        """Solve the LP relaxation of the Restricted Master Problem."""
+        prob = pulp.LpProblem("RMP", pulp.LpMinimize)
+
+        x = {
+            col.route_id: pulp.LpVariable(f"x{col.route_id}", lowBound=0, upBound=1)
+            for col in self._columns
+        }
+        # Artificial slack for coverage (ensures LP feasibility)
+        y = {
+            oid: pulp.LpVariable(f"y_{oid}", lowBound=0)
+            for oid in self._ofrag_ids
+        }
+
+        prob += (
+            pulp.lpSum(col.cost * x[col.route_id] for col in self._columns)
+            + pulp.lpSum(self.cfg.big_m * y[oid] for oid in self._ofrag_ids)
+        )
+
+        for oid in self._ofrag_ids:
+            relevant = [x[c.route_id] for c in self._columns if oid in c.coverage]
+            prob += pulp.lpSum(relevant) + y[oid] == 1, f"cov_{oid}"
+
+        for aid in self._aircraft_ids:
+            relevant = [x[c.route_id] for c in self._columns if c.aircraft_id == aid]
+            prob += pulp.lpSum(relevant) <= 1, f"ac_{aid}"
+
+        prob.solve(pulp.PULP_CBC_CMD(msg=0))
+
+        if pulp.LpStatus[prob.status] not in ("Optimal",):
+            return None
+
+        # Standard LP dual (shadow price): positive for coverage constraints when
+        # covered by artificials (≈ big_M), negative/zero for aircraft-usage constraints.
+        # Pricing reduced cost = c_r - Σ π_j - μ_a  uses these directly.
+        dual_ofrags = {}
+        for oid in self._ofrag_ids:
+            c = prob.constraints.get(f"cov_{oid}")
+            dual_ofrags[oid] = c.pi if (c and c.pi is not None) else 0.0
+
+        dual_aircraft = {}
+        for aid in self._aircraft_ids:
+            c = prob.constraints.get(f"ac_{aid}")
+            dual_aircraft[aid] = c.pi if (c and c.pi is not None) else 0.0
+
+        return {
+            "objective": pulp.value(prob.objective),
+            "dual_ofrags": dual_ofrags,
+            "dual_aircraft": dual_aircraft,
+        }
+
+    # ── Pricing subproblem (label-setting DP) ─────────────────────────────────
+
+    def _price(
+        self,
+        aircraft: pd.Series,
+        dual_ofrags: Dict[str, float],
+        dual_aircraft: float,
+    ) -> List[Route]:
+        """
+        Find all routes for *aircraft* with negative reduced cost.
+
+        State: (ofrag_idx j, check_cycle k)
+        Label: (rc, h, cost, trace)
+          rc    = running reduced cost (cost minus dual contributions so far)
+          h     = accumulated FH since last maintenance in trace
+          cost  = TTM loss accumulated so far (rc + duals; stored separately to
+                  avoid re-deriving cost at harvest time)
+          trace = list of (ofrag_idx, maint_before_flag)
+        """
+        checks = aircraft["checks"]
+        avail = aircraft["available_from"]
+        aid = aircraft["tail_number"]
+        tat = timedelta(minutes=self.cfg.tat_minutes)
+        n = len(self._ofrag_ids)
+        n_checks = len(checks)
+
+        can_follow = self.adj["can_follow"]
+        can_with_check = self.adj["can_follow_with_check"]
+
+        # labels[j][k] = list of (rc, h, cost, trace)
+        labels: List[List[List]] = [[[] for _ in range(n_checks)] for _ in range(n)]
+
+        # --- Seed: single-OFRAG routes starting from SOURCE ---
+        # For cycle k=0: aircraft starts directly.
+        # For cycle k>0: aircraft must complete checks[0..k-1] before OFRAG j departs,
+        #   each with 0 accumulated FH (no prior OFRAGs) → full TTM loss per empty check.
+        #   The prior_loss IS the route cost so far; it is tracked in both rc and cost.
+        for j in range(n):
+            dep_j = self.ofrags.iloc[j]["departure"]
+            fh_j = self.ofrags.iloc[j]["flight_hours"]
+            oid_j = self._ofrag_ids[j]
+
+            earliest = avail      # earliest calendar time to enter the current cycle
+            prior_loss = 0.0      # TTM loss from empty prior checks (= route cost so far)
+
+            for k in range(n_checks):
+                can_reach = earliest + tat <= dep_j
+                fits_ttm  = fh_j <= checks[k]["ttm"] + _EPS
+
+                if can_reach and fits_ttm:
+                    rc = prior_loss - dual_ofrags.get(oid_j, 0.0)
+                    cost = prior_loss  # TTM loss from prior empty checks; no in-trace maint yet
+                    labels[j][k].append((rc, fh_j, cost, [(j, False)]))
+                    break  # lowest feasible cycle with valid timing
+
+                if not can_reach:
+                    break  # no point advancing to later cycles (time only grows)
+
+                # Advance to next cycle: empty check k before OFRAG j
+                if k + 1 < n_checks:
+                    prior_loss += checks[k]["ttm"]
+                    earliest = earliest + timedelta(hours=checks[k]["duration_hours"])
+
+        # --- Forward expansion ---
+        new_routes: List[Route] = []
+
+        for j in range(n):
+            for k in range(n_checks):
+                labels[j][k] = _prune_labels(labels[j][k])
+
+                for rc, h, cost, trace in labels[j][k]:
+                    # Harvest: if this partial route has negative reduced cost, record it
+                    final_rc = rc - dual_aircraft
+                    if final_rc < -self.cfg.cg_tolerance:
+                        new_routes.append(
+                            self._trace_to_route(aid, trace, cost, checks)
+                        )
+
+                    # Extend to OFRAG m
+                    for m in range(j + 1, n):
+                        fh_m = self.ofrags.iloc[m]["flight_hours"]
+                        oid_m = self._ofrag_ids[m]
+                        gain = dual_ofrags.get(oid_m, 0.0)
+
+                        # Option A: direct (no maintenance)
+                        if can_follow[j][m] and h + fh_m <= checks[k]["ttm"] + _EPS:
+                            new_rc = rc - gain
+                            new_h = h + fh_m
+                            new_cost = cost  # no new maintenance event
+                            new_trace = trace + [(m, False)]
+                            labels[m][k].append((new_rc, new_h, new_cost, new_trace))
+
+                        # Option B: maintenance between j and m (cycle k → k+1)
+                        if k + 1 < n_checks:
+                            can_m = can_with_check[k][j][m] if k < len(can_with_check) else False
+                            if can_m and fh_m <= checks[k + 1]["ttm"] + _EPS:
+                                loss = checks[k]["ttm"] - h
+                                new_rc = rc + loss - gain
+                                new_h = fh_m
+                                new_cost = cost + loss  # add maintenance loss
+                                new_trace = trace + [(m, True)]   # True = maint before m
+                                labels[m][k + 1].append((new_rc, new_h, new_cost, new_trace))
+
+        # De-duplicate by coverage set (keep cheapest)
+        seen: Dict[frozenset, Route] = {}
+        for r in new_routes:
+            key = r.coverage
+            if key not in seen or r.ttm_loss < seen[key].ttm_loss:
+                seen[key] = r
+
+        return list(seen.values())
+
+    def _trace_to_route(
+        self,
+        aircraft_id: str,
+        trace: List[Tuple[int, bool]],
+        cost: float,
+        checks: List[Dict],
+    ) -> Route:
+        """
+        Convert a DP trace into a Route object using the pre-computed cost.
+
+        The cost stored in the label already accounts for:
+          - TTM losses from any prior empty checks done before the first OFRAG
+            (seed at cycle k>0)
+          - TTM losses from maintenance events within the trace (Option B expansions)
+        We must NOT re-derive cost from maint_before_index because prior empty checks
+        are not represented in the trace entries.
+        """
+        ofrag_ids = [self._ofrag_ids[idx] for idx, _ in trace]
+        maint_before_index = [pos for pos, (_, mb) in enumerate(trace) if mb]
+        fh_total = sum(self._ofrag_fh[oid] for oid in ofrag_ids)
+        return Route(
+            route_id=self._next_id(),
+            aircraft_id=aircraft_id,
+            ofrag_ids=ofrag_ids,
+            maint_before_index=maint_before_index,
+            flight_hours=fh_total,
+            ttm_loss=round(cost, 6),
+            coverage=frozenset(ofrag_ids),
+        )
+
+    # ── MIP (Branch and Bound) ────────────────────────────────────────────────
+
+    def _solve_mip(self) -> Dict:
+        """Solve the integer Set Partitioning model using PuLP/CBC B&B."""
+        prob = pulp.LpProblem("AircraftRouting_MIP", pulp.LpMinimize)
+
+        x = {
+            col.route_id: pulp.LpVariable(f"x{col.route_id}", cat="Binary")
+            for col in self._columns
+        }
+        y = {
+            oid: pulp.LpVariable(f"y_{oid}", lowBound=0)
+            for oid in self._ofrag_ids
+        }
+
+        prob += (
+            pulp.lpSum(col.cost * x[col.route_id] for col in self._columns)
+            + pulp.lpSum(self.cfg.big_m * y[oid] for oid in self._ofrag_ids)
+        )
+
+        for oid in self._ofrag_ids:
+            relevant = [x[c.route_id] for c in self._columns if oid in c.coverage]
+            prob += pulp.lpSum(relevant) + y[oid] == 1, f"cov_{oid}"
+
+        for aid in self._aircraft_ids:
+            relevant = [x[c.route_id] for c in self._columns if c.aircraft_id == aid]
+            prob += pulp.lpSum(relevant) <= 1, f"ac_{aid}"
+
+        solver = pulp.PULP_CBC_CMD(
+            msg=1,
+            timeLimit=self.cfg.mip_time_limit_seconds,
+            gapRel=self.cfg.mip_gap,
+        )
+        prob.solve(solver)
+
+        status = pulp.LpStatus[prob.status]
+        obj = pulp.value(prob.objective) or 0.0
+
+        selected = [
+            col for col in self._columns
+            if (x[col.route_id].varValue or 0) > 0.5
+        ]
+        uncovered = [
+            oid for oid in self._ofrag_ids
+            if (y[oid].varValue or 0) > 0.5
+        ]
+
+        logger.info("MIP status: %s  objective: %.4f", status, obj)
+        logger.info("Selected routes: %d  Uncovered OFRAGs: %d", len(selected), len(uncovered))
+
+        return {
+            "status": status,
+            "objective": obj,
+            "selected_routes": selected,
+            "uncovered_ofrags": uncovered,
+            "total_columns_generated": len(self._columns),
+        }
+
+    # ── Post-processing ───────────────────────────────────────────────────────
+
+    def _attach_maintenance_events(self, selected_routes: List[Route]) -> List[Dict]:
+        """
+        Run the maintenance monitor on each selected route to get full event detail.
+        Returns a list of dicts (one per route) ready for output / dashboard.
+        """
+        out = []
+        for route in selected_routes:
+            ac_row = self.fleet[self.fleet["tail_number"] == route.aircraft_id].iloc[0]
+            val = validate_route(
+                ofrag_sequence=route.ofrag_ids,
+                ofrags_df=self.ofrags,
+                aircraft_checks=ac_row["checks"],
+                aircraft_available_from=ac_row["available_from"],
+                cfg=self.cfg,
+            )
+            out.append(
+                {
+                    "aircraft_id": route.aircraft_id,
+                    "ofrag_ids": route.ofrag_ids,
+                    "maint_before_index": route.maint_before_index,
+                    "flight_hours": route.flight_hours,
+                    "ttm_loss": route.ttm_loss,
+                    "feasible": val.feasible,
+                    "maintenance_events": val.maintenance_events,
+                }
+            )
+        return out

src/routing_engine/pipeline.py

@@ -0,0 +1,124 @@
+"""
+End-to-end orchestration pipeline.
+
+Usage
+-----
+    from src.routing_engine import RoutingPipeline, DEFAULT_CONFIG
+    pipe = RoutingPipeline(DEFAULT_CONFIG)
+    result = pipe.run()
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import sys
+from pathlib import Path
+from typing import Dict, Optional
+
+import pandas as pd
+
+from .config import RoutingConfig, DEFAULT_CONFIG
+from .ingest import load_all, load_from_dfs
+from .network_generator import build_adjacency
+from .optimizer import AircraftRoutingOptimizer
+from .quality import run_all as quality_check
+from .metrics import (
+    build_schedule_table,
+    build_maintenance_table,
+    build_fleet_summary,
+    solution_summary,
+)
+
+logger = logging.getLogger(__name__)
+
+
+def _setup_logging(level: str):
+    logging.basicConfig(
+        level=getattr(logging, level.upper(), logging.INFO),
+        format="%(asctime)s  %(levelname)-8s  %(name)s  %(message)s",
+        datefmt="%H:%M:%S",
+        handlers=[logging.StreamHandler(sys.stdout)],
+    )
+
+
+class RoutingPipeline:
+    def __init__(self, cfg: RoutingConfig = DEFAULT_CONFIG):
+        self.cfg = cfg
+        _setup_logging(cfg.log_level)
+
+    def run(self, save_outputs: bool = True, raw_dfs=None) -> Dict:
+        """Execute all pipeline stages and return a results dict."""
+        logger.info("-- Stage 1: Ingest --------------------------------------")
+        data = load_from_dfs(raw_dfs, self.cfg) if raw_dfs is not None else load_all(self.cfg)
+        ofrags = data["ofrags"]
+        fleet = data["fleet"]
+
+        logger.info("-- Stage 2: Quality check -------------------------------")
+        qc = quality_check(ofrags, fleet)
+        if not qc["ok"]:
+            logger.warning("Quality issues detected:\n  %s", "\n  ".join(qc["issues"]))
+
+        logger.info("-- Stage 3: Build network -------------------------------")
+        # Collect unique check-cycle durations (for adjacency-with-check matrices)
+        all_durations = set()
+        for _, ac in fleet.iterrows():
+            for c in ac["checks"]:
+                all_durations.add(c["duration_hours"])
+        sorted_durations = sorted(all_durations)
+
+        adj = build_adjacency(ofrags, sorted_durations, self.cfg)
+
+        logger.info("-- Stage 4: Optimise ------------------------------------")
+        opt = AircraftRoutingOptimizer(ofrags, fleet, adj, self.cfg)
+        result = opt.solve()
+
+        logger.info("-- Stage 5: Metrics -------------------------------------")
+        routes = result.get("routes", [])
+        schedule_df = build_schedule_table(routes, ofrags)
+        maint_df = build_maintenance_table(routes)
+        fleet_df = build_fleet_summary(routes, fleet)
+        summary = solution_summary(result, ofrags, fleet)
+
+        logger.info(
+            "\n%s",
+            "\n".join(f"  {k}: {v}" for k, v in summary.items()),
+        )
+
+        if save_outputs:
+            self._save(schedule_df, maint_df, fleet_df, summary)
+
+        return {
+            "summary": summary,
+            "schedule": schedule_df,
+            "maintenance": maint_df,
+            "fleet": fleet_df,
+            "ofrags": ofrags,
+            "raw_result": result,
+            "quality": qc,
+        }
+
+    def _save(
+        self,
+        schedule_df: pd.DataFrame,
+        maint_df: pd.DataFrame,
+        fleet_df: pd.DataFrame,
+        summary: Dict,
+    ):
+        cfg = self.cfg
+        cfg.schedules_dir.mkdir(parents=True, exist_ok=True)
+        cfg.exports_dir.mkdir(parents=True, exist_ok=True)
+        cfg.processed_dir.mkdir(parents=True, exist_ok=True)
+
+        schedule_df.to_csv(cfg.schedules_dir / "flight_strings.csv", index=False)
+        maint_df.to_csv(cfg.schedules_dir / "maintenance_events.csv", index=False)
+        fleet_df.to_csv(cfg.exports_dir / "fleet_utilisation.csv", index=False)
+
+        with open(cfg.exports_dir / "solution_summary.json", "w", encoding="utf-8") as f:
+            json.dump(
+                {k: (str(v) if not isinstance(v, (int, float, str, list, dict, bool, type(None))) else v)
+                 for k, v in summary.items()},
+                f, ensure_ascii=False, indent=2,
+            )
+
+        logger.info("Outputs saved to %s", cfg.schedules_dir)

src/routing_engine/quality.py

@@ -0,0 +1,105 @@
+"""Data quality checks on OFRAGs and fleet tables."""
+
+from __future__ import annotations
+
+import logging
+from typing import Dict, List
+
+import pandas as pd
+
+logger = logging.getLogger(__name__)
+
+
+def check_ofrags(ofrags: pd.DataFrame) -> Dict:
+    issues: List[str] = []
+
+    if ofrags.empty:
+        return {"ok": False, "issues": ["OFRAG table is empty"]}
+
+    missing_dep = ofrags["departure"].isna().sum()
+    if missing_dep:
+        issues.append(f"{missing_dep} OFRAGs have missing departure datetime")
+
+    missing_arr = ofrags["arrival"].isna().sum()
+    if missing_arr:
+        issues.append(f"{missing_arr} OFRAGs have missing arrival datetime")
+
+    neg_fh = (ofrags["flight_hours"] <= 0).sum()
+    if neg_fh:
+        issues.append(f"{neg_fh} OFRAGs have non-positive flight hours")
+
+    inverted = (ofrags["arrival"] <= ofrags["departure"]).sum()
+    if inverted:
+        issues.append(f"{inverted} OFRAGs have arrival ≤ departure")
+
+    not_base = (~ofrags["starts_at_base"] | ~ofrags["ends_at_base"]).sum()
+    if not_base:
+        issues.append(
+            f"{not_base} OFRAGs do not start AND end at the maintenance base "
+            f"(filtered out in load_all)"
+        )
+
+    ok = len(issues) == 0
+    if ok:
+        logger.info("OFRAG quality: OK (%d OFRAGs)", len(ofrags))
+    else:
+        for iss in issues:
+            logger.warning("OFRAG quality: %s", iss)
+
+    return {"ok": ok, "issues": issues}
+
+
+def check_fleet(fleet: pd.DataFrame) -> Dict:
+    issues: List[str] = []
+
+    if fleet.empty:
+        return {"ok": False, "issues": ["Fleet table is empty"]}
+
+    no_checks = fleet["checks"].apply(lambda c: len(c) == 0).sum()
+    if no_checks:
+        issues.append(f"{no_checks} aircraft have no maintenance check cycles")
+
+    zero_ttm = (fleet["ttm_hours"] <= 0).sum()
+    if zero_ttm:
+        issues.append(f"{zero_ttm} aircraft have TTM ≤ 0 (overdue maintenance)")
+
+    ok = len(issues) == 0
+    if ok:
+        logger.info("Fleet quality: OK (%d aircraft)", len(fleet))
+    else:
+        for iss in issues:
+            logger.warning("Fleet quality: %s", iss)
+
+    return {"ok": ok, "issues": issues}
+
+
+def check_feasibility(ofrags: pd.DataFrame, fleet: pd.DataFrame) -> Dict:
+    """High-level feasibility check before running the optimiser."""
+    issues: List[str] = []
+
+    if fleet.empty or ofrags.empty:
+        return {"ok": False, "issues": ["Empty inputs"]}
+
+    max_ttm = fleet["checks"].apply(
+        lambda cycles: max((c["ttm"] for c in cycles), default=0)
+    ).max()
+
+    infeasible_ofrags = ofrags[ofrags["flight_hours"] > max_ttm]
+    if not infeasible_ofrags.empty:
+        ids = infeasible_ofrags["ofrag_id"].tolist()
+        issues.append(
+            f"OFRAGs {ids} have flight_hours > max achievable TTM ({max_ttm:.1f} FH) "
+            f"– these cannot be served by any aircraft."
+        )
+
+    ok = len(issues) == 0
+    return {"ok": ok, "issues": issues}
+
+
+def run_all(ofrags: pd.DataFrame, fleet: pd.DataFrame) -> Dict:
+    r1 = check_ofrags(ofrags)
+    r2 = check_fleet(fleet)
+    r3 = check_feasibility(ofrags, fleet)
+    ok = r1["ok"] and r2["ok"] and r3["ok"]
+    issues = r1["issues"] + r2["issues"] + r3["issues"]
+    return {"ok": ok, "issues": issues}