🏗️ Architectures API

The node_fdm.architectures namespace contains the blueprint definitions for specific flight dynamics problems.

It serves two main purposes: Registration (mapping string names to Python objects) and Implementation (defining the column groups, preprocessing logic, and model stacks for specific datasets like OpenSky or QAR).

---

🧩 Registry

The mapping module is the central lookup table. It allows the ODETrainer and Predictor to instantiate the correct classes based on a configuration string.

mapping

Helpers to dynamically load and assemble architecture components.

get_architecture_from_name(architecture_name)

Return architecture definition, model columns, and custom functions by name.

Parameters:

Name	Type	Description	Default
architecture_name	str	Name of the architecture to load.	required

Returns:

Type	Description
Tuple[Any, Any, Any]	Tuple of (architecture layers, model columns, custom functions).

Source code in src/node_fdm/architectures/mapping.py

def get_architecture_from_name(architecture_name: str) -> Tuple[Any, Any, Any]:
    """Return architecture definition, model columns, and custom functions by name.

    Args:
        architecture_name: Name of the architecture to load.

    Returns:
        Tuple of (architecture layers, model columns, custom functions).
    """
    architecture_dict = get_architecture_module(architecture_name)
    architecture = architecture_dict["model"].ARCHITECTURE
    model_cols = architecture_dict["model"].MODEL_COLS
    custom_fn = architecture_dict["custom_fn"]
    return architecture, model_cols, custom_fn

get_architecture_module(name)

Dynamically import only the architecture requested.

Parameters:

Name	Type	Description	Default
name	str	Architecture package name to load.	required

Returns:

Type	Description
Dict[str, Any]	Dictionary containing imported column, model, and custom function modules.

Raises:

Type	Description
ValueError	If the provided architecture name is not supported.

Source code in src/node_fdm/architectures/mapping.py

def get_architecture_module(name: str) -> Dict[str, Any]:
    """Dynamically import only the architecture requested.

    Args:
        name: Architecture package name to load.

    Returns:
        Dictionary containing imported column, model, and custom function modules.

    Raises:
        ValueError: If the provided architecture name is not supported.
    """
    valid_names = ["opensky_2025", "qar"]

    if name not in valid_names:
        raise ValueError(f"Unknown architecture '{name}'. Valid names: {valid_names}")

    module_root = f"node_fdm.architectures.{name}"

    columns = importlib.import_module(f"{module_root}.columns")
    flight_process = importlib.import_module(f"{module_root}.flight_process")
    model = importlib.import_module(f"{module_root}.model")

    return {
        "columns": columns,
        "custom_fn": (
            flight_process.flight_processing,
            flight_process.segment_filtering,
        ),
        "model": model,
    }

get_architecture_params_from_meta(meta_path)

Load architecture parameters and stats from a meta JSON file.

Parameters:

Name	Type	Description	Default
meta_path	str	Path to the meta JSON file.	required

Returns:

Type	Description
Tuple[Any, Any, Any, Dict[Any, Any]]	Tuple containing architecture, model columns, model parameters, and stats dictionary.

Source code in src/node_fdm/architectures/mapping.py

def get_architecture_params_from_meta(
    meta_path: str,
) -> Tuple[Any, Any, Any, Dict[Any, Any]]:
    """Load architecture parameters and stats from a meta JSON file.

    Args:
        meta_path: Path to the meta JSON file.

    Returns:
        Tuple containing architecture, model columns, model parameters, and stats dictionary.
    """
    with open(meta_path, "r") as f:
        meta = json.load(f)

    architecture, model_cols, _ = get_architecture_from_name(meta["architecture_name"])
    x_cols, u_cols, e0_cols, e_cols, _ = model_cols
    deriv_cols = [col.derivative for col in x_cols]
    model_cols2 = [x_cols, u_cols, e0_cols, e_cols, deriv_cols]

    all_cols_dict = {str(col): col for cols in model_cols2 for col in cols}
    stats_dict = {
        all_cols_dict[str_col]: stats for str_col, stats in meta["stats_dict"].items()
    }

    return architecture, model_cols, meta["model_params"], stats_dict

---

📡 OpenSky 2025

This is the reference implementation for public ADS-B data. It defines a physics-informed architecture capable of handling noisy surveillance data.

