Flight Project

Overview

• End-to-end reference project that ingests file-based aeronautical telemetry, cleans it with Polars and pushes curated artefacts back into the AGILab export directory.

• Demonstrates how to orchestrate file-based distributions, run inside the cluster pool and keep the web interface responsive while the worker pipeline operates asynchronously.

• Bundles web-view lab material (lab_stages.toml) and prompt examples to help you reproduce the workflow showcased in AGILab live demos.

Scientific highlights

The worker derives segment distances from time-stamped telemetry. For two samples with latitude/longitude \((\phi_1, \lambda_1)\) and \((\phi_2, \lambda_2)\), a common great-circle approximation is the haversine formula:

\[d = 2R \arcsin\left(\sqrt{\sin^2\frac{\Delta\phi}{2} + \cos\phi_1\cos\phi_2\sin^2\frac{\Delta\lambda}{2}}\right)\]

with Earth radius \(R\) and angular differences \(\Delta\phi = \phi_2 - \phi_1\), \(\Delta\lambda = \lambda_2 - \lambda_1\). The public demo stores this per-sample segment distance in the historical speed column so existing analysis pages and reducer contracts remain compatible.

Public scope

flight_project is intentionally file-only. The wider AGILab connector catalog supports SQL, object-storage, and OpenSearch-compatible connector definitions, but this built-in app rejects search-index data sources instead of exposing a partially implemented path.

Manager (flight.flight)

• Wraps the runnable application. Converts user-supplied arguments into a validated FlightArgs model, normalises data URIs, and initialises cluster dispatch by seeding WorkDispatcher.

• Provides from_toml / to_toml helpers so snippets on the Orchestrate page can reload configuration and persist overrides.

• Handles managed-PC specifics (path remapping, data directory resets) and keeps the dataframe export folder clean between runs.

Args (flight.flight_args)

• Pydantic models that capture the supported public configuration, including dataset location, slicing parameters and cluster toggles.

• Ships conversion utilities for reading/writing app_settings.toml and merging overrides that are injected by the Orchestrate page.

Worker (flight_worker.flight_worker)

• Extends PolarsWorker to preprocess raw telemetry, compute great-circle segment distances between samples, and partition files across the cluster.

• Can be compiled by the AGILab dispatcher when Cython is enabled; generated .pyx/.c files are build artefacts, not source-of-truth files.

• Demonstrates Windows-friendly path handling and data staging for managed environments.

Assets & Tests

• test/test_flight_project_runtime_args.py covers argument validation, file inventory building, worker defaults, and reduce-artifact emission.

• Cluster validation and UI tests cover the default view_maps route, first-proof flow, and Release Decision reduce-artifact discovery.

API Reference

class flight.flight.Flight(env, args=None, **kwargs)

Bases: BaseWorker

Flight class provides methods to orchestrate the run.

__init__(env, args=None, **kwargs)

as_dict(mode='json')

Return current arguments as a serialisable dictionary.

Return type:

dict[str, Any]

build_distribution(workers)

Build worker -> aircraft -> file batches for file-based telemetry.

static extract_plane_from_file_name(file_path)

provide airplane id from log file name

Parameters:

file_path

Returns:

classmethod from_toml(env, settings_path='app_settings.toml', section='args', **overrides)

Return type:

Flight

get_data_from_files()

Return file slices to distribute across workers.

get_data_from_hawk()

Fail fast for legacy snippets that still select Hawk ingestion.

get_partition_by_planes(df)

build the first level of the distribution tree with planes as atomics partition

Parameters:

• s – df: dataframe containing the output-data to partition

• df

Returns:

ivq_logs = None

to_toml(settings_path='app_settings.toml', section='args', create_missing=True)

Return type:

None

class flight.flight.FlightApp(env, args=None, **kwargs)

Bases: Flight

Alias keeping legacy imports alive.

Shared validation and persistence helpers for Flight project arguments.

flight.flight_args.ArgsModel

alias of FlightArgs

flight.flight_args.ArgsOverrides

alias of FlightArgsTD

class flight.flight_args.FlightArgs(**data)

Bases: BaseModel

Validated configuration for the Flight worker.

data_in

data_out

data_source

datemax

datemin

files

model_config = {'extra': 'forbid', 'validate_assignment': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

nfile

nread

nskip

output_format

reset_target

sampling_rate

to_toml_payload()

Return a TOML-friendly representation (Path/date → str).

Return type:

dict[str, Any]

class flight.flight_args.FlightArgsTD

Bases: TypedDict

data_in

data_out

data_source

datemax

datemin

files

nfile

nread

nskip

output_format

reset_target

sampling_rate

flight.flight_args.apply_source_defaults(args, *, host_ip=None)

Ensure source-specific defaults for missing values.

Return type:

FlightArgs

flight.flight_args.dump_args(args, settings_path, *, section='args', create_missing=True)

Return type:

None

flight.flight_args.dump_args_to_toml(args, settings_path, section='args', create_missing=True)

Persist arguments back to the TOML file (overwriting only the section).

Return type:

None

flight.flight_args.ensure_defaults(args, **kwargs)

Return type:

FlightArgs

flight.flight_args.load_args(settings_path, *, section='args')

Return type:

FlightArgs

flight.flight_args.load_args_from_toml(settings_path, section='args')

Load arguments from a TOML file, applying model defaults when missing.

Return type:

FlightArgs

flight.flight_args.merge_args(base, overrides=None)

Return a new instance with overrides applied on top of base.

Return type:

FlightArgs

class flight_worker.flight_worker.FlightWorker

Bases: PolarsWorker

Class derived from AgiDataWorker

calculate_speed(new_column_name, df)

Compute the segment distance in meters between consecutive coordinate pairs and add it under the legacy speed column name used by this demo. Assumes that the previous coordinate columns are already present.

Return type:

DataFrame

pool_init(worker_vars)

Initialize the pool with worker variables.

Parameters:

worker_vars (dict) – Variables specific to the worker.

pool_vars = {}

preprocess_df(df)

Preprocess the DataFrame by parsing the date column and creating previous coordinate columns. This operation is done once per file.

Return type:

DataFrame

start()

Initialize global variables and setup paths.

stop()

Returns:

work_done(worker_df)

Concatenate dataframe if any and save the results.

Parameters:

worker_df (pl.DataFrame) – Output dataframe for one plane.

work_init()

Initialize work by reading from shared space.

work_pool(file)

Parse IVQ log files.

Parameters:

file (str) – The log file to parse.

Returns:

Parsed data.

Return type:

pl.DataFrame