From b042bcff41846c741c13d14cedb847936637cb89 Mon Sep 17 00:00:00 2001 From: Ramon DALMAU Date: Tue, 20 Oct 2026 10:55:35 +0200 Subject: [PATCH] fix(runners): one flight that cannot be converted no longer stops the run FleetRunner converted the whole JSON input with neats_json_to_flights before any per-flight error handling, so one flight with an invalid trajectory timestamp (or any malformed flight object) raised and stopped the run for every flight. neats_json_to_flights gains an opt-in `failures` list. When given, a flight that cannot be converted is left out and recorded as a FlightConversionFailure (input index, identifying attrs, error); without it, behaviour is unchanged. A top level that is an object or a string is a whole-file error and raises TypeError instead of being recorded flight by flight. Recorded errors carry no tracebacks, so they do not keep the input alive. FleetRunner._load_trajectories and scripts/run_json.py use the list: a failed flight becomes a "Failed at parsing" error record and the other flights go on. If every flight fails, the run stops before loading the weather and returns only error records. The run_json summary now prints the flight id of nested fleet error records. The default conversion path is unchanged: on 148 NEATS JSON files (85,128 flights) its output is identical to before. The units audit report records the status (D2 follow-up, D1 note). Co-Authored-By: Claude Opus 5.5 --- docs/internal/pyneats_units_audit.md | 15 +- scripts/run_json.py | 28 ++- src/pyneats/runners/fleet.py | 35 +++- src/pyneats/steps/parsing/neats_io.py | 270 +++++++++++++++--------- tests/test_invalid_flight_isolation.py | 274 +++++++++++++++++++++++++ 5 files changed, 515 insertions(+), 107 deletions(-) create mode 100644 tests/test_invalid_flight_isolation.py diff --git a/docs/internal/pyneats_units_audit.md b/docs/internal/pyneats_units_audit.md index bf5e735..308a2d6 100644 --- a/docs/internal/pyneats_units_audit.md +++ b/docs/internal/pyneats_units_audit.md @@ -11,7 +11,9 @@ note under its item. Items not listed here are open. | Item | Status | Branch / PR | Date | | --- | --- | --- | --- | -| D2 (trajectory `ts` offsets) | Fixed | `fix/d2-timestamp-utc` | 2026-10-20 | +| D2 (trajectory `ts` offsets) | Fixed | #70 | 2026-10-20 | +| One flight that cannot be converted no longer stops the run (D2 follow-up) | Fixed | `fix/isolate-invalid-flights` | 2026-10-20 | +| D1 (bare single-flight object) | Open; now a clear `TypeError` on the `FleetRunner` JSON path | `fix/isolate-invalid-flights` | 2026-10-20 | ## 1. Outcome @@ -444,6 +446,10 @@ Each entry gives: ### On the AO input path **D1. Bare single-flight object rejected.** +*Still open. On the `FleetRunner` JSON path and in `scripts/run_json.py`, a file whose top +level is an object or a string now raises `TypeError("expected a sequence of flight objects")` +instead of an `AttributeError` (`fix/isolate-invalid-flights`). Accepting a bare object is not +decided.* 1. Top-level JSON → `neats_json_to_flights` (`neats_io.py:45-46`) → `FleetRunner._load_trajectories` (`fleet.py:561-566`) and `scripts/run_json.py:238-240`. 2. Expected: the AO schema's `anyOf` allows one object. Observed: iterating a dict yields its @@ -459,8 +465,11 @@ Each entry gives: *Fixed on branch `fix/d2-timestamp-utc` (see the follow-up status table). The UTC offset is now applied. Fractional seconds are truncated. A lowercase `t` or `z` and a space separator are accepted. A `ts` without offset is read as UTC (maintainer decision). An -invalid `ts` still aborts the batch, now with an error naming the flight and point; isolating -it per flight is a separate change. No effect on existing data: on 148 NEATS JSON files on +invalid `ts` raises an error naming the flight and point. On the `FleetRunner` JSON path and +in `scripts/run_json.py`, such a flight (or any flight object that cannot be converted) becomes +a `Failed at parsing` error record and the other flights go on (`fix/isolate-invalid-flights`); +a direct call to `neats_json_to_flights` without a `failures` list still raises. If every +flight fails, the run returns only error records. No effect on existing data: on 148 NEATS JSON files on the NEATS server (85,128 flights, 10,995,586 points, all `+0000`, `Z` or `+00:00`) the output is identical.* 1. `ts` → `neats_io.py:61-68` → `neats_parser.py:233-242`. diff --git a/scripts/run_json.py b/scripts/run_json.py index 7ff0d66..ea2d5c9 100644 --- a/scripts/run_json.py +++ b/scripts/run_json.py @@ -31,6 +31,7 @@ from pyneats.runners.flight import RunnerConfig from pyneats.runners.large_emitter import FleetRunnerLargeEmitter from pyneats.runners.large_emitter import FlightRunnerLargeEmitter +from pyneats.steps.parsing.neats_io import FlightConversionFailure from pyneats.steps.parsing.neats_io import neats_json_to_flights from pyneats.steps.weather.weather_store import ZarrPaths from pyneats.steps.weather.weather_store import get_weather_from_zarr @@ -89,7 +90,10 @@ def _print_summary( print("Failed flights:") print("-" * 60) for err in errors: - fid = err.get("flight_id", "unknown") + fid = str( + err.get("flight_id") + or err.get("flight_information", {}).get("flight_id", "unknown") + ) msg = err.get("error", err.get("error_msg", str(err))) step = err.get("step_name", "") detail = f"step={step:<20} {msg}" if step else msg @@ -150,13 +154,25 @@ def run_flight_loop( weather = get_weather_from_zarr(zarr_paths) log.info("Weather loaded for flight-mode execution") - # 2. Convert JSON dicts → list[pd.DataFrame] - flight_dfs = neats_json_to_flights(input_flights) - log.info("Converted %d flights from JSON", len(flight_dfs)) + # 2. Convert JSON dicts → list[pd.DataFrame]; a flight that cannot be + # converted is reported as an error and the others go on. + failures: list[FlightConversionFailure] = [] + flight_dfs = neats_json_to_flights(input_flights, failures=failures) + log.info( + "Converted %d flights from JSON (%d could not be converted)", + len(flight_dfs), + len(failures), + ) # 3. Loop over flights succeeded = 0 - errors: list[dict[str, Any]] = [] + errors: list[dict[str, Any]] = [ + { + "flight_id": f.attrs.get("flight_id", f"input_flight_{f.index}"), + "error": str(f.error), + } + for f in failures + ] cfg = RunnerConfig( params={ @@ -186,7 +202,7 @@ def run_flight_loop( errors.append({"flight_id": fid, "error": str(exc)}) # 4. Print summary - _print_summary("FlightRunnerLargeEmitter", len(flight_dfs), succeeded, errors) + _print_summary("FlightRunnerLargeEmitter", len(input_flights), succeeded, errors) # --------------------------------------------------------------------------- diff --git a/src/pyneats/runners/fleet.py b/src/pyneats/runners/fleet.py index 66e5845..6dd7569 100644 --- a/src/pyneats/runners/fleet.py +++ b/src/pyneats/runners/fleet.py @@ -60,6 +60,7 @@ from pyneats.steps.emissions.protocol import EmissionModel from pyneats.steps.emissions.views import FlightWithEmissions from pyneats.steps.interpolation.protocol import TrajectoryInterpolator +from pyneats.steps.parsing.neats_io import FlightConversionFailure from pyneats.steps.parsing.neats_io import neats_json_to_flights from pyneats.steps.parsing.neats_io import split_df_into_flights from pyneats.steps.parsing.protocol import TrajectoryParser @@ -626,9 +627,12 @@ def _parser_params(self) -> dict[str, Any]: # ------------------------------------------------------------------------- def _load_data(self) -> Self: - return ( - self._load_trajectories()._load_weather() # pylint: disable=protected-access # pylint: disable=protected-access - ) + self._load_trajectories() + # Every flight already failed at loading: the run stops here, so + # the weather is not needed. + if self._pipeline_aborted: + return self + return self._load_weather() def _load_trajectories(self) -> Self: # No unit conversion here: true_airspeed must already be m/s and q_fuel @@ -638,8 +642,29 @@ def _load_trajectories(self) -> Self: logger.info("Loading trajectories from %s", path) with path.open("r", encoding="utf-8") as f: json_flights: list[dict[str, Any]] = json.load(f) - self.source_fleet = neats_json_to_flights(json_flights) - logger.info("Loaded %d flights from JSON", len(self.source_fleet)) + # A flight that cannot be converted (e.g. an invalid trajectory + # timestamp) becomes an error record; the other flights go on. + failures: list[FlightConversionFailure] = [] + self.source_fleet = neats_json_to_flights(json_flights, failures=failures) + for failure in failures: + flight_id = failure.attrs.get( + "flight_id", f"input_flight_{failure.index}" + ) + self.error_records.append( + _create_error_record( + failure, flight_id, "parsing", repr(failure.error) + ) + ) + logger.warning( + "Flight %s failed at parsing: %s", flight_id, failure.error + ) + logger.info( + "Loaded %d flights from JSON (%d could not be converted)", + len(self.source_fleet), + len(failures), + ) + if failures and not self.source_fleet: + self._pipeline_aborted = True return self if self.cfg.trajectory_dataframe is not None: diff --git a/src/pyneats/steps/parsing/neats_io.py b/src/pyneats/steps/parsing/neats_io.py index 97b8552..f2e3a73 100644 --- a/src/pyneats/steps/parsing/neats_io.py +++ b/src/pyneats/steps/parsing/neats_io.py @@ -6,9 +6,11 @@ from __future__ import annotations +import logging import re from collections.abc import Mapping from collections.abc import Sequence +from dataclasses import dataclass from datetime import UTC from datetime import datetime from typing import Any @@ -18,6 +20,9 @@ from pyneats.steps.parsing.views import Flight4D +logger = logging.getLogger(__name__) + + # Fractional seconds of the seconds field (".5" in "2025-11-02T10:02:00.5Z"). _FRACTIONAL_SECONDS_RE = re.compile( r"^(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2})\.\d+", re.ASCII @@ -65,10 +70,58 @@ def _ts_to_utc_string(ts_raw: str) -> str: ) from None +@dataclass(frozen=True) +class FlightConversionFailure: + """A flight that ``neats_json_to_flights`` could not convert. + + ``index`` is the flight's position in the input sequence. ``attrs`` holds + the identifying attrs that could be read (``flight_id``, airports, + ``aobt``, ``aircraft_type``, ``engine_uid``); it may be empty. + """ + + index: int + attrs: dict[str, Any] + error: Exception + + +def _identity_attrs(flight: Any) -> dict[str, Any]: + """Read the identifying attrs of a flight object, never raising.""" + try: + fi = flight.get("flight_information", {}) or {} + ap = fi.get("aircraft_properties", {}) or {} + found = { + "flight_id": fi.get("flight_identification"), + "departure_airport": fi.get("departure_airport"), + "arrival_airport": fi.get("arrival_airport"), + "aobt": fi.get("departure_date_time"), + "aircraft_type": ap.get("aircraft_type"), + "engine_uid": ap.get("engine_uid"), + } + except Exception: # malformed flight object + return {} + return {k: v for k, v in found.items() if v is not None} + + +def _without_tracebacks(exc: Exception) -> Exception: + """Drop the tracebacks of ``exc`` and of every exception it chains to. + + A stored traceback keeps its frames alive, and through them the whole + input of ``neats_json_to_flights``. + """ + seen: set[int] = set() + e: BaseException | None = exc + while e is not None and id(e) not in seen: + seen.add(id(e)) + e.__traceback__ = None + e = e.__cause__ or e.__context__ + return exc + + def neats_json_to_flights( flights: Sequence[Mapping[str, Any]], *, model_type: str = "NM", + failures: list[FlightConversionFailure] | None = None, ) -> list[pd.DataFrame]: """ Convert a sequence of NEATS / NM JSON flight objects into a list of @@ -89,11 +142,19 @@ def neats_json_to_flights( docs/internal/pyneats_units_audit.md). model_type : str, optional Model type label to assign to each flight (default is "NM"). + failures : list of FlightConversionFailure, optional + If ``None`` (default), the first flight that cannot be converted + raises and stops the conversion. If a list is given, such a flight is + left out of the result and recorded in ``failures`` instead, so the + other flights are still converted. ``flights`` must then be a + sequence of flight objects; a mapping or a string raises + ``TypeError``. Returns ------- list of pd.DataFrame - List of per-flight DataFrames in Flight4D schema. + List of per-flight DataFrames in Flight4D schema, in input order + (without the failed flights when ``failures`` is given). Notes ----- @@ -167,107 +228,130 @@ def neats_json_to_flights( ``"naphtalene"``, so the ``allowed_attrs`` filter drops the value here (see docs/internal/pyneats_units_audit.md). """ + if failures is not None and isinstance(flights, (Mapping, str, bytes)): + # A whole-file shape error, not a flight error: do not record each + # key or character as a failed flight. + raise TypeError( + f"expected a sequence of flight objects, got {type(flights).__name__}" + ) + result: list[pd.DataFrame] = [] - for flight in flights: - fi = flight.get("flight_information", {}) or {} - ap = fi.get("aircraft_properties", {}) or {} - fp = fi.get("fuel_properties", {}) or {} - - # ...existing code... - - # ----------------------------- - # 2. Per-point rows (canonical column names) - # ----------------------------- - rows: list[dict[str, Any]] = [] - - tr = fi.get("trajectory", {}) or {} - pts = tr.get("trajectory_data", []) or [] - - for i, p in enumerate(pts): - ts_raw = p.get("ts") - - if ts_raw is None: - time_over = None - else: - # "2025-11-02T12:02:00+02:00" → "2025-11-02 10:02:00" (UTC); - # the parser reads the result as UTC. - try: - time_over = _ts_to_utc_string(ts_raw) - except ValueError as exc: - raise ValueError( - f"flight {fi.get('flight_identification')!r}, " - f"trajectory point {i}: {exc}" - ) from exc - - rows.append( - { - # canonical 4D columns - "latitude": p.get("lat"), - "longitude": p.get("lon"), - "time": time_over, - # NB: still FL in hundreds of feet; conversion stays in parser - "altitude": p.get("fl"), - # optional - "fuel_flow": p.get("ff"), - "engine_efficiency": p.get("ee"), - "aircraft_mass": p.get("am"), - # PyNEATS input unit: m/s. The AO JSON field `tas` is in - # knots. The NEATS application converts knots to m/s - # (x 1852 / 3600) before calling PyNEATS. Values received - # here must already be in m/s. - "true_airspeed": p.get("tas"), - } + for index, flight in enumerate(flights): + if failures is None: + result.append(_flight_to_frame(flight, model_type)) + continue + try: + result.append(_flight_to_frame(flight, model_type)) + except Exception as exc: + logger.debug("Flight %d could not be converted", index, exc_info=exc) + failures.append( + FlightConversionFailure( + index=index, + attrs=_identity_attrs(flight), + error=_without_tracebacks(exc), + ) ) - df = pd.DataFrame(rows) + return result + - # Keep only columns that are part of the Flight4D schema - if not df.empty: - allowed_cols = set(Flight4D.REQUIRED) | set(Flight4D.OPTIONAL) - df = df[[c for c in df.columns if c in allowed_cols]] +def _flight_to_frame(flight: Mapping[str, Any], model_type: str) -> pd.DataFrame: + """Convert one NEATS / NM JSON flight object (see ``neats_json_to_flights``).""" + fi = flight.get("flight_information", {}) or {} + ap = fi.get("aircraft_properties", {}) or {} + fp = fi.get("fuel_properties", {}) or {} - # Drop columns that are entirely NaN - df = df.dropna(axis=1, how="all") + # ----------------------------- + # 2. Per-point rows (canonical column names) + # ----------------------------- + rows: list[dict[str, Any]] = [] - # Attach attrs (canonical) - # Reconstruct flight-level attrs as in original logic - allowed_attrs = set(Flight4D.ATTRS_REQUIRED) | set(Flight4D.ATTRS_OPTIONAL) - attrs = { - k: v - for k, v in { - "flight_id": fi.get("flight_identification"), - "departure_airport": fi.get("departure_airport"), - "arrival_airport": fi.get("arrival_airport"), - "model_type": model_type, - "aobt": fi.get("departure_date_time"), - "arrival_date_time": fi.get("arrival_date_time"), - # Aircraft - "aircraft_type": ap.get("aircraft_type"), - "aircraft_series": ap.get("aircraft_version"), - "engine_uid": ap.get("engine_uid"), - "takeoff_weight": ap.get("takeoff_mass"), - "payload_factor": ap.get("load_factor"), - # Fuel (canonical names aligned with NEATSFuel / Flight4D) - "hydrogen_content": fp.get("hydrogen_content"), - "h_c_ratio": fp.get("hydrogen_per_carbon_ratio"), - "aromatic_content": fp.get("aromatic_content"), - # PyNEATS input unit: J/kg of fuel (lower calorific value). - # The AO JSON field `calorific_value` is in MJ/kg of fuel. - # The NEATS application multiplies by 1e6 before calling - # PyNEATS. Values received here must already be in J/kg; do - # not convert again. - "q_fuel": fp.get("calorific_value"), - "sulphur_content": fp.get("sulphur"), - "naphthalene": fp.get("naphthalene"), - }.items() - if v is not None and k in allowed_attrs - } - df.attrs = attrs + tr = fi.get("trajectory", {}) or {} + pts = tr.get("trajectory_data", []) or [] - result.append(df) + for i, p in enumerate(pts): + ts_raw = p.get("ts") - return result + if ts_raw is None: + time_over = None + else: + # "2025-11-02T12:02:00+02:00" → "2025-11-02 10:02:00" (UTC); + # the parser reads the result as UTC. + try: + time_over = _ts_to_utc_string(ts_raw) + except ValueError as exc: + raise ValueError( + f"flight {fi.get('flight_identification')!r}, " + f"trajectory point {i}: {exc}" + ) from exc + + rows.append( + { + # canonical 4D columns + "latitude": p.get("lat"), + "longitude": p.get("lon"), + "time": time_over, + # NB: still FL in hundreds of feet; conversion stays in parser + "altitude": p.get("fl"), + # optional + "fuel_flow": p.get("ff"), + "engine_efficiency": p.get("ee"), + "aircraft_mass": p.get("am"), + # PyNEATS input unit: m/s. The AO JSON field `tas` is in + # knots. The NEATS application converts knots to m/s + # (x 1852 / 3600) before calling PyNEATS. Values received + # here must already be in m/s. + "true_airspeed": p.get("tas"), + } + ) + + df = pd.DataFrame(rows) + + # Keep only columns that are part of the Flight4D schema + if not df.empty: + allowed_cols = set(Flight4D.REQUIRED) | set(Flight4D.OPTIONAL) + df = df[[c for c in df.columns if c in allowed_cols]] + + # Drop columns that are entirely NaN + df = df.dropna(axis=1, how="all") + + # Attach attrs (canonical) + # Reconstruct flight-level attrs as in original logic + allowed_attrs = set(Flight4D.ATTRS_REQUIRED) | set(Flight4D.ATTRS_OPTIONAL) + attrs = { + k: v + for k, v in { + "flight_id": fi.get("flight_identification"), + "departure_airport": fi.get("departure_airport"), + "arrival_airport": fi.get("arrival_airport"), + "model_type": model_type, + "aobt": fi.get("departure_date_time"), + "arrival_date_time": fi.get("arrival_date_time"), + # Aircraft + "aircraft_type": ap.get("aircraft_type"), + "aircraft_series": ap.get("aircraft_version"), + "engine_uid": ap.get("engine_uid"), + "takeoff_weight": ap.get("takeoff_mass"), + "payload_factor": ap.get("load_factor"), + # Fuel (canonical names aligned with NEATSFuel / Flight4D) + "hydrogen_content": fp.get("hydrogen_content"), + "h_c_ratio": fp.get("hydrogen_per_carbon_ratio"), + "aromatic_content": fp.get("aromatic_content"), + # PyNEATS input unit: J/kg of fuel (lower calorific value). + # The AO JSON field `calorific_value` is in MJ/kg of fuel. + # The NEATS application multiplies by 1e6 before calling + # PyNEATS. Values received here must already be in J/kg; do + # not convert again. + "q_fuel": fp.get("calorific_value"), + "sulphur_content": fp.get("sulphur"), + "naphthalene": fp.get("naphthalene"), + }.items() + if v is not None and k in allowed_attrs + } + df.attrs = attrs + + return df def concat_neats_flight( diff --git a/tests/test_invalid_flight_isolation.py b/tests/test_invalid_flight_isolation.py new file mode 100644 index 0000000..f499aff --- /dev/null +++ b/tests/test_invalid_flight_isolation.py @@ -0,0 +1,274 @@ +"""A flight that cannot be converted from JSON fails alone. + +``neats_json_to_flights`` keeps raising by default. With a ``failures`` list, +the flight is left out and recorded instead, and ``FleetRunner`` turns it +into a per-flight error record while the other flights go on. +""" + +from __future__ import annotations + +import copy +import gc +import json +import weakref +from pathlib import Path +from typing import Any + +import pandas as pd +import pytest + +from pyneats.core.views import FlightView +from pyneats.runners.fleet import FleetRunnerParams +from pyneats.runners.large_emitter import FleetRunnerLargeEmitter +from pyneats.steps.parsing.neats_io import FlightConversionFailure +from pyneats.steps.parsing.neats_io import neats_json_to_flights +from pyneats.steps.weather.weather_store import ZarrPaths +from tests.conftest import assert_climate_payload_equal + + +GOLDEN_INPUT = Path(__file__).parent / "data" / "golden" / "fleet_5_flights_input.json" +GOLDEN_OUTPUT = ( + Path(__file__).parent / "data" / "golden" / "fleet_5_flights_output.json" +) + + +def _golden_flights() -> list[dict[str, Any]]: + with GOLDEN_INPUT.open(encoding="utf-8") as fh: + flights: list[dict[str, Any]] = json.load(fh) + return flights + + +def _with_bad_ts(flight: dict[str, Any], flight_id: str) -> dict[str, Any]: + """Copy of ``flight`` renamed ``flight_id``, with one invalid ``ts``.""" + bad = copy.deepcopy(flight) + fi = bad["flight_information"] + fi["flight_identification"] = flight_id + fi["trajectory"]["trajectory_data"][3]["ts"] = "garbage" + return bad + + +# --------------------------------------------------------------------------- +# neats_json_to_flights +# --------------------------------------------------------------------------- + + +def test_default_still_raises() -> None: + good = _golden_flights()[0] + with pytest.raises(ValueError, match="flight 'BADTS', trajectory point 3"): + neats_json_to_flights([good, _with_bad_ts(good, "BADTS")]) + + +def test_failures_list_isolates_the_bad_flight() -> None: + flights = _golden_flights() + failures: list[FlightConversionFailure] = [] + out = neats_json_to_flights( + [flights[0], _with_bad_ts(flights[0], "BADTS"), flights[1]], + failures=failures, + ) + assert [df.attrs["flight_id"] for df in out] == [ + flights[0]["flight_information"]["flight_identification"], + flights[1]["flight_information"]["flight_identification"], + ] + assert len(failures) == 1 + failure = failures[0] + assert failure.index == 1 + assert failure.attrs["flight_id"] == "BADTS" + assert failure.attrs["aircraft_type"] == "A320" + assert isinstance(failure.error, ValueError) + assert "trajectory point 3" in str(failure.error) + # No traceback is kept anywhere in the chain, so the record does not + # keep the input alive. + assert failure.error.__traceback__ is None + assert failure.error.__cause__ is not None + assert failure.error.__cause__.__traceback__ is None + + +def test_failure_does_not_keep_the_input_alive() -> None: + """With cyclic garbage collection off, the input is freed on return.""" + + class Flights(list): # list subclass so it can be weak-referenced + pass + + good = _golden_flights()[0] + flights = Flights([good, _with_bad_ts(good, "BADTS")]) + ref = weakref.ref(flights) + failures: list[FlightConversionFailure] = [] + gc.disable() + try: + neats_json_to_flights(flights, failures=failures) + del flights + assert ref() is None + finally: + gc.enable() + assert len(failures) == 1 + + +@pytest.mark.parametrize( + "not_a_sequence", + [{"flight_information": {}}, {"flights": []}, "a string", b"bytes"], +) +def test_non_sequence_input_is_a_file_error(not_a_sequence: Any) -> None: + """A whole-file shape error raises; it is not recorded as flight failures.""" + failures: list[FlightConversionFailure] = [] + with pytest.raises(TypeError, match="expected a sequence of flight objects"): + neats_json_to_flights(not_a_sequence, failures=failures) + assert failures == [] + + +@pytest.mark.parametrize( + "malformed", + [ + {"flight_information": {"trajectory": [{"lat": 1.0}]}}, # list, not mapping + {"flight_information": "not a mapping"}, + "not a flight object", + ], +) +def test_malformed_flight_is_recorded_not_raised(malformed: Any) -> None: + good = _golden_flights()[0] + failures: list[FlightConversionFailure] = [] + out = neats_json_to_flights([malformed, good], failures=failures) + assert len(out) == 1 + assert [f.index for f in failures] == [0] + assert "flight_id" not in failures[0].attrs + + +def test_failures_list_gives_identical_output_for_valid_flights() -> None: + flights = _golden_flights() + failures: list[FlightConversionFailure] = [] + with_list = neats_json_to_flights(flights, failures=failures) + default = neats_json_to_flights(flights) + assert failures == [] + assert len(with_list) == len(default) + for a, b in zip(with_list, default, strict=True): + pd.testing.assert_frame_equal(a, b, check_exact=True) + assert a.attrs == b.attrs + + +# --------------------------------------------------------------------------- +# FleetRunner loading stage (no BADA or weather needed) +# --------------------------------------------------------------------------- + + +def _runner_for( + flights: list[dict[str, Any]], tmp_path: Path +) -> FleetRunnerLargeEmitter: + path = tmp_path / "input.json" + path.write_text(json.dumps(flights), encoding="utf-8") + cfg = FleetRunnerParams( + trajectory_json_filepath=str(path), + zarr_paths=ZarrPaths(tmp_path / "met.zarr", tmp_path / "rad.zarr", None), + bada_path=str(tmp_path), + params={}, + ) + return FleetRunnerLargeEmitter(cfg) + + +def test_runner_records_the_bad_flight_and_keeps_the_others(tmp_path: Path) -> None: + flights = _golden_flights() + runner = _runner_for([flights[0], _with_bad_ts(flights[0], "BADTS")], tmp_path) + runner._load_trajectories() + + assert runner.source_fleet is not None + assert len(runner.source_fleet) == 1 + assert not runner._pipeline_aborted + (record,) = runner.error_records + assert record["flight_information"]["flight_id"] == "BADTS" + assert ( + record["flight_information"]["departure_airport"] + == (flights[0]["flight_information"]["departure_airport"]) + ) + assert record["error"].startswith("Failed at parsing: ValueError(") + assert "trajectory point 3" in record["error"] + + +def test_runner_aborts_when_every_flight_fails(tmp_path: Path) -> None: + flight = _golden_flights()[0] + runner = _runner_for( + [_with_bad_ts(flight, "BAD1"), _with_bad_ts(flight, "BAD2")], tmp_path + ) + # The weather paths do not exist: eval() must stop before loading them + # and return the error records instead of raising. + runner.eval() + + assert runner._pipeline_aborted + assert runner.met is None + assert runner.results is not None + assert [ + r["flight_information"]["flight_id"] for r in runner.results["flight_results"] + ] == ["BAD1", "BAD2"] + + +def test_runner_rejects_a_json_file_that_is_not_an_array(tmp_path: Path) -> None: + runner = _runner_for([], tmp_path) + (tmp_path / "input.json").write_text( + json.dumps(_golden_flights()[0]), encoding="utf-8" + ) + with pytest.raises(TypeError, match="expected a sequence of flight objects"): + runner._load_trajectories() + assert runner.error_records == [] + + +def test_runner_names_a_flight_without_id_by_position(tmp_path: Path) -> None: + good = _golden_flights()[0] + runner = _runner_for([good, "not a flight object"], tmp_path) # type: ignore[list-item] + runner._load_trajectories() + + (record,) = runner.error_records + assert record["flight_information"]["flight_id"] == "input_flight_1" + + +# --------------------------------------------------------------------------- +# End to end on the golden flights (needs BADA and weather) +# --------------------------------------------------------------------------- + + +@pytest.mark.integration +@pytest.mark.requires_weather +@pytest.mark.requires_bada +@pytest.mark.slow +def test_golden_fleet_with_one_bad_flight( + weather: Any, weather_path: Path | None, bada_path: Path | None, tmp_path: Path +) -> None: + """The five golden flights give their golden results; the added bad + flight gives one error record and does not stop the run.""" + if weather is None or weather_path is None: + pytest.skip("Weather data not available") + if bada_path is None or not bada_path.exists(): + pytest.skip("BADA data not available") + + flights = _golden_flights() + path = tmp_path / "input.json" + path.write_text( + json.dumps(flights[:2] + [_with_bad_ts(flights[0], "BADTS")] + flights[2:]), + encoding="utf-8", + ) + wind = weather_path / "icon_wind.zarr" + cfg = FleetRunnerParams( + trajectory_json_filepath=str(path), + zarr_paths=ZarrPaths( + weather_path / "icon_met.zarr", + weather_path / "icon_rad.zarr", + wind if wind.is_dir() else None, + ), + bada_path=str(bada_path), + params={}, + ) + runner = FleetRunnerLargeEmitter(cfg) + runner.eval() + + assert runner.fleet_with_climate_impact is not None + with GOLDEN_OUTPUT.open(encoding="utf-8") as fh: + golden = [FlightView.from_dict(d) for d in json.load(fh)] + assert len(runner.fleet_with_climate_impact) == len(golden) + for flight, expected in zip(runner.fleet_with_climate_impact, golden, strict=True): + assert_climate_payload_equal( + flight.attrs["climate_impact"], + expected.attrs["climate_impact"], + rtol=1e-3, + ) + + (record,) = runner.error_records + assert record["flight_information"]["flight_id"] == "BADTS" + assert record["error"].startswith("Failed at parsing:") + assert runner.results is not None + assert record in runner.results["flight_results"]