Columns Definition

Defines the input/output variables (State, Control, Environment) and their units.

columns

Column definitions and unit mappings for the OpenSky 2025 architecture.

Processing Hooks

Functions to clean, smooth, and augment raw ADS-B data.

flight_process

Pre-processing utilities for OpenSky 2025 flight data.

flight_processing(df)

Prepare OpenSky flight data by computing altitude differences.

Parameters:

Name	Type	Description	Default
df	DataFrame	Input DataFrame containing flight measurements.	required

Returns:

Type	Description
DataFrame	DataFrame with altitude difference column added.

Source code in src/node_fdm/architectures/opensky_2025/flight_process.py

def flight_processing(df: pd.DataFrame) -> pd.DataFrame:
    """Prepare OpenSky flight data by computing altitude differences.

    Args:
        df: Input DataFrame containing flight measurements.

    Returns:
        DataFrame with altitude difference column added.
    """
    df[col_alt_diff] = df[col_alt_sel] - df[col_alt]

    df[col_vz_sel] = df[col_vz_sel].fillna(0.0)
    df[col_mach_sel] = df[col_mach_sel].fillna(0.0)
    df[col_cas_sel] = df[col_cas_sel].fillna(0.0)

    return df

segment_filtering(f, start_idx, seq_len)

Check whether a segment meets distance variation thresholds.

Parameters:

Name	Type	Description	Default
f	DataFrame	DataFrame containing flight measurements.	required
start_idx	int	Starting index of the segment to evaluate.	required
seq_len	int	Length of the segment to evaluate.	required

Returns:

Type	Description
bool	True if the segment stays within distance thresholds, otherwise False.

Source code in src/node_fdm/architectures/opensky_2025/flight_process.py

def segment_filtering(f: pd.DataFrame, start_idx: int, seq_len: int) -> bool:
    """Check whether a segment meets distance variation thresholds.

    Args:
        f: DataFrame containing flight measurements.
        start_idx: Starting index of the segment to evaluate.
        seq_len: Length of the segment to evaluate.

    Returns:
        True if the segment stays within distance thresholds, otherwise False.
    """
    dist_diff = f[col_dist].diff(1)
    seg = dist_diff.iloc[start_idx : start_idx + seq_len]
    condition = len(seg[(seg < LOW_THR) | (seg > UPPER_THR)]) == 0
    return condition

Model Stack

The assembly of the Neural ODE, connecting physics layers with the learned derivative layer.

model

Layer configuration and column grouping for the OpenSky 2025 architecture.

Trajectory Layer

A specialized physics layer that computes derived kinematic variables.

trajectory_layer

Torch module for computing basic trajectory features for OpenSky 2025 data.

TrajectoryLayer

Bases: Module

Compute trajectory outputs such as vertical speed, Mach, and calibrated airspeed.

Source code in src/node_fdm/architectures/opensky_2025/trajectory_layer.py

class TrajectoryLayer(nn.Module):
    """Compute trajectory outputs such as vertical speed, Mach, and calibrated airspeed."""

    def __init__(self) -> None:
        """Initialize the trajectory layer with base configuration."""
        super().__init__()
        self.alpha = 40

    def forward(self, x: Mapping[Any, torch.Tensor]) -> Dict[Any, torch.Tensor]:
        """Compute derived trajectory quantities from the provided inputs.

        Args:
            x: Mapping from column identifiers to input tensors.

        Returns:
            Dictionary of derived tensors keyed by their column identifiers.
        """
        output_dict = {}
        for col in [col_tas, col_gamma, col_alt, col_long_wind_spd]:
            x[col] = torch.nan_to_num(x[col], nan=0.0, posinf=1e6, neginf=-1e6)
            x[col] = torch.clamp(x[col], min=-1e6, max=1e6)

        tas = x[col_tas]
        gamma = x[col_gamma]
        long_wind = x[col_long_wind_spd]
        alt = x[col_alt]

        output_dict[col_vz] = tas * torch.sin(gamma)

        temp = isa_temperature_torch(alt)

        a = torch.sqrt(torch.clamp(gamma_ratio * R * temp, min=1e-6, max=1e8))

        mach = tas / torch.clamp(a, min=1e-6, max=1e8)
        output_dict[col_mach] = mach

        output_dict[col_gs] = tas - long_wind

        p = isa_pressure_torch(alt)

        pt_over_p = torch.pow(
            torch.clamp(1 + (gamma_ratio - 1) / 2 * mach**2, min=1e-6, max=1e6),
            gamma_ratio / (gamma_ratio - 1),
        )

        qc_p0 = (torch.clamp(p, min=1.0) / p0) * (pt_over_p - 1.0)
        qc_p0 = torch.clamp(qc_p0, min=-0.999, max=1e6)

        CAS_term = torch.clamp(qc_p0 + 1.0, min=1e-8, max=1e6)
        CAS = a0 * torch.sqrt(
            (2.0 / (gamma_ratio - 1.0))
            * (CAS_term ** ((gamma_ratio - 1.0) / gamma_ratio) - 1.0)
        )
        CAS = torch.nan_to_num(CAS, nan=0.0, posinf=1e4, neginf=0.0)
        output_dict[col_cas] = CAS

        # --- Reference differences ---
        ref_alt = x[col_alt_sel]

        alt_diff = ref_alt - alt

        output_dict[col_alt_diff] = torch.nan_to_num(alt_diff, nan=0.0)

        return output_dict

__init__()

Initialize the trajectory layer with base configuration.

Source code in src/node_fdm/architectures/opensky_2025/trajectory_layer.py

def __init__(self) -> None:
    """Initialize the trajectory layer with base configuration."""
    super().__init__()
    self.alpha = 40

forward(x)

Compute derived trajectory quantities from the provided inputs.

Parameters:

Name	Type	Description	Default
x	Mapping[Any, Tensor]	Mapping from column identifiers to input tensors.	required

Returns:

Type	Description
Dict[Any, Tensor]	Dictionary of derived tensors keyed by their column identifiers.

Source code in src/node_fdm/architectures/opensky_2025/trajectory_layer.py

def forward(self, x: Mapping[Any, torch.Tensor]) -> Dict[Any, torch.Tensor]:
    """Compute derived trajectory quantities from the provided inputs.

    Args:
        x: Mapping from column identifiers to input tensors.

    Returns:
        Dictionary of derived tensors keyed by their column identifiers.
    """
    output_dict = {}
    for col in [col_tas, col_gamma, col_alt, col_long_wind_spd]:
        x[col] = torch.nan_to_num(x[col], nan=0.0, posinf=1e6, neginf=-1e6)
        x[col] = torch.clamp(x[col], min=-1e6, max=1e6)

    tas = x[col_tas]
    gamma = x[col_gamma]
    long_wind = x[col_long_wind_spd]
    alt = x[col_alt]

    output_dict[col_vz] = tas * torch.sin(gamma)

    temp = isa_temperature_torch(alt)

    a = torch.sqrt(torch.clamp(gamma_ratio * R * temp, min=1e-6, max=1e8))

    mach = tas / torch.clamp(a, min=1e-6, max=1e8)
    output_dict[col_mach] = mach

    output_dict[col_gs] = tas - long_wind

    p = isa_pressure_torch(alt)

    pt_over_p = torch.pow(
        torch.clamp(1 + (gamma_ratio - 1) / 2 * mach**2, min=1e-6, max=1e6),
        gamma_ratio / (gamma_ratio - 1),
    )

    qc_p0 = (torch.clamp(p, min=1.0) / p0) * (pt_over_p - 1.0)
    qc_p0 = torch.clamp(qc_p0, min=-0.999, max=1e6)

    CAS_term = torch.clamp(qc_p0 + 1.0, min=1e-8, max=1e6)
    CAS = a0 * torch.sqrt(
        (2.0 / (gamma_ratio - 1.0))
        * (CAS_term ** ((gamma_ratio - 1.0) / gamma_ratio) - 1.0)
    )
    CAS = torch.nan_to_num(CAS, nan=0.0, posinf=1e4, neginf=0.0)
    output_dict[col_cas] = CAS

    # --- Reference differences ---
    ref_alt = x[col_alt_sel]

    alt_diff = ref_alt - alt

    output_dict[col_alt_diff] = torch.nan_to_num(alt_diff, nan=0.0)

    return output_dict